summaryrefslogtreecommitdiffstats
path: root/test/units/galaxy/test_collection.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/units/galaxy/test_collection.py')
-rw-r--r--test/units/galaxy/test_collection.py537
1 files changed, 25 insertions, 512 deletions
diff --git a/test/units/galaxy/test_collection.py b/test/units/galaxy/test_collection.py
index 6276fb346c..c86a524a09 100644
--- a/test/units/galaxy/test_collection.py
+++ b/test/units/galaxy/test_collection.py
@@ -9,18 +9,13 @@ __metaclass__ = type
import json
import os
import pytest
-import re
import tarfile
-import tempfile
-import time
import uuid
from hashlib import sha256
-from io import BytesIO, StringIO
+from io import BytesIO
from units.compat.mock import MagicMock
-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
@@ -56,13 +51,6 @@ def collection_input(tmp_path_factory):
@pytest.fixture()
-def galaxy_api_version(monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
-
-@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()
@@ -408,440 +396,53 @@ def test_build_with_symlink_inside_collection(collection_input):
assert actual_file == '63444bfc766154e1bc7557ef6280de20d03fcd81'
-def test_publish_missing_file():
- fake_path = u'/fake/ÅÑŚÌβŁÈ/path'
- expected = to_native("The collection path specified '%s' does not exist." % fake_path)
-
- with pytest.raises(AnsibleError, match=expected):
- collection.publish_collection(fake_path, None, True, 0)
-
-
-def test_publish_not_a_tarball():
- expected = "The collection path specified '{0}' is not a tarball, use 'ansible-galaxy collection build' to " \
- "create a proper release artifact."
-
- 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))):
- collection.publish_collection(temp_file.name, None, True, 0)
-
-
def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
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_open.return_value = StringIO(u'{"task":"%s"}' % fake_import_uri)
- expected_form, expected_content_type = collection._get_mime_data(to_bytes(artifact_path))
+ 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_open.call_count == 1
- assert mock_open.mock_calls[0][1][0] == '%s/api/v2/collections/' % galaxy_server.api_server
- assert mock_open.mock_calls[0][2]['data'] == expected_form
- assert mock_open.mock_calls[0][2]['method'] == 'POST'
- assert mock_open.mock_calls[0][2]['validate_certs'] is True
- assert mock_open.mock_calls[0][2]['headers']['Authorization'] == 'Token key'
- assert mock_open.mock_calls[0][2]['headers']['Content-length'] == len(expected_form)
- assert mock_open.mock_calls[0][2]['headers']['Content-type'] == expected_content_type
-
- assert mock_display.call_count == 2
- assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
- % (artifact_path, galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[1][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_dont_validate_cert(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- galaxy_server.validate_certs = False
- artifact_path, mock_open = collection_artifact
-
- mock_open.return_value = StringIO(u'{"task":"https://galaxy.server.com/api/v2/import/1234"}')
-
- collection.publish_collection(artifact_path, galaxy_server, False, 0)
-
- assert mock_open.call_count == 1
- assert mock_open.mock_calls[0][2]['validate_certs'] is False
-
-
-def test_publish_failure(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- artifact_path, mock_open = collection_artifact
-
- mock_open.side_effect = urllib_error.HTTPError('https://galaxy.server.com', 500, 'msg', {}, StringIO())
-
- expected = 'Error when publishing collection to test_server (https://galaxy.ansible.com) ' \
- '(HTTP Code: 500, Message: Unknown error returned by Galaxy ' \
- 'server. Code: Unknown)'
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
-
-
-def test_publish_failure_with_json_info(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- artifact_path, mock_open = collection_artifact
-
- return_content = StringIO(u'{"message":"Galaxy error message","code":"GWE002"}')
- mock_open.side_effect = urllib_error.HTTPError('https://galaxy.server.com', 503, 'msg', {}, return_content)
-
- expected = 'Error when publishing collection to test_server (https://galaxy.ansible.com) ' \
- '(HTTP Code: 503, Message: Galaxy error message Code: GWE002)'
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
-
-
-@pytest.mark.parametrize("api_version,token_type", [
- ('v2', 'Token'),
- ('v3', 'Bearer')
-])
-def test_publish_with_wait(api_version, token_type, galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {api_version: '/api/%s' % api_version}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- mock_display = MagicMock()
- monkeypatch.setattr(Display, 'display', mock_display)
-
- fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-
- artifact_path, mock_open = collection_artifact
-
- mock_open.side_effect = (
- StringIO(u'{"task":"%s"}' % fake_import_uri),
- StringIO(u'{"finished_at":"some_time","state":"success"}')
- )
-
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
-
- assert mock_open.call_count == 2
- assert mock_open.mock_calls[1][1][0] == fake_import_uri
- assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s key' % token_type
- assert mock_open.mock_calls[1][2]['validate_certs'] is True
- assert mock_open.mock_calls[1][2]['method'] == 'GET'
-
- assert mock_display.call_count == 5
- assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
- % (artifact_path, galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
- % (galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
- assert mock_display.mock_calls[4][1][0] == 'Collection has been successfully published and imported to the ' \
- 'Galaxy server %s %s' % (galaxy_server.name, galaxy_server.api_server)
-
-
-@pytest.mark.parametrize("api_version,exp_api_url,token_type", [
- ('v2', '/api/v2/collections/', 'Token'),
- ('v3', '/api/v3/artifacts/collections/', 'Bearer')
-])
-def test_publish_with_wait_timeout(api_version, exp_api_url, token_type, galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {api_version: '/api/%s' % api_version}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- monkeypatch.setattr(time, 'sleep', MagicMock())
-
- mock_vvv = MagicMock()
- monkeypatch.setattr(Display, 'vvv', mock_vvv)
-
- fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-
- artifact_path, mock_open = collection_artifact
-
- mock_open.side_effect = (
- StringIO(u'{"task":"%s"}' % fake_import_uri),
- StringIO(u'{"finished_at":null}'),
- StringIO(u'{"finished_at":"some_time","state":"success"}')
- )
-
- collection.publish_collection(artifact_path, galaxy_server, True, 60)
-
- assert mock_open.call_count == 3
- assert mock_open.mock_calls[1][1][0] == fake_import_uri
- assert mock_open.mock_calls[1][2]['headers']['Authorization'] == '%s key' % token_type
- assert mock_open.mock_calls[1][2]['validate_certs'] is True
- assert mock_open.mock_calls[1][2]['method'] == 'GET'
- assert mock_open.mock_calls[2][1][0] == fake_import_uri
- assert mock_open.mock_calls[2][2]['headers']['Authorization'] == '%s key' % token_type
- assert mock_open.mock_calls[2][2]['validate_certs'] is True
- assert mock_open.mock_calls[2][2]['method'] == 'GET'
-
- assert mock_vvv.call_count == 2
- assert mock_vvv.mock_calls[1][1][0] == \
- 'Galaxy import process has a status of waiting, wait 2 seconds before trying again'
-
-
-def test_publish_with_wait_timeout_failure(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- monkeypatch.setattr(time, 'sleep', MagicMock())
-
- mock_vvv = MagicMock()
- monkeypatch.setattr(Display, 'vvv', mock_vvv)
+ assert mock_publish.call_count == 1
+ assert mock_publish.mock_calls[0][1][0] == artifact_path
- fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-
- artifact_path, mock_open = collection_artifact
-
- first_call = True
-
- def open_value(*args, **kwargs):
- if first_call:
- return StringIO(u'{"task":"%s"}' % fake_import_uri)
- else:
- return StringIO(u'{"finished_at":null}')
+ 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)
- mock_open.side_effect = open_value
-
- expected = "Timeout while waiting for the Galaxy import process to finish, check progress at '%s'" \
- % fake_import_uri
- with pytest.raises(AnsibleError, match=expected):
- collection.publish_collection(artifact_path, galaxy_server, True, 2)
-
- # While the seconds exceed the time we are testing that the exponential backoff gets to 30 and then sits there
- # Because we mock time.sleep() there should be thousands of calls here
- expected_wait_msg = 'Galaxy import process has a status of waiting, wait {0} seconds before trying again'
- assert mock_vvv.call_count > 9
- 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)
-
-
-def test_publish_with_wait_and_failure(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
+def test_publish_with_wait(galaxy_server, collection_artifact, monkeypatch):
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)
-
- fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-
artifact_path, mock_open = collection_artifact
+ fake_import_uri = 'https://galaxy.server.com/api/v2/import/1234'
- import_stat = {
- 'finished_at': 'some_time',
- 'state': 'failed',
- 'error': {
- 'code': 'GW001',
- 'description': 'Because I said so!',
-
- },
- 'messages': [
- {
- 'level': 'error',
- 'message': 'Some error',
- },
- {
- 'level': 'warning',
- 'message': 'Some warning',
- },
- {
- 'level': 'info',
- 'message': 'Some info',
- },
- ],
- }
-
- mock_open.side_effect = (
- StringIO(u'{"task":"%s"}' % fake_import_uri),
- StringIO(to_text(json.dumps(import_stat)))
- )
-
- expected = 'Galaxy import process failed: Because I said so! (Code: GW001)'
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
-
- assert mock_open.call_count == 2
- assert mock_open.mock_calls[1][1][0] == fake_import_uri
- assert mock_open.mock_calls[1][2]['headers']['Authorization'] == 'Token key'
- assert mock_open.mock_calls[1][2]['validate_certs'] is True
- assert mock_open.mock_calls[1][2]['method'] == 'GET'
-
- assert mock_display.call_count == 4
- assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s"\
- % (artifact_path, galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
- % (galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
-
- assert mock_vvv.call_count == 2
- assert mock_vvv.mock_calls[1][1][0] == 'Galaxy import message: info - Some info'
-
- assert mock_warn.call_count == 1
- assert mock_warn.mock_calls[0][1][0] == 'Galaxy import warning message: Some warning'
-
- assert mock_err.call_count == 1
- assert mock_err.mock_calls[0][1][0] == 'Galaxy import error message: Some error'
-
-
-def test_publish_with_wait_and_failure_and_no_error(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v2': '/api/v2'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- 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)
-
- fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
-
- artifact_path, mock_open = collection_artifact
-
- import_stat = {
- 'finished_at': 'some_time',
- 'state': 'failed',
- 'error': {},
- 'messages': [
- {
- 'level': 'error',
- 'message': 'Some error',
- },
- {
- 'level': 'warning',
- 'message': 'Some warning',
- },
- {
- 'level': 'info',
- 'message': 'Some info',
- },
- ],
- }
-
- mock_open.side_effect = (
- StringIO(u'{"task":"%s"}' % fake_import_uri),
- StringIO(to_text(json.dumps(import_stat)))
- )
-
- expected = 'Galaxy import process failed: Unknown error, see %s for more details (Code: UNKNOWN)' % fake_import_uri
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
-
- assert mock_open.call_count == 2
- assert mock_open.mock_calls[1][1][0] == fake_import_uri
- assert mock_open.mock_calls[1][2]['headers']['Authorization'] == 'Token key'
- assert mock_open.mock_calls[1][2]['validate_certs'] is True
- assert mock_open.mock_calls[1][2]['method'] == 'GET'
-
- assert mock_display.call_count == 4
- assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s"\
- % (artifact_path, galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[1][1][0] == 'Collection has been published to the Galaxy server %s %s'\
- % (galaxy_server.name, galaxy_server.api_server)
- assert mock_display.mock_calls[2][1][0] == 'Waiting until Galaxy import task %s has completed' % fake_import_uri
-
- assert mock_vvv.call_count == 2
- assert mock_vvv.mock_calls[1][1][0] == 'Galaxy import message: info - Some info'
-
- assert mock_warn.call_count == 1
- assert mock_warn.mock_calls[0][1][0] == 'Galaxy import warning message: Some warning'
-
- assert mock_err.call_count == 1
- assert mock_err.mock_calls[0][1][0] == 'Galaxy import error message: Some error'
-
-
-def test_publish_failure_v3_with_json_info_409_conflict(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v3': '/api/v3'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
-
- artifact_path, mock_open = collection_artifact
-
- error_response = {
- "errors": [
- {
- "code": "conflict.collection_exists",
- "detail": 'Collection "testing-ansible_testing_content-4.0.4" already exists.',
- "title": "Conflict.",
- "status": "409",
- },
- ]
- }
-
- return_content = StringIO(to_text(json.dumps(error_response)))
- mock_open.side_effect = urllib_error.HTTPError('https://galaxy.server.com', 409, 'msg', {}, return_content)
+ mock_publish = MagicMock()
+ mock_publish.return_value = fake_import_uri
+ monkeypatch.setattr(galaxy_server, 'publish_collection', mock_publish)
- expected = 'Error when publishing collection to test_server (https://galaxy.ansible.com) ' \
- '(HTTP Code: 409, Message: Collection "testing-ansible_testing_content-4.0.4"' \
- ' already exists. Code: conflict.collection_exists)'
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
+ mock_wait = MagicMock()
+ monkeypatch.setattr(galaxy_server, 'wait_import_task', mock_wait)
+ collection.publish_collection(artifact_path, galaxy_server, True, 0)
-def test_publish_failure_v3_with_json_info_multiple_errors(galaxy_server, collection_artifact, monkeypatch):
- mock_avail_ver = MagicMock()
- mock_avail_ver.return_value = {'v3': '/api/v3'}
- monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
+ assert mock_publish.call_count == 1
+ assert mock_publish.mock_calls[0][1][0] == artifact_path
- artifact_path, mock_open = collection_artifact
+ assert mock_wait.call_count == 1
+ assert mock_wait.mock_calls[0][1][0] == fake_import_uri
- error_response = {
- "errors": [
- {
- "code": "conflict.collection_exists",
- "detail": 'Collection "mynamespace-mycollection-4.1.1" already exists.',
- "title": "Conflict.",
- "status": "400",
- },
- {
- "code": "quantum_improbability",
- "title": "Random(?) quantum improbability.",
- "source": {"parameter": "the_arrow_of_time"},
- "meta": {"remediation": "Try again before"}
- },
- ]
- }
-
- return_content = StringIO(to_text(json.dumps(error_response)))
- mock_open.side_effect = urllib_error.HTTPError('https://galaxy.server.com', 400, 'msg', {}, return_content)
-
- expected = 'Error when publishing collection to test_server (https://galaxy.ansible.com) ' \
- '(HTTP Code: 400, Message: Collection "mynamespace-mycollection-4.1.1"' \
- ' already exists. Code: conflict.collection_exists),' \
- ' (HTTP Code: 400, Message: Random(?) quantum improbability. Code: quantum_improbability)'
- with pytest.raises(AnsibleError, match=re.escape(expected)):
- collection.publish_collection(artifact_path, galaxy_server, True, 0)
+ 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):
@@ -962,91 +563,3 @@ def test_extract_tar_file_missing_parent_dir(tmp_tarfile):
collection._extract_tar_file(tfile, filename, output_dir, temp_dir, checksum)
os.path.isfile(output_file)
-
-
-def test_get_available_api_versions_v2_auth_not_required_without_auth(galaxy_server, collection_artifact, monkeypatch):
- # mock_avail_ver = MagicMock()
- # mock_avail_ver.side_effect = {api_version: '/api/%s' % api_version}
- # monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
- response_obj = {
- "description": "GALAXY REST API",
- "current_version": "v1",
- "available_versions": {
- "v1": "/api/v1/",
- "v2": "/api/v2/"
- },
- "server_version": "3.2.4",
- "version_name": "Doin' it Right",
- "team_members": [
- "chouseknecht",
- "cutwater",
- "alikins",
- "newswangerd",
- "awcrosby",
- "tima",
- "gregdek"
- ]
- }
-
- artifact_path, mock_open = collection_artifact
-
- return_content = StringIO(to_text(json.dumps(response_obj)))
- mock_open.return_value = return_content
- res = collection.get_available_api_versions(galaxy_server)
-
- assert res == {'v1': '/api/v1/', 'v2': '/api/v2/'}
-
-
-def test_get_available_api_versions_v3_auth_required_without_auth(galaxy_server, collection_artifact, monkeypatch):
- # mock_avail_ver = MagicMock()
- # mock_avail_ver.side_effect = {api_version: '/api/%s' % api_version}
- # monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
- error_response = {'code': 'unauthorized', 'detail': 'The request was not authorized'}
- artifact_path, mock_open = collection_artifact
-
- return_content = StringIO(to_text(json.dumps(error_response)))
- mock_open.side_effect = urllib_error.HTTPError('https://galaxy.server.com', 401, 'msg', {'WWW-Authenticate': 'Bearer'}, return_content)
- with pytest.raises(AnsibleError):
- collection.get_available_api_versions(galaxy_server)
-
-
-def test_get_available_api_versions_v3_auth_required_with_auth_on_retry(galaxy_server, collection_artifact, monkeypatch):
- # mock_avail_ver = MagicMock()
- # mock_avail_ver.side_effect = {api_version: '/api/%s' % api_version}
- # monkeypatch.setattr(collection, 'get_available_api_versions', mock_avail_ver)
- error_obj = {'code': 'unauthorized', 'detail': 'The request was not authorized'}
- success_obj = {
- "description": "GALAXY REST API",
- "current_version": "v1",
- "available_versions": {
- "v3": "/api/v3/"
- },
- "server_version": "3.2.4",
- "version_name": "Doin' it Right",
- "team_members": [
- "chouseknecht",
- "cutwater",
- "alikins",
- "newswangerd",
- "awcrosby",
- "tima",
- "gregdek"
- ]
- }
-
- artifact_path, mock_open = collection_artifact
-
- error_response = StringIO(to_text(json.dumps(error_obj)))
- success_response = StringIO(to_text(json.dumps(success_obj)))
- mock_open.side_effect = [
- urllib_error.HTTPError('https://galaxy.server.com', 401, 'msg', {'WWW-Authenticate': 'Bearer'}, error_response),
- success_response,
- ]
-
- try:
- res = collection.get_available_api_versions(galaxy_server)
- except AnsibleError as err:
- print(err)
- raise
-
- assert res == {'v3': '/api/v3/'}