1# -*- coding: utf-8 -*-
2# Copyright: (c) 2019, Ansible Project
3# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
4
5# Make coding more python3-ish
6from __future__ import (absolute_import, division, print_function)
7__metaclass__ = type
8
9import json
10import os
11import re
12import pytest
13import stat
14import tarfile
15import tempfile
16import time
17
18from io import BytesIO, StringIO
19from units.compat.mock import MagicMock
20
21import ansible.constants as C
22from ansible import context
23from ansible.errors import AnsibleError
24from ansible.galaxy import api as galaxy_api
25from ansible.galaxy.api import CollectionVersionMetadata, GalaxyAPI, GalaxyError
26from ansible.galaxy.token import BasicAuthToken, GalaxyToken, KeycloakToken
27from ansible.module_utils._text import to_native, to_text
28from ansible.module_utils.six.moves.urllib import error as urllib_error
29from ansible.utils import context_objects as co
30from ansible.utils.display import Display
31
32
33@pytest.fixture(autouse='function')
34def reset_cli_args():
35    co.GlobalCLIArgs._Singleton__instance = None
36    # Required to initialise the GalaxyAPI object
37    context.CLIARGS._store = {'ignore_certs': False}
38    yield
39    co.GlobalCLIArgs._Singleton__instance = None
40
41
42@pytest.fixture()
43def collection_artifact(tmp_path_factory):
44    ''' Creates a collection artifact tarball that is ready to be published '''
45    output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Output'))
46
47    tar_path = os.path.join(output_dir, 'namespace-collection-v1.0.0.tar.gz')
48    with tarfile.open(tar_path, 'w:gz') as tfile:
49        b_io = BytesIO(b"\x00\x01\x02\x03")
50        tar_info = tarfile.TarInfo('test')
51        tar_info.size = 4
52        tar_info.mode = 0o0644
53        tfile.addfile(tarinfo=tar_info, fileobj=b_io)
54
55    yield tar_path
56
57
58@pytest.fixture()
59def cache_dir(tmp_path_factory, monkeypatch):
60    cache_dir = to_text(tmp_path_factory.mktemp('Test ÅÑŚÌβŁÈ Galaxy Cache'))
61    monkeypatch.setitem(C.config._base_defs, 'GALAXY_CACHE_DIR', {'default': cache_dir})
62
63    yield cache_dir
64
65
66def get_test_galaxy_api(url, version, token_ins=None, token_value=None, no_cache=True):
67    token_value = token_value or "my token"
68    token_ins = token_ins or GalaxyToken(token_value)
69    api = GalaxyAPI(None, "test", url, no_cache=no_cache)
70    # Warning, this doesn't test g_connect() because _availabe_api_versions is set here.  That means
71    # that urls for v2 servers have to append '/api/' themselves in the input data.
72    api._available_api_versions = {version: '%s' % version}
73    api.token = token_ins
74
75    return api
76
77
78def get_collection_versions(namespace='namespace', name='collection'):
79    base_url = 'https://galaxy.server.com/api/v2/collections/{0}/{1}/'.format(namespace, name)
80    versions_url = base_url + 'versions/'
81
82    # Response for collection info
83    responses = [
84        {
85            "id": 1000,
86            "href": base_url,
87            "name": name,
88            "namespace": {
89                "id": 30000,
90                "href": "https://galaxy.ansible.com/api/v1/namespaces/30000/",
91                "name": namespace,
92            },
93            "versions_url": versions_url,
94            "latest_version": {
95                "version": "1.0.5",
96                "href": versions_url + "1.0.5/"
97            },
98            "deprecated": False,
99            "created": "2021-02-09T16:55:42.749915-05:00",
100            "modified": "2021-02-09T16:55:42.749915-05:00",
101        }
102    ]
103
104    # Paginated responses for versions
105    page_versions = (('1.0.0', '1.0.1',), ('1.0.2', '1.0.3',), ('1.0.4', '1.0.5'),)
106    last_page = None
107    for page in range(1, len(page_versions) + 1):
108        if page < len(page_versions):
109            next_page = versions_url + '?page={0}'.format(page + 1)
110        else:
111            next_page = None
112
113        version_results = []
114        for version in page_versions[int(page - 1)]:
115            version_results.append(
116                {'version': version, 'href': versions_url + '{0}/'.format(version)}
117            )
118
119        responses.append(
120            {
121                'count': 6,
122                'next': next_page,
123                'previous': last_page,
124                'results': version_results,
125            }
126        )
127        last_page = page
128
129    return responses
130
131
132def test_api_no_auth():
133    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")
134    actual = {}
135    api._add_auth_token(actual, "")
136    assert actual == {}
137
138
139def test_api_no_auth_but_required():
140    expected = "No access token or username set. A token can be set with --api-key or at "
141    with pytest.raises(AnsibleError, match=expected):
142        GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")._add_auth_token({}, "", required=True)
143
144
145def test_api_token_auth():
146    token = GalaxyToken(token=u"my_token")
147    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
148    actual = {}
149    api._add_auth_token(actual, "", required=True)
150    assert actual == {'Authorization': 'Token my_token'}
151
152
153def test_api_token_auth_with_token_type(monkeypatch):
154    token = KeycloakToken(auth_url='https://api.test/')
155    mock_token_get = MagicMock()
156    mock_token_get.return_value = 'my_token'
157    monkeypatch.setattr(token, 'get', mock_token_get)
158    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
159    actual = {}
160    api._add_auth_token(actual, "", token_type="Bearer", required=True)
161    assert actual == {'Authorization': 'Bearer my_token'}
162
163
164def test_api_token_auth_with_v3_url(monkeypatch):
165    token = KeycloakToken(auth_url='https://api.test/')
166    mock_token_get = MagicMock()
167    mock_token_get.return_value = 'my_token'
168    monkeypatch.setattr(token, 'get', mock_token_get)
169    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
170    actual = {}
171    api._add_auth_token(actual, "https://galaxy.ansible.com/api/v3/resource/name", required=True)
172    assert actual == {'Authorization': 'Bearer my_token'}
173
174
175def test_api_token_auth_with_v2_url():
176    token = GalaxyToken(token=u"my_token")
177    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
178    actual = {}
179    # Add v3 to random part of URL but response should only see the v2 as the full URI path segment.
180    api._add_auth_token(actual, "https://galaxy.ansible.com/api/v2/resourcev3/name", required=True)
181    assert actual == {'Authorization': 'Token my_token'}
182
183
184def test_api_basic_auth_password():
185    token = BasicAuthToken(username=u"user", password=u"pass")
186    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
187    actual = {}
188    api._add_auth_token(actual, "", required=True)
189    assert actual == {'Authorization': 'Basic dXNlcjpwYXNz'}
190
191
192def test_api_basic_auth_no_password():
193    token = BasicAuthToken(username=u"user")
194    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
195    actual = {}
196    api._add_auth_token(actual, "", required=True)
197    assert actual == {'Authorization': 'Basic dXNlcjo='}
198
199
200def test_api_dont_override_auth_header():
201    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")
202    actual = {'Authorization': 'Custom token'}
203    api._add_auth_token(actual, "", required=True)
204    assert actual == {'Authorization': 'Custom token'}
205
206
207def test_initialise_galaxy(monkeypatch):
208    mock_open = MagicMock()
209    mock_open.side_effect = [
210        StringIO(u'{"available_versions":{"v1":"v1/"}}'),
211        StringIO(u'{"token":"my token"}'),
212    ]
213    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
214
215    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")
216    actual = api.authenticate("github_token")
217
218    assert len(api.available_api_versions) == 2
219    assert api.available_api_versions['v1'] == u'v1/'
220    assert api.available_api_versions['v2'] == u'v2/'
221    assert actual == {u'token': u'my token'}
222    assert mock_open.call_count == 2
223    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/'
224    assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent']
225    assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/'
226    assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent']
227    assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token'
228
229
230def test_initialise_galaxy_with_auth(monkeypatch):
231    mock_open = MagicMock()
232    mock_open.side_effect = [
233        StringIO(u'{"available_versions":{"v1":"v1/"}}'),
234        StringIO(u'{"token":"my token"}'),
235    ]
236    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
237
238    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token'))
239    actual = api.authenticate("github_token")
240
241    assert len(api.available_api_versions) == 2
242    assert api.available_api_versions['v1'] == u'v1/'
243    assert api.available_api_versions['v2'] == u'v2/'
244    assert actual == {u'token': u'my token'}
245    assert mock_open.call_count == 2
246    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/'
247    assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent']
248    assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/'
249    assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent']
250    assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token'
251
252
253def test_initialise_automation_hub(monkeypatch):
254    mock_open = MagicMock()
255    mock_open.side_effect = [
256        StringIO(u'{"available_versions":{"v2": "v2/", "v3":"v3/"}}'),
257    ]
258    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
259    token = KeycloakToken(auth_url='https://api.test/')
260    mock_token_get = MagicMock()
261    mock_token_get.return_value = 'my_token'
262    monkeypatch.setattr(token, 'get', mock_token_get)
263
264    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token)
265
266    assert len(api.available_api_versions) == 2
267    assert api.available_api_versions['v2'] == u'v2/'
268    assert api.available_api_versions['v3'] == u'v3/'
269
270    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/'
271    assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent']
272    assert mock_open.mock_calls[0][2]['headers'] == {'Authorization': 'Bearer my_token'}
273
274
275def test_initialise_unknown(monkeypatch):
276    mock_open = MagicMock()
277    mock_open.side_effect = [
278        urllib_error.HTTPError('https://galaxy.ansible.com/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')),
279        urllib_error.HTTPError('https://galaxy.ansible.com/api/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')),
280    ]
281    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
282
283    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token'))
284
285    expected = "Error when finding available api versions from test (%s) (HTTP Code: 500, Message: msg)" \
286        % api.api_server
287    with pytest.raises(AnsibleError, match=re.escape(expected)):
288        api.authenticate("github_token")
289
290
291def test_get_available_api_versions(monkeypatch):
292    mock_open = MagicMock()
293    mock_open.side_effect = [
294        StringIO(u'{"available_versions":{"v1":"v1/","v2":"v2/"}}'),
295    ]
296    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
297
298    api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")
299    actual = api.available_api_versions
300    assert len(actual) == 2
301    assert actual['v1'] == u'v1/'
302    assert actual['v2'] == u'v2/'
303
304    assert mock_open.call_count == 1
305    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/'
306    assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent']
307
308
309def test_publish_collection_missing_file():
310    fake_path = u'/fake/ÅÑŚÌβŁÈ/path'
311    expected = to_native("The collection path specified '%s' does not exist." % fake_path)
312
313    api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2")
314    with pytest.raises(AnsibleError, match=expected):
315        api.publish_collection(fake_path)
316
317
318def test_publish_collection_not_a_tarball():
319    expected = "The collection path specified '{0}' is not a tarball, use 'ansible-galaxy collection build' to " \
320               "create a proper release artifact."
321
322    api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2")
323    with tempfile.NamedTemporaryFile(prefix=u'ÅÑŚÌβŁÈ') as temp_file:
324        temp_file.write(b"\x00")
325        temp_file.flush()
326        with pytest.raises(AnsibleError, match=expected.format(to_native(temp_file.name))):
327            api.publish_collection(temp_file.name)
328
329
330def test_publish_collection_unsupported_version():
331    expected = "Galaxy action publish_collection requires API versions 'v2, v3' but only 'v1' are available on test " \
332               "https://galaxy.ansible.com/api/"
333
334    api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v1")
335    with pytest.raises(AnsibleError, match=expected):
336        api.publish_collection("path")
337
338
339@pytest.mark.parametrize('api_version, collection_url', [
340    ('v2', 'collections'),
341    ('v3', 'artifacts/collections'),
342])
343def test_publish_collection(api_version, collection_url, collection_artifact, monkeypatch):
344    api = get_test_galaxy_api("https://galaxy.ansible.com/api/", api_version)
345
346    mock_call = MagicMock()
347    mock_call.return_value = {'task': 'http://task.url/'}
348    monkeypatch.setattr(api, '_call_galaxy', mock_call)
349
350    actual = api.publish_collection(collection_artifact)
351    assert actual == 'http://task.url/'
352    assert mock_call.call_count == 1
353    assert mock_call.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/%s/%s/' % (api_version, collection_url)
354    assert mock_call.mock_calls[0][2]['headers']['Content-length'] == len(mock_call.mock_calls[0][2]['args'])
355    assert mock_call.mock_calls[0][2]['headers']['Content-type'].startswith(
356        'multipart/form-data; boundary=')
357    assert mock_call.mock_calls[0][2]['args'].startswith(b'--')
358    assert mock_call.mock_calls[0][2]['method'] == 'POST'
359    assert mock_call.mock_calls[0][2]['auth_required'] is True
360
361
362@pytest.mark.parametrize('api_version, collection_url, response, expected', [
363    ('v2', 'collections', {},
364     'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'),
365    ('v2', 'collections', {
366        'message': u'Galaxy error messäge',
367        'code': 'GWE002',
368    }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Galaxy error messäge Code: GWE002)'),
369    ('v3', 'artifact/collections', {},
370     'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'),
371    ('v3', 'artifact/collections', {
372        'errors': [
373            {
374                'code': 'conflict.collection_exists',
375                'detail': 'Collection "mynamespace-mycollection-4.1.1" already exists.',
376                'title': 'Conflict.',
377                'status': '400',
378            },
379            {
380                'code': 'quantum_improbability',
381                'title': u'Rändom(?) quantum improbability.',
382                'source': {'parameter': 'the_arrow_of_time'},
383                'meta': {'remediation': 'Try again before'},
384            },
385        ],
386    }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Collection '
387       u'"mynamespace-mycollection-4.1.1" already exists. Code: conflict.collection_exists), (HTTP Code: 500, '
388       u'Message: Rändom(?) quantum improbability. Code: quantum_improbability)')
389])
390def test_publish_failure(api_version, collection_url, response, expected, collection_artifact, monkeypatch):
391    api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version)
392
393    expected_url = '%s/api/%s/%s' % (api.api_server, api_version, collection_url)
394
395    mock_open = MagicMock()
396    mock_open.side_effect = urllib_error.HTTPError(expected_url, 500, 'msg', {},
397                                                   StringIO(to_text(json.dumps(response))))
398    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
399
400    with pytest.raises(GalaxyError, match=re.escape(to_native(expected % api.api_server))):
401        api.publish_collection(collection_artifact)
402
403
404@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [
405    ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'),
406     '1234',
407     'https://galaxy.server.com/api/v2/collection-imports/1234/'),
408    ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'),
409     '1234',
410     'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'),
411])
412def test_wait_import_task(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch):
413    api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins)
414
415    if token_ins:
416        mock_token_get = MagicMock()
417        mock_token_get.return_value = 'my token'
418        monkeypatch.setattr(token_ins, 'get', mock_token_get)
419
420    mock_open = MagicMock()
421    mock_open.return_value = StringIO(u'{"state":"success","finished_at":"time"}')
422    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
423
424    mock_display = MagicMock()
425    monkeypatch.setattr(Display, 'display', mock_display)
426
427    api.wait_import_task(import_uri)
428
429    assert mock_open.call_count == 1
430    assert mock_open.mock_calls[0][1][0] == full_import_uri
431    assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
432
433    assert mock_display.call_count == 1
434    assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri
435
436
437@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [
438    ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'),
439     '1234',
440     'https://galaxy.server.com/api/v2/collection-imports/1234/'),
441    ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'),
442     '1234',
443     'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'),
444])
445def test_wait_import_task_multiple_requests(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch):
446    api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins)
447
448    if token_ins:
449        mock_token_get = MagicMock()
450        mock_token_get.return_value = 'my token'
451        monkeypatch.setattr(token_ins, 'get', mock_token_get)
452
453    mock_open = MagicMock()
454    mock_open.side_effect = [
455        StringIO(u'{"state":"test"}'),
456        StringIO(u'{"state":"success","finished_at":"time"}'),
457    ]
458    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
459
460    mock_display = MagicMock()
461    monkeypatch.setattr(Display, 'display', mock_display)
462
463    mock_vvv = MagicMock()
464    monkeypatch.setattr(Display, 'vvv', mock_vvv)
465
466    monkeypatch.setattr(time, 'sleep', MagicMock())
467
468    api.wait_import_task(import_uri)
469
470    assert mock_open.call_count == 2
471    assert mock_open.mock_calls[0][1][0] == full_import_uri
472    assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
473    assert mock_open.mock_calls[1][1][0] == full_import_uri
474    assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type
475
476    assert mock_display.call_count == 1
477    assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri
478
479    assert mock_vvv.call_count == 1
480    assert mock_vvv.mock_calls[0][1][0] == \
481        'Galaxy import process has a status of test, wait 2 seconds before trying again'
482
483
484@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri,', [
485    ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'),
486     '1234',
487     'https://galaxy.server.com/api/v2/collection-imports/1234/'),
488    ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'),
489     '1234',
490     'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'),
491])
492def test_wait_import_task_with_failure(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch):
493    api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins)
494
495    if token_ins:
496        mock_token_get = MagicMock()
497        mock_token_get.return_value = 'my token'
498        monkeypatch.setattr(token_ins, 'get', mock_token_get)
499
500    mock_open = MagicMock()
501    mock_open.side_effect = [
502        StringIO(to_text(json.dumps({
503            'finished_at': 'some_time',
504            'state': 'failed',
505            'error': {
506                'code': 'GW001',
507                'description': u'Becäuse I said so!',
508
509            },
510            'messages': [
511                {
512                    'level': 'error',
513                    'message': u'Somé error',
514                },
515                {
516                    'level': 'warning',
517                    'message': u'Some wärning',
518                },
519                {
520                    'level': 'info',
521                    'message': u'Somé info',
522                },
523            ],
524        }))),
525    ]
526    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
527
528    mock_display = MagicMock()
529    monkeypatch.setattr(Display, 'display', mock_display)
530
531    mock_vvv = MagicMock()
532    monkeypatch.setattr(Display, 'vvv', mock_vvv)
533
534    mock_warn = MagicMock()
535    monkeypatch.setattr(Display, 'warning', mock_warn)
536
537    mock_err = MagicMock()
538    monkeypatch.setattr(Display, 'error', mock_err)
539
540    expected = to_native(u'Galaxy import process failed: Becäuse I said so! (Code: GW001)')
541    with pytest.raises(AnsibleError, match=re.escape(expected)):
542        api.wait_import_task(import_uri)
543
544    assert mock_open.call_count == 1
545    assert mock_open.mock_calls[0][1][0] == full_import_uri
546    assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
547
548    assert mock_display.call_count == 1
549    assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri
550
551    assert mock_vvv.call_count == 1
552    assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: info - Somé info'
553
554    assert mock_warn.call_count == 1
555    assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning'
556
557    assert mock_err.call_count == 1
558    assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error'
559
560
561@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [
562    ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my_token'),
563     '1234',
564     'https://galaxy.server.com/api/v2/collection-imports/1234/'),
565    ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'),
566     '1234',
567     'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'),
568])
569def test_wait_import_task_with_failure_no_error(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch):
570    api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins)
571
572    if token_ins:
573        mock_token_get = MagicMock()
574        mock_token_get.return_value = 'my token'
575        monkeypatch.setattr(token_ins, 'get', mock_token_get)
576
577    mock_open = MagicMock()
578    mock_open.side_effect = [
579        StringIO(to_text(json.dumps({
580            'finished_at': 'some_time',
581            'state': 'failed',
582            'error': {},
583            'messages': [
584                {
585                    'level': 'error',
586                    'message': u'Somé error',
587                },
588                {
589                    'level': 'warning',
590                    'message': u'Some wärning',
591                },
592                {
593                    'level': 'info',
594                    'message': u'Somé info',
595                },
596            ],
597        }))),
598    ]
599    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
600
601    mock_display = MagicMock()
602    monkeypatch.setattr(Display, 'display', mock_display)
603
604    mock_vvv = MagicMock()
605    monkeypatch.setattr(Display, 'vvv', mock_vvv)
606
607    mock_warn = MagicMock()
608    monkeypatch.setattr(Display, 'warning', mock_warn)
609
610    mock_err = MagicMock()
611    monkeypatch.setattr(Display, 'error', mock_err)
612
613    expected = 'Galaxy import process failed: Unknown error, see %s for more details \\(Code: UNKNOWN\\)' % full_import_uri
614    with pytest.raises(AnsibleError, match=expected):
615        api.wait_import_task(import_uri)
616
617    assert mock_open.call_count == 1
618    assert mock_open.mock_calls[0][1][0] == full_import_uri
619    assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
620
621    assert mock_display.call_count == 1
622    assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri
623
624    assert mock_vvv.call_count == 1
625    assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: info - Somé info'
626
627    assert mock_warn.call_count == 1
628    assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning'
629
630    assert mock_err.call_count == 1
631    assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error'
632
633
634@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [
635    ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'),
636     '1234',
637     'https://galaxy.server.com/api/v2/collection-imports/1234/'),
638    ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'),
639     '1234',
640     'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'),
641])
642def test_wait_import_task_timeout(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch):
643    api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins)
644
645    if token_ins:
646        mock_token_get = MagicMock()
647        mock_token_get.return_value = 'my token'
648        monkeypatch.setattr(token_ins, 'get', mock_token_get)
649
650    def return_response(*args, **kwargs):
651        return StringIO(u'{"state":"waiting"}')
652
653    mock_open = MagicMock()
654    mock_open.side_effect = return_response
655    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
656
657    mock_display = MagicMock()
658    monkeypatch.setattr(Display, 'display', mock_display)
659
660    mock_vvv = MagicMock()
661    monkeypatch.setattr(Display, 'vvv', mock_vvv)
662
663    monkeypatch.setattr(time, 'sleep', MagicMock())
664
665    expected = "Timeout while waiting for the Galaxy import process to finish, check progress at '%s'" % full_import_uri
666    with pytest.raises(AnsibleError, match=expected):
667        api.wait_import_task(import_uri, 1)
668
669    assert mock_open.call_count > 1
670    assert mock_open.mock_calls[0][1][0] == full_import_uri
671    assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
672    assert mock_open.mock_calls[1][1][0] == full_import_uri
673    assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type
674
675    assert mock_display.call_count == 1
676    assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri
677
678    # expected_wait_msg = 'Galaxy import process has a status of waiting, wait {0} seconds before trying again'
679    assert mock_vvv.call_count > 9  # 1st is opening Galaxy token file.
680
681    # FIXME:
682    # assert mock_vvv.mock_calls[1][1][0] == expected_wait_msg.format(2)
683    # assert mock_vvv.mock_calls[2][1][0] == expected_wait_msg.format(3)
684    # assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(4)
685    # assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(6)
686    # assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(10)
687    # assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(15)
688    # assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(22)
689    # assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(30)
690
691
692@pytest.mark.parametrize('api_version, token_type, version, token_ins', [
693    ('v2', None, 'v2.1.13', None),
694    ('v3', 'Bearer', 'v1.0.0', KeycloakToken(auth_url='https://api.test/api/automation-hub/')),
695])
696def test_get_collection_version_metadata_no_version(api_version, token_type, version, token_ins, monkeypatch):
697    api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins)
698
699    if token_ins:
700        mock_token_get = MagicMock()
701        mock_token_get.return_value = 'my token'
702        monkeypatch.setattr(token_ins, 'get', mock_token_get)
703
704    mock_open = MagicMock()
705    mock_open.side_effect = [
706        StringIO(to_text(json.dumps({
707            'download_url': 'https://downloadme.com',
708            'artifact': {
709                'sha256': 'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f',
710            },
711            'namespace': {
712                'name': 'namespace',
713            },
714            'collection': {
715                'name': 'collection',
716            },
717            'version': version,
718            'metadata': {
719                'dependencies': {},
720            }
721        }))),
722    ]
723    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
724
725    actual = api.get_collection_version_metadata('namespace', 'collection', version)
726
727    assert isinstance(actual, CollectionVersionMetadata)
728    assert actual.namespace == u'namespace'
729    assert actual.name == u'collection'
730    assert actual.download_url == u'https://downloadme.com'
731    assert actual.artifact_sha256 == u'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f'
732    assert actual.version == version
733    assert actual.dependencies == {}
734
735    assert mock_open.call_count == 1
736    assert mock_open.mock_calls[0][1][0] == '%s%s/collections/namespace/collection/versions/%s/' \
737        % (api.api_server, api_version, version)
738
739    # v2 calls dont need auth, so no authz header or token_type
740    if token_type:
741        assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
742
743
744@pytest.mark.parametrize('api_version, token_type, token_ins, response', [
745    ('v2', None, None, {
746        'count': 2,
747        'next': None,
748        'previous': None,
749        'results': [
750            {
751                'version': '1.0.0',
752                'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0',
753            },
754            {
755                'version': '1.0.1',
756                'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1',
757            },
758        ],
759    }),
760    # TODO: Verify this once Automation Hub is actually out
761    ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), {
762        'count': 2,
763        'next': None,
764        'previous': None,
765        'data': [
766            {
767                'version': '1.0.0',
768                'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0',
769            },
770            {
771                'version': '1.0.1',
772                'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1',
773            },
774        ],
775    }),
776])
777def test_get_collection_versions(api_version, token_type, token_ins, response, monkeypatch):
778    api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins)
779
780    if token_ins:
781        mock_token_get = MagicMock()
782        mock_token_get.return_value = 'my token'
783        monkeypatch.setattr(token_ins, 'get', mock_token_get)
784
785    mock_open = MagicMock()
786    mock_open.side_effect = [
787        StringIO(to_text(json.dumps(response))),
788    ]
789    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
790
791    actual = api.get_collection_versions('namespace', 'collection')
792    assert actual == [u'1.0.0', u'1.0.1']
793
794    page_query = '?limit=100' if api_version == 'v3' else '?page_size=100'
795    assert mock_open.call_count == 1
796    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \
797                                            'versions/%s' % (api_version, page_query)
798    if token_ins:
799        assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
800
801
802@pytest.mark.parametrize('api_version, token_type, token_ins, responses', [
803    ('v2', None, None, [
804        {
805            'count': 6,
806            'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100',
807            'previous': None,
808            'results': [  # Pay no mind, using more manageable results than page_size would indicate
809                {
810                    'version': '1.0.0',
811                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0',
812                },
813                {
814                    'version': '1.0.1',
815                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1',
816                },
817            ],
818        },
819        {
820            'count': 6,
821            'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=3&page_size=100',
822            'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions',
823            'results': [
824                {
825                    'version': '1.0.2',
826                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.2',
827                },
828                {
829                    'version': '1.0.3',
830                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.3',
831                },
832            ],
833        },
834        {
835            'count': 6,
836            'next': None,
837            'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100',
838            'results': [
839                {
840                    'version': '1.0.4',
841                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.4',
842                },
843                {
844                    'version': '1.0.5',
845                    'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.5',
846                },
847            ],
848        },
849    ]),
850    ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), [
851        {
852            'count': 6,
853            'links': {
854                # v3 links are relative and the limit is included during pagination
855                'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100',
856                'previous': None,
857            },
858            'data': [
859                {
860                    'version': '1.0.0',
861                    'href': '/api/v3/collections/namespace/collection/versions/1.0.0',
862                },
863                {
864                    'version': '1.0.1',
865                    'href': '/api/v3/collections/namespace/collection/versions/1.0.1',
866                },
867            ],
868        },
869        {
870            'count': 6,
871            'links': {
872                'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=200',
873                'previous': '/api/v3/collections/namespace/collection/versions',
874            },
875            'data': [
876                {
877                    'version': '1.0.2',
878                    'href': '/api/v3/collections/namespace/collection/versions/1.0.2',
879                },
880                {
881                    'version': '1.0.3',
882                    'href': '/api/v3/collections/namespace/collection/versions/1.0.3',
883                },
884            ],
885        },
886        {
887            'count': 6,
888            'links': {
889                'next': None,
890                'previous': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100',
891            },
892            'data': [
893                {
894                    'version': '1.0.4',
895                    'href': '/api/v3/collections/namespace/collection/versions/1.0.4',
896                },
897                {
898                    'version': '1.0.5',
899                    'href': '/api/v3/collections/namespace/collection/versions/1.0.5',
900                },
901            ],
902        },
903    ]),
904])
905def test_get_collection_versions_pagination(api_version, token_type, token_ins, responses, monkeypatch):
906    api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins)
907
908    if token_ins:
909        mock_token_get = MagicMock()
910        mock_token_get.return_value = 'my token'
911        monkeypatch.setattr(token_ins, 'get', mock_token_get)
912
913    mock_open = MagicMock()
914    mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses]
915    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
916
917    actual = api.get_collection_versions('namespace', 'collection')
918    assert actual == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5']
919
920    assert mock_open.call_count == 3
921
922    if api_version == 'v3':
923        query_1 = 'limit=100'
924        query_2 = 'limit=100&offset=100'
925        query_3 = 'limit=100&offset=200'
926    else:
927        query_1 = 'page_size=100'
928        query_2 = 'page=2&page_size=100'
929        query_3 = 'page=3&page_size=100'
930
931    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \
932                                            'versions/?%s' % (api_version, query_1)
933    assert mock_open.mock_calls[1][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \
934                                            'versions/?%s' % (api_version, query_2)
935    assert mock_open.mock_calls[2][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \
936                                            'versions/?%s' % (api_version, query_3)
937
938    if token_type:
939        assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type
940        assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type
941        assert mock_open.mock_calls[2][2]['headers']['Authorization'] == '%s my token' % token_type
942
943
944@pytest.mark.parametrize('responses', [
945    [
946        {
947            'count': 2,
948            'results': [{'name': '3.5.1', }, {'name': '3.5.2'}],
949            'next_link': None,
950            'next': None,
951            'previous_link': None,
952            'previous': None
953        },
954    ],
955    [
956        {
957            'count': 2,
958            'results': [{'name': '3.5.1'}],
959            'next_link': '/api/v1/roles/432/versions/?page=2&page_size=50',
960            'next': '/roles/432/versions/?page=2&page_size=50',
961            'previous_link': None,
962            'previous': None
963        },
964        {
965            'count': 2,
966            'results': [{'name': '3.5.2'}],
967            'next_link': None,
968            'next': None,
969            'previous_link': '/api/v1/roles/432/versions/?&page_size=50',
970            'previous': '/roles/432/versions/?page_size=50',
971        },
972    ]
973])
974def test_get_role_versions_pagination(monkeypatch, responses):
975    api = get_test_galaxy_api('https://galaxy.com/api/', 'v1')
976
977    mock_open = MagicMock()
978    mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses]
979    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
980
981    actual = api.fetch_role_related('versions', 432)
982    assert actual == [{'name': '3.5.1'}, {'name': '3.5.2'}]
983
984    assert mock_open.call_count == len(responses)
985
986    assert mock_open.mock_calls[0][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page_size=50'
987    if len(responses) == 2:
988        assert mock_open.mock_calls[1][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page=2&page_size=50'
989
990
991def test_missing_cache_dir(cache_dir):
992    os.rmdir(cache_dir)
993    GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False)
994
995    assert os.path.isdir(cache_dir)
996    assert stat.S_IMODE(os.stat(cache_dir).st_mode) == 0o700
997
998    cache_file = os.path.join(cache_dir, 'api.json')
999    with open(cache_file) as fd:
1000        actual_cache = fd.read()
1001    assert actual_cache == '{"version": 1}'
1002    assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600
1003
1004
1005def test_existing_cache(cache_dir):
1006    cache_file = os.path.join(cache_dir, 'api.json')
1007    cache_file_contents = '{"version": 1, "test": "json"}'
1008    with open(cache_file, mode='w') as fd:
1009        fd.write(cache_file_contents)
1010        os.chmod(cache_file, 0o655)
1011
1012    GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False)
1013
1014    assert os.path.isdir(cache_dir)
1015    with open(cache_file) as fd:
1016        actual_cache = fd.read()
1017    assert actual_cache == cache_file_contents
1018    assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o655
1019
1020
1021@pytest.mark.parametrize('content', [
1022    '',
1023    'value',
1024    '{"de" "finit" "ely" [\'invalid"]}',
1025    '[]',
1026    '{"version": 2, "test": "json"}',
1027    '{"version": 2, "key": "ÅÑŚÌβŁÈ"}',
1028])
1029def test_cache_invalid_cache_content(content, cache_dir):
1030    cache_file = os.path.join(cache_dir, 'api.json')
1031    with open(cache_file, mode='w') as fd:
1032        fd.write(content)
1033        os.chmod(cache_file, 0o664)
1034
1035    GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False)
1036
1037    with open(cache_file) as fd:
1038        actual_cache = fd.read()
1039    assert actual_cache == '{"version": 1}'
1040    assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o664
1041
1042
1043def test_cache_complete_pagination(cache_dir, monkeypatch):
1044
1045    responses = get_collection_versions()
1046    cache_file = os.path.join(cache_dir, 'api.json')
1047
1048    api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False)
1049
1050    mock_open = MagicMock(
1051        side_effect=[
1052            StringIO(to_text(json.dumps(r)))
1053            for r in responses
1054        ]
1055    )
1056    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
1057
1058    actual_versions = api.get_collection_versions('namespace', 'collection')
1059    assert actual_versions == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5']
1060
1061    with open(cache_file) as fd:
1062        final_cache = json.loads(fd.read())
1063
1064    cached_server = final_cache['galaxy.server.com:']
1065    cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/']
1066    cached_versions = [r['version'] for r in cached_collection['results']]
1067
1068    assert final_cache == api._cache
1069    assert cached_versions == actual_versions
1070
1071
1072def test_cache_flaky_pagination(cache_dir, monkeypatch):
1073
1074    responses = get_collection_versions()
1075    cache_file = os.path.join(cache_dir, 'api.json')
1076
1077    api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False)
1078
1079    # First attempt, fail midway through
1080    mock_open = MagicMock(
1081        side_effect=[
1082            StringIO(to_text(json.dumps(responses[0]))),
1083            StringIO(to_text(json.dumps(responses[1]))),
1084            urllib_error.HTTPError(responses[1]['next'], 500, 'Error', {}, StringIO()),
1085            StringIO(to_text(json.dumps(responses[3]))),
1086        ]
1087    )
1088    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
1089
1090    expected = (
1091        r'Error when getting available collection versions for namespace\.collection '
1092        r'from test \(https://galaxy\.server\.com/api/\) '
1093        r'\(HTTP Code: 500, Message: Error Code: Unknown\)'
1094    )
1095    with pytest.raises(GalaxyError, match=expected):
1096        api.get_collection_versions('namespace', 'collection')
1097
1098    with open(cache_file) as fd:
1099        final_cache = json.loads(fd.read())
1100
1101    assert final_cache == {
1102        'version': 1,
1103        'galaxy.server.com:': {
1104            'modified': {
1105                'namespace.collection': responses[0]['modified']
1106            }
1107        }
1108    }
1109
1110    # Reset API
1111    api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False)
1112
1113    # Second attempt is successful so cache should be populated
1114    mock_open = MagicMock(
1115        side_effect=[
1116            StringIO(to_text(json.dumps(r)))
1117            for r in responses
1118        ]
1119    )
1120    monkeypatch.setattr(galaxy_api, 'open_url', mock_open)
1121
1122    actual_versions = api.get_collection_versions('namespace', 'collection')
1123    assert actual_versions == [u'1.0.0', u'1.0.1', u'1.0.2', u'1.0.3', u'1.0.4', u'1.0.5']
1124
1125    with open(cache_file) as fd:
1126        final_cache = json.loads(fd.read())
1127
1128    cached_server = final_cache['galaxy.server.com:']
1129    cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/']
1130    cached_versions = [r['version'] for r in cached_collection['results']]
1131
1132    assert cached_versions == actual_versions
1133
1134
1135def test_world_writable_cache(cache_dir, monkeypatch):
1136    mock_warning = MagicMock()
1137    monkeypatch.setattr(Display, 'warning', mock_warning)
1138
1139    cache_file = os.path.join(cache_dir, 'api.json')
1140    with open(cache_file, mode='w') as fd:
1141        fd.write('{"version": 2}')
1142        os.chmod(cache_file, 0o666)
1143
1144    api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False)
1145    assert api._cache is None
1146
1147    with open(cache_file) as fd:
1148        actual_cache = fd.read()
1149    assert actual_cache == '{"version": 2}'
1150    assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o666
1151
1152    assert mock_warning.call_count == 1
1153    assert mock_warning.call_args[0][0] == \
1154        'Galaxy cache has world writable access (%s), ignoring it as a cache source.' % cache_file
1155
1156
1157def test_no_cache(cache_dir):
1158    cache_file = os.path.join(cache_dir, 'api.json')
1159    with open(cache_file, mode='w') as fd:
1160        fd.write('random')
1161
1162    api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/')
1163    assert api._cache is None
1164
1165    with open(cache_file) as fd:
1166        actual_cache = fd.read()
1167    assert actual_cache == 'random'
1168
1169
1170def test_clear_cache_with_no_cache(cache_dir):
1171    cache_file = os.path.join(cache_dir, 'api.json')
1172    with open(cache_file, mode='w') as fd:
1173        fd.write('{"version": 1, "key": "value"}')
1174
1175    GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True)
1176    assert not os.path.exists(cache_file)
1177
1178
1179def test_clear_cache(cache_dir):
1180    cache_file = os.path.join(cache_dir, 'api.json')
1181    with open(cache_file, mode='w') as fd:
1182        fd.write('{"version": 1, "key": "value"}')
1183
1184    GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True, no_cache=False)
1185
1186    with open(cache_file) as fd:
1187        actual_cache = fd.read()
1188    assert actual_cache == '{"version": 1}'
1189    assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600
1190
1191
1192@pytest.mark.parametrize(['url', 'expected'], [
1193    ('http://hostname/path', 'hostname:'),
1194    ('http://hostname:80/path', 'hostname:80'),
1195    ('https://testing.com:invalid', 'testing.com:'),
1196    ('https://testing.com:1234', 'testing.com:1234'),
1197    ('https://username:password@testing.com/path', 'testing.com:'),
1198    ('https://username:password@testing.com:443/path', 'testing.com:443'),
1199])
1200def test_cache_id(url, expected):
1201    actual = galaxy_api.get_cache_id(url)
1202    assert actual == expected
1203