diff options
Diffstat (limited to 'test/units/galaxy')
| -rw-r--r-- | test/units/galaxy/__init__.py | 0 | ||||
| -rw-r--r-- | test/units/galaxy/test_api.py | 1362 | ||||
| -rw-r--r-- | test/units/galaxy/test_collection.py | 1217 | ||||
| -rw-r--r-- | test/units/galaxy/test_collection_install.py | 1081 | ||||
| -rw-r--r-- | test/units/galaxy/test_role_install.py | 152 | ||||
| -rw-r--r-- | test/units/galaxy/test_role_requirements.py | 88 | ||||
| -rw-r--r-- | test/units/galaxy/test_token.py | 98 | ||||
| -rw-r--r-- | test/units/galaxy/test_user_agent.py | 18 |
8 files changed, 4016 insertions, 0 deletions
diff --git a/test/units/galaxy/__init__.py b/test/units/galaxy/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/galaxy/__init__.py diff --git a/test/units/galaxy/test_api.py b/test/units/galaxy/test_api.py new file mode 100644 index 0000000..064aff2 --- /dev/null +++ b/test/units/galaxy/test_api.py @@ -0,0 +1,1362 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import json +import os +import re +import pytest +import stat +import tarfile +import tempfile +import time + +from io import BytesIO, StringIO +from unittest.mock import MagicMock + +import ansible.constants as C +from ansible import context +from ansible.errors import AnsibleError +from ansible.galaxy import api as galaxy_api +from ansible.galaxy.api import CollectionVersionMetadata, GalaxyAPI, GalaxyError +from ansible.galaxy.token import BasicAuthToken, GalaxyToken, KeycloakToken +from ansible.module_utils._text import to_native, to_text +from ansible.module_utils.six.moves.urllib import error as urllib_error +from ansible.utils import context_objects as co +from ansible.utils.display import Display + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + # Required to initialise the GalaxyAPI object + context.CLIARGS._store = {'ignore_certs': False} + yield + co.GlobalCLIArgs._Singleton__instance = None + + +@pytest.fixture() +def collection_artifact(tmp_path_factory): + ''' Creates a collection artifact tarball that is ready to be published ''' + output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Output')) + + tar_path = os.path.join(output_dir, 'namespace-collection-v1.0.0.tar.gz') + with tarfile.open(tar_path, 'w:gz') as tfile: + b_io = BytesIO(b"\x00\x01\x02\x03") + tar_info = tarfile.TarInfo('test') + tar_info.size = 4 + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + yield tar_path + + +@pytest.fixture() +def cache_dir(tmp_path_factory, monkeypatch): + cache_dir = to_text(tmp_path_factory.mktemp('Test ÅÑŚÌβŁÈ Galaxy Cache')) + monkeypatch.setattr(C, 'GALAXY_CACHE_DIR', cache_dir) + + yield cache_dir + + +def get_test_galaxy_api(url, version, token_ins=None, token_value=None, no_cache=True): + token_value = token_value or "my token" + token_ins = token_ins or GalaxyToken(token_value) + api = GalaxyAPI(None, "test", url, no_cache=no_cache) + # Warning, this doesn't test g_connect() because _availabe_api_versions is set here. That means + # that urls for v2 servers have to append '/api/' themselves in the input data. + api._available_api_versions = {version: '%s' % version} + api.token = token_ins + + return api + + +def get_v3_collection_versions(namespace='namespace', name='collection'): + pagination_path = f"/api/galaxy/content/community/v3/plugin/{namespace}/content/community/collections/index/{namespace}/{name}/versions" + page_versions = (('1.0.0', '1.0.1',), ('1.0.2', '1.0.3',), ('1.0.4', '1.0.5'),) + responses = [ + {}, # TODO: initial response + ] + + first = f"{pagination_path}/?limit=100" + last = f"{pagination_path}/?limit=100&offset=200" + page_versions = [ + { + "versions": ('1.0.0', '1.0.1',), + "url": first, + }, + { + "versions": ('1.0.2', '1.0.3',), + "url": f"{pagination_path}/?limit=100&offset=100", + }, + { + "versions": ('1.0.4', '1.0.5'), + "url": last, + }, + ] + + previous = None + for page in range(0, len(page_versions)): + data = [] + + if page_versions[page]["url"] == last: + next_page = None + else: + next_page = page_versions[page + 1]["url"] + links = {"first": first, "last": last, "next": next_page, "previous": previous} + + for version in page_versions[page]["versions"]: + data.append( + { + "version": f"{version}", + "href": f"{pagination_path}/{version}/", + "created_at": "2022-05-13T15:55:58.913107Z", + "updated_at": "2022-05-13T15:55:58.913121Z", + "requires_ansible": ">=2.9.10" + } + ) + + responses.append({"meta": {"count": 6}, "links": links, "data": data}) + + previous = page_versions[page]["url"] + return responses + + +def get_collection_versions(namespace='namespace', name='collection'): + base_url = 'https://galaxy.server.com/api/v2/collections/{0}/{1}/'.format(namespace, name) + versions_url = base_url + 'versions/' + + # Response for collection info + responses = [ + { + "id": 1000, + "href": base_url, + "name": name, + "namespace": { + "id": 30000, + "href": "https://galaxy.ansible.com/api/v1/namespaces/30000/", + "name": namespace, + }, + "versions_url": versions_url, + "latest_version": { + "version": "1.0.5", + "href": versions_url + "1.0.5/" + }, + "deprecated": False, + "created": "2021-02-09T16:55:42.749915-05:00", + "modified": "2021-02-09T16:55:42.749915-05:00", + } + ] + + # Paginated responses for versions + page_versions = (('1.0.0', '1.0.1',), ('1.0.2', '1.0.3',), ('1.0.4', '1.0.5'),) + last_page = None + for page in range(1, len(page_versions) + 1): + if page < len(page_versions): + next_page = versions_url + '?page={0}'.format(page + 1) + else: + next_page = None + + version_results = [] + for version in page_versions[int(page - 1)]: + version_results.append( + {'version': version, 'href': versions_url + '{0}/'.format(version)} + ) + + responses.append( + { + 'count': 6, + 'next': next_page, + 'previous': last_page, + 'results': version_results, + } + ) + last_page = page + + return responses + + +def test_api_no_auth(): + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") + actual = {} + api._add_auth_token(actual, "") + assert actual == {} + + +def test_api_no_auth_but_required(): + expected = "No access token or username set. A token can be set with --api-key or at " + with pytest.raises(AnsibleError, match=expected): + GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/")._add_auth_token({}, "", required=True) + + +def test_api_token_auth(): + token = GalaxyToken(token=u"my_token") + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + api._add_auth_token(actual, "", required=True) + assert actual == {'Authorization': 'Token my_token'} + + +def test_api_token_auth_with_token_type(monkeypatch): + token = KeycloakToken(auth_url='https://api.test/') + mock_token_get = MagicMock() + mock_token_get.return_value = 'my_token' + monkeypatch.setattr(token, 'get', mock_token_get) + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + api._add_auth_token(actual, "", token_type="Bearer", required=True) + assert actual == {'Authorization': 'Bearer my_token'} + + +def test_api_token_auth_with_v3_url(monkeypatch): + token = KeycloakToken(auth_url='https://api.test/') + mock_token_get = MagicMock() + mock_token_get.return_value = 'my_token' + monkeypatch.setattr(token, 'get', mock_token_get) + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + api._add_auth_token(actual, "https://galaxy.ansible.com/api/v3/resource/name", required=True) + assert actual == {'Authorization': 'Bearer my_token'} + + +def test_api_token_auth_with_v2_url(): + token = GalaxyToken(token=u"my_token") + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + # Add v3 to random part of URL but response should only see the v2 as the full URI path segment. + api._add_auth_token(actual, "https://galaxy.ansible.com/api/v2/resourcev3/name", required=True) + assert actual == {'Authorization': 'Token my_token'} + + +def test_api_basic_auth_password(): + token = BasicAuthToken(username=u"user", password=u"pass") + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + api._add_auth_token(actual, "", required=True) + assert actual == {'Authorization': 'Basic dXNlcjpwYXNz'} + + +def test_api_basic_auth_no_password(): + token = BasicAuthToken(username=u"user") + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + actual = {} + api._add_auth_token(actual, "", required=True) + assert actual == {'Authorization': 'Basic dXNlcjo='} + + +def test_api_dont_override_auth_header(): + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") + actual = {'Authorization': 'Custom token'} + api._add_auth_token(actual, "", required=True) + assert actual == {'Authorization': 'Custom token'} + + +def test_initialise_galaxy(monkeypatch): + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"token":"my token"}'), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") + actual = api.authenticate("github_token") + + assert len(api.available_api_versions) == 2 + assert api.available_api_versions['v1'] == u'v1/' + assert api.available_api_versions['v2'] == u'v2/' + assert actual == {u'token': u'my token'} + assert mock_open.call_count == 2 + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' + assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] + assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/' + assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent'] + assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token' + + +def test_initialise_galaxy_with_auth(monkeypatch): + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"token":"my token"}'), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token')) + actual = api.authenticate("github_token") + + assert len(api.available_api_versions) == 2 + assert api.available_api_versions['v1'] == u'v1/' + assert api.available_api_versions['v2'] == u'v2/' + assert actual == {u'token': u'my token'} + assert mock_open.call_count == 2 + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' + assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] + assert mock_open.mock_calls[1][1][0] == 'https://galaxy.ansible.com/api/v1/tokens/' + assert 'ansible-galaxy' in mock_open.mock_calls[1][2]['http_agent'] + assert mock_open.mock_calls[1][2]['data'] == 'github_token=github_token' + + +def test_initialise_automation_hub(monkeypatch): + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(u'{"available_versions":{"v2": "v2/", "v3":"v3/"}}'), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + token = KeycloakToken(auth_url='https://api.test/') + mock_token_get = MagicMock() + mock_token_get.return_value = 'my_token' + monkeypatch.setattr(token, 'get', mock_token_get) + + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=token) + + assert len(api.available_api_versions) == 2 + assert api.available_api_versions['v2'] == u'v2/' + assert api.available_api_versions['v3'] == u'v3/' + + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' + assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] + assert mock_open.mock_calls[0][2]['headers'] == {'Authorization': 'Bearer my_token'} + + +def test_initialise_unknown(monkeypatch): + mock_open = MagicMock() + mock_open.side_effect = [ + urllib_error.HTTPError('https://galaxy.ansible.com/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')), + urllib_error.HTTPError('https://galaxy.ansible.com/api/api/', 500, 'msg', {}, StringIO(u'{"msg":"raw error"}')), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/", token=GalaxyToken(token='my_token')) + + expected = "Error when finding available api versions from test (%s) (HTTP Code: 500, Message: msg)" \ + % api.api_server + with pytest.raises(AnsibleError, match=re.escape(expected)): + api.authenticate("github_token") + + +def test_get_available_api_versions(monkeypatch): + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/","v2":"v2/"}}'), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + api = GalaxyAPI(None, "test", "https://galaxy.ansible.com/api/") + actual = api.available_api_versions + assert len(actual) == 2 + assert actual['v1'] == u'v1/' + assert actual['v2'] == u'v2/' + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/' + assert 'ansible-galaxy' in mock_open.mock_calls[0][2]['http_agent'] + + +def test_publish_collection_missing_file(): + fake_path = u'/fake/ÅÑŚÌβŁÈ/path' + expected = to_native("The collection path specified '%s' does not exist." % fake_path) + + api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2") + with pytest.raises(AnsibleError, match=expected): + api.publish_collection(fake_path) + + +def test_publish_collection_not_a_tarball(): + expected = "The collection path specified '{0}' is not a tarball, use 'ansible-galaxy collection build' to " \ + "create a proper release artifact." + + api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v2") + with tempfile.NamedTemporaryFile(prefix=u'ÅÑŚÌβŁÈ') as temp_file: + temp_file.write(b"\x00") + temp_file.flush() + with pytest.raises(AnsibleError, match=expected.format(to_native(temp_file.name))): + api.publish_collection(temp_file.name) + + +def test_publish_collection_unsupported_version(): + expected = "Galaxy action publish_collection requires API versions 'v2, v3' but only 'v1' are available on test " \ + "https://galaxy.ansible.com/api/" + + api = get_test_galaxy_api("https://galaxy.ansible.com/api/", "v1") + with pytest.raises(AnsibleError, match=expected): + api.publish_collection("path") + + +@pytest.mark.parametrize('api_version, collection_url', [ + ('v2', 'collections'), + ('v3', 'artifacts/collections'), +]) +def test_publish_collection(api_version, collection_url, collection_artifact, monkeypatch): + api = get_test_galaxy_api("https://galaxy.ansible.com/api/", api_version) + + mock_call = MagicMock() + mock_call.return_value = {'task': 'http://task.url/'} + monkeypatch.setattr(api, '_call_galaxy', mock_call) + + actual = api.publish_collection(collection_artifact) + assert actual == 'http://task.url/' + assert mock_call.call_count == 1 + assert mock_call.mock_calls[0][1][0] == 'https://galaxy.ansible.com/api/%s/%s/' % (api_version, collection_url) + assert mock_call.mock_calls[0][2]['headers']['Content-length'] == len(mock_call.mock_calls[0][2]['args']) + assert mock_call.mock_calls[0][2]['headers']['Content-type'].startswith( + 'multipart/form-data; boundary=') + assert mock_call.mock_calls[0][2]['args'].startswith(b'--') + assert mock_call.mock_calls[0][2]['method'] == 'POST' + assert mock_call.mock_calls[0][2]['auth_required'] is True + + +@pytest.mark.parametrize('api_version, collection_url, response, expected', [ + ('v2', 'collections', {}, + 'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'), + ('v2', 'collections', { + 'message': u'Galaxy error messäge', + 'code': 'GWE002', + }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Galaxy error messäge Code: GWE002)'), + ('v3', 'artifact/collections', {}, + 'Error when publishing collection to test (%s) (HTTP Code: 500, Message: msg Code: Unknown)'), + ('v3', 'artifact/collections', { + 'errors': [ + { + 'code': 'conflict.collection_exists', + 'detail': 'Collection "mynamespace-mycollection-4.1.1" already exists.', + 'title': 'Conflict.', + 'status': '400', + }, + { + 'code': 'quantum_improbability', + 'title': u'Rändom(?) quantum improbability.', + 'source': {'parameter': 'the_arrow_of_time'}, + 'meta': {'remediation': 'Try again before'}, + }, + ], + }, u'Error when publishing collection to test (%s) (HTTP Code: 500, Message: Collection ' + u'"mynamespace-mycollection-4.1.1" already exists. Code: conflict.collection_exists), (HTTP Code: 500, ' + u'Message: Rändom(?) quantum improbability. Code: quantum_improbability)') +]) +def test_publish_failure(api_version, collection_url, response, expected, collection_artifact, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version) + + expected_url = '%s/api/%s/%s' % (api.api_server, api_version, collection_url) + + mock_open = MagicMock() + mock_open.side_effect = urllib_error.HTTPError(expected_url, 500, 'msg', {}, + StringIO(to_text(json.dumps(response)))) + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + with pytest.raises(GalaxyError, match=re.escape(to_native(expected % api.api_server))): + api.publish_collection(collection_artifact) + + +@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ + ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'), + '1234', + 'https://galaxy.server.com/api/v2/collection-imports/1234/'), + ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), + '1234', + 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), +]) +def test_wait_import_task(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): + api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.return_value = StringIO(u'{"state":"success","finished_at":"time"}') + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + api.wait_import_task(import_uri) + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == full_import_uri + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri + + +@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ + ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'), + '1234', + 'https://galaxy.server.com/api/v2/collection-imports/1234/'), + ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), + '1234', + 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), +]) +def test_wait_import_task_multiple_requests(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): + api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(u'{"state":"test"}'), + StringIO(u'{"state":"success","finished_at":"time"}'), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + mock_vvv = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_vvv) + + monkeypatch.setattr(time, 'sleep', MagicMock()) + + api.wait_import_task(import_uri) + + assert mock_open.call_count == 2 + assert mock_open.mock_calls[0][1][0] == full_import_uri + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + assert mock_open.mock_calls[1][1][0] == full_import_uri + assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri + + assert mock_vvv.call_count == 1 + assert mock_vvv.mock_calls[0][1][0] == \ + 'Galaxy import process has a status of test, wait 2 seconds before trying again' + + +@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri,', [ + ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my token'), + '1234', + 'https://galaxy.server.com/api/v2/collection-imports/1234/'), + ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), + '1234', + 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), +]) +def test_wait_import_task_with_failure(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): + api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(to_text(json.dumps({ + 'finished_at': 'some_time', + 'state': 'failed', + 'error': { + 'code': 'GW001', + 'description': u'Becäuse I said so!', + + }, + 'messages': [ + { + 'level': 'ERrOR', + 'message': u'Somé error', + }, + { + 'level': 'WARNiNG', + 'message': u'Some wärning', + }, + { + 'level': 'INFO', + 'message': u'Somé info', + }, + ], + }))), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + mock_vvv = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_vvv) + + mock_warn = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_warn) + + mock_err = MagicMock() + monkeypatch.setattr(Display, 'error', mock_err) + + expected = to_native(u'Galaxy import process failed: Becäuse I said so! (Code: GW001)') + with pytest.raises(AnsibleError, match=re.escape(expected)): + api.wait_import_task(import_uri) + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == full_import_uri + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri + + assert mock_vvv.call_count == 1 + assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: INFO - Somé info' + + assert mock_warn.call_count == 1 + assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning' + + assert mock_err.call_count == 1 + assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error' + + +@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ + ('https://galaxy.server.com/api/', 'v2', 'Token', GalaxyToken('my_token'), + '1234', + 'https://galaxy.server.com/api/v2/collection-imports/1234/'), + ('https://galaxy.server.com/api/automation-hub/', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), + '1234', + 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), +]) +def test_wait_import_task_with_failure_no_error(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): + api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(to_text(json.dumps({ + 'finished_at': 'some_time', + 'state': 'failed', + 'error': {}, + 'messages': [ + { + 'level': 'ERROR', + 'message': u'Somé error', + }, + { + 'level': 'WARNING', + 'message': u'Some wärning', + }, + { + 'level': 'INFO', + 'message': u'Somé info', + }, + ], + }))), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + mock_vvv = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_vvv) + + mock_warn = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_warn) + + mock_err = MagicMock() + monkeypatch.setattr(Display, 'error', mock_err) + + expected = 'Galaxy import process failed: Unknown error, see %s for more details \\(Code: UNKNOWN\\)' % full_import_uri + with pytest.raises(AnsibleError, match=expected): + api.wait_import_task(import_uri) + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == full_import_uri + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri + + assert mock_vvv.call_count == 1 + assert mock_vvv.mock_calls[0][1][0] == u'Galaxy import message: INFO - Somé info' + + assert mock_warn.call_count == 1 + assert mock_warn.mock_calls[0][1][0] == u'Galaxy import warning message: Some wärning' + + assert mock_err.call_count == 1 + assert mock_err.mock_calls[0][1][0] == u'Galaxy import error message: Somé error' + + +@pytest.mark.parametrize('server_url, api_version, token_type, token_ins, import_uri, full_import_uri', [ + ('https://galaxy.server.com/api', 'v2', 'Token', GalaxyToken('my token'), + '1234', + 'https://galaxy.server.com/api/v2/collection-imports/1234/'), + ('https://galaxy.server.com/api/automation-hub', 'v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), + '1234', + 'https://galaxy.server.com/api/automation-hub/v3/imports/collections/1234/'), +]) +def test_wait_import_task_timeout(server_url, api_version, token_type, token_ins, import_uri, full_import_uri, monkeypatch): + api = get_test_galaxy_api(server_url, api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + def return_response(*args, **kwargs): + return StringIO(u'{"state":"waiting"}') + + mock_open = MagicMock() + mock_open.side_effect = return_response + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + mock_vvv = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_vvv) + + monkeypatch.setattr(time, 'sleep', MagicMock()) + + expected = "Timeout while waiting for the Galaxy import process to finish, check progress at '%s'" % full_import_uri + with pytest.raises(AnsibleError, match=expected): + api.wait_import_task(import_uri, 1) + + assert mock_open.call_count > 1 + assert mock_open.mock_calls[0][1][0] == full_import_uri + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + assert mock_open.mock_calls[1][1][0] == full_import_uri + assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == 'Waiting until Galaxy import task %s has completed' % full_import_uri + + # expected_wait_msg = 'Galaxy import process has a status of waiting, wait {0} seconds before trying again' + assert mock_vvv.call_count > 9 # 1st is opening Galaxy token file. + + # FIXME: + # assert mock_vvv.mock_calls[1][1][0] == expected_wait_msg.format(2) + # assert mock_vvv.mock_calls[2][1][0] == expected_wait_msg.format(3) + # assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(4) + # assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(6) + # assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(10) + # assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(15) + # assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(22) + # assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(30) + + +@pytest.mark.parametrize('api_version, token_type, version, token_ins', [ + ('v2', None, 'v2.1.13', None), + ('v3', 'Bearer', 'v1.0.0', KeycloakToken(auth_url='https://api.test/api/automation-hub/')), +]) +def test_get_collection_version_metadata_no_version(api_version, token_type, version, token_ins, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(to_text(json.dumps({ + 'href': 'https://galaxy.server.com/api/{api}/namespace/name/versions/{version}/'.format(api=api_version, version=version), + 'download_url': 'https://downloadme.com', + 'artifact': { + 'sha256': 'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f', + }, + 'namespace': { + 'name': 'namespace', + }, + 'collection': { + 'name': 'collection', + }, + 'version': version, + 'metadata': { + 'dependencies': {}, + } + }))), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.get_collection_version_metadata('namespace', 'collection', version) + + assert isinstance(actual, CollectionVersionMetadata) + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.download_url == u'https://downloadme.com' + assert actual.artifact_sha256 == u'ac47b6fac117d7c171812750dacda655b04533cf56b31080b82d1c0db3c9d80f' + assert actual.version == version + assert actual.dependencies == {} + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == '%s%s/collections/namespace/collection/versions/%s/' \ + % (api.api_server, api_version, version) + + # v2 calls dont need auth, so no authz header or token_type + if token_type: + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + +@pytest.mark.parametrize('api_version, token_type, token_ins, version', [ + ('v2', None, None, '2.1.13'), + ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/api/automation-hub/'), '1.0.0'), +]) +def test_get_collection_signatures_backwards_compat(api_version, token_type, token_ins, version, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO("{}") + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.get_collection_signatures('namespace', 'collection', version) + assert actual == [] + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == '%s%s/collections/namespace/collection/versions/%s/' \ + % (api.api_server, api_version, version) + + # v2 calls dont need auth, so no authz header or token_type + if token_type: + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + +@pytest.mark.parametrize('api_version, token_type, token_ins, version', [ + ('v2', None, None, '2.1.13'), + ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/api/automation-hub/'), '1.0.0'), +]) +def test_get_collection_signatures(api_version, token_type, token_ins, version, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(to_text(json.dumps({ + 'signatures': [ + { + "signature": "-----BEGIN PGP SIGNATURE-----\nSIGNATURE1\n-----END PGP SIGNATURE-----\n", + "pubkey_fingerprint": "FINGERPRINT", + "signing_service": "ansible-default", + "pulp_created": "2022-01-14T14:05:53.835605Z", + }, + { + "signature": "-----BEGIN PGP SIGNATURE-----\nSIGNATURE2\n-----END PGP SIGNATURE-----\n", + "pubkey_fingerprint": "FINGERPRINT", + "signing_service": "ansible-default", + "pulp_created": "2022-01-14T14:05:53.835605Z", + }, + ], + }))), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.get_collection_signatures('namespace', 'collection', version) + + assert actual == [ + "-----BEGIN PGP SIGNATURE-----\nSIGNATURE1\n-----END PGP SIGNATURE-----\n", + "-----BEGIN PGP SIGNATURE-----\nSIGNATURE2\n-----END PGP SIGNATURE-----\n" + ] + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == '%s%s/collections/namespace/collection/versions/%s/' \ + % (api.api_server, api_version, version) + + # v2 calls dont need auth, so no authz header or token_type + if token_type: + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + +@pytest.mark.parametrize('api_version, token_type, token_ins, response', [ + ('v2', None, None, { + 'count': 2, + 'next': None, + 'previous': None, + 'results': [ + { + 'version': '1.0.0', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', + }, + { + 'version': '1.0.1', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', + }, + ], + }), + # TODO: Verify this once Automation Hub is actually out + ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), { + 'count': 2, + 'next': None, + 'previous': None, + 'data': [ + { + 'version': '1.0.0', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', + }, + { + 'version': '1.0.1', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', + }, + ], + }), +]) +def test_get_collection_versions(api_version, token_type, token_ins, response, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [ + StringIO(to_text(json.dumps(response))), + ] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.get_collection_versions('namespace', 'collection') + assert actual == [u'1.0.0', u'1.0.1'] + + page_query = '?limit=100' if api_version == 'v3' else '?page_size=100' + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ + 'versions/%s' % (api_version, page_query) + if token_ins: + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + + +@pytest.mark.parametrize('api_version, token_type, token_ins, responses', [ + ('v2', None, None, [ + { + 'count': 6, + 'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100', + 'previous': None, + 'results': [ # Pay no mind, using more manageable results than page_size would indicate + { + 'version': '1.0.0', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.0', + }, + { + 'version': '1.0.1', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.1', + }, + ], + }, + { + 'count': 6, + 'next': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=3&page_size=100', + 'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions', + 'results': [ + { + 'version': '1.0.2', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.2', + }, + { + 'version': '1.0.3', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.3', + }, + ], + }, + { + 'count': 6, + 'next': None, + 'previous': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/?page=2&page_size=100', + 'results': [ + { + 'version': '1.0.4', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.4', + }, + { + 'version': '1.0.5', + 'href': 'https://galaxy.server.com/api/v2/collections/namespace/collection/versions/1.0.5', + }, + ], + }, + ]), + ('v3', 'Bearer', KeycloakToken(auth_url='https://api.test/'), [ + { + 'count': 6, + 'links': { + # v3 links are relative and the limit is included during pagination + 'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100', + 'previous': None, + }, + 'data': [ + { + 'version': '1.0.0', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.0', + }, + { + 'version': '1.0.1', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.1', + }, + ], + }, + { + 'count': 6, + 'links': { + 'next': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=200', + 'previous': '/api/v3/collections/namespace/collection/versions', + }, + 'data': [ + { + 'version': '1.0.2', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.2', + }, + { + 'version': '1.0.3', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.3', + }, + ], + }, + { + 'count': 6, + 'links': { + 'next': None, + 'previous': '/api/v3/collections/namespace/collection/versions/?limit=100&offset=100', + }, + 'data': [ + { + 'version': '1.0.4', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.4', + }, + { + 'version': '1.0.5', + 'href': '/api/v3/collections/namespace/collection/versions/1.0.5', + }, + ], + }, + ]), +]) +def test_get_collection_versions_pagination(api_version, token_type, token_ins, responses, monkeypatch): + api = get_test_galaxy_api('https://galaxy.server.com/api/', api_version, token_ins=token_ins) + + if token_ins: + mock_token_get = MagicMock() + mock_token_get.return_value = 'my token' + monkeypatch.setattr(token_ins, 'get', mock_token_get) + + mock_open = MagicMock() + mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.get_collection_versions('namespace', 'collection') + 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'] + + assert mock_open.call_count == 3 + + if api_version == 'v3': + query_1 = 'limit=100' + query_2 = 'limit=100&offset=100' + query_3 = 'limit=100&offset=200' + else: + query_1 = 'page_size=100' + query_2 = 'page=2&page_size=100' + query_3 = 'page=3&page_size=100' + + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ + 'versions/?%s' % (api_version, query_1) + assert mock_open.mock_calls[1][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ + 'versions/?%s' % (api_version, query_2) + assert mock_open.mock_calls[2][1][0] == 'https://galaxy.server.com/api/%s/collections/namespace/collection/' \ + 'versions/?%s' % (api_version, query_3) + + if token_type: + assert mock_open.mock_calls[0][2]['headers']['Authorization'] == '%s my token' % token_type + assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s my token' % token_type + assert mock_open.mock_calls[2][2]['headers']['Authorization'] == '%s my token' % token_type + + +@pytest.mark.parametrize('responses', [ + [ + { + 'count': 2, + 'results': [{'name': '3.5.1', }, {'name': '3.5.2'}], + 'next_link': None, + 'next': None, + 'previous_link': None, + 'previous': None + }, + ], + [ + { + 'count': 2, + 'results': [{'name': '3.5.1'}], + 'next_link': '/api/v1/roles/432/versions/?page=2&page_size=50', + 'next': '/roles/432/versions/?page=2&page_size=50', + 'previous_link': None, + 'previous': None + }, + { + 'count': 2, + 'results': [{'name': '3.5.2'}], + 'next_link': None, + 'next': None, + 'previous_link': '/api/v1/roles/432/versions/?&page_size=50', + 'previous': '/roles/432/versions/?page_size=50', + }, + ] +]) +def test_get_role_versions_pagination(monkeypatch, responses): + api = get_test_galaxy_api('https://galaxy.com/api/', 'v1') + + mock_open = MagicMock() + mock_open.side_effect = [StringIO(to_text(json.dumps(r))) for r in responses] + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual = api.fetch_role_related('versions', 432) + assert actual == [{'name': '3.5.1'}, {'name': '3.5.2'}] + + assert mock_open.call_count == len(responses) + + assert mock_open.mock_calls[0][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page_size=50' + if len(responses) == 2: + assert mock_open.mock_calls[1][1][0] == 'https://galaxy.com/api/v1/roles/432/versions/?page=2&page_size=50' + + +def test_missing_cache_dir(cache_dir): + os.rmdir(cache_dir) + GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) + + assert os.path.isdir(cache_dir) + assert stat.S_IMODE(os.stat(cache_dir).st_mode) == 0o700 + + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == '{"version": 1}' + assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600 + + +def test_existing_cache(cache_dir): + cache_file = os.path.join(cache_dir, 'api.json') + cache_file_contents = '{"version": 1, "test": "json"}' + with open(cache_file, mode='w') as fd: + fd.write(cache_file_contents) + os.chmod(cache_file, 0o655) + + GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) + + assert os.path.isdir(cache_dir) + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == cache_file_contents + assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o655 + + +@pytest.mark.parametrize('content', [ + '', + 'value', + '{"de" "finit" "ely" [\'invalid"]}', + '[]', + '{"version": 2, "test": "json"}', + '{"version": 2, "key": "ÅÑŚÌβŁÈ"}', +]) +def test_cache_invalid_cache_content(content, cache_dir): + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file, mode='w') as fd: + fd.write(content) + os.chmod(cache_file, 0o664) + + GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) + + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == '{"version": 1}' + assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o664 + + +def test_cache_complete_pagination(cache_dir, monkeypatch): + + responses = get_collection_versions() + cache_file = os.path.join(cache_dir, 'api.json') + + api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) + + mock_open = MagicMock( + side_effect=[ + StringIO(to_text(json.dumps(r))) + for r in responses + ] + ) + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual_versions = api.get_collection_versions('namespace', 'collection') + 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'] + + with open(cache_file) as fd: + final_cache = json.loads(fd.read()) + + cached_server = final_cache['galaxy.server.com:'] + cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/'] + cached_versions = [r['version'] for r in cached_collection['results']] + + assert final_cache == api._cache + assert cached_versions == actual_versions + + +def test_cache_complete_pagination_v3(cache_dir, monkeypatch): + + responses = get_v3_collection_versions() + cache_file = os.path.join(cache_dir, 'api.json') + + api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v3', no_cache=False) + + mock_open = MagicMock( + side_effect=[ + StringIO(to_text(json.dumps(r))) + for r in responses + ] + ) + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual_versions = api.get_collection_versions('namespace', 'collection') + 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'] + + with open(cache_file) as fd: + final_cache = json.loads(fd.read()) + + cached_server = final_cache['galaxy.server.com:'] + cached_collection = cached_server['/api/v3/collections/namespace/collection/versions/'] + cached_versions = [r['version'] for r in cached_collection['results']] + + assert final_cache == api._cache + assert cached_versions == actual_versions + + +def test_cache_flaky_pagination(cache_dir, monkeypatch): + + responses = get_collection_versions() + cache_file = os.path.join(cache_dir, 'api.json') + + api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) + + # First attempt, fail midway through + mock_open = MagicMock( + side_effect=[ + StringIO(to_text(json.dumps(responses[0]))), + StringIO(to_text(json.dumps(responses[1]))), + urllib_error.HTTPError(responses[1]['next'], 500, 'Error', {}, StringIO()), + StringIO(to_text(json.dumps(responses[3]))), + ] + ) + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + expected = ( + r'Error when getting available collection versions for namespace\.collection ' + r'from test \(https://galaxy\.server\.com/api/\) ' + r'\(HTTP Code: 500, Message: Error Code: Unknown\)' + ) + with pytest.raises(GalaxyError, match=expected): + api.get_collection_versions('namespace', 'collection') + + with open(cache_file) as fd: + final_cache = json.loads(fd.read()) + + assert final_cache == { + 'version': 1, + 'galaxy.server.com:': { + 'modified': { + 'namespace.collection': responses[0]['modified'] + } + } + } + + # Reset API + api = get_test_galaxy_api('https://galaxy.server.com/api/', 'v2', no_cache=False) + + # Second attempt is successful so cache should be populated + mock_open = MagicMock( + side_effect=[ + StringIO(to_text(json.dumps(r))) + for r in responses + ] + ) + monkeypatch.setattr(galaxy_api, 'open_url', mock_open) + + actual_versions = api.get_collection_versions('namespace', 'collection') + 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'] + + with open(cache_file) as fd: + final_cache = json.loads(fd.read()) + + cached_server = final_cache['galaxy.server.com:'] + cached_collection = cached_server['/api/v2/collections/namespace/collection/versions/'] + cached_versions = [r['version'] for r in cached_collection['results']] + + assert cached_versions == actual_versions + + +def test_world_writable_cache(cache_dir, monkeypatch): + mock_warning = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_warning) + + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file, mode='w') as fd: + fd.write('{"version": 2}') + os.chmod(cache_file, 0o666) + + api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', no_cache=False) + assert api._cache is None + + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == '{"version": 2}' + assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o666 + + assert mock_warning.call_count == 1 + assert mock_warning.call_args[0][0] == \ + 'Galaxy cache has world writable access (%s), ignoring it as a cache source.' % cache_file + + +def test_no_cache(cache_dir): + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file, mode='w') as fd: + fd.write('random') + + api = GalaxyAPI(None, "test", 'https://galaxy.ansible.com/') + assert api._cache is None + + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == 'random' + + +def test_clear_cache_with_no_cache(cache_dir): + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file, mode='w') as fd: + fd.write('{"version": 1, "key": "value"}') + + GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True) + assert not os.path.exists(cache_file) + + +def test_clear_cache(cache_dir): + cache_file = os.path.join(cache_dir, 'api.json') + with open(cache_file, mode='w') as fd: + fd.write('{"version": 1, "key": "value"}') + + GalaxyAPI(None, "test", 'https://galaxy.ansible.com/', clear_response_cache=True, no_cache=False) + + with open(cache_file) as fd: + actual_cache = fd.read() + assert actual_cache == '{"version": 1}' + assert stat.S_IMODE(os.stat(cache_file).st_mode) == 0o600 + + +@pytest.mark.parametrize(['url', 'expected'], [ + ('http://hostname/path', 'hostname:'), + ('http://hostname:80/path', 'hostname:80'), + ('https://testing.com:invalid', 'testing.com:'), + ('https://testing.com:1234', 'testing.com:1234'), + ('https://username:password@testing.com/path', 'testing.com:'), + ('https://username:password@testing.com:443/path', 'testing.com:443'), +]) +def test_cache_id(url, expected): + actual = galaxy_api.get_cache_id(url) + assert actual == expected diff --git a/test/units/galaxy/test_collection.py b/test/units/galaxy/test_collection.py new file mode 100644 index 0000000..106251c --- /dev/null +++ b/test/units/galaxy/test_collection.py @@ -0,0 +1,1217 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import json +import os +import pytest +import re +import tarfile +import tempfile +import uuid + +from hashlib import sha256 +from io import BytesIO +from unittest.mock import MagicMock, mock_open, patch + +import ansible.constants as C +from ansible import context +from ansible.cli.galaxy import GalaxyCLI, SERVER_DEF +from ansible.errors import AnsibleError +from ansible.galaxy import api, collection, token +from ansible.module_utils._text import to_bytes, to_native, to_text +from ansible.module_utils.six.moves import builtins +from ansible.utils import context_objects as co +from ansible.utils.display import Display +from ansible.utils.hashing import secure_hash_s +from ansible.utils.sentinel import Sentinel + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + yield + co.GlobalCLIArgs._Singleton__instance = None + + +@pytest.fixture() +def collection_input(tmp_path_factory): + ''' Creates a collection skeleton directory for build tests ''' + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + namespace = 'ansible_namespace' + collection = 'collection' + skeleton = os.path.join(os.path.dirname(os.path.split(__file__)[0]), 'cli', 'test_data', 'collection_skeleton') + + galaxy_args = ['ansible-galaxy', 'collection', 'init', '%s.%s' % (namespace, collection), + '-c', '--init-path', test_dir, '--collection-skeleton', skeleton] + GalaxyCLI(args=galaxy_args).run() + collection_dir = os.path.join(test_dir, namespace, collection) + output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Output')) + + return collection_dir, output_dir + + +@pytest.fixture() +def collection_artifact(monkeypatch, tmp_path_factory): + ''' Creates a temp collection artifact and mocked open_url instance for publishing tests ''' + mock_open = MagicMock() + monkeypatch.setattr(collection.concrete_artifact_manager, 'open_url', mock_open) + + mock_uuid = MagicMock() + mock_uuid.return_value.hex = 'uuid' + monkeypatch.setattr(uuid, 'uuid4', mock_uuid) + + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + input_file = to_text(tmp_path / 'collection.tar.gz') + + with tarfile.open(input_file, 'w:gz') as tfile: + b_io = BytesIO(b"\x00\x01\x02\x03") + tar_info = tarfile.TarInfo('test') + tar_info.size = 4 + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + return input_file, mock_open + + +@pytest.fixture() +def galaxy_yml_dir(request, tmp_path_factory): + b_test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + b_galaxy_yml = os.path.join(b_test_dir, b'galaxy.yml') + with open(b_galaxy_yml, 'wb') as galaxy_obj: + galaxy_obj.write(to_bytes(request.param)) + + yield b_test_dir + + +@pytest.fixture() +def tmp_tarfile(tmp_path_factory, manifest_info): + ''' Creates a temporary tar file for _extract_tar_file tests ''' + filename = u'ÅÑŚÌβŁÈ' + temp_dir = to_bytes(tmp_path_factory.mktemp('test-%s Collections' % to_native(filename))) + tar_file = os.path.join(temp_dir, to_bytes('%s.tar.gz' % filename)) + data = os.urandom(8) + + with tarfile.open(tar_file, 'w:gz') as tfile: + b_io = BytesIO(data) + tar_info = tarfile.TarInfo(filename) + tar_info.size = len(data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + b_data = to_bytes(json.dumps(manifest_info, indent=True), errors='surrogate_or_strict') + b_io = BytesIO(b_data) + tar_info = tarfile.TarInfo('MANIFEST.json') + tar_info.size = len(b_data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + sha256_hash = sha256() + sha256_hash.update(data) + + with tarfile.open(tar_file, 'r') as tfile: + yield temp_dir, tfile, filename, sha256_hash.hexdigest() + + +@pytest.fixture() +def galaxy_server(): + context.CLIARGS._store = {'ignore_certs': False} + galaxy_api = api.GalaxyAPI(None, 'test_server', 'https://galaxy.ansible.com', + token=token.GalaxyToken(token='key')) + return galaxy_api + + +@pytest.fixture() +def manifest_template(): + def get_manifest_info(namespace='ansible_namespace', name='collection', version='0.1.0'): + return { + "collection_info": { + "namespace": namespace, + "name": name, + "version": version, + "authors": [ + "shertel" + ], + "readme": "README.md", + "tags": [ + "test", + "collection" + ], + "description": "Test", + "license": [ + "MIT" + ], + "license_file": None, + "dependencies": {}, + "repository": "https://github.com/{0}/{1}".format(namespace, name), + "documentation": None, + "homepage": None, + "issues": None + }, + "file_manifest_file": { + "name": "FILES.json", + "ftype": "file", + "chksum_type": "sha256", + "chksum_sha256": "files_manifest_checksum", + "format": 1 + }, + "format": 1 + } + + return get_manifest_info + + +@pytest.fixture() +def manifest_info(manifest_template): + return manifest_template() + + +@pytest.fixture() +def files_manifest_info(): + return { + "files": [ + { + "name": ".", + "ftype": "dir", + "chksum_type": None, + "chksum_sha256": None, + "format": 1 + }, + { + "name": "README.md", + "ftype": "file", + "chksum_type": "sha256", + "chksum_sha256": "individual_file_checksum", + "format": 1 + } + ], + "format": 1} + + +@pytest.fixture() +def manifest(manifest_info): + b_data = to_bytes(json.dumps(manifest_info)) + + with patch.object(builtins, 'open', mock_open(read_data=b_data)) as m: + with open('MANIFEST.json', mode='rb') as fake_file: + yield fake_file, sha256(b_data).hexdigest() + + +@pytest.mark.parametrize( + 'required_signature_count,valid', + [ + ("1", True), + ("+1", True), + ("all", True), + ("+all", True), + ("-1", False), + ("invalid", False), + ("1.5", False), + ("+", False), + ] +) +def test_cli_options(required_signature_count, valid, monkeypatch): + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + '--keyring', + '~/.ansible/pubring.kbx', + '--required-valid-signature-count', + required_signature_count + ] + + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + + if valid: + galaxy_cli.run() + else: + with pytest.raises(SystemExit, match='2') as error: + galaxy_cli.run() + + +@pytest.mark.parametrize( + "config,server", + [ + ( + # Options to create ini config + { + 'url': 'https://galaxy.ansible.com', + 'validate_certs': 'False', + 'v3': 'False', + }, + # Expected server attributes + { + 'validate_certs': False, + '_available_api_versions': {}, + }, + ), + ( + { + 'url': 'https://galaxy.ansible.com', + 'validate_certs': 'True', + 'v3': 'True', + }, + { + 'validate_certs': True, + '_available_api_versions': {'v3': '/v3'}, + }, + ), + ], +) +def test_bool_type_server_config_options(config, server, monkeypatch): + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + ] + + config_lines = [ + "[galaxy]", + "server_list=server1\n", + "[galaxy_server.server1]", + "url=%s" % config['url'], + "v3=%s" % config['v3'], + "validate_certs=%s\n" % config['validate_certs'], + ] + + with tempfile.NamedTemporaryFile(suffix='.cfg') as tmp_file: + tmp_file.write( + to_bytes('\n'.join(config_lines)) + ) + tmp_file.flush() + + with patch.object(C, 'GALAXY_SERVER_LIST', ['server1']): + with patch.object(C.config, '_config_file', tmp_file.name): + C.config._parse_config_file() + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + galaxy_cli.run() + + assert galaxy_cli.api_servers[0].name == 'server1' + assert galaxy_cli.api_servers[0].validate_certs == server['validate_certs'] + assert galaxy_cli.api_servers[0]._available_api_versions == server['_available_api_versions'] + + +@pytest.mark.parametrize('global_ignore_certs', [True, False]) +def test_validate_certs(global_ignore_certs, monkeypatch): + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + ] + if global_ignore_certs: + cli_args.append('--ignore-certs') + + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + galaxy_cli.run() + + assert len(galaxy_cli.api_servers) == 1 + assert galaxy_cli.api_servers[0].validate_certs is not global_ignore_certs + + +@pytest.mark.parametrize( + ["ignore_certs_cli", "ignore_certs_cfg", "expected_validate_certs"], + [ + (None, None, True), + (None, True, False), + (None, False, True), + (True, None, False), + (True, True, False), + (True, False, False), + ] +) +def test_validate_certs_with_server_url(ignore_certs_cli, ignore_certs_cfg, expected_validate_certs, monkeypatch): + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + '-s', + 'https://galaxy.ansible.com' + ] + if ignore_certs_cli: + cli_args.append('--ignore-certs') + if ignore_certs_cfg is not None: + monkeypatch.setattr(C, 'GALAXY_IGNORE_CERTS', ignore_certs_cfg) + + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + galaxy_cli.run() + + assert len(galaxy_cli.api_servers) == 1 + assert galaxy_cli.api_servers[0].validate_certs == expected_validate_certs + + +@pytest.mark.parametrize( + ["ignore_certs_cli", "ignore_certs_cfg", "expected_server2_validate_certs", "expected_server3_validate_certs"], + [ + (None, None, True, True), + (None, True, True, False), + (None, False, True, True), + (True, None, False, False), + (True, True, False, False), + (True, False, False, False), + ] +) +def test_validate_certs_server_config(ignore_certs_cfg, ignore_certs_cli, expected_server2_validate_certs, expected_server3_validate_certs, monkeypatch): + server_names = ['server1', 'server2', 'server3'] + cfg_lines = [ + "[galaxy]", + "server_list=server1,server2,server3", + "[galaxy_server.server1]", + "url=https://galaxy.ansible.com/api/", + "validate_certs=False", + "[galaxy_server.server2]", + "url=https://galaxy.ansible.com/api/", + "validate_certs=True", + "[galaxy_server.server3]", + "url=https://galaxy.ansible.com/api/", + ] + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + ] + if ignore_certs_cli: + cli_args.append('--ignore-certs') + if ignore_certs_cfg is not None: + monkeypatch.setattr(C, 'GALAXY_IGNORE_CERTS', ignore_certs_cfg) + + monkeypatch.setattr(C, 'GALAXY_SERVER_LIST', server_names) + + with tempfile.NamedTemporaryFile(suffix='.cfg') as tmp_file: + tmp_file.write(to_bytes('\n'.join(cfg_lines), errors='surrogate_or_strict')) + tmp_file.flush() + + monkeypatch.setattr(C.config, '_config_file', tmp_file.name) + C.config._parse_config_file() + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + galaxy_cli.run() + + # (not) --ignore-certs > server's validate_certs > (not) GALAXY_IGNORE_CERTS > True + assert galaxy_cli.api_servers[0].validate_certs is False + assert galaxy_cli.api_servers[1].validate_certs is expected_server2_validate_certs + assert galaxy_cli.api_servers[2].validate_certs is expected_server3_validate_certs + + +def test_build_collection_no_galaxy_yaml(): + fake_path = u'/fake/ÅÑŚÌβŁÈ/path' + expected = to_native("The collection galaxy.yml path '%s/galaxy.yml' does not exist." % fake_path) + + with pytest.raises(AnsibleError, match=expected): + collection.build_collection(fake_path, u'output', False) + + +def test_build_existing_output_file(collection_input): + input_dir, output_dir = collection_input + + existing_output_dir = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz') + os.makedirs(existing_output_dir) + + expected = "The output collection artifact '%s' already exists, but is a directory - aborting" \ + % to_native(existing_output_dir) + with pytest.raises(AnsibleError, match=expected): + collection.build_collection(to_text(input_dir, errors='surrogate_or_strict'), to_text(output_dir, errors='surrogate_or_strict'), False) + + +def test_build_existing_output_without_force(collection_input): + input_dir, output_dir = collection_input + + existing_output = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz') + with open(existing_output, 'w+') as out_file: + out_file.write("random garbage") + out_file.flush() + + expected = "The file '%s' already exists. You can use --force to re-create the collection artifact." \ + % to_native(existing_output) + with pytest.raises(AnsibleError, match=expected): + collection.build_collection(to_text(input_dir, errors='surrogate_or_strict'), to_text(output_dir, errors='surrogate_or_strict'), False) + + +def test_build_existing_output_with_force(collection_input): + input_dir, output_dir = collection_input + + existing_output = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz') + with open(existing_output, 'w+') as out_file: + out_file.write("random garbage") + out_file.flush() + + collection.build_collection(to_text(input_dir, errors='surrogate_or_strict'), to_text(output_dir, errors='surrogate_or_strict'), True) + + # Verify the file was replaced with an actual tar file + assert tarfile.is_tarfile(existing_output) + + +def test_build_with_existing_files_and_manifest(collection_input): + input_dir, output_dir = collection_input + + with open(os.path.join(input_dir, 'MANIFEST.json'), "wb") as fd: + fd.write(b'{"collection_info": {"version": "6.6.6"}, "version": 1}') + + with open(os.path.join(input_dir, 'FILES.json'), "wb") as fd: + fd.write(b'{"files": [], "format": 1}') + + with open(os.path.join(input_dir, "plugins", "MANIFEST.json"), "wb") as fd: + fd.write(b"test data that should be in build") + + collection.build_collection(to_text(input_dir, errors='surrogate_or_strict'), to_text(output_dir, errors='surrogate_or_strict'), False) + + output_artifact = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz') + assert tarfile.is_tarfile(output_artifact) + + with tarfile.open(output_artifact, mode='r') as actual: + members = actual.getmembers() + + manifest_file = next(m for m in members if m.path == "MANIFEST.json") + manifest_file_obj = actual.extractfile(manifest_file.name) + manifest_file_text = manifest_file_obj.read() + manifest_file_obj.close() + assert manifest_file_text != b'{"collection_info": {"version": "6.6.6"}, "version": 1}' + + json_file = next(m for m in members if m.path == "MANIFEST.json") + json_file_obj = actual.extractfile(json_file.name) + json_file_text = json_file_obj.read() + json_file_obj.close() + assert json_file_text != b'{"files": [], "format": 1}' + + sub_manifest_file = next(m for m in members if m.path == "plugins/MANIFEST.json") + sub_manifest_file_obj = actual.extractfile(sub_manifest_file.name) + sub_manifest_file_text = sub_manifest_file_obj.read() + sub_manifest_file_obj.close() + assert sub_manifest_file_text == b"test data that should be in build" + + +@pytest.mark.parametrize('galaxy_yml_dir', [b'namespace: value: broken'], indirect=True) +def test_invalid_yaml_galaxy_file(galaxy_yml_dir): + galaxy_file = os.path.join(galaxy_yml_dir, b'galaxy.yml') + expected = to_native(b"Failed to parse the galaxy.yml at '%s' with the following error:" % galaxy_file) + + with pytest.raises(AnsibleError, match=expected): + collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + + +@pytest.mark.parametrize('galaxy_yml_dir', [b'namespace: test_namespace'], indirect=True) +def test_missing_required_galaxy_key(galaxy_yml_dir): + galaxy_file = os.path.join(galaxy_yml_dir, b'galaxy.yml') + expected = "The collection galaxy.yml at '%s' is missing the following mandatory keys: authors, name, " \ + "readme, version" % to_native(galaxy_file) + + with pytest.raises(AnsibleError, match=expected): + collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + + +@pytest.mark.parametrize('galaxy_yml_dir', [b'namespace: test_namespace'], indirect=True) +def test_galaxy_yaml_no_mandatory_keys(galaxy_yml_dir): + expected = "The collection galaxy.yml at '%s/galaxy.yml' is missing the " \ + "following mandatory keys: authors, name, readme, version" % to_native(galaxy_yml_dir) + + with pytest.raises(ValueError, match=expected): + assert collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir, require_build_metadata=False) == expected + + +@pytest.mark.parametrize('galaxy_yml_dir', [b'My life story is so very interesting'], indirect=True) +def test_galaxy_yaml_no_mandatory_keys_bad_yaml(galaxy_yml_dir): + expected = "The collection galaxy.yml at '%s/galaxy.yml' is incorrectly formatted." % to_native(galaxy_yml_dir) + + with pytest.raises(AnsibleError, match=expected): + collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + + +@pytest.mark.parametrize('galaxy_yml_dir', [b""" +namespace: namespace +name: collection +authors: Jordan +version: 0.1.0 +readme: README.md +invalid: value"""], indirect=True) +def test_warning_extra_keys(galaxy_yml_dir, monkeypatch): + display_mock = MagicMock() + monkeypatch.setattr(Display, 'warning', display_mock) + + collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + + assert display_mock.call_count == 1 + assert display_mock.call_args[0][0] == "Found unknown keys in collection galaxy.yml at '%s/galaxy.yml': invalid"\ + % to_text(galaxy_yml_dir) + + +@pytest.mark.parametrize('galaxy_yml_dir', [b""" +namespace: namespace +name: collection +authors: Jordan +version: 0.1.0 +readme: README.md"""], indirect=True) +def test_defaults_galaxy_yml(galaxy_yml_dir): + actual = collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + + assert actual['namespace'] == 'namespace' + assert actual['name'] == 'collection' + assert actual['authors'] == ['Jordan'] + assert actual['version'] == '0.1.0' + assert actual['readme'] == 'README.md' + assert actual['description'] is None + assert actual['repository'] is None + assert actual['documentation'] is None + assert actual['homepage'] is None + assert actual['issues'] is None + assert actual['tags'] == [] + assert actual['dependencies'] == {} + assert actual['license'] == [] + + +@pytest.mark.parametrize('galaxy_yml_dir', [(b""" +namespace: namespace +name: collection +authors: Jordan +version: 0.1.0 +readme: README.md +license: MIT"""), (b""" +namespace: namespace +name: collection +authors: Jordan +version: 0.1.0 +readme: README.md +license: +- MIT""")], indirect=True) +def test_galaxy_yml_list_value(galaxy_yml_dir): + actual = collection.concrete_artifact_manager._get_meta_from_src_dir(galaxy_yml_dir) + assert actual['license'] == ['MIT'] + + +def test_build_ignore_files_and_folders(collection_input, monkeypatch): + input_dir = collection_input[0] + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_display) + + git_folder = os.path.join(input_dir, '.git') + retry_file = os.path.join(input_dir, 'ansible.retry') + + tests_folder = os.path.join(input_dir, 'tests', 'output') + tests_output_file = os.path.join(tests_folder, 'result.txt') + + os.makedirs(git_folder) + os.makedirs(tests_folder) + + with open(retry_file, 'w+') as ignore_file: + ignore_file.write('random') + ignore_file.flush() + + with open(tests_output_file, 'w+') as tests_file: + tests_file.write('random') + tests_file.flush() + + actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection', [], Sentinel) + + assert actual['format'] == 1 + for manifest_entry in actual['files']: + assert manifest_entry['name'] not in ['.git', 'ansible.retry', 'galaxy.yml', 'tests/output', 'tests/output/result.txt'] + + expected_msgs = [ + "Skipping '%s/galaxy.yml' for collection build" % to_text(input_dir), + "Skipping '%s' for collection build" % to_text(retry_file), + "Skipping '%s' for collection build" % to_text(git_folder), + "Skipping '%s' for collection build" % to_text(tests_folder), + ] + assert mock_display.call_count == 4 + assert mock_display.mock_calls[0][1][0] in expected_msgs + assert mock_display.mock_calls[1][1][0] in expected_msgs + assert mock_display.mock_calls[2][1][0] in expected_msgs + assert mock_display.mock_calls[3][1][0] in expected_msgs + + +def test_build_ignore_older_release_in_root(collection_input, monkeypatch): + input_dir = collection_input[0] + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_display) + + # This is expected to be ignored because it is in the root collection dir. + release_file = os.path.join(input_dir, 'namespace-collection-0.0.0.tar.gz') + + # This is not expected to be ignored because it is not in the root collection dir. + fake_release_file = os.path.join(input_dir, 'plugins', 'namespace-collection-0.0.0.tar.gz') + + for filename in [release_file, fake_release_file]: + with open(filename, 'w+') as file_obj: + file_obj.write('random') + file_obj.flush() + + actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection', [], Sentinel) + assert actual['format'] == 1 + + plugin_release_found = False + for manifest_entry in actual['files']: + assert manifest_entry['name'] != 'namespace-collection-0.0.0.tar.gz' + if manifest_entry['name'] == 'plugins/namespace-collection-0.0.0.tar.gz': + plugin_release_found = True + + assert plugin_release_found + + expected_msgs = [ + "Skipping '%s/galaxy.yml' for collection build" % to_text(input_dir), + "Skipping '%s' for collection build" % to_text(release_file) + ] + assert mock_display.call_count == 2 + assert mock_display.mock_calls[0][1][0] in expected_msgs + assert mock_display.mock_calls[1][1][0] in expected_msgs + + +def test_build_ignore_patterns(collection_input, monkeypatch): + input_dir = collection_input[0] + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_display) + + actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection', + ['*.md', 'plugins/action', 'playbooks/*.j2'], + Sentinel) + assert actual['format'] == 1 + + expected_missing = [ + 'README.md', + 'docs/My Collection.md', + 'plugins/action', + 'playbooks/templates/test.conf.j2', + 'playbooks/templates/subfolder/test.conf.j2', + ] + + # Files or dirs that are close to a match but are not, make sure they are present + expected_present = [ + 'docs', + 'roles/common/templates/test.conf.j2', + 'roles/common/templates/subfolder/test.conf.j2', + ] + + actual_files = [e['name'] for e in actual['files']] + for m in expected_missing: + assert m not in actual_files + + for p in expected_present: + assert p in actual_files + + expected_msgs = [ + "Skipping '%s/galaxy.yml' for collection build" % to_text(input_dir), + "Skipping '%s/README.md' for collection build" % to_text(input_dir), + "Skipping '%s/docs/My Collection.md' for collection build" % to_text(input_dir), + "Skipping '%s/plugins/action' for collection build" % to_text(input_dir), + "Skipping '%s/playbooks/templates/test.conf.j2' for collection build" % to_text(input_dir), + "Skipping '%s/playbooks/templates/subfolder/test.conf.j2' for collection build" % to_text(input_dir), + ] + assert mock_display.call_count == len(expected_msgs) + assert mock_display.mock_calls[0][1][0] in expected_msgs + assert mock_display.mock_calls[1][1][0] in expected_msgs + assert mock_display.mock_calls[2][1][0] in expected_msgs + assert mock_display.mock_calls[3][1][0] in expected_msgs + assert mock_display.mock_calls[4][1][0] in expected_msgs + assert mock_display.mock_calls[5][1][0] in expected_msgs + + +def test_build_ignore_symlink_target_outside_collection(collection_input, monkeypatch): + input_dir, outside_dir = collection_input + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_display) + + link_path = os.path.join(input_dir, 'plugins', 'connection') + os.symlink(outside_dir, link_path) + + actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection', [], Sentinel) + for manifest_entry in actual['files']: + assert manifest_entry['name'] != 'plugins/connection' + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == "Skipping '%s' as it is a symbolic link to a directory outside " \ + "the collection" % to_text(link_path) + + +def test_build_copy_symlink_target_inside_collection(collection_input): + input_dir = collection_input[0] + + os.makedirs(os.path.join(input_dir, 'playbooks', 'roles')) + roles_link = os.path.join(input_dir, 'playbooks', 'roles', 'linked') + + roles_target = os.path.join(input_dir, 'roles', 'linked') + roles_target_tasks = os.path.join(roles_target, 'tasks') + os.makedirs(roles_target_tasks) + with open(os.path.join(roles_target_tasks, 'main.yml'), 'w+') as tasks_main: + tasks_main.write("---\n- hosts: localhost\n tasks:\n - ping:") + tasks_main.flush() + + os.symlink(roles_target, roles_link) + + actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection', [], Sentinel) + + linked_entries = [e for e in actual['files'] if e['name'].startswith('playbooks/roles/linked')] + assert len(linked_entries) == 1 + assert linked_entries[0]['name'] == 'playbooks/roles/linked' + assert linked_entries[0]['ftype'] == 'dir' + + +def test_build_with_symlink_inside_collection(collection_input): + input_dir, output_dir = collection_input + + os.makedirs(os.path.join(input_dir, 'playbooks', 'roles')) + roles_link = os.path.join(input_dir, 'playbooks', 'roles', 'linked') + file_link = os.path.join(input_dir, 'docs', 'README.md') + + roles_target = os.path.join(input_dir, 'roles', 'linked') + roles_target_tasks = os.path.join(roles_target, 'tasks') + os.makedirs(roles_target_tasks) + with open(os.path.join(roles_target_tasks, 'main.yml'), 'w+') as tasks_main: + tasks_main.write("---\n- hosts: localhost\n tasks:\n - ping:") + tasks_main.flush() + + os.symlink(roles_target, roles_link) + os.symlink(os.path.join(input_dir, 'README.md'), file_link) + + collection.build_collection(to_text(input_dir, errors='surrogate_or_strict'), to_text(output_dir, errors='surrogate_or_strict'), False) + + output_artifact = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz') + assert tarfile.is_tarfile(output_artifact) + + with tarfile.open(output_artifact, mode='r') as actual: + members = actual.getmembers() + + linked_folder = next(m for m in members if m.path == 'playbooks/roles/linked') + assert linked_folder.type == tarfile.SYMTYPE + assert linked_folder.linkname == '../../roles/linked' + + linked_file = next(m for m in members if m.path == 'docs/README.md') + assert linked_file.type == tarfile.SYMTYPE + assert linked_file.linkname == '../README.md' + + linked_file_obj = actual.extractfile(linked_file.name) + actual_file = secure_hash_s(linked_file_obj.read()) + linked_file_obj.close() + + assert actual_file == '63444bfc766154e1bc7557ef6280de20d03fcd81' + + +def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch): + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + artifact_path, mock_open = collection_artifact + fake_import_uri = 'https://galaxy.server.com/api/v2/import/1234' + + mock_publish = MagicMock() + mock_publish.return_value = fake_import_uri + monkeypatch.setattr(galaxy_server, 'publish_collection', mock_publish) + + collection.publish_collection(artifact_path, galaxy_server, False, 0) + + assert mock_publish.call_count == 1 + assert mock_publish.mock_calls[0][1][0] == artifact_path + + assert mock_display.call_count == 1 + assert mock_display.mock_calls[0][1][0] == \ + "Collection has been pushed to the Galaxy server %s %s, not waiting until import has completed due to " \ + "--no-wait being set. Import task results can be found at %s" % (galaxy_server.name, galaxy_server.api_server, + fake_import_uri) + + +def test_publish_with_wait(galaxy_server, collection_artifact, monkeypatch): + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + artifact_path, mock_open = collection_artifact + fake_import_uri = 'https://galaxy.server.com/api/v2/import/1234' + + mock_publish = MagicMock() + mock_publish.return_value = fake_import_uri + monkeypatch.setattr(galaxy_server, 'publish_collection', mock_publish) + + mock_wait = MagicMock() + monkeypatch.setattr(galaxy_server, 'wait_import_task', mock_wait) + + collection.publish_collection(artifact_path, galaxy_server, True, 0) + + assert mock_publish.call_count == 1 + assert mock_publish.mock_calls[0][1][0] == artifact_path + + assert mock_wait.call_count == 1 + assert mock_wait.mock_calls[0][1][0] == '1234' + + assert mock_display.mock_calls[0][1][0] == "Collection has been published to the Galaxy server test_server %s" \ + % galaxy_server.api_server + + +def test_find_existing_collections(tmp_path_factory, monkeypatch): + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + collection1 = os.path.join(test_dir, 'namespace1', 'collection1') + collection2 = os.path.join(test_dir, 'namespace2', 'collection2') + fake_collection1 = os.path.join(test_dir, 'namespace3', 'collection3') + fake_collection2 = os.path.join(test_dir, 'namespace4') + os.makedirs(collection1) + os.makedirs(collection2) + os.makedirs(os.path.split(fake_collection1)[0]) + + open(fake_collection1, 'wb+').close() + open(fake_collection2, 'wb+').close() + + collection1_manifest = json.dumps({ + 'collection_info': { + 'namespace': 'namespace1', + 'name': 'collection1', + 'version': '1.2.3', + 'authors': ['Jordan Borean'], + 'readme': 'README.md', + 'dependencies': {}, + }, + 'format': 1, + }) + with open(os.path.join(collection1, 'MANIFEST.json'), 'wb') as manifest_obj: + manifest_obj.write(to_bytes(collection1_manifest)) + + mock_warning = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_warning) + + actual = list(collection.find_existing_collections(test_dir, artifacts_manager=concrete_artifact_cm)) + + assert len(actual) == 2 + for actual_collection in actual: + if '%s.%s' % (actual_collection.namespace, actual_collection.name) == 'namespace1.collection1': + assert actual_collection.namespace == 'namespace1' + assert actual_collection.name == 'collection1' + assert actual_collection.ver == '1.2.3' + assert to_text(actual_collection.src) == collection1 + else: + assert actual_collection.namespace == 'namespace2' + assert actual_collection.name == 'collection2' + assert actual_collection.ver == '*' + assert to_text(actual_collection.src) == collection2 + + assert mock_warning.call_count == 1 + assert mock_warning.mock_calls[0][1][0] == "Collection at '%s' does not have a MANIFEST.json file, nor has it galaxy.yml: " \ + "cannot detect version." % to_text(collection2) + + +def test_download_file(tmp_path_factory, monkeypatch): + temp_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + + data = b"\x00\x01\x02\x03" + sha256_hash = sha256() + sha256_hash.update(data) + + mock_open = MagicMock() + mock_open.return_value = BytesIO(data) + monkeypatch.setattr(collection.concrete_artifact_manager, 'open_url', mock_open) + + expected = temp_dir + actual = collection._download_file('http://google.com/file', temp_dir, sha256_hash.hexdigest(), True) + + assert actual.startswith(expected) + assert os.path.isfile(actual) + with open(actual, 'rb') as file_obj: + assert file_obj.read() == data + + assert mock_open.call_count == 1 + assert mock_open.mock_calls[0][1][0] == 'http://google.com/file' + + +def test_download_file_hash_mismatch(tmp_path_factory, monkeypatch): + temp_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + + data = b"\x00\x01\x02\x03" + + mock_open = MagicMock() + mock_open.return_value = BytesIO(data) + monkeypatch.setattr(collection.concrete_artifact_manager, 'open_url', mock_open) + + expected = "Mismatch artifact hash with downloaded file" + with pytest.raises(AnsibleError, match=expected): + collection._download_file('http://google.com/file', temp_dir, 'bad', True) + + +def test_extract_tar_file_invalid_hash(tmp_tarfile): + temp_dir, tfile, filename, dummy = tmp_tarfile + + expected = "Checksum mismatch for '%s' inside collection at '%s'" % (to_native(filename), to_native(tfile.name)) + with pytest.raises(AnsibleError, match=expected): + collection._extract_tar_file(tfile, filename, temp_dir, temp_dir, "fakehash") + + +def test_extract_tar_file_missing_member(tmp_tarfile): + temp_dir, tfile, dummy, dummy = tmp_tarfile + + expected = "Collection tar at '%s' does not contain the expected file 'missing'." % to_native(tfile.name) + with pytest.raises(AnsibleError, match=expected): + collection._extract_tar_file(tfile, 'missing', temp_dir, temp_dir) + + +def test_extract_tar_file_missing_parent_dir(tmp_tarfile): + temp_dir, tfile, filename, checksum = tmp_tarfile + output_dir = os.path.join(temp_dir, b'output') + output_file = os.path.join(output_dir, to_bytes(filename)) + + collection._extract_tar_file(tfile, filename, output_dir, temp_dir, checksum) + os.path.isfile(output_file) + + +def test_extract_tar_file_outside_dir(tmp_path_factory): + filename = u'ÅÑŚÌβŁÈ' + temp_dir = to_bytes(tmp_path_factory.mktemp('test-%s Collections' % to_native(filename))) + tar_file = os.path.join(temp_dir, to_bytes('%s.tar.gz' % filename)) + data = os.urandom(8) + + tar_filename = '../%s.sh' % filename + with tarfile.open(tar_file, 'w:gz') as tfile: + b_io = BytesIO(data) + tar_info = tarfile.TarInfo(tar_filename) + tar_info.size = len(data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + expected = re.escape("Cannot extract tar entry '%s' as it will be placed outside the collection directory" + % to_native(tar_filename)) + with tarfile.open(tar_file, 'r') as tfile: + with pytest.raises(AnsibleError, match=expected): + collection._extract_tar_file(tfile, tar_filename, os.path.join(temp_dir, to_bytes(filename)), temp_dir) + + +def test_require_one_of_collections_requirements_with_both(): + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'verify', 'namespace.collection', '-r', 'requirements.yml']) + + with pytest.raises(AnsibleError) as req_err: + cli._require_one_of_collections_requirements(('namespace.collection',), 'requirements.yml') + + with pytest.raises(AnsibleError) as cli_err: + cli.run() + + assert req_err.value.message == cli_err.value.message == 'The positional collection_name arg and --requirements-file are mutually exclusive.' + + +def test_require_one_of_collections_requirements_with_neither(): + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'verify']) + + with pytest.raises(AnsibleError) as req_err: + cli._require_one_of_collections_requirements((), '') + + with pytest.raises(AnsibleError) as cli_err: + cli.run() + + assert req_err.value.message == cli_err.value.message == 'You must specify a collection name or a requirements file.' + + +def test_require_one_of_collections_requirements_with_collections(): + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'verify', 'namespace1.collection1', 'namespace2.collection1:1.0.0']) + collections = ('namespace1.collection1', 'namespace2.collection1:1.0.0',) + + requirements = cli._require_one_of_collections_requirements(collections, '')['collections'] + + req_tuples = [('%s.%s' % (req.namespace, req.name), req.ver, req.src, req.type,) for req in requirements] + assert req_tuples == [('namespace1.collection1', '*', None, 'galaxy'), ('namespace2.collection1', '1.0.0', None, 'galaxy')] + + +@patch('ansible.cli.galaxy.GalaxyCLI._parse_requirements_file') +def test_require_one_of_collections_requirements_with_requirements(mock_parse_requirements_file, galaxy_server): + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'verify', '-r', 'requirements.yml', 'namespace.collection']) + mock_parse_requirements_file.return_value = {'collections': [('namespace.collection', '1.0.5', galaxy_server)]} + requirements = cli._require_one_of_collections_requirements((), 'requirements.yml')['collections'] + + assert mock_parse_requirements_file.call_count == 1 + assert requirements == [('namespace.collection', '1.0.5', galaxy_server)] + + +@patch('ansible.cli.galaxy.GalaxyCLI.execute_verify', spec=True) +def test_call_GalaxyCLI(execute_verify): + galaxy_args = ['ansible-galaxy', 'collection', 'verify', 'namespace.collection'] + + GalaxyCLI(args=galaxy_args).run() + + assert execute_verify.call_count == 1 + + +@patch('ansible.cli.galaxy.GalaxyCLI.execute_verify') +def test_call_GalaxyCLI_with_implicit_role(execute_verify): + galaxy_args = ['ansible-galaxy', 'verify', 'namespace.implicit_role'] + + with pytest.raises(SystemExit): + GalaxyCLI(args=galaxy_args).run() + + assert not execute_verify.called + + +@patch('ansible.cli.galaxy.GalaxyCLI.execute_verify') +def test_call_GalaxyCLI_with_role(execute_verify): + galaxy_args = ['ansible-galaxy', 'role', 'verify', 'namespace.role'] + + with pytest.raises(SystemExit): + GalaxyCLI(args=galaxy_args).run() + + assert not execute_verify.called + + +@patch('ansible.cli.galaxy.verify_collections', spec=True) +def test_execute_verify_with_defaults(mock_verify_collections): + galaxy_args = ['ansible-galaxy', 'collection', 'verify', 'namespace.collection:1.0.4'] + GalaxyCLI(args=galaxy_args).run() + + assert mock_verify_collections.call_count == 1 + + print("Call args {0}".format(mock_verify_collections.call_args[0])) + requirements, search_paths, galaxy_apis, ignore_errors = mock_verify_collections.call_args[0] + + assert [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type) for r in requirements] == [('namespace.collection', '1.0.4', None, 'galaxy')] + for install_path in search_paths: + assert install_path.endswith('ansible_collections') + assert galaxy_apis[0].api_server == 'https://galaxy.ansible.com' + assert ignore_errors is False + + +@patch('ansible.cli.galaxy.verify_collections', spec=True) +def test_execute_verify(mock_verify_collections): + GalaxyCLI(args=[ + 'ansible-galaxy', 'collection', 'verify', 'namespace.collection:1.0.4', '--ignore-certs', + '-p', '~/.ansible', '--ignore-errors', '--server', 'http://galaxy-dev.com', + ]).run() + + assert mock_verify_collections.call_count == 1 + + requirements, search_paths, galaxy_apis, ignore_errors = mock_verify_collections.call_args[0] + + assert [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type) for r in requirements] == [('namespace.collection', '1.0.4', None, 'galaxy')] + for install_path in search_paths: + assert install_path.endswith('ansible_collections') + assert galaxy_apis[0].api_server == 'http://galaxy-dev.com' + assert ignore_errors is True + + +def test_verify_file_hash_deleted_file(manifest_info): + data = to_bytes(json.dumps(manifest_info)) + digest = sha256(data).hexdigest() + + namespace = manifest_info['collection_info']['namespace'] + name = manifest_info['collection_info']['name'] + version = manifest_info['collection_info']['version'] + server = 'http://galaxy.ansible.com' + + error_queue = [] + + with patch.object(builtins, 'open', mock_open(read_data=data)) as m: + with patch.object(collection.os.path, 'isfile', MagicMock(return_value=False)) as mock_isfile: + collection._verify_file_hash(b'path/', 'file', digest, error_queue) + + assert mock_isfile.called_once + + assert len(error_queue) == 1 + assert error_queue[0].installed is None + assert error_queue[0].expected == digest + + +def test_verify_file_hash_matching_hash(manifest_info): + + data = to_bytes(json.dumps(manifest_info)) + digest = sha256(data).hexdigest() + + namespace = manifest_info['collection_info']['namespace'] + name = manifest_info['collection_info']['name'] + version = manifest_info['collection_info']['version'] + server = 'http://galaxy.ansible.com' + + error_queue = [] + + with patch.object(builtins, 'open', mock_open(read_data=data)) as m: + with patch.object(collection.os.path, 'isfile', MagicMock(return_value=True)) as mock_isfile: + collection._verify_file_hash(b'path/', 'file', digest, error_queue) + + assert mock_isfile.called_once + + assert error_queue == [] + + +def test_verify_file_hash_mismatching_hash(manifest_info): + + data = to_bytes(json.dumps(manifest_info)) + digest = sha256(data).hexdigest() + different_digest = 'not_{0}'.format(digest) + + namespace = manifest_info['collection_info']['namespace'] + name = manifest_info['collection_info']['name'] + version = manifest_info['collection_info']['version'] + server = 'http://galaxy.ansible.com' + + error_queue = [] + + with patch.object(builtins, 'open', mock_open(read_data=data)) as m: + with patch.object(collection.os.path, 'isfile', MagicMock(return_value=True)) as mock_isfile: + collection._verify_file_hash(b'path/', 'file', different_digest, error_queue) + + assert mock_isfile.called_once + + assert len(error_queue) == 1 + assert error_queue[0].installed == digest + assert error_queue[0].expected == different_digest + + +def test_consume_file(manifest): + + manifest_file, checksum = manifest + assert checksum == collection._consume_file(manifest_file) + + +def test_consume_file_and_write_contents(manifest, manifest_info): + + manifest_file, checksum = manifest + + write_to = BytesIO() + actual_hash = collection._consume_file(manifest_file, write_to) + + write_to.seek(0) + assert to_bytes(json.dumps(manifest_info)) == write_to.read() + assert actual_hash == checksum + + +def test_get_tar_file_member(tmp_tarfile): + + temp_dir, tfile, filename, checksum = tmp_tarfile + + with collection._get_tar_file_member(tfile, filename) as (tar_file_member, tar_file_obj): + assert isinstance(tar_file_member, tarfile.TarInfo) + assert isinstance(tar_file_obj, tarfile.ExFileObject) + + +def test_get_nonexistent_tar_file_member(tmp_tarfile): + temp_dir, tfile, filename, checksum = tmp_tarfile + + file_does_not_exist = filename + 'nonexistent' + + with pytest.raises(AnsibleError) as err: + collection._get_tar_file_member(tfile, file_does_not_exist) + + assert to_text(err.value.message) == "Collection tar at '%s' does not contain the expected file '%s'." % (to_text(tfile.name), file_does_not_exist) + + +def test_get_tar_file_hash(tmp_tarfile): + temp_dir, tfile, filename, checksum = tmp_tarfile + + assert checksum == collection._get_tar_file_hash(tfile.name, filename) + + +def test_get_json_from_tar_file(tmp_tarfile): + temp_dir, tfile, filename, checksum = tmp_tarfile + + assert 'MANIFEST.json' in tfile.getnames() + + data = collection._get_json_from_tar_file(tfile.name, 'MANIFEST.json') + + assert isinstance(data, dict) diff --git a/test/units/galaxy/test_collection_install.py b/test/units/galaxy/test_collection_install.py new file mode 100644 index 0000000..2118f0e --- /dev/null +++ b/test/units/galaxy/test_collection_install.py @@ -0,0 +1,1081 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import copy +import json +import os +import pytest +import re +import shutil +import stat +import tarfile +import yaml + +from io import BytesIO, StringIO +from unittest.mock import MagicMock, patch +from unittest import mock + +import ansible.module_utils.six.moves.urllib.error as urllib_error + +from ansible import context +from ansible.cli.galaxy import GalaxyCLI +from ansible.errors import AnsibleError +from ansible.galaxy import collection, api, dependency_resolution +from ansible.galaxy.dependency_resolution.dataclasses import Candidate, Requirement +from ansible.module_utils._text import to_bytes, to_native, to_text +from ansible.module_utils.common.process import get_bin_path +from ansible.utils import context_objects as co +from ansible.utils.display import Display + + +class RequirementCandidates(): + def __init__(self): + self.candidates = [] + + def func_wrapper(self, func): + def run(*args, **kwargs): + self.candidates = func(*args, **kwargs) + return self.candidates + return run + + +def call_galaxy_cli(args): + orig = co.GlobalCLIArgs._Singleton__instance + co.GlobalCLIArgs._Singleton__instance = None + try: + GalaxyCLI(args=['ansible-galaxy', 'collection'] + args).run() + finally: + co.GlobalCLIArgs._Singleton__instance = orig + + +def artifact_json(namespace, name, version, dependencies, server): + json_str = json.dumps({ + 'artifact': { + 'filename': '%s-%s-%s.tar.gz' % (namespace, name, version), + 'sha256': '2d76f3b8c4bab1072848107fb3914c345f71a12a1722f25c08f5d3f51f4ab5fd', + 'size': 1234, + }, + 'download_url': '%s/download/%s-%s-%s.tar.gz' % (server, namespace, name, version), + 'metadata': { + 'namespace': namespace, + 'name': name, + 'dependencies': dependencies, + }, + 'version': version + }) + return to_text(json_str) + + +def artifact_versions_json(namespace, name, versions, galaxy_api, available_api_versions=None): + results = [] + available_api_versions = available_api_versions or {} + api_version = 'v2' + if 'v3' in available_api_versions: + api_version = 'v3' + for version in versions: + results.append({ + 'href': '%s/api/%s/%s/%s/versions/%s/' % (galaxy_api.api_server, api_version, namespace, name, version), + 'version': version, + }) + + if api_version == 'v2': + json_str = json.dumps({ + 'count': len(versions), + 'next': None, + 'previous': None, + 'results': results + }) + + if api_version == 'v3': + response = {'meta': {'count': len(versions)}, + 'data': results, + 'links': {'first': None, + 'last': None, + 'next': None, + 'previous': None}, + } + json_str = json.dumps(response) + return to_text(json_str) + + +def error_json(galaxy_api, errors_to_return=None, available_api_versions=None): + errors_to_return = errors_to_return or [] + available_api_versions = available_api_versions or {} + + response = {} + + api_version = 'v2' + if 'v3' in available_api_versions: + api_version = 'v3' + + if api_version == 'v2': + assert len(errors_to_return) <= 1 + if errors_to_return: + response = errors_to_return[0] + + if api_version == 'v3': + response['errors'] = errors_to_return + + json_str = json.dumps(response) + return to_text(json_str) + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + yield + co.GlobalCLIArgs._Singleton__instance = None + + +@pytest.fixture() +def collection_artifact(request, tmp_path_factory): + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + namespace = 'ansible_namespace' + collection = 'collection' + + skeleton_path = os.path.join(os.path.dirname(os.path.split(__file__)[0]), 'cli', 'test_data', 'collection_skeleton') + collection_path = os.path.join(test_dir, namespace, collection) + + call_galaxy_cli(['init', '%s.%s' % (namespace, collection), '-c', '--init-path', test_dir, + '--collection-skeleton', skeleton_path]) + dependencies = getattr(request, 'param', {}) + + galaxy_yml = os.path.join(collection_path, 'galaxy.yml') + with open(galaxy_yml, 'rb+') as galaxy_obj: + existing_yaml = yaml.safe_load(galaxy_obj) + existing_yaml['dependencies'] = dependencies + + galaxy_obj.seek(0) + galaxy_obj.write(to_bytes(yaml.safe_dump(existing_yaml))) + galaxy_obj.truncate() + + # Create a file with +x in the collection so we can test the permissions + execute_path = os.path.join(collection_path, 'runme.sh') + with open(execute_path, mode='wb') as fd: + fd.write(b"echo hi") + os.chmod(execute_path, os.stat(execute_path).st_mode | stat.S_IEXEC) + + call_galaxy_cli(['build', collection_path, '--output-path', test_dir]) + + collection_tar = os.path.join(test_dir, '%s-%s-0.1.0.tar.gz' % (namespace, collection)) + return to_bytes(collection_path), to_bytes(collection_tar) + + +@pytest.fixture() +def galaxy_server(): + context.CLIARGS._store = {'ignore_certs': False} + galaxy_api = api.GalaxyAPI(None, 'test_server', 'https://galaxy.ansible.com') + galaxy_api.get_collection_signatures = MagicMock(return_value=[]) + return galaxy_api + + +def test_concrete_artifact_manager_scm_no_executable(monkeypatch): + url = 'https://github.com/org/repo' + version = 'commitish' + mock_subprocess_check_call = MagicMock() + monkeypatch.setattr(collection.concrete_artifact_manager.subprocess, 'check_call', mock_subprocess_check_call) + mock_mkdtemp = MagicMock(return_value='') + monkeypatch.setattr(collection.concrete_artifact_manager, 'mkdtemp', mock_mkdtemp) + mock_get_bin_path = MagicMock(side_effect=[ValueError('Failed to find required executable')]) + monkeypatch.setattr(collection.concrete_artifact_manager, 'get_bin_path', mock_get_bin_path) + + error = re.escape( + "Could not find git executable to extract the collection from the Git repository `https://github.com/org/repo`" + ) + with pytest.raises(AnsibleError, match=error): + collection.concrete_artifact_manager._extract_collection_from_git(url, version, b'path') + + +@pytest.mark.parametrize( + 'url,version,trailing_slash', + [ + ('https://github.com/org/repo', 'commitish', False), + ('https://github.com/org/repo,commitish', None, False), + ('https://github.com/org/repo/,commitish', None, True), + ('https://github.com/org/repo#,commitish', None, False), + ] +) +def test_concrete_artifact_manager_scm_cmd(url, version, trailing_slash, monkeypatch): + mock_subprocess_check_call = MagicMock() + monkeypatch.setattr(collection.concrete_artifact_manager.subprocess, 'check_call', mock_subprocess_check_call) + mock_mkdtemp = MagicMock(return_value='') + monkeypatch.setattr(collection.concrete_artifact_manager, 'mkdtemp', mock_mkdtemp) + + collection.concrete_artifact_manager._extract_collection_from_git(url, version, b'path') + + assert mock_subprocess_check_call.call_count == 2 + + repo = 'https://github.com/org/repo' + if trailing_slash: + repo += '/' + + git_executable = get_bin_path('git') + clone_cmd = (git_executable, 'clone', repo, '') + + assert mock_subprocess_check_call.call_args_list[0].args[0] == clone_cmd + assert mock_subprocess_check_call.call_args_list[1].args[0] == (git_executable, 'checkout', 'commitish') + + +@pytest.mark.parametrize( + 'url,version,trailing_slash', + [ + ('https://github.com/org/repo', 'HEAD', False), + ('https://github.com/org/repo,HEAD', None, False), + ('https://github.com/org/repo/,HEAD', None, True), + ('https://github.com/org/repo#,HEAD', None, False), + ('https://github.com/org/repo', None, False), + ] +) +def test_concrete_artifact_manager_scm_cmd_shallow(url, version, trailing_slash, monkeypatch): + mock_subprocess_check_call = MagicMock() + monkeypatch.setattr(collection.concrete_artifact_manager.subprocess, 'check_call', mock_subprocess_check_call) + mock_mkdtemp = MagicMock(return_value='') + monkeypatch.setattr(collection.concrete_artifact_manager, 'mkdtemp', mock_mkdtemp) + + collection.concrete_artifact_manager._extract_collection_from_git(url, version, b'path') + + assert mock_subprocess_check_call.call_count == 2 + + repo = 'https://github.com/org/repo' + if trailing_slash: + repo += '/' + git_executable = get_bin_path('git') + shallow_clone_cmd = (git_executable, 'clone', '--depth=1', repo, '') + + assert mock_subprocess_check_call.call_args_list[0].args[0] == shallow_clone_cmd + assert mock_subprocess_check_call.call_args_list[1].args[0] == (git_executable, 'checkout', 'HEAD') + + +def test_build_requirement_from_path(collection_artifact): + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + actual = Requirement.from_dir_path_as_unknown(collection_artifact[0], concrete_artifact_cm) + + assert actual.namespace == u'ansible_namespace' + assert actual.name == u'collection' + assert actual.src == collection_artifact[0] + assert actual.ver == u'0.1.0' + + +@pytest.mark.parametrize('version', ['1.1.1', '1.1.0', '1.0.0']) +def test_build_requirement_from_path_with_manifest(version, collection_artifact): + manifest_path = os.path.join(collection_artifact[0], b'MANIFEST.json') + manifest_value = json.dumps({ + 'collection_info': { + 'namespace': 'namespace', + 'name': 'name', + 'version': version, + 'dependencies': { + 'ansible_namespace.collection': '*' + } + } + }) + with open(manifest_path, 'wb') as manifest_obj: + manifest_obj.write(to_bytes(manifest_value)) + + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + actual = Requirement.from_dir_path_as_unknown(collection_artifact[0], concrete_artifact_cm) + + # While the folder name suggests a different collection, we treat MANIFEST.json as the source of truth. + assert actual.namespace == u'namespace' + assert actual.name == u'name' + assert actual.src == collection_artifact[0] + assert actual.ver == to_text(version) + + +def test_build_requirement_from_path_invalid_manifest(collection_artifact): + manifest_path = os.path.join(collection_artifact[0], b'MANIFEST.json') + with open(manifest_path, 'wb') as manifest_obj: + manifest_obj.write(b"not json") + + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + + expected = "Collection tar file member MANIFEST.json does not contain a valid json string." + with pytest.raises(AnsibleError, match=expected): + Requirement.from_dir_path_as_unknown(collection_artifact[0], concrete_artifact_cm) + + +def test_build_artifact_from_path_no_version(collection_artifact, monkeypatch): + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + # a collection artifact should always contain a valid version + manifest_path = os.path.join(collection_artifact[0], b'MANIFEST.json') + manifest_value = json.dumps({ + 'collection_info': { + 'namespace': 'namespace', + 'name': 'name', + 'version': '', + 'dependencies': {} + } + }) + with open(manifest_path, 'wb') as manifest_obj: + manifest_obj.write(to_bytes(manifest_value)) + + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + + expected = ( + '^Collection metadata file `.*` at `.*` is expected to have a valid SemVer ' + 'version value but got {empty_unicode_string!r}$'. + format(empty_unicode_string=u'') + ) + with pytest.raises(AnsibleError, match=expected): + Requirement.from_dir_path_as_unknown(collection_artifact[0], concrete_artifact_cm) + + +def test_build_requirement_from_path_no_version(collection_artifact, monkeypatch): + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + # version may be falsey/arbitrary strings for collections in development + manifest_path = os.path.join(collection_artifact[0], b'galaxy.yml') + metadata = { + 'authors': ['Ansible'], + 'readme': 'README.md', + 'namespace': 'namespace', + 'name': 'name', + 'version': '', + 'dependencies': {}, + } + with open(manifest_path, 'wb') as manifest_obj: + manifest_obj.write(to_bytes(yaml.safe_dump(metadata))) + + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + actual = Requirement.from_dir_path_as_unknown(collection_artifact[0], concrete_artifact_cm) + + # While the folder name suggests a different collection, we treat MANIFEST.json as the source of truth. + assert actual.namespace == u'namespace' + assert actual.name == u'name' + assert actual.src == collection_artifact[0] + assert actual.ver == u'*' + + +def test_build_requirement_from_tar(collection_artifact): + tmp_path = os.path.join(os.path.split(collection_artifact[1])[0], b'temp') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + + actual = Requirement.from_requirement_dict({'name': to_text(collection_artifact[1])}, concrete_artifact_cm) + + assert actual.namespace == u'ansible_namespace' + assert actual.name == u'collection' + assert actual.src == to_text(collection_artifact[1]) + assert actual.ver == u'0.1.0' + + +def test_build_requirement_from_tar_fail_not_tar(tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + test_file = os.path.join(test_dir, b'fake.tar.gz') + with open(test_file, 'wb') as test_obj: + test_obj.write(b"\x00\x01\x02\x03") + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + expected = "Collection artifact at '%s' is not a valid tar file." % to_native(test_file) + with pytest.raises(AnsibleError, match=expected): + Requirement.from_requirement_dict({'name': to_text(test_file)}, concrete_artifact_cm) + + +def test_build_requirement_from_tar_no_manifest(tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + + json_data = to_bytes(json.dumps( + { + 'files': [], + 'format': 1, + } + )) + + tar_path = os.path.join(test_dir, b'ansible-collections.tar.gz') + with tarfile.open(tar_path, 'w:gz') as tfile: + b_io = BytesIO(json_data) + tar_info = tarfile.TarInfo('FILES.json') + tar_info.size = len(json_data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + expected = "Collection at '%s' does not contain the required file MANIFEST.json." % to_native(tar_path) + with pytest.raises(AnsibleError, match=expected): + Requirement.from_requirement_dict({'name': to_text(tar_path)}, concrete_artifact_cm) + + +def test_build_requirement_from_tar_no_files(tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + + json_data = to_bytes(json.dumps( + { + 'collection_info': {}, + } + )) + + tar_path = os.path.join(test_dir, b'ansible-collections.tar.gz') + with tarfile.open(tar_path, 'w:gz') as tfile: + b_io = BytesIO(json_data) + tar_info = tarfile.TarInfo('MANIFEST.json') + tar_info.size = len(json_data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + with pytest.raises(KeyError, match='namespace'): + Requirement.from_requirement_dict({'name': to_text(tar_path)}, concrete_artifact_cm) + + +def test_build_requirement_from_tar_invalid_manifest(tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + + json_data = b"not a json" + + tar_path = os.path.join(test_dir, b'ansible-collections.tar.gz') + with tarfile.open(tar_path, 'w:gz') as tfile: + b_io = BytesIO(json_data) + tar_info = tarfile.TarInfo('MANIFEST.json') + tar_info.size = len(json_data) + tar_info.mode = 0o0644 + tfile.addfile(tarinfo=tar_info, fileobj=b_io) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + expected = "Collection tar file member MANIFEST.json does not contain a valid json string." + with pytest.raises(AnsibleError, match=expected): + Requirement.from_requirement_dict({'name': to_text(tar_path)}, concrete_artifact_cm) + + +def test_build_requirement_from_name(galaxy_server, monkeypatch, tmp_path_factory): + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['2.1.9', '2.1.10'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_version_metadata = MagicMock( + namespace='namespace', name='collection', + version='2.1.10', artifact_sha256='', dependencies={} + ) + monkeypatch.setattr(api.GalaxyAPI, 'get_collection_version_metadata', mock_version_metadata) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + collections = ['namespace.collection'] + requirements_file = None + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', collections[0]]) + requirements = cli._require_one_of_collections_requirements( + collections, requirements_file, artifacts_manager=concrete_artifact_cm + )['collections'] + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, True, False, False, False, False + )['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.ver == u'2.1.10' + assert actual.src == galaxy_server + + assert mock_get_versions.call_count == 1 + assert mock_get_versions.mock_calls[0][1] == ('namespace', 'collection') + + +def test_build_requirement_from_name_with_prerelease(galaxy_server, monkeypatch, tmp_path_factory): + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['1.0.1', '2.0.1-beta.1', '2.0.1'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.1', None, None, {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, True, False, False, False, False + )['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'2.0.1' + + assert mock_get_versions.call_count == 1 + assert mock_get_versions.mock_calls[0][1] == ('namespace', 'collection') + + +def test_build_requirment_from_name_with_prerelease_explicit(galaxy_server, monkeypatch, tmp_path_factory): + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['1.0.1', '2.0.1-beta.1', '2.0.1'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.1-beta.1', None, None, + {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:2.0.1-beta.1']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:2.0.1-beta.1'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, True, False, False, False, False + )['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'2.0.1-beta.1' + + assert mock_get_info.call_count == 1 + assert mock_get_info.mock_calls[0][1] == ('namespace', 'collection', '2.0.1-beta.1') + + +def test_build_requirement_from_name_second_server(galaxy_server, monkeypatch, tmp_path_factory): + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['1.0.1', '1.0.2', '1.0.3'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '1.0.3', None, None, {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + broken_server = copy.copy(galaxy_server) + broken_server.api_server = 'https://broken.com/' + mock_version_list = MagicMock() + mock_version_list.return_value = [] + monkeypatch.setattr(broken_server, 'get_collection_versions', mock_version_list) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:>1.0.1']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:>1.0.1'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + actual = collection._resolve_depenency_map( + requirements, [broken_server, galaxy_server], concrete_artifact_cm, None, True, False, False, False, False + )['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'1.0.3' + + assert mock_version_list.call_count == 1 + assert mock_version_list.mock_calls[0][1] == ('namespace', 'collection') + + assert mock_get_versions.call_count == 1 + assert mock_get_versions.mock_calls[0][1] == ('namespace', 'collection') + + +def test_build_requirement_from_name_missing(galaxy_server, monkeypatch, tmp_path_factory): + mock_open = MagicMock() + mock_open.return_value = [] + + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_open) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:>1.0.1']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + expected = "Failed to resolve the requested dependencies map. Could not satisfy the following requirements:\n* namespace.collection:* (direct request)" + with pytest.raises(AnsibleError, match=re.escape(expected)): + collection._resolve_depenency_map(requirements, [galaxy_server, galaxy_server], concrete_artifact_cm, None, False, True, False, False, False) + + +def test_build_requirement_from_name_401_unauthorized(galaxy_server, monkeypatch, tmp_path_factory): + mock_open = MagicMock() + mock_open.side_effect = api.GalaxyError(urllib_error.HTTPError('https://galaxy.server.com', 401, 'msg', {}, + StringIO()), "error") + + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_open) + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:>1.0.1']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + expected = "error (HTTP Code: 401, Message: msg)" + with pytest.raises(api.GalaxyError, match=re.escape(expected)): + collection._resolve_depenency_map(requirements, [galaxy_server, galaxy_server], concrete_artifact_cm, None, False, False, False, False, False) + + +def test_build_requirement_from_name_single_version(galaxy_server, monkeypatch, tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + multi_api_proxy = collection.galaxy_api_proxy.MultiGalaxyAPIProxy([galaxy_server], concrete_artifact_cm) + dep_provider = dependency_resolution.providers.CollectionDependencyProvider(apis=multi_api_proxy, concrete_artifacts_manager=concrete_artifact_cm) + + matches = RequirementCandidates() + mock_find_matches = MagicMock(side_effect=matches.func_wrapper(dep_provider.find_matches), autospec=True) + monkeypatch.setattr(dependency_resolution.providers.CollectionDependencyProvider, 'find_matches', mock_find_matches) + + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['2.0.0'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.0', None, None, + {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:==2.0.0']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:==2.0.0'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, False, True, False, False, False)['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'2.0.0' + assert [c.ver for c in matches.candidates] == [u'2.0.0'] + + assert mock_get_info.call_count == 1 + assert mock_get_info.mock_calls[0][1] == ('namespace', 'collection', '2.0.0') + + +def test_build_requirement_from_name_multiple_versions_one_match(galaxy_server, monkeypatch, tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + multi_api_proxy = collection.galaxy_api_proxy.MultiGalaxyAPIProxy([galaxy_server], concrete_artifact_cm) + dep_provider = dependency_resolution.providers.CollectionDependencyProvider(apis=multi_api_proxy, concrete_artifacts_manager=concrete_artifact_cm) + + matches = RequirementCandidates() + mock_find_matches = MagicMock(side_effect=matches.func_wrapper(dep_provider.find_matches), autospec=True) + monkeypatch.setattr(dependency_resolution.providers.CollectionDependencyProvider, 'find_matches', mock_find_matches) + + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['2.0.0', '2.0.1', '2.0.2'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.1', None, None, + {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:>=2.0.1,<2.0.2']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:>=2.0.1,<2.0.2'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, False, True, False, False, False)['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'2.0.1' + assert [c.ver for c in matches.candidates] == [u'2.0.1'] + + assert mock_get_versions.call_count == 1 + assert mock_get_versions.mock_calls[0][1] == ('namespace', 'collection') + + assert mock_get_info.call_count == 1 + assert mock_get_info.mock_calls[0][1] == ('namespace', 'collection', '2.0.1') + + +def test_build_requirement_from_name_multiple_version_results(galaxy_server, monkeypatch, tmp_path_factory): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + multi_api_proxy = collection.galaxy_api_proxy.MultiGalaxyAPIProxy([galaxy_server], concrete_artifact_cm) + dep_provider = dependency_resolution.providers.CollectionDependencyProvider(apis=multi_api_proxy, concrete_artifacts_manager=concrete_artifact_cm) + + matches = RequirementCandidates() + mock_find_matches = MagicMock(side_effect=matches.func_wrapper(dep_provider.find_matches), autospec=True) + monkeypatch.setattr(dependency_resolution.providers.CollectionDependencyProvider, 'find_matches', mock_find_matches) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.5', None, None, {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['1.0.1', '1.0.2', '1.0.3'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + mock_get_versions.return_value = ['2.0.0', '2.0.1', '2.0.2', '2.0.3', '2.0.4', '2.0.5'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:!=2.0.2']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:!=2.0.2'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + actual = collection._resolve_depenency_map( + requirements, [galaxy_server], concrete_artifact_cm, None, False, True, False, False, False)['namespace.collection'] + + assert actual.namespace == u'namespace' + assert actual.name == u'collection' + assert actual.src == galaxy_server + assert actual.ver == u'2.0.5' + # should be ordered latest to earliest + assert [c.ver for c in matches.candidates] == [u'2.0.5', u'2.0.4', u'2.0.3', u'2.0.1', u'2.0.0'] + + assert mock_get_versions.call_count == 1 + assert mock_get_versions.mock_calls[0][1] == ('namespace', 'collection') + + +def test_candidate_with_conflict(monkeypatch, tmp_path_factory, galaxy_server): + + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '2.0.5', None, None, {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + mock_get_versions = MagicMock() + mock_get_versions.return_value = ['2.0.5'] + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection:!=2.0.5']) + requirements = cli._require_one_of_collections_requirements( + ['namespace.collection:!=2.0.5'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + expected = "Failed to resolve the requested dependencies map. Could not satisfy the following requirements:\n" + expected += "* namespace.collection:!=2.0.5 (direct request)" + with pytest.raises(AnsibleError, match=re.escape(expected)): + collection._resolve_depenency_map(requirements, [galaxy_server], concrete_artifact_cm, None, False, True, False, False, False) + + +def test_dep_candidate_with_conflict(monkeypatch, tmp_path_factory, galaxy_server): + test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + mock_get_info_return = [ + api.CollectionVersionMetadata('parent', 'collection', '2.0.5', None, None, {'namespace.collection': '!=1.0.0'}, None, None), + api.CollectionVersionMetadata('namespace', 'collection', '1.0.0', None, None, {}, None, None), + ] + mock_get_info = MagicMock(side_effect=mock_get_info_return) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + mock_get_versions = MagicMock(side_effect=[['2.0.5'], ['1.0.0']]) + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'parent.collection:2.0.5']) + requirements = cli._require_one_of_collections_requirements( + ['parent.collection:2.0.5'], None, artifacts_manager=concrete_artifact_cm + )['collections'] + + expected = "Failed to resolve the requested dependencies map. Could not satisfy the following requirements:\n" + expected += "* namespace.collection:!=1.0.0 (dependency of parent.collection:2.0.5)" + with pytest.raises(AnsibleError, match=re.escape(expected)): + collection._resolve_depenency_map(requirements, [galaxy_server], concrete_artifact_cm, None, False, True, False, False, False) + + +def test_install_installed_collection(monkeypatch, tmp_path_factory, galaxy_server): + + mock_installed_collections = MagicMock(return_value=[Candidate('namespace.collection', '1.2.3', None, 'dir', None)]) + + monkeypatch.setattr(collection, 'find_existing_collections', mock_installed_collections) + + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(test_dir, validate_certs=False) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + mock_get_info = MagicMock() + mock_get_info.return_value = api.CollectionVersionMetadata('namespace', 'collection', '1.2.3', None, None, {}, None, None) + monkeypatch.setattr(galaxy_server, 'get_collection_version_metadata', mock_get_info) + + mock_get_versions = MagicMock(return_value=['1.2.3', '1.3.0']) + monkeypatch.setattr(galaxy_server, 'get_collection_versions', mock_get_versions) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection']) + cli.run() + + expected = "Nothing to do. All requested collections are already installed. If you want to reinstall them, consider using `--force`." + assert mock_display.mock_calls[1][1][0] == expected + + +def test_install_collection(collection_artifact, monkeypatch): + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + collection_tar = collection_artifact[1] + + temp_path = os.path.join(os.path.split(collection_tar)[0], b'temp') + os.makedirs(temp_path) + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + + output_path = os.path.join(os.path.split(collection_tar)[0]) + collection_path = os.path.join(output_path, b'ansible_namespace', b'collection') + os.makedirs(os.path.join(collection_path, b'delete_me')) # Create a folder to verify the install cleans out the dir + + candidate = Candidate('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None) + collection.install(candidate, to_text(output_path), concrete_artifact_cm) + + # Ensure the temp directory is empty, nothing is left behind + assert os.listdir(temp_path) == [] + + actual_files = os.listdir(collection_path) + actual_files.sort() + assert actual_files == [b'FILES.json', b'MANIFEST.json', b'README.md', b'docs', b'playbooks', b'plugins', b'roles', + b'runme.sh'] + + assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'plugins')).st_mode) == 0o0755 + assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'README.md')).st_mode) == 0o0644 + assert stat.S_IMODE(os.stat(os.path.join(collection_path, b'runme.sh')).st_mode) == 0o0755 + + assert mock_display.call_count == 2 + assert mock_display.mock_calls[0][1][0] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" \ + % to_text(collection_path) + assert mock_display.mock_calls[1][1][0] == "ansible_namespace.collection:0.1.0 was installed successfully" + + +def test_install_collection_with_download(galaxy_server, collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + shutil.rmtree(collection_path) + + collections_dir = ('%s' % os.path.sep).join(to_text(collection_path).split('%s' % os.path.sep)[:-2]) + + temp_path = os.path.join(os.path.split(collection_tar)[0], b'temp') + os.makedirs(temp_path) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + + mock_download = MagicMock() + mock_download.return_value = collection_tar + monkeypatch.setattr(concrete_artifact_cm, 'get_galaxy_artifact_path', mock_download) + + req = Candidate('ansible_namespace.collection', '0.1.0', 'https://downloadme.com', 'galaxy', None) + collection.install(req, to_text(collections_dir), concrete_artifact_cm) + + actual_files = os.listdir(collection_path) + actual_files.sort() + assert actual_files == [b'FILES.json', b'MANIFEST.json', b'README.md', b'docs', b'playbooks', b'plugins', b'roles', + b'runme.sh'] + + assert mock_display.call_count == 2 + assert mock_display.mock_calls[0][1][0] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" \ + % to_text(collection_path) + assert mock_display.mock_calls[1][1][0] == "ansible_namespace.collection:0.1.0 was installed successfully" + + assert mock_download.call_count == 1 + assert mock_download.mock_calls[0][1][0].src == 'https://downloadme.com' + assert mock_download.mock_calls[0][1][0].type == 'galaxy' + + +def test_install_collections_from_tar(collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + temp_path = os.path.split(collection_tar)[0] + shutil.rmtree(collection_path) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + + requirements = [Requirement('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None)] + collection.install_collections(requirements, to_text(temp_path), [], False, False, False, False, False, False, concrete_artifact_cm, True, False) + + assert os.path.isdir(collection_path) + + actual_files = os.listdir(collection_path) + actual_files.sort() + assert actual_files == [b'FILES.json', b'MANIFEST.json', b'README.md', b'docs', b'playbooks', b'plugins', b'roles', + b'runme.sh'] + + with open(os.path.join(collection_path, b'MANIFEST.json'), 'rb') as manifest_obj: + actual_manifest = json.loads(to_text(manifest_obj.read())) + + assert actual_manifest['collection_info']['namespace'] == 'ansible_namespace' + assert actual_manifest['collection_info']['name'] == 'collection' + assert actual_manifest['collection_info']['version'] == '0.1.0' + + # Filter out the progress cursor display calls. + display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2] and len(m[1]) == 1] + assert len(display_msgs) == 4 + assert display_msgs[0] == "Process install dependency map" + assert display_msgs[1] == "Starting collection install process" + assert display_msgs[2] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" % to_text(collection_path) + + +def test_install_collections_existing_without_force(collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + temp_path = os.path.split(collection_tar)[0] + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + + assert os.path.isdir(collection_path) + + requirements = [Requirement('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None)] + collection.install_collections(requirements, to_text(temp_path), [], False, False, False, False, False, False, concrete_artifact_cm, True, False) + + assert os.path.isdir(collection_path) + + actual_files = os.listdir(collection_path) + actual_files.sort() + assert actual_files == [b'README.md', b'docs', b'galaxy.yml', b'playbooks', b'plugins', b'roles', b'runme.sh'] + + # Filter out the progress cursor display calls. + display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2] and len(m[1]) == 1] + assert len(display_msgs) == 1 + + assert display_msgs[0] == 'Nothing to do. All requested collections are already installed. If you want to reinstall them, consider using `--force`.' + + for msg in display_msgs: + assert 'WARNING' not in msg + + +def test_install_missing_metadata_warning(collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + temp_path = os.path.split(collection_tar)[0] + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + for file in [b'MANIFEST.json', b'galaxy.yml']: + b_path = os.path.join(collection_path, file) + if os.path.isfile(b_path): + os.unlink(b_path) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + requirements = [Requirement('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None)] + collection.install_collections(requirements, to_text(temp_path), [], False, False, False, False, False, False, concrete_artifact_cm, True, False) + + display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2] and len(m[1]) == 1] + + assert 'WARNING' in display_msgs[0] + + +# Makes sure we don't get stuck in some recursive loop +@pytest.mark.parametrize('collection_artifact', [ + {'ansible_namespace.collection': '>=0.0.1'}, +], indirect=True) +def test_install_collection_with_circular_dependency(collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + temp_path = os.path.split(collection_tar)[0] + shutil.rmtree(collection_path) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + requirements = [Requirement('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None)] + collection.install_collections(requirements, to_text(temp_path), [], False, False, False, False, False, False, concrete_artifact_cm, True, False) + + assert os.path.isdir(collection_path) + + actual_files = os.listdir(collection_path) + actual_files.sort() + assert actual_files == [b'FILES.json', b'MANIFEST.json', b'README.md', b'docs', b'playbooks', b'plugins', b'roles', + b'runme.sh'] + + with open(os.path.join(collection_path, b'MANIFEST.json'), 'rb') as manifest_obj: + actual_manifest = json.loads(to_text(manifest_obj.read())) + + assert actual_manifest['collection_info']['namespace'] == 'ansible_namespace' + assert actual_manifest['collection_info']['name'] == 'collection' + assert actual_manifest['collection_info']['version'] == '0.1.0' + assert actual_manifest['collection_info']['dependencies'] == {'ansible_namespace.collection': '>=0.0.1'} + + # Filter out the progress cursor display calls. + display_msgs = [m[1][0] for m in mock_display.mock_calls if 'newline' not in m[2] and len(m[1]) == 1] + assert len(display_msgs) == 4 + assert display_msgs[0] == "Process install dependency map" + assert display_msgs[1] == "Starting collection install process" + assert display_msgs[2] == "Installing 'ansible_namespace.collection:0.1.0' to '%s'" % to_text(collection_path) + assert display_msgs[3] == "ansible_namespace.collection:0.1.0 was installed successfully" + + +@pytest.mark.parametrize('collection_artifact', [ + None, + {}, +], indirect=True) +def test_install_collection_with_no_dependency(collection_artifact, monkeypatch): + collection_path, collection_tar = collection_artifact + temp_path = os.path.split(collection_tar)[0] + shutil.rmtree(collection_path) + + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(temp_path, validate_certs=False) + requirements = [Requirement('ansible_namespace.collection', '0.1.0', to_text(collection_tar), 'file', None)] + collection.install_collections(requirements, to_text(temp_path), [], False, False, False, False, False, False, concrete_artifact_cm, True, False) + + assert os.path.isdir(collection_path) + + with open(os.path.join(collection_path, b'MANIFEST.json'), 'rb') as manifest_obj: + actual_manifest = json.loads(to_text(manifest_obj.read())) + + assert not actual_manifest['collection_info']['dependencies'] + assert actual_manifest['collection_info']['namespace'] == 'ansible_namespace' + assert actual_manifest['collection_info']['name'] == 'collection' + assert actual_manifest['collection_info']['version'] == '0.1.0' + + +@pytest.mark.parametrize( + "signatures,required_successful_count,ignore_errors,expected_success", + [ + ([], 'all', [], True), + (["good_signature"], 'all', [], True), + (["good_signature", collection.gpg.GpgBadArmor(status='failed')], 'all', [], False), + ([collection.gpg.GpgBadArmor(status='failed')], 'all', [], False), + # This is expected to succeed because ignored does not increment failed signatures. + # "all" signatures is not a specific number, so all == no (non-ignored) signatures in this case. + ([collection.gpg.GpgBadArmor(status='failed')], 'all', ["BADARMOR"], True), + ([collection.gpg.GpgBadArmor(status='failed'), "good_signature"], 'all', ["BADARMOR"], True), + ([], '+all', [], False), + ([collection.gpg.GpgBadArmor(status='failed')], '+all', ["BADARMOR"], False), + ([], '1', [], True), + ([], '+1', [], False), + (["good_signature"], '2', [], False), + (["good_signature", collection.gpg.GpgBadArmor(status='failed')], '2', [], False), + # This is expected to fail because ignored does not increment successful signatures. + # 2 signatures are required, but only 1 is successful. + (["good_signature", collection.gpg.GpgBadArmor(status='failed')], '2', ["BADARMOR"], False), + (["good_signature", "good_signature"], '2', [], True), + ] +) +def test_verify_file_signatures(signatures, required_successful_count, ignore_errors, expected_success): + # type: (List[bool], int, bool, bool) -> None + + def gpg_error_generator(results): + for result in results: + if isinstance(result, collection.gpg.GpgBaseError): + yield result + + fqcn = 'ns.coll' + manifest_file = 'MANIFEST.json' + keyring = '~/.ansible/pubring.kbx' + + with patch.object(collection, 'run_gpg_verify', MagicMock(return_value=("somestdout", 0,))): + with patch.object(collection, 'parse_gpg_errors', MagicMock(return_value=gpg_error_generator(signatures))): + assert collection.verify_file_signatures( + fqcn, + manifest_file, + signatures, + keyring, + required_successful_count, + ignore_errors + ) == expected_success diff --git a/test/units/galaxy/test_role_install.py b/test/units/galaxy/test_role_install.py new file mode 100644 index 0000000..687fcac --- /dev/null +++ b/test/units/galaxy/test_role_install.py @@ -0,0 +1,152 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + + +import os +import functools +import pytest +import tempfile + +from io import StringIO +from ansible import context +from ansible.cli.galaxy import GalaxyCLI +from ansible.galaxy import api, role, Galaxy +from ansible.module_utils._text import to_text +from ansible.utils import context_objects as co + + +def call_galaxy_cli(args): + orig = co.GlobalCLIArgs._Singleton__instance + co.GlobalCLIArgs._Singleton__instance = None + try: + GalaxyCLI(args=['ansible-galaxy', 'role'] + args).run() + finally: + co.GlobalCLIArgs._Singleton__instance = orig + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + yield + co.GlobalCLIArgs._Singleton__instance = None + + +@pytest.fixture(autouse=True) +def galaxy_server(): + context.CLIARGS._store = {'ignore_certs': False} + galaxy_api = api.GalaxyAPI(None, 'test_server', 'https://galaxy.ansible.com') + return galaxy_api + + +@pytest.fixture(autouse=True) +def init_role_dir(tmp_path_factory): + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Roles Input')) + namespace = 'ansible_namespace' + role = 'role' + skeleton_path = os.path.join(os.path.dirname(os.path.split(__file__)[0]), 'cli', 'test_data', 'role_skeleton') + call_galaxy_cli(['init', '%s.%s' % (namespace, role), '-c', '--init-path', test_dir, '--role-skeleton', skeleton_path]) + + +def mock_NamedTemporaryFile(mocker, **args): + mock_ntf = mocker.MagicMock() + mock_ntf.write = mocker.MagicMock() + mock_ntf.close = mocker.MagicMock() + mock_ntf.name = None + return mock_ntf + + +@pytest.fixture +def init_mock_temp_file(mocker, monkeypatch): + monkeypatch.setattr(tempfile, 'NamedTemporaryFile', functools.partial(mock_NamedTemporaryFile, mocker)) + + +@pytest.fixture(autouse=True) +def mock_role_download_api(mocker, monkeypatch): + mock_role_api = mocker.MagicMock() + mock_role_api.side_effect = [ + StringIO(u''), + ] + monkeypatch.setattr(role, 'open_url', mock_role_api) + return mock_role_api + + +def test_role_download_github(init_mock_temp_file, mocker, galaxy_server, mock_role_download_api, monkeypatch): + mock_api = mocker.MagicMock() + mock_api.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"results":[{"id":"123","github_user":"test_owner","github_repo": "test_role"}]}'), + StringIO(u'{"results":[{"name": "0.0.1"},{"name": "0.0.2"}]}'), + ] + monkeypatch.setattr(api, 'open_url', mock_api) + + role.GalaxyRole(Galaxy(), galaxy_server, 'test_owner.test_role', version="0.0.1").install() + + assert mock_role_download_api.call_count == 1 + assert mock_role_download_api.mock_calls[0][1][0] == 'https://github.com/test_owner/test_role/archive/0.0.1.tar.gz' + + +def test_role_download_github_default_version(init_mock_temp_file, mocker, galaxy_server, mock_role_download_api, monkeypatch): + mock_api = mocker.MagicMock() + mock_api.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"results":[{"id":"123","github_user":"test_owner","github_repo": "test_role"}]}'), + StringIO(u'{"results":[{"name": "0.0.1"},{"name": "0.0.2"}]}'), + ] + monkeypatch.setattr(api, 'open_url', mock_api) + + role.GalaxyRole(Galaxy(), galaxy_server, 'test_owner.test_role').install() + + assert mock_role_download_api.call_count == 1 + assert mock_role_download_api.mock_calls[0][1][0] == 'https://github.com/test_owner/test_role/archive/0.0.2.tar.gz' + + +def test_role_download_github_no_download_url_for_version(init_mock_temp_file, mocker, galaxy_server, mock_role_download_api, monkeypatch): + mock_api = mocker.MagicMock() + mock_api.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"results":[{"id":"123","github_user":"test_owner","github_repo": "test_role"}]}'), + StringIO(u'{"results":[{"name": "0.0.1"},{"name": "0.0.2","download_url":"http://localhost:8080/test_owner/test_role/0.0.2.tar.gz"}]}'), + ] + monkeypatch.setattr(api, 'open_url', mock_api) + + role.GalaxyRole(Galaxy(), galaxy_server, 'test_owner.test_role', version="0.0.1").install() + + assert mock_role_download_api.call_count == 1 + assert mock_role_download_api.mock_calls[0][1][0] == 'https://github.com/test_owner/test_role/archive/0.0.1.tar.gz' + + +def test_role_download_url(init_mock_temp_file, mocker, galaxy_server, mock_role_download_api, monkeypatch): + mock_api = mocker.MagicMock() + mock_api.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"results":[{"id":"123","github_user":"test_owner","github_repo": "test_role"}]}'), + StringIO(u'{"results":[{"name": "0.0.1","download_url":"http://localhost:8080/test_owner/test_role/0.0.1.tar.gz"},' + u'{"name": "0.0.2","download_url":"http://localhost:8080/test_owner/test_role/0.0.2.tar.gz"}]}'), + ] + monkeypatch.setattr(api, 'open_url', mock_api) + + role.GalaxyRole(Galaxy(), galaxy_server, 'test_owner.test_role', version="0.0.1").install() + + assert mock_role_download_api.call_count == 1 + assert mock_role_download_api.mock_calls[0][1][0] == 'http://localhost:8080/test_owner/test_role/0.0.1.tar.gz' + + +def test_role_download_url_default_version(init_mock_temp_file, mocker, galaxy_server, mock_role_download_api, monkeypatch): + mock_api = mocker.MagicMock() + mock_api.side_effect = [ + StringIO(u'{"available_versions":{"v1":"v1/"}}'), + StringIO(u'{"results":[{"id":"123","github_user":"test_owner","github_repo": "test_role"}]}'), + StringIO(u'{"results":[{"name": "0.0.1","download_url":"http://localhost:8080/test_owner/test_role/0.0.1.tar.gz"},' + u'{"name": "0.0.2","download_url":"http://localhost:8080/test_owner/test_role/0.0.2.tar.gz"}]}'), + ] + monkeypatch.setattr(api, 'open_url', mock_api) + + role.GalaxyRole(Galaxy(), galaxy_server, 'test_owner.test_role').install() + + assert mock_role_download_api.call_count == 1 + assert mock_role_download_api.mock_calls[0][1][0] == 'http://localhost:8080/test_owner/test_role/0.0.2.tar.gz' diff --git a/test/units/galaxy/test_role_requirements.py b/test/units/galaxy/test_role_requirements.py new file mode 100644 index 0000000..a84bbb5 --- /dev/null +++ b/test/units/galaxy/test_role_requirements.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2020, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + + +import pytest +from ansible.playbook.role.requirement import RoleRequirement + + +def test_null_role_url(): + role = RoleRequirement.role_yaml_parse('') + assert role['src'] == '' + assert role['name'] == '' + assert role['scm'] is None + assert role['version'] is None + + +def test_git_file_role_url(): + role = RoleRequirement.role_yaml_parse('git+file:///home/bennojoy/nginx') + assert role['src'] == 'file:///home/bennojoy/nginx' + assert role['name'] == 'nginx' + assert role['scm'] == 'git' + assert role['version'] is None + + +def test_https_role_url(): + role = RoleRequirement.role_yaml_parse('https://github.com/bennojoy/nginx') + assert role['src'] == 'https://github.com/bennojoy/nginx' + assert role['name'] == 'nginx' + assert role['scm'] is None + assert role['version'] is None + + +def test_git_https_role_url(): + role = RoleRequirement.role_yaml_parse('git+https://github.com/geerlingguy/ansible-role-composer.git') + assert role['src'] == 'https://github.com/geerlingguy/ansible-role-composer.git' + assert role['name'] == 'ansible-role-composer' + assert role['scm'] == 'git' + assert role['version'] is None + + +def test_git_version_role_url(): + role = RoleRequirement.role_yaml_parse('git+https://github.com/geerlingguy/ansible-role-composer.git,main') + assert role['src'] == 'https://github.com/geerlingguy/ansible-role-composer.git' + assert role['name'] == 'ansible-role-composer' + assert role['scm'] == 'git' + assert role['version'] == 'main' + + +@pytest.mark.parametrize("url", [ + ('https://some.webserver.example.com/files/main.tar.gz'), + ('https://some.webserver.example.com/files/main.tar.bz2'), + ('https://some.webserver.example.com/files/main.tar.xz'), +]) +def test_tar_role_url(url): + role = RoleRequirement.role_yaml_parse(url) + assert role['src'] == url + assert role['name'].startswith('main') + assert role['scm'] is None + assert role['version'] is None + + +def test_git_ssh_role_url(): + role = RoleRequirement.role_yaml_parse('git@gitlab.company.com:mygroup/ansible-base.git') + assert role['src'] == 'git@gitlab.company.com:mygroup/ansible-base.git' + assert role['name'].startswith('ansible-base') + assert role['scm'] is None + assert role['version'] is None + + +def test_token_role_url(): + role = RoleRequirement.role_yaml_parse('git+https://gitlab+deploy-token-312644:_aJQ9c3HWzmRR4knBNyx@gitlab.com/akasurde/ansible-demo') + assert role['src'] == 'https://gitlab+deploy-token-312644:_aJQ9c3HWzmRR4knBNyx@gitlab.com/akasurde/ansible-demo' + assert role['name'].startswith('ansible-demo') + assert role['scm'] == 'git' + assert role['version'] is None + + +def test_token_new_style_role_url(): + role = RoleRequirement.role_yaml_parse({"src": "git+https://gitlab+deploy-token-312644:_aJQ9c3HWzmRR4knBNyx@gitlab.com/akasurde/ansible-demo"}) + assert role['src'] == 'https://gitlab+deploy-token-312644:_aJQ9c3HWzmRR4knBNyx@gitlab.com/akasurde/ansible-demo' + assert role['name'].startswith('ansible-demo') + assert role['scm'] == 'git' + assert role['version'] == '' diff --git a/test/units/galaxy/test_token.py b/test/units/galaxy/test_token.py new file mode 100644 index 0000000..24af386 --- /dev/null +++ b/test/units/galaxy/test_token.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import os +import pytest +from unittest.mock import MagicMock + +import ansible.constants as C +from ansible.cli.galaxy import GalaxyCLI, SERVER_DEF +from ansible.galaxy.token import GalaxyToken, NoTokenSentinel +from ansible.module_utils._text import to_bytes, to_text + + +@pytest.fixture() +def b_token_file(request, tmp_path_factory): + b_test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Token')) + b_token_path = os.path.join(b_test_dir, b"token.yml") + + token = getattr(request, 'param', None) + if token: + with open(b_token_path, 'wb') as token_fd: + token_fd.write(b"token: %s" % to_bytes(token)) + + orig_token_path = C.GALAXY_TOKEN_PATH + C.GALAXY_TOKEN_PATH = to_text(b_token_path) + try: + yield b_token_path + finally: + C.GALAXY_TOKEN_PATH = orig_token_path + + +def test_client_id(monkeypatch): + monkeypatch.setattr(C, 'GALAXY_SERVER_LIST', ['server1', 'server2']) + + test_server_config = {option[0]: None for option in SERVER_DEF} + test_server_config.update( + { + 'url': 'http://my_galaxy_ng:8000/api/automation-hub/', + 'auth_url': 'http://my_keycloak:8080/auth/realms/myco/protocol/openid-connect/token', + 'client_id': 'galaxy-ng', + 'token': 'access_token', + } + ) + + test_server_default = {option[0]: None for option in SERVER_DEF} + test_server_default.update( + { + 'url': 'https://cloud.redhat.com/api/automation-hub/', + 'auth_url': 'https://sso.redhat.com/auth/realms/redhat-external/protocol/openid-connect/token', + 'token': 'access_token', + } + ) + + get_plugin_options = MagicMock(side_effect=[test_server_config, test_server_default]) + monkeypatch.setattr(C.config, 'get_plugin_options', get_plugin_options) + + cli_args = [ + 'ansible-galaxy', + 'collection', + 'install', + 'namespace.collection:1.0.0', + ] + + galaxy_cli = GalaxyCLI(args=cli_args) + mock_execute_install = MagicMock() + monkeypatch.setattr(galaxy_cli, '_execute_install_collection', mock_execute_install) + galaxy_cli.run() + + assert galaxy_cli.api_servers[0].token.client_id == 'galaxy-ng' + assert galaxy_cli.api_servers[1].token.client_id == 'cloud-services' + + +def test_token_explicit(b_token_file): + assert GalaxyToken(token="explicit").get() == "explicit" + + +@pytest.mark.parametrize('b_token_file', ['file'], indirect=True) +def test_token_explicit_override_file(b_token_file): + assert GalaxyToken(token="explicit").get() == "explicit" + + +@pytest.mark.parametrize('b_token_file', ['file'], indirect=True) +def test_token_from_file(b_token_file): + assert GalaxyToken().get() == "file" + + +def test_token_from_file_missing(b_token_file): + assert GalaxyToken().get() is None + + +@pytest.mark.parametrize('b_token_file', ['file'], indirect=True) +def test_token_none(b_token_file): + assert GalaxyToken(token=NoTokenSentinel).get() is None diff --git a/test/units/galaxy/test_user_agent.py b/test/units/galaxy/test_user_agent.py new file mode 100644 index 0000000..da0103f --- /dev/null +++ b/test/units/galaxy/test_user_agent.py @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2019, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import platform + +from ansible.galaxy import user_agent +from ansible.module_utils.ansible_release import __version__ as ansible_version + + +def test_user_agent(): + res = user_agent.user_agent() + assert res.startswith('ansible-galaxy/%s' % ansible_version) + assert platform.system() in res + assert 'python:' in res |