# -*- 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 annotations 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.common.file import S_IRWU_RG_RO from ansible.module_utils.common.text.converters import to_native, to_text import urllib.error from ansible.utils import context_objects as co from ansible.utils.display import Display @pytest.fixture(autouse=True) 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 = S_IRWU_RG_RO 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 _available_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) 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) 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) 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) 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) 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