e747487720
* ansible-galaxy: support multiple servers on install * Added docs for the server configuration file * Fix up doc string for requirements file format * Fix bugs after testing * Fix kwarg doc and added version * Fix typo and doc improvement * Fix base64 encoding and allow --server to override list
879 lines
34 KiB
Python
879 lines
34 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright: (c) 2019, Ansible Project
|
|
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
|
|
|
# Make coding more python3-ish
|
|
from __future__ import (absolute_import, division, print_function)
|
|
__metaclass__ = type
|
|
|
|
import json
|
|
import os
|
|
import pytest
|
|
import re
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
|
|
from hashlib import sha256
|
|
from io import BytesIO, StringIO
|
|
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
|
|
from ansible.galaxy import api, collection, token
|
|
from ansible.module_utils._text import to_bytes, to_native, to_text
|
|
from ansible.utils import context_objects as co
|
|
from ansible.utils.display import Display
|
|
from ansible.utils.hashing import secure_hash_s
|
|
|
|
|
|
@pytest.fixture(autouse='function')
|
|
def reset_cli_args():
|
|
co.GlobalCLIArgs._Singleton__instance = None
|
|
yield
|
|
co.GlobalCLIArgs._Singleton__instance = None
|
|
|
|
|
|
@pytest.fixture()
|
|
def collection_input(tmp_path_factory):
|
|
''' Creates a collection skeleton directory for build tests '''
|
|
test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Input'))
|
|
namespace = 'ansible_namespace'
|
|
collection = 'collection'
|
|
skeleton = os.path.join(os.path.dirname(os.path.split(__file__)[0]), 'cli', 'test_data', 'collection_skeleton')
|
|
|
|
galaxy_args = ['ansible-galaxy', 'collection', 'init', '%s.%s' % (namespace, collection),
|
|
'-c', '--init-path', test_dir, '--collection-skeleton', skeleton]
|
|
GalaxyCLI(args=galaxy_args).run()
|
|
collection_dir = os.path.join(test_dir, namespace, collection)
|
|
output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Output'))
|
|
|
|
return collection_dir, output_dir
|
|
|
|
|
|
@pytest.fixture()
|
|
def collection_artifact(monkeypatch, tmp_path_factory):
|
|
''' Creates a temp collection artifact and mocked open_url instance for publishing tests '''
|
|
mock_open = MagicMock()
|
|
monkeypatch.setattr(collection, 'open_url', mock_open)
|
|
|
|
mock_uuid = MagicMock()
|
|
mock_uuid.return_value.hex = 'uuid'
|
|
monkeypatch.setattr(uuid, 'uuid4', mock_uuid)
|
|
|
|
tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')
|
|
input_file = to_text(tmp_path / 'collection.tar.gz')
|
|
|
|
with tarfile.open(input_file, 'w:gz') as tfile:
|
|
b_io = BytesIO(b"\x00\x01\x02\x03")
|
|
tar_info = tarfile.TarInfo('test')
|
|
tar_info.size = 4
|
|
tar_info.mode = 0o0644
|
|
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
|
|
|
|
return input_file, mock_open
|
|
|
|
|
|
@pytest.fixture()
|
|
def galaxy_yml(request, tmp_path_factory):
|
|
b_test_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections'))
|
|
b_galaxy_yml = os.path.join(b_test_dir, b'galaxy.yml')
|
|
with open(b_galaxy_yml, 'wb') as galaxy_obj:
|
|
galaxy_obj.write(to_bytes(request.param))
|
|
|
|
yield b_galaxy_yml
|
|
|
|
|
|
@pytest.fixture()
|
|
def tmp_tarfile(tmp_path_factory):
|
|
''' Creates a temporary tar file for _extract_tar_file tests '''
|
|
filename = u'ÅÑŚÌβŁÈ'
|
|
temp_dir = to_bytes(tmp_path_factory.mktemp('test-%s Collections' % to_native(filename)))
|
|
tar_file = os.path.join(temp_dir, to_bytes('%s.tar.gz' % filename))
|
|
data = os.urandom(8)
|
|
|
|
with tarfile.open(tar_file, 'w:gz') as tfile:
|
|
b_io = BytesIO(data)
|
|
tar_info = tarfile.TarInfo(filename)
|
|
tar_info.size = len(data)
|
|
tar_info.mode = 0o0644
|
|
tfile.addfile(tarinfo=tar_info, fileobj=b_io)
|
|
|
|
sha256_hash = sha256()
|
|
sha256_hash.update(data)
|
|
|
|
with tarfile.open(tar_file, 'r') as tfile:
|
|
yield temp_dir, tfile, filename, sha256_hash.hexdigest()
|
|
|
|
|
|
@pytest.fixture()
|
|
def galaxy_server():
|
|
context.CLIARGS._store = {'ignore_certs': False}
|
|
galaxy_api = api.GalaxyAPI(None, 'test_server', 'https://galaxy.ansible.com',
|
|
token=token.GalaxyToken(token='key'))
|
|
return galaxy_api
|
|
|
|
|
|
def test_build_collection_no_galaxy_yaml():
|
|
fake_path = u'/fake/ÅÑŚÌβŁÈ/path'
|
|
expected = to_native("The collection galaxy.yml path '%s/galaxy.yml' does not exist." % fake_path)
|
|
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection.build_collection(fake_path, 'output', False)
|
|
|
|
|
|
def test_build_existing_output_file(collection_input):
|
|
input_dir, output_dir = collection_input
|
|
|
|
existing_output_dir = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz')
|
|
os.makedirs(existing_output_dir)
|
|
|
|
expected = "The output collection artifact '%s' already exists, but is a directory - aborting" \
|
|
% to_native(existing_output_dir)
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection.build_collection(input_dir, output_dir, False)
|
|
|
|
|
|
def test_build_existing_output_without_force(collection_input):
|
|
input_dir, output_dir = collection_input
|
|
|
|
existing_output = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz')
|
|
with open(existing_output, 'w+') as out_file:
|
|
out_file.write("random garbage")
|
|
out_file.flush()
|
|
|
|
expected = "The file '%s' already exists. You can use --force to re-create the collection artifact." \
|
|
% to_native(existing_output)
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection.build_collection(input_dir, output_dir, False)
|
|
|
|
|
|
def test_build_existing_output_with_force(collection_input):
|
|
input_dir, output_dir = collection_input
|
|
|
|
existing_output = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz')
|
|
with open(existing_output, 'w+') as out_file:
|
|
out_file.write("random garbage")
|
|
out_file.flush()
|
|
|
|
collection.build_collection(input_dir, output_dir, True)
|
|
|
|
# Verify the file was replaced with an actual tar file
|
|
assert tarfile.is_tarfile(existing_output)
|
|
|
|
|
|
@pytest.mark.parametrize('galaxy_yml', [b'namespace: value: broken'], indirect=True)
|
|
def test_invalid_yaml_galaxy_file(galaxy_yml):
|
|
expected = to_native(b"Failed to parse the galaxy.yml at '%s' with the following error:" % galaxy_yml)
|
|
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection._get_galaxy_yml(galaxy_yml)
|
|
|
|
|
|
@pytest.mark.parametrize('galaxy_yml', [b'namespace: test_namespace'], indirect=True)
|
|
def test_missing_required_galaxy_key(galaxy_yml):
|
|
expected = "The collection galaxy.yml at '%s' is missing the following mandatory keys: authors, name, " \
|
|
"readme, version" % to_native(galaxy_yml)
|
|
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection._get_galaxy_yml(galaxy_yml)
|
|
|
|
|
|
@pytest.mark.parametrize('galaxy_yml', [b"""
|
|
namespace: namespace
|
|
name: collection
|
|
authors: Jordan
|
|
version: 0.1.0
|
|
readme: README.md
|
|
invalid: value"""], indirect=True)
|
|
def test_warning_extra_keys(galaxy_yml, monkeypatch):
|
|
display_mock = MagicMock()
|
|
monkeypatch.setattr(Display, 'warning', display_mock)
|
|
|
|
collection._get_galaxy_yml(galaxy_yml)
|
|
|
|
assert display_mock.call_count == 1
|
|
assert display_mock.call_args[0][0] == "Found unknown keys in collection galaxy.yml at '%s': invalid"\
|
|
% to_text(galaxy_yml)
|
|
|
|
|
|
@pytest.mark.parametrize('galaxy_yml', [b"""
|
|
namespace: namespace
|
|
name: collection
|
|
authors: Jordan
|
|
version: 0.1.0
|
|
readme: README.md"""], indirect=True)
|
|
def test_defaults_galaxy_yml(galaxy_yml):
|
|
actual = collection._get_galaxy_yml(galaxy_yml)
|
|
|
|
assert actual['namespace'] == 'namespace'
|
|
assert actual['name'] == 'collection'
|
|
assert actual['authors'] == ['Jordan']
|
|
assert actual['version'] == '0.1.0'
|
|
assert actual['readme'] == 'README.md'
|
|
assert actual['description'] is None
|
|
assert actual['repository'] is None
|
|
assert actual['documentation'] is None
|
|
assert actual['homepage'] is None
|
|
assert actual['issues'] is None
|
|
assert actual['tags'] == []
|
|
assert actual['dependencies'] == {}
|
|
assert actual['license_ids'] == []
|
|
|
|
|
|
@pytest.mark.parametrize('galaxy_yml', [(b"""
|
|
namespace: namespace
|
|
name: collection
|
|
authors: Jordan
|
|
version: 0.1.0
|
|
readme: README.md
|
|
license: MIT"""), (b"""
|
|
namespace: namespace
|
|
name: collection
|
|
authors: Jordan
|
|
version: 0.1.0
|
|
readme: README.md
|
|
license:
|
|
- MIT""")], indirect=True)
|
|
def test_galaxy_yml_list_value(galaxy_yml):
|
|
actual = collection._get_galaxy_yml(galaxy_yml)
|
|
assert actual['license_ids'] == ['MIT']
|
|
|
|
|
|
def test_build_ignore_files_and_folders(collection_input, monkeypatch):
|
|
input_dir = collection_input[0]
|
|
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'vvv', mock_display)
|
|
|
|
git_folder = os.path.join(input_dir, '.git')
|
|
retry_file = os.path.join(input_dir, 'ansible.retry')
|
|
|
|
os.makedirs(git_folder)
|
|
with open(retry_file, 'w+') as ignore_file:
|
|
ignore_file.write('random')
|
|
ignore_file.flush()
|
|
|
|
actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection')
|
|
|
|
assert actual['format'] == 1
|
|
for manifest_entry in actual['files']:
|
|
assert manifest_entry['name'] not in ['.git', 'ansible.retry', 'galaxy.yml']
|
|
|
|
expected_msgs = [
|
|
"Skipping '%s' for collection build" % to_text(retry_file),
|
|
"Skipping '%s' for collection build" % to_text(git_folder),
|
|
]
|
|
assert mock_display.call_count == 2
|
|
assert mock_display.mock_calls[0][1][0] in expected_msgs
|
|
assert mock_display.mock_calls[1][1][0] in expected_msgs
|
|
|
|
|
|
def test_build_ignore_older_release_in_root(collection_input, monkeypatch):
|
|
input_dir = collection_input[0]
|
|
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'vvv', mock_display)
|
|
|
|
# This is expected to be ignored because it is in the root collection dir.
|
|
release_file = os.path.join(input_dir, 'namespace-collection-0.0.0.tar.gz')
|
|
|
|
# This is not expected to be ignored because it is not in the root collection dir.
|
|
fake_release_file = os.path.join(input_dir, 'plugins', 'namespace-collection-0.0.0.tar.gz')
|
|
|
|
for filename in [release_file, fake_release_file]:
|
|
with open(filename, 'w+') as file_obj:
|
|
file_obj.write('random')
|
|
file_obj.flush()
|
|
|
|
actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection')
|
|
assert actual['format'] == 1
|
|
|
|
plugin_release_found = False
|
|
for manifest_entry in actual['files']:
|
|
assert manifest_entry['name'] != 'namespace-collection-0.0.0.tar.gz'
|
|
if manifest_entry['name'] == 'plugins/namespace-collection-0.0.0.tar.gz':
|
|
plugin_release_found = True
|
|
|
|
assert plugin_release_found
|
|
|
|
assert mock_display.call_count == 1
|
|
assert mock_display.mock_calls[0][1][0] == "Skipping '%s' for collection build" % to_text(release_file)
|
|
|
|
|
|
def test_build_ignore_symlink_target_outside_collection(collection_input, monkeypatch):
|
|
input_dir, outside_dir = collection_input
|
|
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'warning', mock_display)
|
|
|
|
link_path = os.path.join(input_dir, 'plugins', 'connection')
|
|
os.symlink(outside_dir, link_path)
|
|
|
|
actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection')
|
|
for manifest_entry in actual['files']:
|
|
assert manifest_entry['name'] != 'plugins/connection'
|
|
|
|
assert mock_display.call_count == 1
|
|
assert mock_display.mock_calls[0][1][0] == "Skipping '%s' as it is a symbolic link to a directory outside " \
|
|
"the collection" % to_text(link_path)
|
|
|
|
|
|
def test_build_copy_symlink_target_inside_collection(collection_input):
|
|
input_dir = collection_input[0]
|
|
|
|
os.makedirs(os.path.join(input_dir, 'playbooks', 'roles'))
|
|
roles_link = os.path.join(input_dir, 'playbooks', 'roles', 'linked')
|
|
|
|
roles_target = os.path.join(input_dir, 'roles', 'linked')
|
|
roles_target_tasks = os.path.join(roles_target, 'tasks')
|
|
os.makedirs(roles_target_tasks)
|
|
with open(os.path.join(roles_target_tasks, 'main.yml'), 'w+') as tasks_main:
|
|
tasks_main.write("---\n- hosts: localhost\n tasks:\n - ping:")
|
|
tasks_main.flush()
|
|
|
|
os.symlink(roles_target, roles_link)
|
|
|
|
actual = collection._build_files_manifest(to_bytes(input_dir), 'namespace', 'collection')
|
|
|
|
linked_entries = [e for e in actual['files'] if e['name'].startswith('playbooks/roles/linked')]
|
|
assert len(linked_entries) == 3
|
|
assert linked_entries[0]['name'] == 'playbooks/roles/linked'
|
|
assert linked_entries[0]['ftype'] == 'dir'
|
|
assert linked_entries[1]['name'] == 'playbooks/roles/linked/tasks'
|
|
assert linked_entries[1]['ftype'] == 'dir'
|
|
assert linked_entries[2]['name'] == 'playbooks/roles/linked/tasks/main.yml'
|
|
assert linked_entries[2]['ftype'] == 'file'
|
|
assert linked_entries[2]['chksum_sha256'] == '9c97a1633c51796999284c62236b8d5462903664640079b80c37bf50080fcbc3'
|
|
|
|
|
|
def test_build_with_symlink_inside_collection(collection_input):
|
|
input_dir, output_dir = collection_input
|
|
|
|
os.makedirs(os.path.join(input_dir, 'playbooks', 'roles'))
|
|
roles_link = os.path.join(input_dir, 'playbooks', 'roles', 'linked')
|
|
file_link = os.path.join(input_dir, 'docs', 'README.md')
|
|
|
|
roles_target = os.path.join(input_dir, 'roles', 'linked')
|
|
roles_target_tasks = os.path.join(roles_target, 'tasks')
|
|
os.makedirs(roles_target_tasks)
|
|
with open(os.path.join(roles_target_tasks, 'main.yml'), 'w+') as tasks_main:
|
|
tasks_main.write("---\n- hosts: localhost\n tasks:\n - ping:")
|
|
tasks_main.flush()
|
|
|
|
os.symlink(roles_target, roles_link)
|
|
os.symlink(os.path.join(input_dir, 'README.md'), file_link)
|
|
|
|
collection.build_collection(input_dir, output_dir, False)
|
|
|
|
output_artifact = os.path.join(output_dir, 'ansible_namespace-collection-0.1.0.tar.gz')
|
|
assert tarfile.is_tarfile(output_artifact)
|
|
|
|
with tarfile.open(output_artifact, mode='r') as actual:
|
|
members = actual.getmembers()
|
|
|
|
linked_members = [m for m in members if m.path.startswith('playbooks/roles/linked/tasks')]
|
|
assert len(linked_members) == 2
|
|
assert linked_members[0].name == 'playbooks/roles/linked/tasks'
|
|
assert linked_members[0].isdir()
|
|
|
|
assert linked_members[1].name == 'playbooks/roles/linked/tasks/main.yml'
|
|
assert linked_members[1].isreg()
|
|
|
|
linked_task = actual.extractfile(linked_members[1].name)
|
|
actual_task = secure_hash_s(linked_task.read())
|
|
linked_task.close()
|
|
|
|
assert actual_task == 'f4dcc52576b6c2cd8ac2832c52493881c4e54226'
|
|
|
|
linked_file = [m for m in members if m.path == 'docs/README.md']
|
|
assert len(linked_file) == 1
|
|
assert linked_file[0].isreg()
|
|
|
|
linked_file_obj = actual.extractfile(linked_file[0].name)
|
|
actual_file = secure_hash_s(linked_file_obj.read())
|
|
linked_file_obj.close()
|
|
|
|
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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def test_publish_no_wait(galaxy_server, collection_artifact, monkeypatch):
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'display', mock_display)
|
|
|
|
artifact_path, mock_open = collection_artifact
|
|
fake_import_uri = 'https://galaxy.server.com/api/v2/import/1234'
|
|
|
|
mock_open.return_value = StringIO(u'{"task":"%s"}' % fake_import_uri)
|
|
expected_form, expected_content_type = collection._get_mime_data(to_bytes(artifact_path))
|
|
|
|
collection.publish_collection(artifact_path, galaxy_server, False)
|
|
|
|
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, not waiting until import has completed due to --no-wait " \
|
|
"being set. Import task results can be found at %s" % fake_import_uri
|
|
|
|
|
|
def test_publish_dont_validate_cert(galaxy_server, collection_artifact):
|
|
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)
|
|
|
|
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):
|
|
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 (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)
|
|
|
|
|
|
def test_publish_failure_with_json_info(galaxy_server, collection_artifact):
|
|
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 (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)
|
|
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
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 == 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 successfully published to the Galaxy server'
|
|
|
|
assert mock_vvv.call_count == 3
|
|
assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
|
|
% (galaxy_server.name, galaxy_server.api_server)
|
|
assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
|
|
|
|
|
|
def test_publish_with_wait_timeout(collection_artifact, monkeypatch):
|
|
monkeypatch.setattr(time, 'sleep', MagicMock())
|
|
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'display', mock_display)
|
|
|
|
mock_vvv = MagicMock()
|
|
monkeypatch.setattr(Display, 'vvv', mock_vvv)
|
|
|
|
fake_import_uri = 'https://galaxy-server/api/v2/import/1234'
|
|
server = 'https://galaxy.server.com'
|
|
|
|
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, server, 'key', True, True)
|
|
|
|
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'] == 'Token key'
|
|
assert mock_open.mock_calls[1][2]['validate_certs'] is False
|
|
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'] == 'Token key'
|
|
assert mock_open.mock_calls[2][2]['validate_certs'] is False
|
|
assert mock_open.mock_calls[2][2]['method'] == 'GET'
|
|
|
|
assert mock_display.call_count == 2
|
|
assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s" % (artifact_path, server)
|
|
assert mock_display.mock_calls[1][1][0] == 'Collection has been successfully published to the Galaxy server'
|
|
|
|
assert mock_vvv.call_count == 3
|
|
assert mock_vvv.mock_calls[0][1][0] == 'Collection has been pushed to the Galaxy server %s' % server
|
|
assert mock_vvv.mock_calls[1][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
|
|
assert mock_vvv.mock_calls[2][1][0] == \
|
|
'Galaxy import process has a status of waiting, wait 2 seconds before trying again'
|
|
|
|
|
|
def test_publish_with_wait_timeout(galaxy_server, collection_artifact, monkeypatch):
|
|
galaxy_server.validate_certs = False
|
|
|
|
monkeypatch.setattr(time, 'sleep', MagicMock())
|
|
|
|
mock_display = MagicMock()
|
|
monkeypatch.setattr(Display, 'display', mock_display)
|
|
|
|
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":null}'),
|
|
StringIO(u'{"finished_at":null}'),
|
|
StringIO(u'{"finished_at":null}'),
|
|
StringIO(u'{"finished_at":null}'),
|
|
StringIO(u'{"finished_at":null}'),
|
|
StringIO(u'{"finished_at":null}'),
|
|
)
|
|
|
|
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)
|
|
|
|
assert mock_open.call_count == 8
|
|
for i in range(7):
|
|
mock_call = mock_open.mock_calls[i + 1]
|
|
assert mock_call[1][0] == fake_import_uri
|
|
assert mock_call[2]['headers']['Authorization'] == 'Token key'
|
|
assert mock_call[2]['validate_certs'] is False
|
|
assert mock_call[2]['method'] == 'GET'
|
|
|
|
assert mock_display.call_count == 1
|
|
assert mock_display.mock_calls[0][1][0] == "Publishing collection artifact '%s' to %s %s" \
|
|
% (artifact_path, galaxy_server.name, galaxy_server.api_server)
|
|
|
|
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] == 'Collection has been pushed to the Galaxy server %s %s' \
|
|
% (galaxy_server.name, galaxy_server.api_server)
|
|
assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
|
|
assert mock_vvv.mock_calls[3][1][0] == expected_wait_msg.format(2)
|
|
assert mock_vvv.mock_calls[4][1][0] == expected_wait_msg.format(3)
|
|
assert mock_vvv.mock_calls[5][1][0] == expected_wait_msg.format(4)
|
|
assert mock_vvv.mock_calls[6][1][0] == expected_wait_msg.format(6)
|
|
assert mock_vvv.mock_calls[7][1][0] == expected_wait_msg.format(10)
|
|
assert mock_vvv.mock_calls[8][1][0] == expected_wait_msg.format(15)
|
|
|
|
|
|
def test_publish_with_wait_and_failure(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
|
|
|
|
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)
|
|
|
|
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 == 1
|
|
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_vvv.call_count == 4
|
|
assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
|
|
% (galaxy_server.name, galaxy_server.api_server)
|
|
assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
|
|
assert mock_vvv.mock_calls[3][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_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)
|
|
|
|
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 == 1
|
|
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_vvv.call_count == 4
|
|
assert mock_vvv.mock_calls[1][1][0] == 'Collection has been pushed to the Galaxy server %s %s' \
|
|
% (galaxy_server.name, galaxy_server.api_server)
|
|
assert mock_vvv.mock_calls[2][1][0] == 'Waiting until galaxy import task %s has completed' % fake_import_uri
|
|
assert mock_vvv.mock_calls[3][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_find_existing_collections(tmp_path_factory, monkeypatch):
|
|
test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections'))
|
|
collection1 = os.path.join(test_dir, 'namespace1', 'collection1')
|
|
collection2 = os.path.join(test_dir, 'namespace2', 'collection2')
|
|
fake_collection1 = os.path.join(test_dir, 'namespace3', 'collection3')
|
|
fake_collection2 = os.path.join(test_dir, 'namespace4')
|
|
os.makedirs(collection1)
|
|
os.makedirs(collection2)
|
|
os.makedirs(os.path.split(fake_collection1)[0])
|
|
|
|
open(fake_collection1, 'wb+').close()
|
|
open(fake_collection2, 'wb+').close()
|
|
|
|
collection1_manifest = json.dumps({
|
|
'collection_info': {
|
|
'namespace': 'namespace1',
|
|
'name': 'collection1',
|
|
'version': '1.2.3',
|
|
'authors': ['Jordan Borean'],
|
|
'readme': 'README.md',
|
|
'dependencies': {},
|
|
},
|
|
'format': 1,
|
|
})
|
|
with open(os.path.join(collection1, 'MANIFEST.json'), 'wb') as manifest_obj:
|
|
manifest_obj.write(to_bytes(collection1_manifest))
|
|
|
|
mock_warning = MagicMock()
|
|
monkeypatch.setattr(Display, 'warning', mock_warning)
|
|
|
|
actual = collection._find_existing_collections(test_dir)
|
|
|
|
assert len(actual) == 2
|
|
for actual_collection in actual:
|
|
assert actual_collection.skip is True
|
|
|
|
if str(actual_collection) == 'namespace1.collection1':
|
|
assert actual_collection.namespace == 'namespace1'
|
|
assert actual_collection.name == 'collection1'
|
|
assert actual_collection.b_path == to_bytes(collection1)
|
|
assert actual_collection.api is None
|
|
assert actual_collection.versions == set(['1.2.3'])
|
|
assert actual_collection.latest_version == '1.2.3'
|
|
assert actual_collection.dependencies == {}
|
|
else:
|
|
assert actual_collection.namespace == 'namespace2'
|
|
assert actual_collection.name == 'collection2'
|
|
assert actual_collection.b_path == to_bytes(collection2)
|
|
assert actual_collection.api is None
|
|
assert actual_collection.versions == set(['*'])
|
|
assert actual_collection.latest_version == '*'
|
|
assert actual_collection.dependencies == {}
|
|
|
|
assert mock_warning.call_count == 1
|
|
assert mock_warning.mock_calls[0][1][0] == "Collection at '%s' does not have a MANIFEST.json file, cannot " \
|
|
"detect version." % to_text(collection2)
|
|
|
|
|
|
def test_download_file(tmp_path_factory, monkeypatch):
|
|
temp_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections'))
|
|
|
|
data = b"\x00\x01\x02\x03"
|
|
sha256_hash = sha256()
|
|
sha256_hash.update(data)
|
|
|
|
mock_open = MagicMock()
|
|
mock_open.return_value = BytesIO(data)
|
|
monkeypatch.setattr(collection, 'open_url', mock_open)
|
|
|
|
expected = os.path.join(temp_dir, b'file')
|
|
actual = collection._download_file('http://google.com/file', temp_dir, sha256_hash.hexdigest(), True)
|
|
|
|
assert actual.startswith(expected)
|
|
assert os.path.isfile(actual)
|
|
with open(actual, 'rb') as file_obj:
|
|
assert file_obj.read() == data
|
|
|
|
assert mock_open.call_count == 1
|
|
assert mock_open.mock_calls[0][1][0] == 'http://google.com/file'
|
|
|
|
|
|
def test_download_file_hash_mismatch(tmp_path_factory, monkeypatch):
|
|
temp_dir = to_bytes(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections'))
|
|
|
|
data = b"\x00\x01\x02\x03"
|
|
|
|
mock_open = MagicMock()
|
|
mock_open.return_value = BytesIO(data)
|
|
monkeypatch.setattr(collection, 'open_url', mock_open)
|
|
|
|
expected = "Mismatch artifact hash with downloaded file"
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection._download_file('http://google.com/file', temp_dir, 'bad', True)
|
|
|
|
|
|
def test_extract_tar_file_invalid_hash(tmp_tarfile):
|
|
temp_dir, tfile, filename, dummy = tmp_tarfile
|
|
|
|
expected = "Checksum mismatch for '%s' inside collection at '%s'" % (to_native(filename), to_native(tfile.name))
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection._extract_tar_file(tfile, filename, temp_dir, temp_dir, "fakehash")
|
|
|
|
|
|
def test_extract_tar_file_missing_member(tmp_tarfile):
|
|
temp_dir, tfile, dummy, dummy = tmp_tarfile
|
|
|
|
expected = "Collection tar at '%s' does not contain the expected file 'missing'." % to_native(tfile.name)
|
|
with pytest.raises(AnsibleError, match=expected):
|
|
collection._extract_tar_file(tfile, 'missing', temp_dir, temp_dir)
|
|
|
|
|
|
def test_extract_tar_file_missing_parent_dir(tmp_tarfile):
|
|
temp_dir, tfile, filename, checksum = tmp_tarfile
|
|
output_dir = os.path.join(temp_dir, b'output')
|
|
output_file = os.path.join(output_dir, to_bytes(filename))
|
|
|
|
collection._extract_tar_file(tfile, filename, output_dir, temp_dir, checksum)
|
|
os.path.isfile(output_file)
|