ansible/test/units/galaxy/test_api.py

# -*- 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.common.text.converters 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)

    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