diff options
Diffstat (limited to 'test/units/cli')
44 files changed, 3011 insertions, 0 deletions
diff --git a/test/units/cli/__init__.py b/test/units/cli/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/__init__.py diff --git a/test/units/cli/arguments/test_optparse_helpers.py b/test/units/cli/arguments/test_optparse_helpers.py new file mode 100644 index 0000000..082c9be --- /dev/null +++ b/test/units/cli/arguments/test_optparse_helpers.py @@ -0,0 +1,38 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2018, Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import sys + +import pytest + +from ansible import constants as C +from ansible.cli.arguments import option_helpers as opt_help +from ansible import __path__ as ansible_path +from ansible.release import __version__ as ansible_version + +if C.DEFAULT_MODULE_PATH is None: + cpath = u'Default w/o overrides' +else: + cpath = C.DEFAULT_MODULE_PATH + +FAKE_PROG = u'ansible-cli-test' +VERSION_OUTPUT = opt_help.version(prog=FAKE_PROG) + + +@pytest.mark.parametrize( + 'must_have', [ + FAKE_PROG + u' [core %s]' % ansible_version, + u'config file = %s' % C.CONFIG_FILE, + u'configured module search path = %s' % cpath, + u'ansible python module location = %s' % ':'.join(ansible_path), + u'ansible collection location = %s' % ':'.join(C.COLLECTIONS_PATHS), + u'executable location = ', + u'python version = %s' % ''.join(sys.version.splitlines()), + ] +) +def test_option_helper_version(must_have): + assert must_have in VERSION_OUTPUT diff --git a/test/units/cli/galaxy/test_collection_extract_tar.py b/test/units/cli/galaxy/test_collection_extract_tar.py new file mode 100644 index 0000000..526442c --- /dev/null +++ b/test/units/cli/galaxy/test_collection_extract_tar.py @@ -0,0 +1,61 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import pytest + +from ansible.errors import AnsibleError +from ansible.galaxy.collection import _extract_tar_dir + + +@pytest.fixture +def fake_tar_obj(mocker): + m_tarfile = mocker.Mock() + m_tarfile.type = mocker.Mock(return_value=b'99') + m_tarfile.SYMTYPE = mocker.Mock(return_value=b'22') + + return m_tarfile + + +def test_extract_tar_member_trailing_sep(mocker): + m_tarfile = mocker.Mock() + m_tarfile.getmember = mocker.Mock(side_effect=KeyError) + + with pytest.raises(AnsibleError, match='Unable to extract'): + _extract_tar_dir(m_tarfile, '/some/dir/', b'/some/dest') + + assert m_tarfile.getmember.call_count == 1 + + +def test_extract_tar_member_no_trailing_sep(mocker): + m_tarfile = mocker.Mock() + m_tarfile.getmember = mocker.Mock(side_effect=KeyError) + + with pytest.raises(AnsibleError, match='Unable to extract'): + _extract_tar_dir(m_tarfile, '/some/dir', b'/some/dest') + + assert m_tarfile.getmember.call_count == 2 + + +def test_extract_tar_dir_exists(mocker, fake_tar_obj): + mocker.patch('os.makedirs', return_value=None) + m_makedir = mocker.patch('os.mkdir', return_value=None) + mocker.patch('os.path.isdir', return_value=True) + + _extract_tar_dir(fake_tar_obj, '/some/dir', b'/some/dest') + + assert not m_makedir.called + + +def test_extract_tar_dir_does_not_exist(mocker, fake_tar_obj): + mocker.patch('os.makedirs', return_value=None) + m_makedir = mocker.patch('os.mkdir', return_value=None) + mocker.patch('os.path.isdir', return_value=False) + + _extract_tar_dir(fake_tar_obj, '/some/dir', b'/some/dest') + + assert m_makedir.called + assert m_makedir.call_args[0] == (b'/some/dir', 0o0755) diff --git a/test/units/cli/galaxy/test_display_collection.py b/test/units/cli/galaxy/test_display_collection.py new file mode 100644 index 0000000..c86227b --- /dev/null +++ b/test/units/cli/galaxy/test_display_collection.py @@ -0,0 +1,46 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import pytest + +from ansible.cli.galaxy import _display_collection +from ansible.galaxy.dependency_resolution.dataclasses import Requirement + + +@pytest.fixture +def collection_object(): + def _cobj(fqcn='sandwiches.ham'): + return Requirement(fqcn, '1.5.0', None, 'galaxy', None) + return _cobj + + +def test_display_collection(capsys, collection_object): + _display_collection(collection_object()) + out, err = capsys.readouterr() + + assert out == 'sandwiches.ham 1.5.0 \n' + + +def test_display_collections_small_max_widths(capsys, collection_object): + _display_collection(collection_object(), 1, 1) + out, err = capsys.readouterr() + + assert out == 'sandwiches.ham 1.5.0 \n' + + +def test_display_collections_large_max_widths(capsys, collection_object): + _display_collection(collection_object(), 20, 20) + out, err = capsys.readouterr() + + assert out == 'sandwiches.ham 1.5.0 \n' + + +def test_display_collection_small_minimum_widths(capsys, collection_object): + _display_collection(collection_object('a.b'), min_cwidth=0, min_vwidth=0) + out, err = capsys.readouterr() + + assert out == 'a.b 1.5.0 \n' diff --git a/test/units/cli/galaxy/test_display_header.py b/test/units/cli/galaxy/test_display_header.py new file mode 100644 index 0000000..ae926b0 --- /dev/null +++ b/test/units/cli/galaxy/test_display_header.py @@ -0,0 +1,41 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +from ansible.cli.galaxy import _display_header + + +def test_display_header_default(capsys): + _display_header('/collections/path', 'h1', 'h2') + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '' + assert out_lines[1] == '# /collections/path' + assert out_lines[2] == 'h1 h2 ' + assert out_lines[3] == '---------- -------' + + +def test_display_header_widths(capsys): + _display_header('/collections/path', 'Collection', 'Version', 18, 18) + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '' + assert out_lines[1] == '# /collections/path' + assert out_lines[2] == 'Collection Version ' + assert out_lines[3] == '------------------ ------------------' + + +def test_display_header_small_widths(capsys): + _display_header('/collections/path', 'Col', 'Ver', 1, 1) + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '' + assert out_lines[1] == '# /collections/path' + assert out_lines[2] == 'Col Ver' + assert out_lines[3] == '--- ---' diff --git a/test/units/cli/galaxy/test_display_role.py b/test/units/cli/galaxy/test_display_role.py new file mode 100644 index 0000000..e23a772 --- /dev/null +++ b/test/units/cli/galaxy/test_display_role.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +from ansible.cli.galaxy import _display_role + + +def test_display_role(mocker, capsys): + mocked_galaxy_role = mocker.Mock(install_info=None) + mocked_galaxy_role.name = 'testrole' + _display_role(mocked_galaxy_role) + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '- testrole, (unknown version)' + + +def test_display_role_known_version(mocker, capsys): + mocked_galaxy_role = mocker.Mock(install_info={'version': '1.0.0'}) + mocked_galaxy_role.name = 'testrole' + _display_role(mocked_galaxy_role) + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '- testrole, 1.0.0' diff --git a/test/units/cli/galaxy/test_execute_list.py b/test/units/cli/galaxy/test_execute_list.py new file mode 100644 index 0000000..41fee0b --- /dev/null +++ b/test/units/cli/galaxy/test_execute_list.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import pytest + +from ansible import context +from ansible.cli.galaxy import GalaxyCLI + + +def test_execute_list_role_called(mocker): + """Make sure the correct method is called for a role""" + + gc = GalaxyCLI(['ansible-galaxy', 'role', 'list']) + context.CLIARGS._store = {'type': 'role'} + execute_list_role_mock = mocker.patch('ansible.cli.galaxy.GalaxyCLI.execute_list_role', side_effect=AttributeError('raised intentionally')) + execute_list_collection_mock = mocker.patch('ansible.cli.galaxy.GalaxyCLI.execute_list_collection', side_effect=AttributeError('raised intentionally')) + with pytest.raises(AttributeError): + gc.execute_list() + + assert execute_list_role_mock.call_count == 1 + assert execute_list_collection_mock.call_count == 0 + + +def test_execute_list_collection_called(mocker): + """Make sure the correct method is called for a collection""" + + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list']) + context.CLIARGS._store = {'type': 'collection'} + execute_list_role_mock = mocker.patch('ansible.cli.galaxy.GalaxyCLI.execute_list_role', side_effect=AttributeError('raised intentionally')) + execute_list_collection_mock = mocker.patch('ansible.cli.galaxy.GalaxyCLI.execute_list_collection', side_effect=AttributeError('raised intentionally')) + with pytest.raises(AttributeError): + gc.execute_list() + + assert execute_list_role_mock.call_count == 0 + assert execute_list_collection_mock.call_count == 1 diff --git a/test/units/cli/galaxy/test_execute_list_collection.py b/test/units/cli/galaxy/test_execute_list_collection.py new file mode 100644 index 0000000..e8a834d --- /dev/null +++ b/test/units/cli/galaxy/test_execute_list_collection.py @@ -0,0 +1,284 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import pytest + +from ansible import context +from ansible.cli.galaxy import GalaxyCLI +from ansible.errors import AnsibleError, AnsibleOptionsError +from ansible.galaxy import collection +from ansible.galaxy.dependency_resolution.dataclasses import Requirement +from ansible.module_utils._text import to_native + + +def path_exists(path): + if to_native(path) == '/root/.ansible/collections/ansible_collections/sandwiches/ham': + return False + elif to_native(path) == '/usr/share/ansible/collections/ansible_collections/sandwiches/reuben': + return False + elif to_native(path) == 'nope': + return False + else: + return True + + +def isdir(path): + if to_native(path) == 'nope': + return False + else: + return True + + +def cliargs(collections_paths=None, collection_name=None): + if collections_paths is None: + collections_paths = ['~/root/.ansible/collections', '/usr/share/ansible/collections'] + + context.CLIARGS._store = { + 'collections_path': collections_paths, + 'collection': collection_name, + 'type': 'collection', + 'output_format': 'human' + } + + +@pytest.fixture +def mock_collection_objects(mocker): + mocker.patch('ansible.cli.galaxy.GalaxyCLI._resolve_path', side_effect=['/root/.ansible/collections', '/usr/share/ansible/collections']) + mocker.patch('ansible.cli.galaxy.validate_collection_path', + side_effect=['/root/.ansible/collections/ansible_collections', '/usr/share/ansible/collections/ansible_collections']) + + collection_args_1 = ( + ( + 'sandwiches.pbj', + '1.5.0', + None, + 'dir', + None, + ), + ( + 'sandwiches.reuben', + '2.5.0', + None, + 'dir', + None, + ), + ) + + collection_args_2 = ( + ( + 'sandwiches.pbj', + '1.0.0', + None, + 'dir', + None, + ), + ( + 'sandwiches.ham', + '1.0.0', + None, + 'dir', + None, + ), + ) + + collections_path_1 = [Requirement(*cargs) for cargs in collection_args_1] + collections_path_2 = [Requirement(*cargs) for cargs in collection_args_2] + + mocker.patch('ansible.cli.galaxy.find_existing_collections', side_effect=[collections_path_1, collections_path_2]) + + +@pytest.fixture +def mock_from_path(mocker): + def _from_path(collection_name='pbj'): + collection_args = { + 'sandwiches.pbj': ( + ( + 'sandwiches.pbj', + '1.5.0', + None, + 'dir', + None, + ), + ( + 'sandwiches.pbj', + '1.0.0', + None, + 'dir', + None, + ), + ), + 'sandwiches.ham': ( + ( + 'sandwiches.ham', + '1.0.0', + None, + 'dir', + None, + ), + ), + } + + from_path_objects = [Requirement(*args) for args in collection_args[collection_name]] + mocker.patch('ansible.cli.galaxy.Requirement.from_dir_path_as_unknown', side_effect=from_path_objects) + + return _from_path + + +def test_execute_list_collection_all(mocker, capsys, mock_collection_objects, tmp_path_factory): + """Test listing all collections from multiple paths""" + + cliargs() + + mocker.patch('os.path.exists', return_value=True) + mocker.patch('os.path.isdir', return_value=True) + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list']) + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert len(out_lines) == 12 + assert out_lines[0] == '' + assert out_lines[1] == '# /root/.ansible/collections/ansible_collections' + assert out_lines[2] == 'Collection Version' + assert out_lines[3] == '----------------- -------' + assert out_lines[4] == 'sandwiches.pbj 1.5.0 ' + assert out_lines[5] == 'sandwiches.reuben 2.5.0 ' + assert out_lines[6] == '' + assert out_lines[7] == '# /usr/share/ansible/collections/ansible_collections' + assert out_lines[8] == 'Collection Version' + assert out_lines[9] == '-------------- -------' + assert out_lines[10] == 'sandwiches.ham 1.0.0 ' + assert out_lines[11] == 'sandwiches.pbj 1.0.0 ' + + +def test_execute_list_collection_specific(mocker, capsys, mock_collection_objects, mock_from_path, tmp_path_factory): + """Test listing a specific collection""" + + collection_name = 'sandwiches.ham' + mock_from_path(collection_name) + + cliargs(collection_name=collection_name) + mocker.patch('os.path.exists', path_exists) + mocker.patch('os.path.isdir', return_value=True) + mocker.patch('ansible.galaxy.collection.validate_collection_name', collection_name) + mocker.patch('ansible.cli.galaxy._get_collection_widths', return_value=(14, 5)) + + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list', collection_name]) + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert len(out_lines) == 5 + assert out_lines[0] == '' + assert out_lines[1] == '# /usr/share/ansible/collections/ansible_collections' + assert out_lines[2] == 'Collection Version' + assert out_lines[3] == '-------------- -------' + assert out_lines[4] == 'sandwiches.ham 1.0.0 ' + + +def test_execute_list_collection_specific_duplicate(mocker, capsys, mock_collection_objects, mock_from_path, tmp_path_factory): + """Test listing a specific collection that exists at multiple paths""" + + collection_name = 'sandwiches.pbj' + mock_from_path(collection_name) + + cliargs(collection_name=collection_name) + mocker.patch('os.path.exists', path_exists) + mocker.patch('os.path.isdir', return_value=True) + mocker.patch('ansible.galaxy.collection.validate_collection_name', collection_name) + + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list', collection_name]) + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert len(out_lines) == 10 + assert out_lines[0] == '' + assert out_lines[1] == '# /root/.ansible/collections/ansible_collections' + assert out_lines[2] == 'Collection Version' + assert out_lines[3] == '-------------- -------' + assert out_lines[4] == 'sandwiches.pbj 1.5.0 ' + assert out_lines[5] == '' + assert out_lines[6] == '# /usr/share/ansible/collections/ansible_collections' + assert out_lines[7] == 'Collection Version' + assert out_lines[8] == '-------------- -------' + assert out_lines[9] == 'sandwiches.pbj 1.0.0 ' + + +def test_execute_list_collection_specific_invalid_fqcn(mocker, tmp_path_factory): + """Test an invalid fully qualified collection name (FQCN)""" + + collection_name = 'no.good.name' + + cliargs(collection_name=collection_name) + mocker.patch('os.path.exists', return_value=True) + mocker.patch('os.path.isdir', return_value=True) + + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list', collection_name]) + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + with pytest.raises(AnsibleError, match='Invalid collection name'): + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + +def test_execute_list_collection_no_valid_paths(mocker, capsys, tmp_path_factory): + """Test listing collections when no valid paths are given""" + + cliargs() + + mocker.patch('os.path.exists', return_value=True) + mocker.patch('os.path.isdir', return_value=False) + mocker.patch('ansible.utils.color.ANSIBLE_COLOR', False) + mocker.patch('ansible.cli.galaxy.display.columns', 79) + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list']) + + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + + with pytest.raises(AnsibleOptionsError, match=r'None of the provided paths were usable.'): + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + out, err = capsys.readouterr() + + assert '[WARNING]: - the configured path' in err + assert 'exists, but it\nis not a directory.' in err + + +def test_execute_list_collection_one_invalid_path(mocker, capsys, mock_collection_objects, tmp_path_factory): + """Test listing all collections when one invalid path is given""" + + cliargs() + mocker.patch('os.path.exists', return_value=True) + mocker.patch('os.path.isdir', isdir) + mocker.patch('ansible.cli.galaxy.GalaxyCLI._resolve_path', side_effect=['/root/.ansible/collections', 'nope']) + mocker.patch('ansible.utils.color.ANSIBLE_COLOR', False) + + gc = GalaxyCLI(['ansible-galaxy', 'collection', 'list', '-p', 'nope']) + tmp_path = tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections') + concrete_artifact_cm = collection.concrete_artifact_manager.ConcreteArtifactsManager(tmp_path, validate_certs=False) + gc.execute_list_collection(artifacts_manager=concrete_artifact_cm) + + out, err = capsys.readouterr() + out_lines = out.splitlines() + + assert out_lines[0] == '' + assert out_lines[1] == '# /root/.ansible/collections/ansible_collections' + assert out_lines[2] == 'Collection Version' + assert out_lines[3] == '----------------- -------' + assert out_lines[4] == 'sandwiches.pbj 1.5.0 ' + # Only a partial test of the output + + assert err == '[WARNING]: - the configured path nope, exists, but it is not a directory.\n' diff --git a/test/units/cli/galaxy/test_get_collection_widths.py b/test/units/cli/galaxy/test_get_collection_widths.py new file mode 100644 index 0000000..6e1cbf5 --- /dev/null +++ b/test/units/cli/galaxy/test_get_collection_widths.py @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 Ansible Project +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +import pytest + +from ansible.cli.galaxy import _get_collection_widths +from ansible.galaxy.dependency_resolution.dataclasses import Requirement + + +@pytest.fixture +def collection_objects(): + collection_ham = Requirement('sandwiches.ham', '1.5.0', None, 'galaxy', None) + + collection_pbj = Requirement('sandwiches.pbj', '2.5', None, 'galaxy', None) + + collection_reuben = Requirement('sandwiches.reuben', '4', None, 'galaxy', None) + + return [collection_ham, collection_pbj, collection_reuben] + + +def test_get_collection_widths(collection_objects): + assert _get_collection_widths(collection_objects) == (17, 5) + + +def test_get_collection_widths_single_collection(mocker): + mocked_collection = Requirement('sandwiches.club', '3.0.0', None, 'galaxy', None) + # Make this look like it is not iterable + mocker.patch('ansible.cli.galaxy.is_iterable', return_value=False) + + assert _get_collection_widths(mocked_collection) == (15, 5) diff --git a/test/units/cli/test_adhoc.py b/test/units/cli/test_adhoc.py new file mode 100644 index 0000000..18775f5 --- /dev/null +++ b/test/units/cli/test_adhoc.py @@ -0,0 +1,116 @@ +# Copyright: (c) 2018, Abhijeet Kasurde <akasurde@redhat.com> +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import pytest +import re + +from ansible import context +from ansible.cli.adhoc import AdHocCLI, display +from ansible.errors import AnsibleOptionsError + + +def test_parse(): + """ Test adhoc parse""" + with pytest.raises(ValueError, match='A non-empty list for args is required'): + adhoc_cli = AdHocCLI([]) + + adhoc_cli = AdHocCLI(['ansibletest']) + with pytest.raises(SystemExit): + adhoc_cli.parse() + + +def test_with_command(): + """ Test simple adhoc command""" + module_name = 'command' + adhoc_cli = AdHocCLI(args=['ansible', '-m', module_name, '-vv', 'localhost']) + adhoc_cli.parse() + assert context.CLIARGS['module_name'] == module_name + assert display.verbosity == 2 + + +def test_simple_command(): + """ Test valid command and its run""" + adhoc_cli = AdHocCLI(['/bin/ansible', '-m', 'command', 'localhost', '-a', 'echo "hi"']) + adhoc_cli.parse() + ret = adhoc_cli.run() + assert ret == 0 + + +def test_no_argument(): + """ Test no argument command""" + adhoc_cli = AdHocCLI(['/bin/ansible', '-m', 'command', 'localhost']) + adhoc_cli.parse() + with pytest.raises(AnsibleOptionsError) as exec_info: + adhoc_cli.run() + assert 'No argument passed to command module' == str(exec_info.value) + + +def test_did_you_mean_playbook(): + """ Test adhoc with yml file as argument parameter""" + adhoc_cli = AdHocCLI(['/bin/ansible', '-m', 'command', 'localhost.yml']) + adhoc_cli.parse() + with pytest.raises(AnsibleOptionsError) as exec_info: + adhoc_cli.run() + assert 'No argument passed to command module (did you mean to run ansible-playbook?)' == str(exec_info.value) + + +def test_play_ds_positive(): + """ Test _play_ds""" + adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost', '-m', 'command']) + adhoc_cli.parse() + ret = adhoc_cli._play_ds('command', 10, 2) + assert ret['name'] == 'Ansible Ad-Hoc' + assert ret['tasks'] == [{'action': {'module': 'command', 'args': {}}, 'async_val': 10, 'poll': 2, 'timeout': 0}] + + +def test_play_ds_with_include_role(): + """ Test include_role command with poll""" + adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost', '-m', 'include_role']) + adhoc_cli.parse() + ret = adhoc_cli._play_ds('include_role', None, 2) + assert ret['name'] == 'Ansible Ad-Hoc' + assert ret['gather_facts'] == 'no' + + +def test_run_import_playbook(): + """ Test import_playbook which is not allowed with ad-hoc command""" + import_playbook = 'import_playbook' + adhoc_cli = AdHocCLI(args=['/bin/ansible', '-m', import_playbook, 'localhost']) + adhoc_cli.parse() + with pytest.raises(AnsibleOptionsError) as exec_info: + adhoc_cli.run() + assert context.CLIARGS['module_name'] == import_playbook + assert "'%s' is not a valid action for ad-hoc commands" % import_playbook == str(exec_info.value) + + +def test_run_no_extra_vars(): + adhoc_cli = AdHocCLI(args=['/bin/ansible', 'localhost', '-e']) + with pytest.raises(SystemExit) as exec_info: + adhoc_cli.parse() + assert exec_info.value.code == 2 + + +def test_ansible_version(capsys, mocker): + adhoc_cli = AdHocCLI(args=['/bin/ansible', '--version']) + with pytest.raises(SystemExit): + adhoc_cli.run() + version = capsys.readouterr() + try: + version_lines = version.out.splitlines() + except AttributeError: + # Python 2.6 does return a named tuple, so get the first item + version_lines = version[0].splitlines() + + assert len(version_lines) == 9, 'Incorrect number of lines in "ansible --version" output' + assert re.match(r'ansible \[core [0-9.a-z]+\]$', version_lines[0]), 'Incorrect ansible version line in "ansible --version" output' + assert re.match(' config file = .*$', version_lines[1]), 'Incorrect config file line in "ansible --version" output' + assert re.match(' configured module search path = .*$', version_lines[2]), 'Incorrect module search path in "ansible --version" output' + assert re.match(' ansible python module location = .*$', version_lines[3]), 'Incorrect python module location in "ansible --version" output' + assert re.match(' ansible collection location = .*$', version_lines[4]), 'Incorrect collection location in "ansible --version" output' + assert re.match(' executable location = .*$', version_lines[5]), 'Incorrect executable locaction in "ansible --version" output' + assert re.match(' python version = .*$', version_lines[6]), 'Incorrect python version in "ansible --version" output' + assert re.match(' jinja version = .*$', version_lines[7]), 'Incorrect jinja version in "ansible --version" output' + assert re.match(' libyaml = .*$', version_lines[8]), 'Missing libyaml in "ansible --version" output' diff --git a/test/units/cli/test_cli.py b/test/units/cli/test_cli.py new file mode 100644 index 0000000..79c2b8f --- /dev/null +++ b/test/units/cli/test_cli.py @@ -0,0 +1,381 @@ +# (c) 2017, Adrian Likins <alikins@redhat.com> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +from units.compat import unittest +from unittest.mock import patch, MagicMock + +from units.mock.loader import DictDataLoader + +from ansible.release import __version__ +from ansible.parsing import vault +from ansible import cli + + +class TestCliVersion(unittest.TestCase): + + def test_version_info(self): + version_info = cli.CLI.version_info() + self.assertEqual(version_info['string'], __version__) + + def test_version_info_gitinfo(self): + version_info = cli.CLI.version_info(gitinfo=True) + self.assertIn('python version', version_info['string']) + + +class TestCliBuildVaultIds(unittest.TestCase): + def setUp(self): + self.tty_patcher = patch('ansible.cli.sys.stdin.isatty', return_value=True) + self.mock_isatty = self.tty_patcher.start() + + def tearDown(self): + self.tty_patcher.stop() + + def test(self): + res = cli.CLI.build_vault_ids(['foo@bar']) + self.assertEqual(res, ['foo@bar']) + + def test_create_new_password_no_vault_id(self): + res = cli.CLI.build_vault_ids([], create_new_password=True) + self.assertEqual(res, ['default@prompt_ask_vault_pass']) + + def test_create_new_password_no_vault_id_no_auto_prompt(self): + res = cli.CLI.build_vault_ids([], auto_prompt=False, create_new_password=True) + self.assertEqual(res, []) + + def test_no_vault_id_no_auto_prompt(self): + # simulate 'ansible-playbook site.yml' with out --ask-vault-pass, should not prompt + res = cli.CLI.build_vault_ids([], auto_prompt=False) + self.assertEqual(res, []) + + def test_no_vault_ids_auto_prompt(self): + # create_new_password=False + # simulate 'ansible-vault edit encrypted.yml' + res = cli.CLI.build_vault_ids([], auto_prompt=True) + self.assertEqual(res, ['default@prompt_ask_vault_pass']) + + def test_no_vault_ids_auto_prompt_ask_vault_pass(self): + # create_new_password=False + # simulate 'ansible-vault edit --ask-vault-pass encrypted.yml' + res = cli.CLI.build_vault_ids([], auto_prompt=True, ask_vault_pass=True) + self.assertEqual(res, ['default@prompt_ask_vault_pass']) + + def test_create_new_password_auto_prompt(self): + # simulate 'ansible-vault encrypt somefile.yml' + res = cli.CLI.build_vault_ids([], auto_prompt=True, create_new_password=True) + self.assertEqual(res, ['default@prompt_ask_vault_pass']) + + def test_create_new_password_no_vault_id_ask_vault_pass(self): + res = cli.CLI.build_vault_ids([], ask_vault_pass=True, + create_new_password=True) + self.assertEqual(res, ['default@prompt_ask_vault_pass']) + + def test_create_new_password_with_vault_ids(self): + res = cli.CLI.build_vault_ids(['foo@bar'], create_new_password=True) + self.assertEqual(res, ['foo@bar']) + + def test_create_new_password_no_vault_ids_password_files(self): + res = cli.CLI.build_vault_ids([], vault_password_files=['some-password-file'], + create_new_password=True) + self.assertEqual(res, ['default@some-password-file']) + + def test_everything(self): + res = cli.CLI.build_vault_ids(['blip@prompt', 'baz@prompt_ask_vault_pass', + 'some-password-file', 'qux@another-password-file'], + vault_password_files=['yet-another-password-file', + 'one-more-password-file'], + ask_vault_pass=True, + create_new_password=True, + auto_prompt=False) + + self.assertEqual(set(res), set(['blip@prompt', 'baz@prompt_ask_vault_pass', + 'default@prompt_ask_vault_pass', + 'some-password-file', 'qux@another-password-file', + 'default@yet-another-password-file', + 'default@one-more-password-file'])) + + +class TestCliSetupVaultSecrets(unittest.TestCase): + def setUp(self): + self.fake_loader = DictDataLoader({}) + self.tty_patcher = patch('ansible.cli.sys.stdin.isatty', return_value=True) + self.mock_isatty = self.tty_patcher.start() + + self.display_v_patcher = patch('ansible.cli.display.verbosity', return_value=6) + self.mock_display_v = self.display_v_patcher.start() + cli.display.verbosity = 5 + + def tearDown(self): + self.tty_patcher.stop() + self.display_v_patcher.stop() + cli.display.verbosity = 0 + + def test(self): + res = cli.CLI.setup_vault_secrets(None, None, auto_prompt=False) + self.assertIsInstance(res, list) + + @patch('ansible.cli.get_file_vault_secret') + def test_password_file(self, mock_file_secret): + filename = '/dev/null/secret' + mock_file_secret.return_value = MagicMock(bytes=b'file1_password', + vault_id='file1', + filename=filename) + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['secret1@%s' % filename, 'secret2'], + vault_password_files=[filename]) + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['secret1']) + self.assertIn('secret1', [x[0] for x in matches]) + match = matches[0][1] + self.assertEqual(match.bytes, b'file1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='prompt1') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['prompt1@prompt'], + ask_vault_pass=True, + auto_prompt=False) + + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['prompt1']) + self.assertIn('prompt1', [x[0] for x in matches]) + match = matches[0][1] + self.assertEqual(match.bytes, b'prompt1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_no_tty(self, mock_prompt_secret): + self.mock_isatty.return_value = False + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='prompt1', + name='bytes_should_be_prompt1_password', + spec=vault.PromptVaultSecret) + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['prompt1@prompt'], + ask_vault_pass=True, + auto_prompt=False) + + self.assertIsInstance(res, list) + self.assertEqual(len(res), 2) + matches = vault.match_secrets(res, ['prompt1']) + self.assertIn('prompt1', [x[0] for x in matches]) + self.assertEqual(len(matches), 1) + + @patch('ansible.cli.get_file_vault_secret') + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_no_tty_and_password_file(self, mock_prompt_secret, mock_file_secret): + self.mock_isatty.return_value = False + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='prompt1') + filename = '/dev/null/secret' + mock_file_secret.return_value = MagicMock(bytes=b'file1_password', + vault_id='file1', + filename=filename) + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['prompt1@prompt', 'file1@/dev/null/secret'], + ask_vault_pass=True) + + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['file1']) + self.assertIn('file1', [x[0] for x in matches]) + self.assertNotIn('prompt1', [x[0] for x in matches]) + match = matches[0][1] + self.assertEqual(match.bytes, b'file1_password') + + def _assert_ids(self, vault_id_names, res, password=b'prompt1_password'): + self.assertIsInstance(res, list) + len_ids = len(vault_id_names) + matches = vault.match_secrets(res, vault_id_names) + self.assertEqual(len(res), len_ids, 'len(res):%s does not match len_ids:%s' % (len(res), len_ids)) + self.assertEqual(len(matches), len_ids) + for index, prompt in enumerate(vault_id_names): + self.assertIn(prompt, [x[0] for x in matches]) + # simple mock, same password/prompt for each mock_prompt_secret + self.assertEqual(matches[index][1].bytes, password) + + @patch('ansible.cli.PromptVaultSecret') + def test_multiple_prompts(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='prompt1') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['prompt1@prompt', + 'prompt2@prompt'], + ask_vault_pass=False) + + vault_id_names = ['prompt1', 'prompt2'] + self._assert_ids(vault_id_names, res) + + @patch('ansible.cli.PromptVaultSecret') + def test_multiple_prompts_and_ask_vault_pass(self, mock_prompt_secret): + self.mock_isatty.return_value = False + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='prompt1') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['prompt1@prompt', + 'prompt2@prompt', + 'prompt3@prompt_ask_vault_pass'], + ask_vault_pass=True) + + # We provide some vault-ids and secrets, so auto_prompt shouldn't get triggered, + # so there is + vault_id_names = ['prompt1', 'prompt2', 'prompt3', 'default'] + self._assert_ids(vault_id_names, res) + + @patch('ansible.cli.C') + @patch('ansible.cli.get_file_vault_secret') + @patch('ansible.cli.PromptVaultSecret') + def test_default_file_vault(self, mock_prompt_secret, + mock_file_secret, + mock_config): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='default') + mock_file_secret.return_value = MagicMock(bytes=b'file1_password', + vault_id='default') + mock_config.DEFAULT_VAULT_PASSWORD_FILE = '/dev/null/faux/vault_password_file' + mock_config.DEFAULT_VAULT_IDENTITY = 'default' + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=[], + create_new_password=False, + ask_vault_pass=False) + + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['default']) + # --vault-password-file/DEFAULT_VAULT_PASSWORD_FILE is higher precendce than prompts + # if the same vault-id ('default') regardless of cli order since it didn't matter in 2.3 + + self.assertEqual(matches[0][1].bytes, b'file1_password') + self.assertEqual(len(matches), 1) + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=[], + create_new_password=False, + ask_vault_pass=True, + auto_prompt=True) + + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['default']) + self.assertEqual(matches[0][1].bytes, b'file1_password') + self.assertEqual(matches[1][1].bytes, b'prompt1_password') + self.assertEqual(len(matches), 2) + + @patch('ansible.cli.get_file_vault_secret') + @patch('ansible.cli.PromptVaultSecret') + def test_default_file_vault_identity_list(self, mock_prompt_secret, + mock_file_secret): + default_vault_ids = ['some_prompt@prompt', + 'some_file@/dev/null/secret'] + + mock_prompt_secret.return_value = MagicMock(bytes=b'some_prompt_password', + vault_id='some_prompt') + + filename = '/dev/null/secret' + mock_file_secret.return_value = MagicMock(bytes=b'some_file_password', + vault_id='some_file', + filename=filename) + + vault_ids = default_vault_ids + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=vault_ids, + create_new_password=False, + ask_vault_pass=True) + + self.assertIsInstance(res, list) + matches = vault.match_secrets(res, ['some_file']) + # --vault-password-file/DEFAULT_VAULT_PASSWORD_FILE is higher precendce than prompts + # if the same vault-id ('default') regardless of cli order since it didn't matter in 2.3 + self.assertEqual(matches[0][1].bytes, b'some_file_password') + matches = vault.match_secrets(res, ['some_prompt']) + self.assertEqual(matches[0][1].bytes, b'some_prompt_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_just_ask_vault_pass(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='default') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=[], + create_new_password=False, + ask_vault_pass=True) + + self.assertIsInstance(res, list) + match = vault.match_secrets(res, ['default'])[0][1] + self.assertEqual(match.bytes, b'prompt1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_new_password_ask_vault_pass(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='default') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=[], + create_new_password=True, + ask_vault_pass=True) + + self.assertIsInstance(res, list) + match = vault.match_secrets(res, ['default'])[0][1] + self.assertEqual(match.bytes, b'prompt1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_new_password_vault_id_prompt(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='some_vault_id') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['some_vault_id@prompt'], + create_new_password=True, + ask_vault_pass=False) + + self.assertIsInstance(res, list) + match = vault.match_secrets(res, ['some_vault_id'])[0][1] + self.assertEqual(match.bytes, b'prompt1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_new_password_vault_id_prompt_ask_vault_pass(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='default') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['some_vault_id@prompt_ask_vault_pass'], + create_new_password=True, + ask_vault_pass=False) + + self.assertIsInstance(res, list) + match = vault.match_secrets(res, ['some_vault_id'])[0][1] + self.assertEqual(match.bytes, b'prompt1_password') + + @patch('ansible.cli.PromptVaultSecret') + def test_prompt_new_password_vault_id_prompt_ask_vault_pass_ask_vault_pass(self, mock_prompt_secret): + mock_prompt_secret.return_value = MagicMock(bytes=b'prompt1_password', + vault_id='default') + + res = cli.CLI.setup_vault_secrets(loader=self.fake_loader, + vault_ids=['some_vault_id@prompt_ask_vault_pass'], + create_new_password=True, + ask_vault_pass=True) + + self.assertIsInstance(res, list) + match = vault.match_secrets(res, ['some_vault_id'])[0][1] + self.assertEqual(match.bytes, b'prompt1_password') diff --git a/test/units/cli/test_console.py b/test/units/cli/test_console.py new file mode 100644 index 0000000..4fc05dd --- /dev/null +++ b/test/units/cli/test_console.py @@ -0,0 +1,51 @@ +# (c) 2016, Thilo Uttendorfer <tlo@sengaya.de> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +from units.compat import unittest +from unittest.mock import patch + +from ansible.cli.console import ConsoleCLI + + +class TestConsoleCLI(unittest.TestCase): + def test_parse(self): + cli = ConsoleCLI(['ansible test']) + cli.parse() + self.assertTrue(cli.parser is not None) + + def test_module_args(self): + cli = ConsoleCLI(['ansible test']) + cli.parse() + res = cli.module_args('copy') + self.assertTrue(cli.parser is not None) + self.assertIn('src', res) + self.assertIn('backup', res) + self.assertIsInstance(res, list) + + @patch('ansible.utils.display.Display.display') + def test_helpdefault(self, mock_display): + cli = ConsoleCLI(['ansible test']) + cli.parse() + cli.modules = set(['copy']) + cli.helpdefault('copy') + self.assertTrue(cli.parser is not None) + self.assertTrue(len(mock_display.call_args_list) > 0, + "display.display should have been called but was not") diff --git a/test/units/cli/test_data/collection_skeleton/README.md b/test/units/cli/test_data/collection_skeleton/README.md new file mode 100644 index 0000000..4cfd8af --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/README.md @@ -0,0 +1 @@ +A readme
\ No newline at end of file diff --git a/test/units/cli/test_data/collection_skeleton/docs/My Collection.md b/test/units/cli/test_data/collection_skeleton/docs/My Collection.md new file mode 100644 index 0000000..6fa917f --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/docs/My Collection.md @@ -0,0 +1 @@ +Welcome to my test collection doc for {{ namespace }}.
\ No newline at end of file diff --git a/test/units/cli/test_data/collection_skeleton/galaxy.yml.j2 b/test/units/cli/test_data/collection_skeleton/galaxy.yml.j2 new file mode 100644 index 0000000..b1da267 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/galaxy.yml.j2 @@ -0,0 +1,7 @@ +namespace: '{{ namespace }}' +name: '{{ collection_name }}' +version: 0.1.0 +readme: README.md +authors: +- Ansible Cow <acow@bovineuniversity.edu> +- Tu Cow <tucow@bovineuniversity.edu> diff --git a/test/units/cli/test_data/collection_skeleton/playbooks/main.yml b/test/units/cli/test_data/collection_skeleton/playbooks/main.yml new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/playbooks/main.yml diff --git a/test/units/cli/test_data/collection_skeleton/playbooks/templates/subfolder/test.conf.j2 b/test/units/cli/test_data/collection_skeleton/playbooks/templates/subfolder/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/playbooks/templates/subfolder/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/collection_skeleton/playbooks/templates/test.conf.j2 b/test/units/cli/test_data/collection_skeleton/playbooks/templates/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/playbooks/templates/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/collection_skeleton/plugins/action/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/action/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/action/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/plugins/filter/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/filter/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/filter/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/plugins/inventory/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/inventory/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/inventory/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/plugins/lookup/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/lookup/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/lookup/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/plugins/module_utils/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/module_utils/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/module_utils/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/plugins/modules/.git_keep b/test/units/cli/test_data/collection_skeleton/plugins/modules/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/plugins/modules/.git_keep diff --git a/test/units/cli/test_data/collection_skeleton/roles/common/tasks/main.yml.j2 b/test/units/cli/test_data/collection_skeleton/roles/common/tasks/main.yml.j2 new file mode 100644 index 0000000..77adf2e --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/roles/common/tasks/main.yml.j2 @@ -0,0 +1,3 @@ +- name: test collection skeleton + debug: + msg: "Namespace: {{ namespace }}"
\ No newline at end of file diff --git a/test/units/cli/test_data/collection_skeleton/roles/common/templates/subfolder/test.conf.j2 b/test/units/cli/test_data/collection_skeleton/roles/common/templates/subfolder/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/roles/common/templates/subfolder/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/collection_skeleton/roles/common/templates/test.conf.j2 b/test/units/cli/test_data/collection_skeleton/roles/common/templates/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/collection_skeleton/roles/common/templates/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/role_skeleton/README.md b/test/units/cli/test_data/role_skeleton/README.md new file mode 100644 index 0000000..225dd44 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/README.md @@ -0,0 +1,38 @@ +Role Name +========= + +A brief description of the role goes here. + +Requirements +------------ + +Any pre-requisites that may not be covered by Ansible itself or the role should be mentioned here. For instance, if the role uses the EC2 module, it may be a good idea to mention in this section that the boto package is required. + +Role Variables +-------------- + +A description of the settable variables for this role should go here, including any variables that are in defaults/main.yml, vars/main.yml, and any variables that can/should be set via parameters to the role. Any variables that are read from other roles and/or the global scope (ie. hostvars, group vars, etc.) should be mentioned here as well. + +Dependencies +------------ + +A list of other roles hosted on Galaxy should go here, plus any details in regards to parameters that may need to be set for other roles, or variables that are used from other roles. + +Example Playbook +---------------- + +Including an example of how to use your role (for instance, with variables passed in as parameters) is always nice for users too: + + - hosts: servers + roles: + - { role: username.rolename, x: 42 } + +License +------- + +BSD + +Author Information +------------------ + +An optional section for the role authors to include contact information, or a website (HTML is not allowed). diff --git a/test/units/cli/test_data/role_skeleton/defaults/main.yml.j2 b/test/units/cli/test_data/role_skeleton/defaults/main.yml.j2 new file mode 100644 index 0000000..3818e64 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/defaults/main.yml.j2 @@ -0,0 +1,2 @@ +--- +# defaults file for {{ role_name }} diff --git a/test/units/cli/test_data/role_skeleton/files/.git_keep b/test/units/cli/test_data/role_skeleton/files/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/files/.git_keep diff --git a/test/units/cli/test_data/role_skeleton/handlers/main.yml.j2 b/test/units/cli/test_data/role_skeleton/handlers/main.yml.j2 new file mode 100644 index 0000000..3f4c496 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/handlers/main.yml.j2 @@ -0,0 +1,2 @@ +--- +# handlers file for {{ role_name }} diff --git a/test/units/cli/test_data/role_skeleton/inventory b/test/units/cli/test_data/role_skeleton/inventory new file mode 100644 index 0000000..2fbb50c --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/inventory @@ -0,0 +1 @@ +localhost diff --git a/test/units/cli/test_data/role_skeleton/meta/main.yml.j2 b/test/units/cli/test_data/role_skeleton/meta/main.yml.j2 new file mode 100644 index 0000000..2fc53cb --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/meta/main.yml.j2 @@ -0,0 +1,62 @@ +galaxy_info: + author: {{ author }} + description: {{ description }} + company: {{ company }} + + # If the issue tracker for your role is not on github, uncomment the + # next line and provide a value + # issue_tracker_url: {{ issue_tracker_url }} + + # Some suggested licenses: + # - BSD (default) + # - MIT + # - GPLv2 + # - GPLv3 + # - Apache + # - CC-BY + license: {{ license }} + + min_ansible_version: {{ min_ansible_version }} + + # Optionally specify the branch Galaxy will use when accessing the GitHub + # repo for this role. During role install, if no tags are available, + # Galaxy will use this branch. During import Galaxy will access files on + # this branch. If travis integration is configured, only notification for this + # branch will be accepted. Otherwise, in all cases, the repo's default branch + # (usually master) will be used. + #github_branch: + + # + # Provide a list of supported platforms, and for each platform a list of versions. + # If you don't wish to enumerate all versions for a particular platform, use 'all'. + # To view available platforms and versions (or releases), visit: + # https://galaxy.ansible.com/api/v1/platforms/ + # + # platforms: + # - name: Fedora + # versions: + # - all + # - 25 + # - name: SomePlatform + # versions: + # - all + # - 1.0 + # - 7 + # - 99.99 + + galaxy_tags: [] + # List tags for your role here, one per line. A tag is + # a keyword that describes and categorizes the role. + # Users find roles by searching for tags. Be sure to + # remove the '[]' above if you add tags to this list. + # + # NOTE: A tag is limited to a single word comprised of + # alphanumeric characters. Maximum 20 tags per role. + +dependencies: [] + # List your role dependencies here, one per line. + # Be sure to remove the '[]' above if you add dependencies + # to this list. +{%- for dependency in dependencies %} + #- {{ dependency }} +{%- endfor %} diff --git a/test/units/cli/test_data/role_skeleton/tasks/main.yml.j2 b/test/units/cli/test_data/role_skeleton/tasks/main.yml.j2 new file mode 100644 index 0000000..a988065 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/tasks/main.yml.j2 @@ -0,0 +1,2 @@ +--- +# tasks file for {{ role_name }} diff --git a/test/units/cli/test_data/role_skeleton/templates/.git_keep b/test/units/cli/test_data/role_skeleton/templates/.git_keep new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/templates/.git_keep diff --git a/test/units/cli/test_data/role_skeleton/templates/subfolder/test.conf.j2 b/test/units/cli/test_data/role_skeleton/templates/subfolder/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/templates/subfolder/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/role_skeleton/templates/test.conf.j2 b/test/units/cli/test_data/role_skeleton/templates/test.conf.j2 new file mode 100644 index 0000000..b4e3364 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/templates/test.conf.j2 @@ -0,0 +1,2 @@ +[defaults] +test_key = {{ test_variable }} diff --git a/test/units/cli/test_data/role_skeleton/templates_extra/templates.txt.j2 b/test/units/cli/test_data/role_skeleton/templates_extra/templates.txt.j2 new file mode 100644 index 0000000..143d630 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/templates_extra/templates.txt.j2 @@ -0,0 +1 @@ +{{ role_name }} diff --git a/test/units/cli/test_data/role_skeleton/tests/test.yml.j2 b/test/units/cli/test_data/role_skeleton/tests/test.yml.j2 new file mode 100644 index 0000000..0c40f95 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/tests/test.yml.j2 @@ -0,0 +1,5 @@ +--- +- hosts: localhost + remote_user: root + roles: + - {{ role_name }} diff --git a/test/units/cli/test_data/role_skeleton/vars/main.yml.j2 b/test/units/cli/test_data/role_skeleton/vars/main.yml.j2 new file mode 100644 index 0000000..092d511 --- /dev/null +++ b/test/units/cli/test_data/role_skeleton/vars/main.yml.j2 @@ -0,0 +1,2 @@ +--- +# vars file for {{ role_name }} diff --git a/test/units/cli/test_doc.py b/test/units/cli/test_doc.py new file mode 100644 index 0000000..b10f088 --- /dev/null +++ b/test/units/cli/test_doc.py @@ -0,0 +1,130 @@ +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import pytest + +from ansible.cli.doc import DocCLI, RoleMixin +from ansible.plugins.loader import module_loader + + +TTY_IFY_DATA = { + # No substitutions + 'no-op': 'no-op', + 'no-op Z(test)': 'no-op Z(test)', + # Simple cases of all substitutions + 'I(italic)': "`italic'", + 'B(bold)': '*bold*', + 'M(ansible.builtin.module)': '[ansible.builtin.module]', + 'U(https://docs.ansible.com)': 'https://docs.ansible.com', + 'L(the user guide,https://docs.ansible.com/user-guide.html)': 'the user guide <https://docs.ansible.com/user-guide.html>', + 'R(the user guide,user-guide)': 'the user guide', + 'C(/usr/bin/file)': "`/usr/bin/file'", + 'HORIZONTALLINE': '\n{0}\n'.format('-' * 13), + # Multiple substitutions + 'The M(ansible.builtin.yum) module B(MUST) be given the C(package) parameter. See the R(looping docs,using-loops) for more info': + "The [ansible.builtin.yum] module *MUST* be given the `package' parameter. See the looping docs for more info", + # Problem cases + 'IBM(International Business Machines)': 'IBM(International Business Machines)', + 'L(the user guide, https://docs.ansible.com/)': 'the user guide <https://docs.ansible.com/>', + 'R(the user guide, user-guide)': 'the user guide', + # de-rsty refs and anchors + 'yolo :ref:`my boy` does stuff': 'yolo `my boy` does stuff', + '.. seealso:: Something amazing': 'See also: Something amazing', + '.. seealso:: Troublesome multiline\n Stuff goes htere': 'See also: Troublesome multiline\n Stuff goes htere', + '.. note:: boring stuff': 'Note: boring stuff', +} + + +@pytest.mark.parametrize('text, expected', sorted(TTY_IFY_DATA.items())) +def test_ttyify(text, expected): + assert DocCLI.tty_ify(text) == expected + + +def test_rolemixin__build_summary(): + obj = RoleMixin() + role_name = 'test_role' + collection_name = 'test.units' + argspec = { + 'main': {'short_description': 'main short description'}, + 'alternate': {'short_description': 'alternate short description'}, + } + expected = { + 'collection': collection_name, + 'entry_points': { + 'main': argspec['main']['short_description'], + 'alternate': argspec['alternate']['short_description'], + } + } + + fqcn, summary = obj._build_summary(role_name, collection_name, argspec) + assert fqcn == '.'.join([collection_name, role_name]) + assert summary == expected + + +def test_rolemixin__build_summary_empty_argspec(): + obj = RoleMixin() + role_name = 'test_role' + collection_name = 'test.units' + argspec = {} + expected = { + 'collection': collection_name, + 'entry_points': {} + } + + fqcn, summary = obj._build_summary(role_name, collection_name, argspec) + assert fqcn == '.'.join([collection_name, role_name]) + assert summary == expected + + +def test_rolemixin__build_doc(): + obj = RoleMixin() + role_name = 'test_role' + path = '/a/b/c' + collection_name = 'test.units' + entrypoint_filter = 'main' + argspec = { + 'main': {'short_description': 'main short description'}, + 'alternate': {'short_description': 'alternate short description'}, + } + expected = { + 'path': path, + 'collection': collection_name, + 'entry_points': { + 'main': argspec['main'], + } + } + fqcn, doc = obj._build_doc(role_name, path, collection_name, argspec, entrypoint_filter) + assert fqcn == '.'.join([collection_name, role_name]) + assert doc == expected + + +def test_rolemixin__build_doc_no_filter_match(): + obj = RoleMixin() + role_name = 'test_role' + path = '/a/b/c' + collection_name = 'test.units' + entrypoint_filter = 'doesNotExist' + argspec = { + 'main': {'short_description': 'main short description'}, + 'alternate': {'short_description': 'alternate short description'}, + } + fqcn, doc = obj._build_doc(role_name, path, collection_name, argspec, entrypoint_filter) + assert fqcn == '.'.join([collection_name, role_name]) + assert doc is None + + +def test_builtin_modules_list(): + args = ['ansible-doc', '-l', 'ansible.builtin', '-t', 'module'] + obj = DocCLI(args=args) + obj.parse() + result = obj._list_plugins('module', module_loader) + assert len(result) > 0 + + +def test_legacy_modules_list(): + args = ['ansible-doc', '-l', 'ansible.legacy', '-t', 'module'] + obj = DocCLI(args=args) + obj.parse() + result = obj._list_plugins('module', module_loader) + assert len(result) > 0 diff --git a/test/units/cli/test_galaxy.py b/test/units/cli/test_galaxy.py new file mode 100644 index 0000000..8ff5640 --- /dev/null +++ b/test/units/cli/test_galaxy.py @@ -0,0 +1,1346 @@ +# -*- coding: utf-8 -*- +# (c) 2016, Adrian Likins <alikins@redhat.com> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import ansible +from io import BytesIO +import json +import os +import pytest +import shutil +import stat +import tarfile +import tempfile +import yaml + +import ansible.constants as C +from ansible import context +from ansible.cli.galaxy import GalaxyCLI +from ansible.galaxy import collection +from ansible.galaxy.api import GalaxyAPI +from ansible.errors import AnsibleError +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 units.compat import unittest +from unittest.mock import patch, MagicMock + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + yield + co.GlobalCLIArgs._Singleton__instance = None + + +class TestGalaxy(unittest.TestCase): + @classmethod + def setUpClass(cls): + '''creating prerequisites for installing a role; setUpClass occurs ONCE whereas setUp occurs with every method tested.''' + # class data for easy viewing: role_dir, role_tar, role_name, role_req, role_path + + cls.temp_dir = tempfile.mkdtemp(prefix='ansible-test_galaxy-') + os.chdir(cls.temp_dir) + + if os.path.exists("./delete_me"): + shutil.rmtree("./delete_me") + + # creating framework for a role + gc = GalaxyCLI(args=["ansible-galaxy", "init", "--offline", "delete_me"]) + gc.run() + cls.role_dir = "./delete_me" + cls.role_name = "delete_me" + + # making a temp dir for role installation + cls.role_path = os.path.join(tempfile.mkdtemp(), "roles") + if not os.path.isdir(cls.role_path): + os.makedirs(cls.role_path) + + # creating a tar file name for class data + cls.role_tar = './delete_me.tar.gz' + cls.makeTar(cls.role_tar, cls.role_dir) + + # creating a temp file with installation requirements + cls.role_req = './delete_me_requirements.yml' + fd = open(cls.role_req, "w") + fd.write("- 'src': '%s'\n 'name': '%s'\n 'path': '%s'" % (cls.role_tar, cls.role_name, cls.role_path)) + fd.close() + + @classmethod + def makeTar(cls, output_file, source_dir): + ''' used for making a tarfile from a role directory ''' + # adding directory into a tar file + try: + tar = tarfile.open(output_file, "w:gz") + tar.add(source_dir, arcname=os.path.basename(source_dir)) + except AttributeError: # tarfile obj. has no attribute __exit__ prior to python 2. 7 + pass + finally: # ensuring closure of tarfile obj + tar.close() + + @classmethod + def tearDownClass(cls): + '''After tests are finished removes things created in setUpClass''' + # deleting the temp role directory + if os.path.exists(cls.role_dir): + shutil.rmtree(cls.role_dir) + if os.path.exists(cls.role_req): + os.remove(cls.role_req) + if os.path.exists(cls.role_tar): + os.remove(cls.role_tar) + if os.path.isdir(cls.role_path): + shutil.rmtree(cls.role_path) + + os.chdir('/') + shutil.rmtree(cls.temp_dir) + + def setUp(self): + # Reset the stored command line args + co.GlobalCLIArgs._Singleton__instance = None + self.default_args = ['ansible-galaxy'] + + def tearDown(self): + # Reset the stored command line args + co.GlobalCLIArgs._Singleton__instance = None + + def test_init(self): + galaxy_cli = GalaxyCLI(args=self.default_args) + self.assertTrue(isinstance(galaxy_cli, GalaxyCLI)) + + def test_display_min(self): + gc = GalaxyCLI(args=self.default_args) + role_info = {'name': 'some_role_name'} + display_result = gc._display_role_info(role_info) + self.assertTrue(display_result.find('some_role_name') > -1) + + def test_display_galaxy_info(self): + gc = GalaxyCLI(args=self.default_args) + galaxy_info = {} + role_info = {'name': 'some_role_name', + 'galaxy_info': galaxy_info} + display_result = gc._display_role_info(role_info) + if display_result.find('\n\tgalaxy_info:') == -1: + self.fail('Expected galaxy_info to be indented once') + + def test_run(self): + ''' verifies that the GalaxyCLI object's api is created and that execute() is called. ''' + gc = GalaxyCLI(args=["ansible-galaxy", "install", "--ignore-errors", "imaginary_role"]) + gc.parse() + with patch.object(ansible.cli.CLI, "run", return_value=None) as mock_run: + gc.run() + # testing + self.assertIsInstance(gc.galaxy, ansible.galaxy.Galaxy) + self.assertEqual(mock_run.call_count, 1) + self.assertTrue(isinstance(gc.api, ansible.galaxy.api.GalaxyAPI)) + + def test_execute_remove(self): + # installing role + gc = GalaxyCLI(args=["ansible-galaxy", "install", "-p", self.role_path, "-r", self.role_req, '--force']) + gc.run() + + # location where the role was installed + role_file = os.path.join(self.role_path, self.role_name) + + # removing role + # Have to reset the arguments in the context object manually since we're doing the + # equivalent of running the command line program twice + co.GlobalCLIArgs._Singleton__instance = None + gc = GalaxyCLI(args=["ansible-galaxy", "remove", role_file, self.role_name]) + gc.run() + + # testing role was removed + removed_role = not os.path.exists(role_file) + self.assertTrue(removed_role) + + def test_exit_without_ignore_without_flag(self): + ''' tests that GalaxyCLI exits with the error specified if the --ignore-errors flag is not used ''' + gc = GalaxyCLI(args=["ansible-galaxy", "install", "--server=None", "fake_role_name"]) + with patch.object(ansible.utils.display.Display, "display", return_value=None) as mocked_display: + # testing that error expected is raised + self.assertRaises(AnsibleError, gc.run) + self.assertTrue(mocked_display.called_once_with("- downloading role 'fake_role_name', owned by ")) + + def test_exit_without_ignore_with_flag(self): + ''' tests that GalaxyCLI exits without the error specified if the --ignore-errors flag is used ''' + # testing with --ignore-errors flag + gc = GalaxyCLI(args=["ansible-galaxy", "install", "--server=None", "fake_role_name", "--ignore-errors"]) + with patch.object(ansible.utils.display.Display, "display", return_value=None) as mocked_display: + gc.run() + self.assertTrue(mocked_display.called_once_with("- downloading role 'fake_role_name', owned by ")) + + def test_parse_no_action(self): + ''' testing the options parser when no action is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", ""]) + self.assertRaises(SystemExit, gc.parse) + + def test_parse_invalid_action(self): + ''' testing the options parser when an invalid action is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "NOT_ACTION"]) + self.assertRaises(SystemExit, gc.parse) + + def test_parse_delete(self): + ''' testing the options parser when the action 'delete' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "delete", "foo", "bar"]) + gc.parse() + self.assertEqual(context.CLIARGS['verbosity'], 0) + + def test_parse_import(self): + ''' testing the options parser when the action 'import' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "import", "foo", "bar"]) + gc.parse() + self.assertEqual(context.CLIARGS['wait'], True) + self.assertEqual(context.CLIARGS['reference'], None) + self.assertEqual(context.CLIARGS['check_status'], False) + self.assertEqual(context.CLIARGS['verbosity'], 0) + + def test_parse_info(self): + ''' testing the options parser when the action 'info' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "info", "foo", "bar"]) + gc.parse() + self.assertEqual(context.CLIARGS['offline'], False) + + def test_parse_init(self): + ''' testing the options parser when the action 'init' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "init", "foo"]) + gc.parse() + self.assertEqual(context.CLIARGS['offline'], False) + self.assertEqual(context.CLIARGS['force'], False) + + def test_parse_install(self): + ''' testing the options parser when the action 'install' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "install"]) + gc.parse() + self.assertEqual(context.CLIARGS['ignore_errors'], False) + self.assertEqual(context.CLIARGS['no_deps'], False) + self.assertEqual(context.CLIARGS['requirements'], None) + self.assertEqual(context.CLIARGS['force'], False) + + def test_parse_list(self): + ''' testing the options parser when the action 'list' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "list"]) + gc.parse() + self.assertEqual(context.CLIARGS['verbosity'], 0) + + def test_parse_remove(self): + ''' testing the options parser when the action 'remove' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "remove", "foo"]) + gc.parse() + self.assertEqual(context.CLIARGS['verbosity'], 0) + + def test_parse_search(self): + ''' testing the options parswer when the action 'search' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "search"]) + gc.parse() + self.assertEqual(context.CLIARGS['platforms'], None) + self.assertEqual(context.CLIARGS['galaxy_tags'], None) + self.assertEqual(context.CLIARGS['author'], None) + + def test_parse_setup(self): + ''' testing the options parser when the action 'setup' is given ''' + gc = GalaxyCLI(args=["ansible-galaxy", "setup", "source", "github_user", "github_repo", "secret"]) + gc.parse() + self.assertEqual(context.CLIARGS['verbosity'], 0) + self.assertEqual(context.CLIARGS['remove_id'], None) + self.assertEqual(context.CLIARGS['setup_list'], False) + + +class ValidRoleTests(object): + + expected_role_dirs = ('defaults', 'files', 'handlers', 'meta', 'tasks', 'templates', 'vars', 'tests') + + @classmethod + def setUpRole(cls, role_name, galaxy_args=None, skeleton_path=None, use_explicit_type=False): + if galaxy_args is None: + galaxy_args = [] + + if skeleton_path is not None: + cls.role_skeleton_path = skeleton_path + galaxy_args += ['--role-skeleton', skeleton_path] + + # Make temp directory for testing + cls.test_dir = tempfile.mkdtemp() + if not os.path.isdir(cls.test_dir): + os.makedirs(cls.test_dir) + + cls.role_dir = os.path.join(cls.test_dir, role_name) + cls.role_name = role_name + + # create role using default skeleton + args = ['ansible-galaxy'] + if use_explicit_type: + args += ['role'] + args += ['init', '-c', '--offline'] + galaxy_args + ['--init-path', cls.test_dir, cls.role_name] + + gc = GalaxyCLI(args=args) + gc.run() + cls.gc = gc + + if skeleton_path is None: + cls.role_skeleton_path = gc.galaxy.default_role_skeleton_path + + @classmethod + def tearDownClass(cls): + if os.path.isdir(cls.test_dir): + shutil.rmtree(cls.test_dir) + + def test_metadata(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertIn('galaxy_info', metadata, msg='unable to find galaxy_info in metadata') + self.assertIn('dependencies', metadata, msg='unable to find dependencies in metadata') + + def test_readme(self): + readme_path = os.path.join(self.role_dir, 'README.md') + self.assertTrue(os.path.exists(readme_path), msg='Readme doesn\'t exist') + + def test_main_ymls(self): + need_main_ymls = set(self.expected_role_dirs) - set(['meta', 'tests', 'files', 'templates']) + for d in need_main_ymls: + main_yml = os.path.join(self.role_dir, d, 'main.yml') + self.assertTrue(os.path.exists(main_yml)) + expected_string = "---\n# {0} file for {1}".format(d, self.role_name) + with open(main_yml, 'r') as f: + self.assertEqual(expected_string, f.read().strip()) + + def test_role_dirs(self): + for d in self.expected_role_dirs: + self.assertTrue(os.path.isdir(os.path.join(self.role_dir, d)), msg="Expected role subdirectory {0} doesn't exist".format(d)) + + def test_readme_contents(self): + with open(os.path.join(self.role_dir, 'README.md'), 'r') as readme: + contents = readme.read() + + with open(os.path.join(self.role_skeleton_path, 'README.md'), 'r') as f: + expected_contents = f.read() + + self.assertEqual(expected_contents, contents, msg='README.md does not match expected') + + def test_test_yml(self): + with open(os.path.join(self.role_dir, 'tests', 'test.yml'), 'r') as f: + test_playbook = yaml.safe_load(f) + print(test_playbook) + self.assertEqual(len(test_playbook), 1) + self.assertEqual(test_playbook[0]['hosts'], 'localhost') + self.assertEqual(test_playbook[0]['remote_user'], 'root') + self.assertListEqual(test_playbook[0]['roles'], [self.role_name], msg='The list of roles included in the test play doesn\'t match') + + +class TestGalaxyInitDefault(unittest.TestCase, ValidRoleTests): + + @classmethod + def setUpClass(cls): + cls.setUpRole(role_name='delete_me') + + def test_metadata_contents(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertEqual(metadata.get('galaxy_info', dict()).get('author'), 'your name', msg='author was not set properly in metadata') + + +class TestGalaxyInitAPB(unittest.TestCase, ValidRoleTests): + + @classmethod + def setUpClass(cls): + cls.setUpRole('delete_me_apb', galaxy_args=['--type=apb']) + + def test_metadata_apb_tag(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertIn('apb', metadata.get('galaxy_info', dict()).get('galaxy_tags', []), msg='apb tag not set in role metadata') + + def test_metadata_contents(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertEqual(metadata.get('galaxy_info', dict()).get('author'), 'your name', msg='author was not set properly in metadata') + + def test_apb_yml(self): + self.assertTrue(os.path.exists(os.path.join(self.role_dir, 'apb.yml')), msg='apb.yml was not created') + + def test_test_yml(self): + with open(os.path.join(self.role_dir, 'tests', 'test.yml'), 'r') as f: + test_playbook = yaml.safe_load(f) + print(test_playbook) + self.assertEqual(len(test_playbook), 1) + self.assertEqual(test_playbook[0]['hosts'], 'localhost') + self.assertFalse(test_playbook[0]['gather_facts']) + self.assertEqual(test_playbook[0]['connection'], 'local') + self.assertIsNone(test_playbook[0]['tasks'], msg='We\'re expecting an unset list of tasks in test.yml') + + +class TestGalaxyInitContainer(unittest.TestCase, ValidRoleTests): + + @classmethod + def setUpClass(cls): + cls.setUpRole('delete_me_container', galaxy_args=['--type=container']) + + def test_metadata_container_tag(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertIn('container', metadata.get('galaxy_info', dict()).get('galaxy_tags', []), msg='container tag not set in role metadata') + + def test_metadata_contents(self): + with open(os.path.join(self.role_dir, 'meta', 'main.yml'), 'r') as mf: + metadata = yaml.safe_load(mf) + self.assertEqual(metadata.get('galaxy_info', dict()).get('author'), 'your name', msg='author was not set properly in metadata') + + def test_meta_container_yml(self): + self.assertTrue(os.path.exists(os.path.join(self.role_dir, 'meta', 'container.yml')), msg='container.yml was not created') + + def test_test_yml(self): + with open(os.path.join(self.role_dir, 'tests', 'test.yml'), 'r') as f: + test_playbook = yaml.safe_load(f) + print(test_playbook) + self.assertEqual(len(test_playbook), 1) + self.assertEqual(test_playbook[0]['hosts'], 'localhost') + self.assertFalse(test_playbook[0]['gather_facts']) + self.assertEqual(test_playbook[0]['connection'], 'local') + self.assertIsNone(test_playbook[0]['tasks'], msg='We\'re expecting an unset list of tasks in test.yml') + + +class TestGalaxyInitSkeleton(unittest.TestCase, ValidRoleTests): + + @classmethod + def setUpClass(cls): + role_skeleton_path = os.path.join(os.path.split(__file__)[0], 'test_data', 'role_skeleton') + cls.setUpRole('delete_me_skeleton', skeleton_path=role_skeleton_path, use_explicit_type=True) + + def test_empty_files_dir(self): + files_dir = os.path.join(self.role_dir, 'files') + self.assertTrue(os.path.isdir(files_dir)) + self.assertListEqual(os.listdir(files_dir), [], msg='we expect the files directory to be empty, is ignore working?') + + def test_template_ignore_jinja(self): + test_conf_j2 = os.path.join(self.role_dir, 'templates', 'test.conf.j2') + self.assertTrue(os.path.exists(test_conf_j2), msg="The test.conf.j2 template doesn't seem to exist, is it being rendered as test.conf?") + with open(test_conf_j2, 'r') as f: + contents = f.read() + expected_contents = '[defaults]\ntest_key = {{ test_variable }}' + self.assertEqual(expected_contents, contents.strip(), msg="test.conf.j2 doesn't contain what it should, is it being rendered?") + + def test_template_ignore_jinja_subfolder(self): + test_conf_j2 = os.path.join(self.role_dir, 'templates', 'subfolder', 'test.conf.j2') + self.assertTrue(os.path.exists(test_conf_j2), msg="The test.conf.j2 template doesn't seem to exist, is it being rendered as test.conf?") + with open(test_conf_j2, 'r') as f: + contents = f.read() + expected_contents = '[defaults]\ntest_key = {{ test_variable }}' + self.assertEqual(expected_contents, contents.strip(), msg="test.conf.j2 doesn't contain what it should, is it being rendered?") + + def test_template_ignore_similar_folder(self): + self.assertTrue(os.path.exists(os.path.join(self.role_dir, 'templates_extra', 'templates.txt'))) + + def test_skeleton_option(self): + self.assertEqual(self.role_skeleton_path, context.CLIARGS['role_skeleton'], msg='Skeleton path was not parsed properly from the command line') + + +@pytest.mark.parametrize('cli_args, expected', [ + (['ansible-galaxy', 'collection', 'init', 'abc._def'], 0), + (['ansible-galaxy', 'collection', 'init', 'abc._def', '-vvv'], 3), + (['ansible-galaxy', 'collection', 'init', 'abc._def', '-vv'], 2), +]) +def test_verbosity_arguments(cli_args, expected, monkeypatch): + # Mock out the functions so we don't actually execute anything + for func_name in [f for f in dir(GalaxyCLI) if f.startswith("execute_")]: + monkeypatch.setattr(GalaxyCLI, func_name, MagicMock()) + + cli = GalaxyCLI(args=cli_args) + cli.run() + + assert context.CLIARGS['verbosity'] == expected + + +@pytest.fixture() +def collection_skeleton(request, tmp_path_factory): + name, skeleton_path = request.param + + galaxy_args = ['ansible-galaxy', 'collection', 'init', '-c'] + + if skeleton_path is not None: + galaxy_args += ['--collection-skeleton', skeleton_path] + + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + galaxy_args += ['--init-path', test_dir, name] + + GalaxyCLI(args=galaxy_args).run() + namespace_name, collection_name = name.split('.', 1) + collection_dir = os.path.join(test_dir, namespace_name, collection_name) + + return collection_dir + + +@pytest.mark.parametrize('collection_skeleton', [ + ('ansible_test.my_collection', None), +], indirect=True) +def test_collection_default(collection_skeleton): + meta_path = os.path.join(collection_skeleton, 'galaxy.yml') + + with open(meta_path, 'r') as galaxy_meta: + metadata = yaml.safe_load(galaxy_meta) + + assert metadata['namespace'] == 'ansible_test' + assert metadata['name'] == 'my_collection' + assert metadata['authors'] == ['your name <example@domain.com>'] + assert metadata['readme'] == 'README.md' + assert metadata['version'] == '1.0.0' + assert metadata['description'] == 'your collection description' + assert metadata['license'] == ['GPL-2.0-or-later'] + assert metadata['tags'] == [] + assert metadata['dependencies'] == {} + assert metadata['documentation'] == 'http://docs.example.com' + assert metadata['repository'] == 'http://example.com/repository' + assert metadata['homepage'] == 'http://example.com' + assert metadata['issues'] == 'http://example.com/issue/tracker' + + for d in ['docs', 'plugins', 'roles']: + assert os.path.isdir(os.path.join(collection_skeleton, d)), \ + "Expected collection subdirectory {0} doesn't exist".format(d) + + +@pytest.mark.parametrize('collection_skeleton', [ + ('ansible_test.delete_me_skeleton', os.path.join(os.path.split(__file__)[0], 'test_data', 'collection_skeleton')), +], indirect=True) +def test_collection_skeleton(collection_skeleton): + meta_path = os.path.join(collection_skeleton, 'galaxy.yml') + + with open(meta_path, 'r') as galaxy_meta: + metadata = yaml.safe_load(galaxy_meta) + + assert metadata['namespace'] == 'ansible_test' + assert metadata['name'] == 'delete_me_skeleton' + assert metadata['authors'] == ['Ansible Cow <acow@bovineuniversity.edu>', 'Tu Cow <tucow@bovineuniversity.edu>'] + assert metadata['version'] == '0.1.0' + assert metadata['readme'] == 'README.md' + assert len(metadata) == 5 + + assert os.path.exists(os.path.join(collection_skeleton, 'README.md')) + + # Test empty directories exist and are empty + for empty_dir in ['plugins/action', 'plugins/filter', 'plugins/inventory', 'plugins/lookup', + 'plugins/module_utils', 'plugins/modules']: + + assert os.listdir(os.path.join(collection_skeleton, empty_dir)) == [] + + # Test files that don't end with .j2 were not templated + doc_file = os.path.join(collection_skeleton, 'docs', 'My Collection.md') + with open(doc_file, 'r') as f: + doc_contents = f.read() + assert doc_contents.strip() == 'Welcome to my test collection doc for {{ namespace }}.' + + # Test files that end with .j2 but are in the templates directory were not templated + for template_dir in ['playbooks/templates', 'playbooks/templates/subfolder', + 'roles/common/templates', 'roles/common/templates/subfolder']: + test_conf_j2 = os.path.join(collection_skeleton, template_dir, 'test.conf.j2') + assert os.path.exists(test_conf_j2) + + with open(test_conf_j2, 'r') as f: + contents = f.read() + expected_contents = '[defaults]\ntest_key = {{ test_variable }}' + + assert expected_contents == contents.strip() + + +@pytest.fixture() +def collection_artifact(collection_skeleton, tmp_path_factory): + ''' Creates a collection artifact tarball that is ready to be published and installed ''' + output_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Output')) + + # Create a file with +x in the collection so we can test the permissions + execute_path = os.path.join(collection_skeleton, 'runme.sh') + with open(execute_path, mode='wb') as fd: + fd.write(b"echo hi") + + # S_ISUID should not be present on extraction. + os.chmod(execute_path, os.stat(execute_path).st_mode | stat.S_ISUID | stat.S_IEXEC) + + # Because we call GalaxyCLI in collection_skeleton we need to reset the singleton back to None so it uses the new + # args, we reset the original args once it is done. + orig_cli_args = co.GlobalCLIArgs._Singleton__instance + try: + co.GlobalCLIArgs._Singleton__instance = None + galaxy_args = ['ansible-galaxy', 'collection', 'build', collection_skeleton, '--output-path', output_dir] + gc = GalaxyCLI(args=galaxy_args) + gc.run() + + yield output_dir + finally: + co.GlobalCLIArgs._Singleton__instance = orig_cli_args + + +def test_invalid_skeleton_path(): + expected = "- the skeleton path '/fake/path' does not exist, cannot init collection" + + gc = GalaxyCLI(args=['ansible-galaxy', 'collection', 'init', 'my.collection', '--collection-skeleton', + '/fake/path']) + with pytest.raises(AnsibleError, match=expected): + gc.run() + + +@pytest.mark.parametrize("name", [ + "", + "invalid", + "hypen-ns.collection", + "ns.hyphen-collection", + "ns.collection.weird", +]) +def test_invalid_collection_name_init(name): + expected = "Invalid collection name '%s', name must be in the format <namespace>.<collection>" % name + + gc = GalaxyCLI(args=['ansible-galaxy', 'collection', 'init', name]) + with pytest.raises(AnsibleError, match=expected): + gc.run() + + +@pytest.mark.parametrize("name, expected", [ + ("", ""), + ("invalid", "invalid"), + ("invalid:1.0.0", "invalid"), + ("hypen-ns.collection", "hypen-ns.collection"), + ("ns.hyphen-collection", "ns.hyphen-collection"), + ("ns.collection.weird", "ns.collection.weird"), +]) +def test_invalid_collection_name_install(name, expected, tmp_path_factory): + install_path = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections')) + + # FIXME: we should add the collection name in the error message + # Used to be: expected = "Invalid collection name '%s', name must be in the format <namespace>.<collection>" % expected + expected = "Neither the collection requirement entry key 'name', nor 'source' point to a concrete resolvable collection artifact. " + expected += r"Also 'name' is not an FQCN\. A valid collection name must be in the format <namespace>\.<collection>\. " + expected += r"Please make sure that the namespace and the collection name contain characters from \[a\-zA\-Z0\-9_\] only\." + + gc = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', name, '-p', os.path.join(install_path, 'install')]) + with pytest.raises(AnsibleError, match=expected): + gc.run() + + +@pytest.mark.parametrize('collection_skeleton', [ + ('ansible_test.build_collection', None), +], indirect=True) +def test_collection_build(collection_artifact): + tar_path = os.path.join(collection_artifact, 'ansible_test-build_collection-1.0.0.tar.gz') + assert tarfile.is_tarfile(tar_path) + + with tarfile.open(tar_path, mode='r') as tar: + tar_members = tar.getmembers() + + valid_files = ['MANIFEST.json', 'FILES.json', 'roles', 'docs', 'plugins', 'plugins/README.md', 'README.md', + 'runme.sh', 'meta', 'meta/runtime.yml'] + assert len(tar_members) == len(valid_files) + + # Verify the uid and gid is 0 and the correct perms are set + for member in tar_members: + assert member.name in valid_files + + assert member.gid == 0 + assert member.gname == '' + assert member.uid == 0 + assert member.uname == '' + if member.isdir() or member.name == 'runme.sh': + assert member.mode == 0o0755 + else: + assert member.mode == 0o0644 + + manifest_file = tar.extractfile(tar_members[0]) + try: + manifest = json.loads(to_text(manifest_file.read())) + finally: + manifest_file.close() + + coll_info = manifest['collection_info'] + file_manifest = manifest['file_manifest_file'] + assert manifest['format'] == 1 + assert len(manifest.keys()) == 3 + + assert coll_info['namespace'] == 'ansible_test' + assert coll_info['name'] == 'build_collection' + assert coll_info['version'] == '1.0.0' + assert coll_info['authors'] == ['your name <example@domain.com>'] + assert coll_info['readme'] == 'README.md' + assert coll_info['tags'] == [] + assert coll_info['description'] == 'your collection description' + assert coll_info['license'] == ['GPL-2.0-or-later'] + assert coll_info['license_file'] is None + assert coll_info['dependencies'] == {} + assert coll_info['repository'] == 'http://example.com/repository' + assert coll_info['documentation'] == 'http://docs.example.com' + assert coll_info['homepage'] == 'http://example.com' + assert coll_info['issues'] == 'http://example.com/issue/tracker' + assert len(coll_info.keys()) == 14 + + assert file_manifest['name'] == 'FILES.json' + assert file_manifest['ftype'] == 'file' + assert file_manifest['chksum_type'] == 'sha256' + assert file_manifest['chksum_sha256'] is not None # Order of keys makes it hard to verify the checksum + assert file_manifest['format'] == 1 + assert len(file_manifest.keys()) == 5 + + files_file = tar.extractfile(tar_members[1]) + try: + files = json.loads(to_text(files_file.read())) + finally: + files_file.close() + + assert len(files['files']) == 9 + assert files['format'] == 1 + assert len(files.keys()) == 2 + + valid_files_entries = ['.', 'roles', 'docs', 'plugins', 'plugins/README.md', 'README.md', 'runme.sh', 'meta', 'meta/runtime.yml'] + for file_entry in files['files']: + assert file_entry['name'] in valid_files_entries + assert file_entry['format'] == 1 + + if file_entry['name'] in ['plugins/README.md', 'runme.sh', 'meta/runtime.yml']: + assert file_entry['ftype'] == 'file' + assert file_entry['chksum_type'] == 'sha256' + # Can't test the actual checksum as the html link changes based on the version or the file contents + # don't matter + assert file_entry['chksum_sha256'] is not None + elif file_entry['name'] == 'README.md': + assert file_entry['ftype'] == 'file' + assert file_entry['chksum_type'] == 'sha256' + assert file_entry['chksum_sha256'] == '6d8b5f9b5d53d346a8cd7638a0ec26e75e8d9773d952162779a49d25da6ef4f5' + else: + assert file_entry['ftype'] == 'dir' + assert file_entry['chksum_type'] is None + assert file_entry['chksum_sha256'] is None + + assert len(file_entry.keys()) == 5 + + +@pytest.fixture() +def collection_install(reset_cli_args, tmp_path_factory, monkeypatch): + mock_install = MagicMock() + monkeypatch.setattr(ansible.cli.galaxy, 'install_collections', mock_install) + + mock_warning = MagicMock() + monkeypatch.setattr(ansible.utils.display.Display, 'warning', mock_warning) + + output_dir = to_text((tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Output'))) + yield mock_install, mock_warning, output_dir + + +def test_collection_install_with_names(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', 'namespace2.collection:1.0.1', + '--collections-path', output_dir] + GalaxyCLI(args=galaxy_args).run() + + collection_path = os.path.join(output_dir, 'ansible_collections') + assert os.path.isdir(collection_path) + + assert mock_warning.call_count == 1 + assert "The specified collections path '%s' is not part of the configured Ansible collections path" % output_dir \ + in mock_warning.call_args[0][0] + + assert mock_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_install.call_args[0][0]] + assert requirements == [('namespace.collection', '*', None, 'galaxy'), + ('namespace2.collection', '1.0.1', None, 'galaxy')] + assert mock_install.call_args[0][1] == collection_path + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + +def test_collection_install_with_requirements_file(collection_install): + mock_install, mock_warning, output_dir = collection_install + + requirements_file = os.path.join(output_dir, 'requirements.yml') + with open(requirements_file, 'wb') as req_obj: + req_obj.write(b'''--- +collections: +- namespace.coll +- name: namespace2.coll + version: '>2.0.1' +''') + + galaxy_args = ['ansible-galaxy', 'collection', 'install', '--requirements-file', requirements_file, + '--collections-path', output_dir] + GalaxyCLI(args=galaxy_args).run() + + collection_path = os.path.join(output_dir, 'ansible_collections') + assert os.path.isdir(collection_path) + + assert mock_warning.call_count == 1 + assert "The specified collections path '%s' is not part of the configured Ansible collections path" % output_dir \ + in mock_warning.call_args[0][0] + + assert mock_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_install.call_args[0][0]] + assert requirements == [('namespace.coll', '*', None, 'galaxy'), + ('namespace2.coll', '>2.0.1', None, 'galaxy')] + assert mock_install.call_args[0][1] == collection_path + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + +def test_collection_install_with_relative_path(collection_install, monkeypatch): + mock_install = collection_install[0] + + mock_req = MagicMock() + mock_req.return_value = {'collections': [('namespace.coll', '*', None, None)], 'roles': []} + monkeypatch.setattr(ansible.cli.galaxy.GalaxyCLI, '_parse_requirements_file', mock_req) + + monkeypatch.setattr(os, 'makedirs', MagicMock()) + + requirements_file = './requirements.myl' + collections_path = './ansible_collections' + galaxy_args = ['ansible-galaxy', 'collection', 'install', '--requirements-file', requirements_file, + '--collections-path', collections_path] + GalaxyCLI(args=galaxy_args).run() + + assert mock_install.call_count == 1 + assert mock_install.call_args[0][0] == [('namespace.coll', '*', None, None)] + assert mock_install.call_args[0][1] == os.path.abspath(collections_path) + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + assert mock_req.call_count == 1 + assert mock_req.call_args[0][0] == os.path.abspath(requirements_file) + + +def test_collection_install_with_unexpanded_path(collection_install, monkeypatch): + mock_install = collection_install[0] + + mock_req = MagicMock() + mock_req.return_value = {'collections': [('namespace.coll', '*', None, None)], 'roles': []} + monkeypatch.setattr(ansible.cli.galaxy.GalaxyCLI, '_parse_requirements_file', mock_req) + + monkeypatch.setattr(os, 'makedirs', MagicMock()) + + requirements_file = '~/requirements.myl' + collections_path = '~/ansible_collections' + galaxy_args = ['ansible-galaxy', 'collection', 'install', '--requirements-file', requirements_file, + '--collections-path', collections_path] + GalaxyCLI(args=galaxy_args).run() + + assert mock_install.call_count == 1 + assert mock_install.call_args[0][0] == [('namespace.coll', '*', None, None)] + assert mock_install.call_args[0][1] == os.path.expanduser(os.path.expandvars(collections_path)) + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + assert mock_req.call_count == 1 + assert mock_req.call_args[0][0] == os.path.expanduser(os.path.expandvars(requirements_file)) + + +def test_collection_install_in_collection_dir(collection_install, monkeypatch): + mock_install, mock_warning, output_dir = collection_install + + collections_path = C.COLLECTIONS_PATHS[0] + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', 'namespace2.collection:1.0.1', + '--collections-path', collections_path] + GalaxyCLI(args=galaxy_args).run() + + assert mock_warning.call_count == 0 + + assert mock_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_install.call_args[0][0]] + assert requirements == [('namespace.collection', '*', None, 'galaxy'), + ('namespace2.collection', '1.0.1', None, 'galaxy')] + assert mock_install.call_args[0][1] == os.path.join(collections_path, 'ansible_collections') + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + +def test_collection_install_with_url(monkeypatch, collection_install): + mock_install, dummy, output_dir = collection_install + + mock_open = MagicMock(return_value=BytesIO()) + monkeypatch.setattr(collection.concrete_artifact_manager, 'open_url', mock_open) + + mock_metadata = MagicMock(return_value={'namespace': 'foo', 'name': 'bar', 'version': 'v1.0.0'}) + monkeypatch.setattr(collection.concrete_artifact_manager, '_get_meta_from_tar', mock_metadata) + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'https://foo/bar/foo-bar-v1.0.0.tar.gz', + '--collections-path', output_dir] + GalaxyCLI(args=galaxy_args).run() + + collection_path = os.path.join(output_dir, 'ansible_collections') + assert os.path.isdir(collection_path) + + assert mock_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_install.call_args[0][0]] + assert requirements == [('foo.bar', 'v1.0.0', 'https://foo/bar/foo-bar-v1.0.0.tar.gz', 'url')] + assert mock_install.call_args[0][1] == collection_path + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + +def test_collection_install_name_and_requirements_fail(collection_install): + test_path = collection_install[2] + expected = 'The positional collection_name arg and --requirements-file are mutually exclusive.' + + with pytest.raises(AnsibleError, match=expected): + GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', + test_path, '--requirements-file', test_path]).run() + + +def test_collection_install_no_name_and_requirements_fail(collection_install): + test_path = collection_install[2] + expected = 'You must specify a collection name or a requirements file.' + + with pytest.raises(AnsibleError, match=expected): + GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', '--collections-path', test_path]).run() + + +def test_collection_install_path_with_ansible_collections(collection_install): + mock_install, mock_warning, output_dir = collection_install + + collection_path = os.path.join(output_dir, 'ansible_collections') + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', 'namespace2.collection:1.0.1', + '--collections-path', collection_path] + GalaxyCLI(args=galaxy_args).run() + + assert os.path.isdir(collection_path) + + assert mock_warning.call_count == 1 + assert "The specified collections path '%s' is not part of the configured Ansible collections path" \ + % collection_path in mock_warning.call_args[0][0] + + assert mock_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_install.call_args[0][0]] + assert requirements == [('namespace.collection', '*', None, 'galaxy'), + ('namespace2.collection', '1.0.1', None, 'galaxy')] + assert mock_install.call_args[0][1] == collection_path + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + assert mock_install.call_args[0][3] is False # ignore_errors + assert mock_install.call_args[0][4] is False # no_deps + assert mock_install.call_args[0][5] is False # force + assert mock_install.call_args[0][6] is False # force_deps + + +def test_collection_install_ignore_certs(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--ignore-certs'] + GalaxyCLI(args=galaxy_args).run() + + assert mock_install.call_args[0][3] is False + + +def test_collection_install_force(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--force'] + GalaxyCLI(args=galaxy_args).run() + + # mock_install args: collections, output_path, apis, ignore_errors, no_deps, force, force_deps + assert mock_install.call_args[0][5] is True + + +def test_collection_install_force_deps(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--force-with-deps'] + GalaxyCLI(args=galaxy_args).run() + + # mock_install args: collections, output_path, apis, ignore_errors, no_deps, force, force_deps + assert mock_install.call_args[0][6] is True + + +def test_collection_install_no_deps(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--no-deps'] + GalaxyCLI(args=galaxy_args).run() + + # mock_install args: collections, output_path, apis, ignore_errors, no_deps, force, force_deps + assert mock_install.call_args[0][4] is True + + +def test_collection_install_ignore(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--ignore-errors'] + GalaxyCLI(args=galaxy_args).run() + + # mock_install args: collections, output_path, apis, ignore_errors, no_deps, force, force_deps + assert mock_install.call_args[0][3] is True + + +def test_collection_install_custom_server(collection_install): + mock_install, mock_warning, output_dir = collection_install + + galaxy_args = ['ansible-galaxy', 'collection', 'install', 'namespace.collection', '--collections-path', output_dir, + '--server', 'https://galaxy-dev.ansible.com'] + GalaxyCLI(args=galaxy_args).run() + + assert len(mock_install.call_args[0][2]) == 1 + assert mock_install.call_args[0][2][0].api_server == 'https://galaxy-dev.ansible.com' + assert mock_install.call_args[0][2][0].validate_certs is True + + +@pytest.fixture() +def requirements_file(request, tmp_path_factory): + content = request.param + + test_dir = to_text(tmp_path_factory.mktemp('test-ÅÑŚÌβŁÈ Collections Requirements')) + requirements_file = os.path.join(test_dir, 'requirements.yml') + + if content: + with open(requirements_file, 'wb') as req_obj: + req_obj.write(to_bytes(content)) + + yield requirements_file + + +@pytest.fixture() +def requirements_cli(monkeypatch): + monkeypatch.setattr(GalaxyCLI, 'execute_install', MagicMock()) + cli = GalaxyCLI(args=['ansible-galaxy', 'install']) + cli.run() + return cli + + +@pytest.mark.parametrize('requirements_file', [None], indirect=True) +def test_parse_requirements_file_that_doesnt_exist(requirements_cli, requirements_file): + expected = "The requirements file '%s' does not exist." % to_native(requirements_file) + with pytest.raises(AnsibleError, match=expected): + requirements_cli._parse_requirements_file(requirements_file) + + +@pytest.mark.parametrize('requirements_file', ['not a valid yml file: hi: world'], indirect=True) +def test_parse_requirements_file_that_isnt_yaml(requirements_cli, requirements_file): + expected = "Failed to parse the requirements yml at '%s' with the following error" % to_native(requirements_file) + with pytest.raises(AnsibleError, match=expected): + requirements_cli._parse_requirements_file(requirements_file) + + +@pytest.mark.parametrize('requirements_file', [(''' +# Older role based requirements.yml +- galaxy.role +- anotherrole +''')], indirect=True) +def test_parse_requirements_in_older_format_illega(requirements_cli, requirements_file): + expected = "Expecting requirements file to be a dict with the key 'collections' that contains a list of " \ + "collections to install" + + with pytest.raises(AnsibleError, match=expected): + requirements_cli._parse_requirements_file(requirements_file, allow_old_format=False) + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- version: 1.0.0 +'''], indirect=True) +def test_parse_requirements_without_mandatory_name_key(requirements_cli, requirements_file): + # Used to be "Collections requirement entry should contain the key name." + # Should we check that either source or name is provided before using the dep resolver? + + expected = "Neither the collection requirement entry key 'name', nor 'source' point to a concrete resolvable collection artifact. " + expected += r"Also 'name' is not an FQCN\. A valid collection name must be in the format <namespace>\.<collection>\. " + expected += r"Please make sure that the namespace and the collection name contain characters from \[a\-zA\-Z0\-9_\] only\." + + with pytest.raises(AnsibleError, match=expected): + requirements_cli._parse_requirements_file(requirements_file) + + +@pytest.mark.parametrize('requirements_file', [(''' +collections: +- namespace.collection1 +- namespace.collection2 +'''), (''' +collections: +- name: namespace.collection1 +- name: namespace.collection2 +''')], indirect=True) +def test_parse_requirements(requirements_cli, requirements_file): + expected = { + 'roles': [], + 'collections': [('namespace.collection1', '*', None, 'galaxy'), ('namespace.collection2', '*', None, 'galaxy')] + } + actual = requirements_cli._parse_requirements_file(requirements_file) + actual['collections'] = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in actual.get('collections', [])] + + assert actual == expected + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- name: namespace.collection1 + version: ">=1.0.0,<=2.0.0" + source: https://galaxy-dev.ansible.com +- namespace.collection2'''], indirect=True) +def test_parse_requirements_with_extra_info(requirements_cli, requirements_file): + actual = requirements_cli._parse_requirements_file(requirements_file) + actual['collections'] = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in actual.get('collections', [])] + + assert len(actual['roles']) == 0 + assert len(actual['collections']) == 2 + assert actual['collections'][0][0] == 'namespace.collection1' + assert actual['collections'][0][1] == '>=1.0.0,<=2.0.0' + assert actual['collections'][0][2].api_server == 'https://galaxy-dev.ansible.com' + + assert actual['collections'][1] == ('namespace.collection2', '*', None, 'galaxy') + + +@pytest.mark.parametrize('requirements_file', [''' +roles: +- username.role_name +- src: username2.role_name2 +- src: ssh://github.com/user/repo + scm: git + +collections: +- namespace.collection2 +'''], indirect=True) +def test_parse_requirements_with_roles_and_collections(requirements_cli, requirements_file): + actual = requirements_cli._parse_requirements_file(requirements_file) + actual['collections'] = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in actual.get('collections', [])] + + assert len(actual['roles']) == 3 + assert actual['roles'][0].name == 'username.role_name' + assert actual['roles'][1].name == 'username2.role_name2' + assert actual['roles'][2].name == 'repo' + assert actual['roles'][2].src == 'ssh://github.com/user/repo' + + assert len(actual['collections']) == 1 + assert actual['collections'][0] == ('namespace.collection2', '*', None, 'galaxy') + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- name: namespace.collection +- name: namespace2.collection2 + source: https://galaxy-dev.ansible.com/ +- name: namespace3.collection3 + source: server +'''], indirect=True) +def test_parse_requirements_with_collection_source(requirements_cli, requirements_file): + galaxy_api = GalaxyAPI(requirements_cli.api, 'server', 'https://config-server') + requirements_cli.api_servers.append(galaxy_api) + + actual = requirements_cli._parse_requirements_file(requirements_file) + actual['collections'] = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in actual.get('collections', [])] + + assert actual['roles'] == [] + assert len(actual['collections']) == 3 + assert actual['collections'][0] == ('namespace.collection', '*', None, 'galaxy') + + assert actual['collections'][1][0] == 'namespace2.collection2' + assert actual['collections'][1][1] == '*' + assert actual['collections'][1][2].api_server == 'https://galaxy-dev.ansible.com/' + + assert actual['collections'][2][0] == 'namespace3.collection3' + assert actual['collections'][2][1] == '*' + assert actual['collections'][2][2].api_server == 'https://config-server' + + +@pytest.mark.parametrize('requirements_file', [''' +- username.included_role +- src: https://github.com/user/repo +'''], indirect=True) +def test_parse_requirements_roles_with_include(requirements_cli, requirements_file): + reqs = [ + 'ansible.role', + {'include': requirements_file}, + ] + parent_requirements = os.path.join(os.path.dirname(requirements_file), 'parent.yaml') + with open(to_bytes(parent_requirements), 'wb') as req_fd: + req_fd.write(to_bytes(yaml.safe_dump(reqs))) + + actual = requirements_cli._parse_requirements_file(parent_requirements) + + assert len(actual['roles']) == 3 + assert actual['collections'] == [] + assert actual['roles'][0].name == 'ansible.role' + assert actual['roles'][1].name == 'username.included_role' + assert actual['roles'][2].name == 'repo' + assert actual['roles'][2].src == 'https://github.com/user/repo' + + +@pytest.mark.parametrize('requirements_file', [''' +- username.role +- include: missing.yml +'''], indirect=True) +def test_parse_requirements_roles_with_include_missing(requirements_cli, requirements_file): + expected = "Failed to find include requirements file 'missing.yml' in '%s'" % to_native(requirements_file) + + with pytest.raises(AnsibleError, match=expected): + requirements_cli._parse_requirements_file(requirements_file) + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- namespace.name +roles: +- namespace.name +'''], indirect=True) +def test_install_implicit_role_with_collections(requirements_file, monkeypatch): + mock_collection_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_collection', mock_collection_install) + mock_role_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_role', mock_role_install) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'display', mock_display) + + cli = GalaxyCLI(args=['ansible-galaxy', 'install', '-r', requirements_file]) + cli.run() + + assert mock_collection_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_collection_install.call_args[0][0]] + assert requirements == [('namespace.name', '*', None, 'galaxy')] + assert mock_collection_install.call_args[0][1] == cli._get_default_collection_path() + + assert mock_role_install.call_count == 1 + assert len(mock_role_install.call_args[0][0]) == 1 + assert str(mock_role_install.call_args[0][0][0]) == 'namespace.name' + + found = False + for mock_call in mock_display.mock_calls: + if 'contains collections which will be ignored' in mock_call[1][0]: + found = True + break + assert not found + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- namespace.name +roles: +- namespace.name +'''], indirect=True) +def test_install_explicit_role_with_collections(requirements_file, monkeypatch): + mock_collection_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_collection', mock_collection_install) + mock_role_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_role', mock_role_install) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_display) + + cli = GalaxyCLI(args=['ansible-galaxy', 'role', 'install', '-r', requirements_file]) + cli.run() + + assert mock_collection_install.call_count == 0 + + assert mock_role_install.call_count == 1 + assert len(mock_role_install.call_args[0][0]) == 1 + assert str(mock_role_install.call_args[0][0][0]) == 'namespace.name' + + found = False + for mock_call in mock_display.mock_calls: + if 'contains collections which will be ignored' in mock_call[1][0]: + found = True + break + assert found + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- namespace.name +roles: +- namespace.name +'''], indirect=True) +def test_install_role_with_collections_and_path(requirements_file, monkeypatch): + mock_collection_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_collection', mock_collection_install) + mock_role_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_role', mock_role_install) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'warning', mock_display) + + cli = GalaxyCLI(args=['ansible-galaxy', 'install', '-p', 'path', '-r', requirements_file]) + cli.run() + + assert mock_collection_install.call_count == 0 + + assert mock_role_install.call_count == 1 + assert len(mock_role_install.call_args[0][0]) == 1 + assert str(mock_role_install.call_args[0][0][0]) == 'namespace.name' + + found = False + for mock_call in mock_display.mock_calls: + if 'contains collections which will be ignored' in mock_call[1][0]: + found = True + break + assert found + + +@pytest.mark.parametrize('requirements_file', [''' +collections: +- namespace.name +roles: +- namespace.name +'''], indirect=True) +def test_install_collection_with_roles(requirements_file, monkeypatch): + mock_collection_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_collection', mock_collection_install) + mock_role_install = MagicMock() + monkeypatch.setattr(GalaxyCLI, '_execute_install_role', mock_role_install) + + mock_display = MagicMock() + monkeypatch.setattr(Display, 'vvv', mock_display) + + cli = GalaxyCLI(args=['ansible-galaxy', 'collection', 'install', '-r', requirements_file]) + cli.run() + + assert mock_collection_install.call_count == 1 + requirements = [('%s.%s' % (r.namespace, r.name), r.ver, r.src, r.type,) for r in mock_collection_install.call_args[0][0]] + assert requirements == [('namespace.name', '*', None, 'galaxy')] + + assert mock_role_install.call_count == 0 + + found = False + for mock_call in mock_display.mock_calls: + if 'contains roles which will be ignored' in mock_call[1][0]: + found = True + break + assert found diff --git a/test/units/cli/test_playbook.py b/test/units/cli/test_playbook.py new file mode 100644 index 0000000..f25e54d --- /dev/null +++ b/test/units/cli/test_playbook.py @@ -0,0 +1,46 @@ +# (c) 2016, Adrian Likins <alikins@redhat.com> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +from units.compat import unittest +from units.mock.loader import DictDataLoader + +from ansible import context +from ansible.inventory.manager import InventoryManager +from ansible.vars.manager import VariableManager + +from ansible.cli.playbook import PlaybookCLI + + +class TestPlaybookCLI(unittest.TestCase): + def test_flush_cache(self): + cli = PlaybookCLI(args=["ansible-playbook", "--flush-cache", "foobar.yml"]) + cli.parse() + self.assertTrue(context.CLIARGS['flush_cache']) + + variable_manager = VariableManager() + fake_loader = DictDataLoader({'foobar.yml': ""}) + inventory = InventoryManager(loader=fake_loader, sources='testhost,') + + variable_manager.set_host_facts('testhost', {'canary': True}) + self.assertTrue('testhost' in variable_manager._fact_cache) + + cli._flush_cache(inventory, variable_manager) + self.assertFalse('testhost' in variable_manager._fact_cache) diff --git a/test/units/cli/test_vault.py b/test/units/cli/test_vault.py new file mode 100644 index 0000000..2304f4d --- /dev/null +++ b/test/units/cli/test_vault.py @@ -0,0 +1,230 @@ +# -*- coding: utf-8 -*- +# (c) 2017, Adrian Likins <alikins@redhat.com> +# +# This file is part of Ansible +# +# Ansible is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# Ansible is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ansible. If not, see <http://www.gnu.org/licenses/>. + +# Make coding more python3-ish +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +import os +import pytest + +from units.compat import unittest +from unittest.mock import patch, MagicMock +from units.mock.vault_helper import TextVaultSecret + +from ansible import context, errors +from ansible.cli.vault import VaultCLI +from ansible.module_utils._text import to_text +from ansible.utils import context_objects as co + + +# TODO: make these tests assert something, likely by verifing +# mock calls + + +@pytest.fixture(autouse='function') +def reset_cli_args(): + co.GlobalCLIArgs._Singleton__instance = None + yield + co.GlobalCLIArgs._Singleton__instance = None + + +class TestVaultCli(unittest.TestCase): + def setUp(self): + self.tty_patcher = patch('ansible.cli.sys.stdin.isatty', return_value=False) + self.mock_isatty = self.tty_patcher.start() + + def tearDown(self): + self.tty_patcher.stop() + + def test_parse_empty(self): + cli = VaultCLI(['vaultcli']) + self.assertRaises(SystemExit, + cli.parse) + + # FIXME: something weird seems to be afoot when parsing actions + # cli = VaultCLI(args=['view', '/dev/null/foo', 'mysecret3']) + # will skip '/dev/null/foo'. something in cli.CLI.set_action() ? + # maybe we self.args gets modified in a loop? + def test_parse_view_file(self): + cli = VaultCLI(args=['ansible-vault', 'view', '/dev/null/foo']) + cli.parse() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + def test_view_missing_file_no_secret(self, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [] + cli = VaultCLI(args=['ansible-vault', 'view', '/dev/null/foo']) + cli.parse() + self.assertRaisesRegex(errors.AnsibleOptionsError, + "A vault password is required to use Ansible's Vault", + cli.run) + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + def test_encrypt_missing_file_no_secret(self, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [] + cli = VaultCLI(args=['ansible-vault', 'encrypt', '/dev/null/foo']) + cli.parse() + self.assertRaisesRegex(errors.AnsibleOptionsError, + "A vault password is required to use Ansible's Vault", + cli.run) + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_encrypt(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'encrypt', '/dev/null/foo']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_encrypt_string(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'encrypt_string', + 'some string to encrypt']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + @patch('ansible.cli.vault.display.prompt', return_value='a_prompt') + def test_encrypt_string_prompt(self, mock_display, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', + 'encrypt_string', + '--prompt', + '--show-input', + 'some string to encrypt']) + cli.parse() + cli.run() + args, kwargs = mock_display.call_args + assert kwargs["private"] is False + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + @patch('ansible.cli.vault.display.prompt', return_value='a_prompt') + def test_shadowed_encrypt_string_prompt(self, mock_display, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', + 'encrypt_string', + '--prompt', + 'some string to encrypt']) + cli.parse() + cli.run() + args, kwargs = mock_display.call_args + assert kwargs["private"] + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + @patch('ansible.cli.vault.sys.stdin.read', return_value='This is data from stdin') + def test_encrypt_string_stdin(self, mock_stdin_read, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', + 'encrypt_string', + '--stdin-name', + 'the_var_from_stdin', + '-']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_encrypt_string_names(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'encrypt_string', + '--name', 'foo1', + '--name', 'foo2', + 'some string to encrypt']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_encrypt_string_more_args_than_names(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'encrypt_string', + '--name', 'foo1', + 'some string to encrypt', + 'other strings', + 'a few more string args']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_create(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'create', '/dev/null/foo']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_edit(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'edit', '/dev/null/foo']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_decrypt(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'decrypt', '/dev/null/foo']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_view(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'view', '/dev/null/foo']) + cli.parse() + cli.run() + + @patch('ansible.cli.vault.VaultCLI.setup_vault_secrets') + @patch('ansible.cli.vault.VaultEditor') + def test_rekey(self, mock_vault_editor, mock_setup_vault_secrets): + mock_setup_vault_secrets.return_value = [('default', TextVaultSecret('password'))] + cli = VaultCLI(args=['ansible-vault', 'rekey', '/dev/null/foo']) + cli.parse() + cli.run() + + +@pytest.mark.parametrize('cli_args, expected', [ + (['ansible-vault', 'view', 'vault.txt'], 0), + (['ansible-vault', 'view', 'vault.txt', '-vvv'], 3), + (['ansible-vault', 'view', 'vault.txt', '-vv'], 2), +]) +def test_verbosity_arguments(cli_args, expected, tmp_path_factory, monkeypatch): + # Add a password file so we don't get a prompt in the test + test_dir = to_text(tmp_path_factory.mktemp('test-ansible-vault')) + pass_file = os.path.join(test_dir, 'pass.txt') + with open(pass_file, 'w') as pass_fd: + pass_fd.write('password') + + cli_args.extend(['--vault-id', pass_file]) + + # Mock out the functions so we don't actually execute anything + for func_name in [f for f in dir(VaultCLI) if f.startswith("execute_")]: + monkeypatch.setattr(VaultCLI, func_name, MagicMock()) + + cli = VaultCLI(args=cli_args) + cli.run() + + assert context.CLIARGS['verbosity'] == expected |