diff options
Diffstat (limited to 'test/units/plugins')
19 files changed, 0 insertions, 2440 deletions
diff --git a/test/units/plugins/become/test_doas.py b/test/units/plugins/become/test_doas.py deleted file mode 100644 index 00b1b1b7fb..0000000000 --- a/test/units/plugins/become/test_doas.py +++ /dev/null @@ -1,39 +0,0 @@ -# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> -# (c) 2020 Ansible Project -# -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -# Make coding more python3-ish -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import re - -from ansible import context -from ansible.playbook.play_context import PlayContext -from ansible.plugins.loader import become_loader - - -def test_doas(mocker, parser, reset_cli_args): - options = parser.parse_args([]) - context._init_global_context(options) - play_context = PlayContext() - - default_cmd = "/bin/foo" - default_exe = "/bin/bash" - doas_exe = 'doas' - doas_flags = '-n' - - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert cmd == default_cmd - - success = 'BECOME-SUCCESS-.+?' - - play_context.become = True - play_context.become_user = 'foo' - play_context.set_become_plugin(become_loader.get('doas')) - play_context.become_method = 'doas' - play_context.become_flags = doas_flags - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert (re.match("""%s %s -u %s %s -c 'echo %s; %s'""" % (doas_exe, doas_flags, play_context.become_user, default_exe, success, - default_cmd), cmd) is not None) diff --git a/test/units/plugins/become/test_dzdo.py b/test/units/plugins/become/test_dzdo.py deleted file mode 100644 index 2110a7dcbe..0000000000 --- a/test/units/plugins/become/test_dzdo.py +++ /dev/null @@ -1,44 +0,0 @@ -# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> -# (c) 2020 Ansible Project -# -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -# Make coding more python3-ish -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import re - -from ansible import context -from ansible.playbook.play_context import PlayContext -from ansible.plugins.loader import become_loader - - -def test_dzdo(mocker, parser, reset_cli_args): - options = parser.parse_args([]) - context._init_global_context(options) - play_context = PlayContext() - - default_cmd = "/bin/foo" - default_exe = "/bin/bash" - dzdo_exe = 'dzdo' - dzdo_flags = '' - - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert cmd == default_cmd - - success = 'BECOME-SUCCESS-.+?' - - play_context.become = True - play_context.become_user = 'foo' - play_context.set_become_plugin(become_loader.get('dzdo')) - play_context.become_method = 'dzdo' - play_context.become_flags = dzdo_flags - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert re.match("""%s %s -u %s %s -c 'echo %s; %s'""" % (dzdo_exe, dzdo_flags, play_context.become_user, default_exe, - success, default_cmd), cmd) is not None - play_context.become_pass = 'testpass' - play_context.set_become_plugin(become_loader.get('dzdo')) - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert re.match("""%s %s -p %s -u %s %s -c 'echo %s; %s'""" % (dzdo_exe, dzdo_flags, r'\"\[dzdo via ansible, key=.+?\] password:\"', - play_context.become_user, default_exe, success, default_cmd), cmd) is not None diff --git a/test/units/plugins/become/test_ksu.py b/test/units/plugins/become/test_ksu.py deleted file mode 100644 index 7906fe1dda..0000000000 --- a/test/units/plugins/become/test_ksu.py +++ /dev/null @@ -1,39 +0,0 @@ -# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> -# (c) 2020 Ansible Project -# -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -# Make coding more python3-ish -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import re - -from ansible import context -from ansible.playbook.play_context import PlayContext -from ansible.plugins.loader import become_loader - - -def test_ksu(mocker, parser, reset_cli_args): - options = parser.parse_args([]) - context._init_global_context(options) - play_context = PlayContext() - - default_cmd = "/bin/foo" - default_exe = "/bin/bash" - ksu_exe = 'ksu' - ksu_flags = '' - - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert cmd == default_cmd - - success = 'BECOME-SUCCESS-.+?' - - play_context.become = True - play_context.become_user = 'foo' - play_context.set_become_plugin(become_loader.get('ksu')) - play_context.become_method = 'ksu' - play_context.become_flags = ksu_flags - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert (re.match("""%s %s %s -e %s -c 'echo %s; %s'""" % (ksu_exe, play_context.become_user, ksu_flags, - default_exe, success, default_cmd), cmd) is not None) diff --git a/test/units/plugins/become/test_pbrun.py b/test/units/plugins/become/test_pbrun.py deleted file mode 100644 index a008b8ae75..0000000000 --- a/test/units/plugins/become/test_pbrun.py +++ /dev/null @@ -1,39 +0,0 @@ -# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> -# (c) 2020 Ansible Project -# -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -# Make coding more python3-ish -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import re - -from ansible import context -from ansible.playbook.play_context import PlayContext -from ansible.plugins.loader import become_loader - - -def test_pbrun(mocker, parser, reset_cli_args): - options = parser.parse_args([]) - context._init_global_context(options) - play_context = PlayContext() - - default_cmd = "/bin/foo" - default_exe = "/bin/bash" - pbrun_exe = 'pbrun' - pbrun_flags = '' - - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert cmd == default_cmd - - success = 'BECOME-SUCCESS-.+?' - - play_context.become = True - play_context.become_user = 'foo' - play_context.set_become_plugin(become_loader.get('pbrun')) - play_context.become_method = 'pbrun' - play_context.become_flags = pbrun_flags - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert re.match("""%s %s -u %s 'echo %s; %s'""" % (pbrun_exe, pbrun_flags, play_context.become_user, - success, default_cmd), cmd) is not None diff --git a/test/units/plugins/become/test_pfexec.py b/test/units/plugins/become/test_pfexec.py deleted file mode 100644 index e1c54e78b5..0000000000 --- a/test/units/plugins/become/test_pfexec.py +++ /dev/null @@ -1,38 +0,0 @@ -# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com> -# (c) 2020 Ansible Project -# -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -# Make coding more python3-ish -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import re - -from ansible import context -from ansible.playbook.play_context import PlayContext -from ansible.plugins.loader import become_loader - - -def test_pfexec(mocker, parser, reset_cli_args): - options = parser.parse_args([]) - context._init_global_context(options) - play_context = PlayContext() - - default_cmd = "/bin/foo" - default_exe = "/bin/bash" - pfexec_exe = 'pfexec' - pfexec_flags = '' - - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert cmd == default_cmd - - success = 'BECOME-SUCCESS-.+?' - - play_context.become = True - play_context.become_user = 'foo' - play_context.set_become_plugin(become_loader.get('pfexec')) - play_context.become_method = 'pfexec' - play_context.become_flags = pfexec_flags - cmd = play_context.make_become_cmd(cmd=default_cmd, executable=default_exe) - assert re.match('''%s %s "'echo %s; %s'"''' % (pfexec_exe, pfexec_flags, success, default_cmd), cmd) is not None diff --git a/test/units/plugins/cache/test_memcached.py b/test/units/plugins/cache/test_memcached.py deleted file mode 100644 index 5e997a7729..0000000000 --- a/test/units/plugins/cache/test_memcached.py +++ /dev/null @@ -1,35 +0,0 @@ -# (c) 2012-2015, Michael DeHaan <michael.dehaan@gmail.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 pytest - -pytest.importorskip('memcache') - -from ansible.plugins.loader import cache_loader -from ansible.plugins.cache.memcached import CacheModule as MemcachedCache - - -def test_memcached_cachemodule(): - assert isinstance(MemcachedCache(), MemcachedCache) - - -def test_memcached_cachemodule_with_loader(): - assert isinstance(cache_loader.get('memcached'), MemcachedCache) diff --git a/test/units/plugins/cache/test_redis.py b/test/units/plugins/cache/test_redis.py deleted file mode 100644 index 3b242304f9..0000000000 --- a/test/units/plugins/cache/test_redis.py +++ /dev/null @@ -1,36 +0,0 @@ -# (c) 2012-2015, Michael DeHaan <michael.dehaan@gmail.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 pytest - -pytest.importorskip('redis') - -from ansible.plugins.loader import cache_loader -from ansible.plugins.cache.redis import CacheModule as RedisCache - - -def test_redis_cachemodule(): - assert isinstance(RedisCache(), RedisCache) - - -def test_redis_cachemodule_with_loader(): - # The _uri option is required for the redis plugin - assert isinstance(cache_loader.get('redis', **{'_uri': '127.0.0.1:6379:1'}), RedisCache) diff --git a/test/units/plugins/cliconf/test_nos.py b/test/units/plugins/cliconf/test_nos.py deleted file mode 100644 index 22d0be4468..0000000000 --- a/test/units/plugins/cliconf/test_nos.py +++ /dev/null @@ -1,137 +0,0 @@ -# -# (c) 2018 Extreme Networks Inc. -# -# 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/>. -# -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -from os import path -import json - -from mock import MagicMock, call - -from units.compat import unittest -from ansible.plugins.cliconf import nos - -FIXTURE_DIR = b'%s/fixtures/nos' % ( - path.dirname(path.abspath(__file__)).encode('utf-8') -) - - -def _connection_side_effect(*args, **kwargs): - try: - if args: - value = args[0] - else: - value = kwargs.get('command') - - fixture_path = path.abspath( - b'%s/%s' % (FIXTURE_DIR, b'_'.join(value.split(b' '))) - ) - with open(fixture_path, 'rb') as file_desc: - return file_desc.read() - except (OSError, IOError): - if args: - value = args[0] - return value - elif kwargs.get('command'): - value = kwargs.get('command') - return value - - return 'Nope' - - -class TestPluginCLIConfNOS(unittest.TestCase): - """ Test class for NOS CLI Conf Methods - """ - def setUp(self): - self._mock_connection = MagicMock() - self._mock_connection.send.side_effect = _connection_side_effect - self._cliconf = nos.Cliconf(self._mock_connection) - self.maxDiff = None - - def tearDown(self): - pass - - def test_get_device_info(self): - """ Test get_device_info - """ - device_info = self._cliconf.get_device_info() - - mock_device_info = { - 'network_os': 'nos', - 'network_os_model': 'BR-VDX6740', - 'network_os_version': '7.2.0', - } - - self.assertEqual(device_info, mock_device_info) - - def test_get_config(self): - """ Test get_config - """ - running_config = self._cliconf.get_config() - - fixture_path = path.abspath(b'%s/show_running-config' % FIXTURE_DIR) - with open(fixture_path, 'rb') as file_desc: - mock_running_config = file_desc.read() - self.assertEqual(running_config, mock_running_config) - - def test_edit_config(self): - """ Test edit_config - """ - test_config_command = b'this\nis\nthe\nsong\nthat\nnever\nends' - - self._cliconf.edit_config(test_config_command) - - send_calls = [] - - for command in [b'configure terminal', test_config_command, b'end']: - send_calls.append(call( - command=command, - prompt_retry_check=False, - sendonly=False, - newline=True, - check_all=False - )) - - self._mock_connection.send.assert_has_calls(send_calls) - - def test_get_capabilities(self): - """ Test get_capabilities - """ - capabilities = json.loads(self._cliconf.get_capabilities()) - mock_capabilities = { - 'network_api': 'cliconf', - 'rpc': [ - 'get_config', - 'edit_config', - 'get_capabilities', - 'get', - 'enable_response_logging', - 'disable_response_logging' - ], - 'device_info': { - 'network_os_model': 'BR-VDX6740', - 'network_os_version': '7.2.0', - 'network_os': 'nos' - } - } - - self.assertEqual( - mock_capabilities, - capabilities - ) diff --git a/test/units/plugins/cliconf/test_slxos.py b/test/units/plugins/cliconf/test_slxos.py deleted file mode 100644 index 5befc768f5..0000000000 --- a/test/units/plugins/cliconf/test_slxos.py +++ /dev/null @@ -1,144 +0,0 @@ -# -# (c) 2018 Extreme Networks Inc. -# -# 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/>. -# -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -from os import path -import json - -from mock import MagicMock, call - -from units.compat import unittest -from ansible.plugins.cliconf import slxos - -FIXTURE_DIR = b'%s/fixtures/slxos' % ( - path.dirname(path.abspath(__file__)).encode('utf-8') -) - - -def _connection_side_effect(*args, **kwargs): - try: - if args: - value = args[0] - else: - value = kwargs.get('command') - - fixture_path = path.abspath( - b'%s/%s' % (FIXTURE_DIR, b'_'.join(value.split(b' '))) - ) - with open(fixture_path, 'rb') as file_desc: - return file_desc.read() - except (OSError, IOError): - if args: - value = args[0] - return value - elif kwargs.get('command'): - value = kwargs.get('command') - return value - - return 'Nope' - - -class TestPluginCLIConfSLXOS(unittest.TestCase): - """ Test class for SLX-OS CLI Conf Methods - """ - def setUp(self): - self._mock_connection = MagicMock() - self._mock_connection.send.side_effect = _connection_side_effect - self._cliconf = slxos.Cliconf(self._mock_connection) - self.maxDiff = None - - def tearDown(self): - pass - - def test_get_device_info(self): - """ Test get_device_info - """ - device_info = self._cliconf.get_device_info() - - mock_device_info = { - 'network_os': 'slxos', - 'network_os_model': 'BR-SLX9140', - 'network_os_version': '17s.1.02', - } - - self.assertEqual(device_info, mock_device_info) - - def test_get_config(self): - """ Test get_config - """ - running_config = self._cliconf.get_config() - - fixture_path = path.abspath(b'%s/show_running-config' % FIXTURE_DIR) - with open(fixture_path, 'rb') as file_desc: - mock_running_config = file_desc.read() - self.assertEqual(running_config, mock_running_config) - - startup_config = self._cliconf.get_config() - - fixture_path = path.abspath(b'%s/show_running-config' % FIXTURE_DIR) - with open(fixture_path, 'rb') as file_desc: - mock_startup_config = file_desc.read() - self.assertEqual(startup_config, mock_startup_config) - - def test_edit_config(self): - """ Test edit_config - """ - test_config_command = b'this\nis\nthe\nsong\nthat\nnever\nends' - - self._cliconf.edit_config(test_config_command) - - send_calls = [] - - for command in [b'configure terminal', test_config_command, b'end']: - send_calls.append(call( - command=command, - prompt_retry_check=False, - sendonly=False, - newline=True, - check_all=False - )) - - self._mock_connection.send.assert_has_calls(send_calls) - - def test_get_capabilities(self): - """ Test get_capabilities - """ - capabilities = json.loads(self._cliconf.get_capabilities()) - mock_capabilities = { - 'network_api': 'cliconf', - 'rpc': [ - 'get_config', - 'edit_config', - 'get_capabilities', - 'get', - 'enable_response_logging', - 'disable_response_logging' - ], - 'device_info': { - 'network_os_model': 'BR-SLX9140', - 'network_os_version': '17s.1.02', - 'network_os': 'slxos' - } - } - - self.assertEqual( - mock_capabilities, - capabilities - ) diff --git a/test/units/plugins/connection/test_docker.py b/test/units/plugins/connection/test_docker.py deleted file mode 100644 index d371550214..0000000000 --- a/test/units/plugins/connection/test_docker.py +++ /dev/null @@ -1,61 +0,0 @@ -# (c) 2020 Red Hat, Inc. -# -# 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 io import StringIO -import pytest - -from units.compat import mock -from units.compat import unittest -from ansible.errors import AnsibleError -from ansible.playbook.play_context import PlayContext -from ansible.plugins.connection.docker import Connection as DockerConnection - - -class TestDockerConnectionClass(unittest.TestCase): - - def setUp(self): - self.play_context = PlayContext() - self.play_context.prompt = ( - '[sudo via ansible, key=ouzmdnewuhucvuaabtjmweasarviygqq] password: ' - ) - self.in_stream = StringIO() - - def tearDown(self): - pass - - @mock.patch('ansible.plugins.connection.docker.Connection._old_docker_version', return_value=('false', 'garbage', '', 1)) - @mock.patch('ansible.plugins.connection.docker.Connection._new_docker_version', return_value=('docker version', '1.2.3', '', 0)) - def test_docker_connection_module_too_old(self, mock_new_docker_verison, mock_old_docker_version): - self.assertRaisesRegexp(AnsibleError, '^docker connection type requires docker 1.3 or higher$', - DockerConnection, self.play_context, self.in_stream, docker_command='/fake/docker') - - @mock.patch('ansible.plugins.connection.docker.Connection._old_docker_version', return_value=('false', 'garbage', '', 1)) - @mock.patch('ansible.plugins.connection.docker.Connection._new_docker_version', return_value=('docker version', '1.3.4', '', 0)) - def test_docker_connection_module(self, mock_new_docker_verison, mock_old_docker_version): - self.assertIsInstance(DockerConnection(self.play_context, self.in_stream, docker_command='/fake/docker'), - DockerConnection) - - # old version and new version fail - @mock.patch('ansible.plugins.connection.docker.Connection._old_docker_version', return_value=('false', 'garbage', '', 1)) - @mock.patch('ansible.plugins.connection.docker.Connection._new_docker_version', return_value=('false', 'garbage', '', 1)) - def test_docker_connection_module_wrong_cmd(self, mock_new_docker_version, mock_old_docker_version): - self.assertRaisesRegexp(AnsibleError, '^Docker version check (.*?) failed: ', - DockerConnection, self.play_context, self.in_stream, docker_command='/fake/docker') diff --git a/test/units/plugins/connection/test_lxc.py b/test/units/plugins/connection/test_lxc.py deleted file mode 100644 index f754406c5d..0000000000 --- a/test/units/plugins/connection/test_lxc.py +++ /dev/null @@ -1,40 +0,0 @@ -# -# (c) 2020 Red Hat Inc. -# -# 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 io import StringIO -import pytest - -from units.compat import unittest -from ansible.plugins.connection import lxc -from ansible.playbook.play_context import PlayContext - - -class TestLXCConnectionClass(unittest.TestCase): - - def test_lxc_connection_module(self): - play_context = PlayContext() - play_context.prompt = ( - '[sudo via ansible, key=ouzmdnewuhucvuaabtjmweasarviygqq] password: ' - ) - in_stream = StringIO() - - self.assertIsInstance(lxc.Connection(play_context, in_stream), lxc.Connection) diff --git a/test/units/plugins/httpapi/test_ftd.py b/test/units/plugins/httpapi/test_ftd.py deleted file mode 100644 index 7767ea2db8..0000000000 --- a/test/units/plugins/httpapi/test_ftd.py +++ /dev/null @@ -1,413 +0,0 @@ -# Copyright (c) 2018 Cisco and/or its affiliates. -# -# 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/>. -# - -import json - -from ansible.module_utils.six.moves.urllib.error import HTTPError -from units.compat import mock -from units.compat import unittest -from units.compat.mock import mock_open, patch - -from ansible.errors import AnsibleConnectionFailure -from ansible.module_utils.connection import ConnectionError -from ansible.module_utils.network.ftd.common import HTTPMethod, ResponseParams -from ansible.module_utils.network.ftd.fdm_swagger_client import SpecProp, FdmSwaggerParser -from ansible.module_utils.six import BytesIO, PY3, StringIO -from ansible.plugins.httpapi.ftd import HttpApi, BASE_HEADERS, TOKEN_PATH_TEMPLATE, DEFAULT_API_VERSIONS - - -if PY3: - BUILTINS_NAME = 'builtins' -else: - BUILTINS_NAME = '__builtin__' - - -class FakeFtdHttpApiPlugin(HttpApi): - def __init__(self, conn): - super(FakeFtdHttpApiPlugin, self).__init__(conn) - self.hostvars = { - 'token_path': '/testLoginUrl', - 'spec_path': '/testSpecUrl' - } - - def get_option(self, var): - return self.hostvars[var] - - def set_option(self, var, val): - self.hostvars[var] = val - - -class TestFtdHttpApi(unittest.TestCase): - - def setUp(self): - self.connection_mock = mock.Mock() - self.ftd_plugin = FakeFtdHttpApiPlugin(self.connection_mock) - self.ftd_plugin.access_token = 'ACCESS_TOKEN' - self.ftd_plugin._load_name = 'httpapi' - - def test_login_should_request_tokens_when_no_refresh_token(self): - self.connection_mock.send.return_value = self._connection_response( - {'access_token': 'ACCESS_TOKEN', 'refresh_token': 'REFRESH_TOKEN'} - ) - - self.ftd_plugin.login('foo', 'bar') - - assert 'ACCESS_TOKEN' == self.ftd_plugin.access_token - assert 'REFRESH_TOKEN' == self.ftd_plugin.refresh_token - assert {'Authorization': 'Bearer ACCESS_TOKEN'} == self.ftd_plugin.connection._auth - expected_body = json.dumps({'grant_type': 'password', 'username': 'foo', 'password': 'bar'}) - self.connection_mock.send.assert_called_once_with(mock.ANY, expected_body, headers=mock.ANY, method=mock.ANY) - - def test_login_should_update_tokens_when_refresh_token_exists(self): - self.ftd_plugin.refresh_token = 'REFRESH_TOKEN' - self.connection_mock.send.return_value = self._connection_response( - {'access_token': 'NEW_ACCESS_TOKEN', 'refresh_token': 'NEW_REFRESH_TOKEN'} - ) - - self.ftd_plugin.login('foo', 'bar') - - assert 'NEW_ACCESS_TOKEN' == self.ftd_plugin.access_token - assert 'NEW_REFRESH_TOKEN' == self.ftd_plugin.refresh_token - assert {'Authorization': 'Bearer NEW_ACCESS_TOKEN'} == self.ftd_plugin.connection._auth - expected_body = json.dumps({'grant_type': 'refresh_token', 'refresh_token': 'REFRESH_TOKEN'}) - self.connection_mock.send.assert_called_once_with(mock.ANY, expected_body, headers=mock.ANY, method=mock.ANY) - - def test_login_should_use_env_variable_when_set(self): - temp_token_path = self.ftd_plugin.hostvars['token_path'] - self.ftd_plugin.hostvars['token_path'] = '/testFakeLoginUrl' - self.connection_mock.send.return_value = self._connection_response( - {'access_token': 'ACCESS_TOKEN', 'refresh_token': 'REFRESH_TOKEN'} - ) - - self.ftd_plugin.login('foo', 'bar') - - self.connection_mock.send.assert_called_once_with('/testFakeLoginUrl', mock.ANY, headers=mock.ANY, - method=mock.ANY) - self.ftd_plugin.hostvars['token_path'] = temp_token_path - - def test_login_raises_exception_when_no_refresh_token_and_no_credentials(self): - with self.assertRaises(AnsibleConnectionFailure) as res: - self.ftd_plugin.login(None, None) - assert 'Username and password are required' in str(res.exception) - - def test_login_raises_exception_when_invalid_response(self): - self.connection_mock.send.return_value = self._connection_response( - {'no_access_token': 'ACCESS_TOKEN'} - ) - - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin.login('foo', 'bar') - - assert 'Server returned response without token info during connection authentication' in str(res.exception) - - def test_login_raises_exception_when_http_error(self): - self.connection_mock.send.side_effect = HTTPError('http://testhost.com', 400, '', {}, - StringIO('{"message": "Failed to authenticate user"}')) - - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin.login('foo', 'bar') - - assert 'Failed to authenticate user' in str(res.exception) - - def test_logout_should_revoke_tokens(self): - self.ftd_plugin.access_token = 'ACCESS_TOKEN_TO_REVOKE' - self.ftd_plugin.refresh_token = 'REFRESH_TOKEN_TO_REVOKE' - self.connection_mock.send.return_value = self._connection_response(None) - - self.ftd_plugin.logout() - - assert self.ftd_plugin.access_token is None - assert self.ftd_plugin.refresh_token is None - expected_body = json.dumps({'grant_type': 'revoke_token', 'access_token': 'ACCESS_TOKEN_TO_REVOKE', - 'token_to_revoke': 'REFRESH_TOKEN_TO_REVOKE'}) - self.connection_mock.send.assert_called_once_with(mock.ANY, expected_body, headers=mock.ANY, method=mock.ANY) - - def test_send_request_should_send_correct_request(self): - exp_resp = {'id': '123', 'name': 'foo'} - self.connection_mock.send.return_value = self._connection_response(exp_resp) - - resp = self.ftd_plugin.send_request('/test/{objId}', HTTPMethod.PUT, - body_params={'name': 'foo'}, - path_params={'objId': '123'}, - query_params={'at': 0}) - - assert {ResponseParams.SUCCESS: True, ResponseParams.STATUS_CODE: 200, - ResponseParams.RESPONSE: exp_resp} == resp - self.connection_mock.send.assert_called_once_with('/test/123?at=0', '{"name": "foo"}', method=HTTPMethod.PUT, - headers=BASE_HEADERS) - - def test_send_request_should_return_empty_dict_when_no_response_data(self): - self.connection_mock.send.return_value = self._connection_response(None) - - resp = self.ftd_plugin.send_request('/test', HTTPMethod.GET) - - assert {ResponseParams.SUCCESS: True, ResponseParams.STATUS_CODE: 200, ResponseParams.RESPONSE: {}} == resp - self.connection_mock.send.assert_called_once_with('/test', None, method=HTTPMethod.GET, - headers=BASE_HEADERS) - - def test_send_request_should_return_error_info_when_http_error_raises(self): - self.connection_mock.send.side_effect = HTTPError('http://testhost.com', 500, '', {}, - StringIO('{"errorMessage": "ERROR"}')) - - resp = self.ftd_plugin.send_request('/test', HTTPMethod.GET) - - assert {ResponseParams.SUCCESS: False, ResponseParams.STATUS_CODE: 500, - ResponseParams.RESPONSE: {'errorMessage': 'ERROR'}} == resp - - def test_send_request_raises_exception_when_invalid_response(self): - self.connection_mock.send.return_value = self._connection_response('nonValidJson') - - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin.send_request('/test', HTTPMethod.GET) - - assert 'Invalid JSON response' in str(res.exception) - - def test_handle_httperror_should_update_tokens_and_retry_on_auth_errors(self): - self.ftd_plugin.refresh_token = 'REFRESH_TOKEN' - self.connection_mock.send.return_value = self._connection_response( - {'access_token': 'NEW_ACCESS_TOKEN', 'refresh_token': 'NEW_REFRESH_TOKEN'} - ) - - retry = self.ftd_plugin.handle_httperror(HTTPError('http://testhost.com', 401, '', {}, None)) - - assert retry - assert 'NEW_ACCESS_TOKEN' == self.ftd_plugin.access_token - assert 'NEW_REFRESH_TOKEN' == self.ftd_plugin.refresh_token - - def test_handle_httperror_should_not_retry_on_non_auth_errors(self): - assert not self.ftd_plugin.handle_httperror(HTTPError('http://testhost.com', 500, '', {}, None)) - - def test_handle_httperror_should_not_retry_when_ignoring_http_errors(self): - self.ftd_plugin._ignore_http_errors = True - assert not self.ftd_plugin.handle_httperror(HTTPError('http://testhost.com', 401, '', {}, None)) - - @patch('os.path.isdir', mock.Mock(return_value=False)) - def test_download_file(self): - self.connection_mock.send.return_value = self._connection_response('File content') - - open_mock = mock_open() - with patch('%s.open' % BUILTINS_NAME, open_mock): - self.ftd_plugin.download_file('/files/1', '/tmp/test.txt') - - open_mock.assert_called_once_with('/tmp/test.txt', 'wb') - open_mock().write.assert_called_once_with(b'File content') - - @patch('os.path.isdir', mock.Mock(return_value=True)) - def test_download_file_should_extract_filename_from_headers(self): - filename = 'test_file.txt' - response = mock.Mock() - response.info.return_value = {'Content-Disposition': 'attachment; filename="%s"' % filename} - dummy, response_data = self._connection_response('File content') - self.connection_mock.send.return_value = response, response_data - - open_mock = mock_open() - with patch('%s.open' % BUILTINS_NAME, open_mock): - self.ftd_plugin.download_file('/files/1', '/tmp/') - - open_mock.assert_called_once_with('/tmp/%s' % filename, 'wb') - open_mock().write.assert_called_once_with(b'File content') - - @patch('os.path.basename', mock.Mock(return_value='test.txt')) - @patch('ansible.plugins.httpapi.ftd.encode_multipart_formdata', - mock.Mock(return_value=('--Encoded data--', 'multipart/form-data'))) - def test_upload_file(self): - self.connection_mock.send.return_value = self._connection_response({'id': '123'}) - - open_mock = mock_open() - with patch('%s.open' % BUILTINS_NAME, open_mock): - resp = self.ftd_plugin.upload_file('/tmp/test.txt', '/files') - - assert {'id': '123'} == resp - exp_headers = dict(BASE_HEADERS) - exp_headers['Content-Length'] = len('--Encoded data--') - exp_headers['Content-Type'] = 'multipart/form-data' - self.connection_mock.send.assert_called_once_with('/files', data='--Encoded data--', - headers=exp_headers, method=HTTPMethod.POST) - open_mock.assert_called_once_with('/tmp/test.txt', 'rb') - - @patch('os.path.basename', mock.Mock(return_value='test.txt')) - @patch('ansible.plugins.httpapi.ftd.encode_multipart_formdata', - mock.Mock(return_value=('--Encoded data--', 'multipart/form-data'))) - def test_upload_file_raises_exception_when_invalid_response(self): - self.connection_mock.send.return_value = self._connection_response('invalidJsonResponse') - - open_mock = mock_open() - with patch('%s.open' % BUILTINS_NAME, open_mock): - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin.upload_file('/tmp/test.txt', '/files') - - assert 'Invalid JSON response' in str(res.exception) - - @patch.object(FdmSwaggerParser, 'parse_spec') - def test_get_operation_spec(self, parse_spec_mock): - self.connection_mock.send.return_value = self._connection_response(None) - parse_spec_mock.return_value = { - SpecProp.OPERATIONS: {'testOp': 'Specification for testOp'} - } - - assert 'Specification for testOp' == self.ftd_plugin.get_operation_spec('testOp') - assert self.ftd_plugin.get_operation_spec('nonExistingTestOp') is None - - @patch.object(FdmSwaggerParser, 'parse_spec') - def test_get_model_spec(self, parse_spec_mock): - self.connection_mock.send.return_value = self._connection_response(None) - parse_spec_mock.return_value = { - SpecProp.MODELS: {'TestModel': 'Specification for TestModel'} - } - - assert 'Specification for TestModel' == self.ftd_plugin.get_model_spec('TestModel') - assert self.ftd_plugin.get_model_spec('NonExistingTestModel') is None - - @patch.object(FdmSwaggerParser, 'parse_spec') - def test_get_operation_spec_by_model_name(self, parse_spec_mock): - self.connection_mock.send.return_value = self._connection_response(None) - operation1 = {'modelName': 'TestModel'} - op_model_name_is_none = {'modelName': None} - op_without_model_name = {'url': 'testUrl'} - - parse_spec_mock.return_value = { - SpecProp.MODEL_OPERATIONS: { - 'TestModel': { - 'testOp1': operation1, - 'testOp2': 'spec2' - }, - 'TestModel2': { - 'testOp10': 'spec10', - 'testOp20': 'spec20' - } - }, - SpecProp.OPERATIONS: { - 'testOp1': operation1, - 'testOp10': { - 'modelName': 'TestModel2' - }, - 'testOpWithoutModelName': op_without_model_name, - 'testOpModelNameIsNone': op_model_name_is_none - } - } - - assert {'testOp1': operation1, 'testOp2': 'spec2'} == self.ftd_plugin.get_operation_specs_by_model_name( - 'TestModel') - assert None is self.ftd_plugin.get_operation_specs_by_model_name( - 'testOpModelNameIsNone') - - assert None is self.ftd_plugin.get_operation_specs_by_model_name( - 'testOpWithoutModelName') - - assert self.ftd_plugin.get_operation_specs_by_model_name('nonExistingOperation') is None - - @staticmethod - def _connection_response(response, status=200): - response_mock = mock.Mock() - response_mock.getcode.return_value = status - response_text = json.dumps(response) if type(response) is dict else response - response_data = BytesIO(response_text.encode() if response_text else ''.encode()) - return response_mock, response_data - - def test_get_list_of_supported_api_versions_with_failed_http_request(self): - error_msg = "Invalid Credentials" - fp = mock.MagicMock() - fp.read.return_value = '{{"error-msg": "{0}"}}'.format(error_msg) - send_mock = mock.MagicMock(side_effect=HTTPError('url', 400, 'msg', 'hdrs', fp)) - with mock.patch.object(self.ftd_plugin.connection, 'send', send_mock): - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin._get_supported_api_versions() - - assert error_msg in str(res.exception) - - def test_get_list_of_supported_api_versions_with_buggy_response(self): - error_msg = "Non JSON value" - http_response_mock = mock.MagicMock() - http_response_mock.getvalue.return_value = error_msg - - send_mock = mock.MagicMock(return_value=(None, http_response_mock)) - - with mock.patch.object(self.ftd_plugin.connection, 'send', send_mock): - with self.assertRaises(ConnectionError) as res: - self.ftd_plugin._get_supported_api_versions() - assert error_msg in str(res.exception) - - def test_get_list_of_supported_api_versions_with_positive_response(self): - http_response_mock = mock.MagicMock() - http_response_mock.getvalue.return_value = '{"supportedVersions": ["v1"]}' - - send_mock = mock.MagicMock(return_value=(None, http_response_mock)) - with mock.patch.object(self.ftd_plugin.connection, 'send', send_mock): - supported_versions = self.ftd_plugin._get_supported_api_versions() - assert supported_versions == ['v1'] - - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_api_token_path', mock.MagicMock(return_value=None)) - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_known_token_paths') - def test_lookup_login_url_with_empty_response(self, get_known_token_paths_mock): - payload = mock.MagicMock() - get_known_token_paths_mock.return_value = [] - self.assertRaises( - ConnectionError, - self.ftd_plugin._lookup_login_url, - payload - ) - - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_known_token_paths') - @patch('ansible.plugins.httpapi.ftd.HttpApi._send_login_request') - def test_lookup_login_url_with_failed_request(self, api_request_mock, get_known_token_paths_mock): - payload = mock.MagicMock() - url = mock.MagicMock() - get_known_token_paths_mock.return_value = [url] - api_request_mock.side_effect = ConnectionError('Error message') - with mock.patch.object(self.ftd_plugin.connection, 'queue_message') as display_mock: - self.assertRaises( - ConnectionError, - self.ftd_plugin._lookup_login_url, - payload - ) - assert display_mock.called - - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_api_token_path', mock.MagicMock(return_value=None)) - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_known_token_paths') - @patch('ansible.plugins.httpapi.ftd.HttpApi._send_login_request') - @patch('ansible.plugins.httpapi.ftd.HttpApi._set_api_token_path') - def test_lookup_login_url_with_positive_result(self, set_api_token_mock, api_request_mock, - get_known_token_paths_mock): - payload = mock.MagicMock() - url = mock.MagicMock() - get_known_token_paths_mock.return_value = [url] - response_mock = mock.MagicMock() - api_request_mock.return_value = response_mock - - resp = self.ftd_plugin._lookup_login_url(payload) - - set_api_token_mock.assert_called_once_with(url) - assert resp == response_mock - - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_supported_api_versions') - def test_get_known_token_paths_with_positive_response(self, get_list_of_supported_api_versions_mock): - test_versions = ['v1', 'v2'] - get_list_of_supported_api_versions_mock.return_value = test_versions - result = self.ftd_plugin._get_known_token_paths() - assert result == [TOKEN_PATH_TEMPLATE.format(version) for version in test_versions] - - @patch('ansible.plugins.httpapi.ftd.HttpApi._get_supported_api_versions') - def test_get_known_token_paths_with_failed_api_call(self, get_list_of_supported_api_versions_mock): - get_list_of_supported_api_versions_mock.side_effect = ConnectionError('test error message') - result = self.ftd_plugin._get_known_token_paths() - assert result == [TOKEN_PATH_TEMPLATE.format(version) for version in DEFAULT_API_VERSIONS] - - def test_set_api_token_path(self): - url = mock.MagicMock() - self.ftd_plugin._set_api_token_path(url) - assert self.ftd_plugin._get_api_token_path() == url diff --git a/test/units/plugins/inventory/test_linode.py b/test/units/plugins/inventory/test_linode.py deleted file mode 100644 index b192414cf8..0000000000 --- a/test/units/plugins/inventory/test_linode.py +++ /dev/null @@ -1,76 +0,0 @@ -# -*- coding: utf-8 -*- - -# Copyright 2018 Luke Murphy <lukewm@riseup.net> -# -# 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/>. - -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -import pytest -import sys - -linode_apiv4 = pytest.importorskip('linode_api4') -mandatory_py_version = pytest.mark.skipif( - sys.version_info < (2, 7), - reason='The linode_api4 dependency requires python2.7 or higher' -) - - -from ansible.errors import AnsibleError, AnsibleParserError -from ansible.plugins.inventory.linode import InventoryModule - - -@pytest.fixture(scope="module") -def inventory(): - return InventoryModule() - - -def test_access_token_lookup(inventory): - inventory._options = {'access_token': None} - with pytest.raises(AnsibleError) as error_message: - inventory._build_client() - assert 'Could not retrieve Linode access token' in error_message - - -def test_validate_option(inventory): - assert ['eu-west'] == inventory._validate_option('regions', list, 'eu-west') - assert ['eu-west'] == inventory._validate_option('regions', list, ['eu-west']) - - -def test_validation_option_bad_option(inventory): - with pytest.raises(AnsibleParserError) as error_message: - inventory._validate_option('regions', dict, []) - assert "The option filters ([]) must be a <class 'dict'>" == error_message - - -def test_empty_config_query_options(inventory): - regions, types = inventory._get_query_options({}) - assert regions == types == [] - - -def test_conig_query_options(inventory): - regions, types = inventory._get_query_options({ - 'regions': ['eu-west', 'us-east'], - 'types': ['g5-standard-2', 'g6-standard-2'], - }) - - assert regions == ['eu-west', 'us-east'] - assert types == ['g5-standard-2', 'g6-standard-2'] - - -def test_verify_file_bad_config(inventory): - assert inventory.verify_file('foobar.linde.yml') is False diff --git a/test/units/plugins/lookup/test_avi.py b/test/units/plugins/lookup/test_avi.py deleted file mode 100644 index 4619cba735..0000000000 --- a/test/units/plugins/lookup/test_avi.py +++ /dev/null @@ -1,92 +0,0 @@ -# -*- coding: utf-8 -*- -# (c) 2019, Sandeep Bandi <sandeepb@avinetworks.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 -import json - -from units.compat.mock import patch, MagicMock - -from ansible.errors import AnsibleError -from ansible.plugins.loader import lookup_loader -from ansible.plugins.lookup import avi - - -try: - import builtins as __builtin__ -except ImportError: - import __builtin__ - - -fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') - -with open(fixture_path + '/avi.json') as json_file: - data = json.load(json_file) - - -@pytest.fixture -def dummy_credentials(): - dummy_credentials = {} - dummy_credentials['controller'] = "192.0.2.13" - dummy_credentials['username'] = "admin" - dummy_credentials['password'] = "password" - dummy_credentials['api_version'] = "17.2.14" - dummy_credentials['tenant'] = 'admin' - return dummy_credentials - - -@pytest.fixture -def super_switcher(scope="function", autouse=True): - # Mocking the inbuilt super as it is used in ApiSession initialization - original_super = __builtin__.super - __builtin__.super = MagicMock() - yield - # Revert the super to default state - __builtin__.super = original_super - - -def test_lookup_multiple_obj(dummy_credentials): - avi_lookup = lookup_loader.get('avi') - avi_mock = MagicMock() - avi_mock.return_value.get.return_value.json.return_value = data["mock_multiple_obj"] - with patch.object(avi, 'ApiSession', avi_mock): - retval = avi_lookup.run([], {}, avi_credentials=dummy_credentials, - obj_type="network") - assert retval == data["mock_multiple_obj"]["results"] - - -def test_lookup_single_obj(dummy_credentials): - avi_lookup = lookup_loader.get('avi') - avi_mock = MagicMock() - avi_mock.return_value.get_object_by_name.return_value = data["mock_single_obj"] - with patch.object(avi, 'ApiSession', avi_mock): - retval = avi_lookup.run([], {}, avi_credentials=dummy_credentials, - obj_type="network", obj_name='PG-123') - assert retval[0] == data["mock_single_obj"] - - -def test_invalid_lookup(dummy_credentials): - avi_lookup = lookup_loader.get('avi') - avi_mock = MagicMock() - with pytest.raises(AnsibleError): - with patch.object(avi, 'ApiSession', avi_mock): - avi_lookup.run([], {}, avi_credentials=dummy_credentials) diff --git a/test/units/plugins/lookup/test_conjur_variable.py b/test/units/plugins/lookup/test_conjur_variable.py deleted file mode 100644 index 98890f82ab..0000000000 --- a/test/units/plugins/lookup/test_conjur_variable.py +++ /dev/null @@ -1,110 +0,0 @@ -# -*- coding: utf-8 -*- -# (c) 2018, Jason Vanderhoof <jason.vanderhoof@cyberark.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 pytest -from units.compat.mock import MagicMock -from ansible.errors import AnsibleError -from ansible.module_utils.six.moves import http_client -from ansible.plugins.lookup import conjur_variable -import tempfile - - -class TestLookupModule: - def test_valid_netrc_file(self): - with tempfile.NamedTemporaryFile() as temp_netrc: - temp_netrc.write(b"machine http://localhost/authn\n") - temp_netrc.write(b" login admin\n") - temp_netrc.write(b" password my-pass\n") - temp_netrc.seek(0) - - results = conjur_variable._load_identity_from_file(temp_netrc.name, 'http://localhost') - - assert results['id'] == 'admin' - assert results['api_key'] == 'my-pass' - - def test_netrc_without_host_file(self): - with tempfile.NamedTemporaryFile() as temp_netrc: - temp_netrc.write(b"machine http://localhost/authn\n") - temp_netrc.write(b" login admin\n") - temp_netrc.write(b" password my-pass\n") - temp_netrc.seek(0) - - with pytest.raises(AnsibleError): - conjur_variable._load_identity_from_file(temp_netrc.name, 'http://foo') - - def test_valid_configuration(self): - with tempfile.NamedTemporaryFile() as configuration_file: - configuration_file.write(b"---\n") - configuration_file.write(b"account: demo-policy\n") - configuration_file.write(b"plugins: []\n") - configuration_file.write(b"appliance_url: http://localhost:8080\n") - configuration_file.seek(0) - - results = conjur_variable._load_conf_from_file(configuration_file.name) - assert results['account'] == 'demo-policy' - assert results['appliance_url'] == 'http://localhost:8080' - - def test_valid_token_retrieval(self, mocker): - mock_response = MagicMock(spec_set=http_client.HTTPResponse) - try: - mock_response.getcode.return_value = 200 - except Exception: - # HTTPResponse is a Python 3 only feature. This uses a generic mock for python 2.6 - mock_response = MagicMock() - mock_response.getcode.return_value = 200 - - mock_response.read.return_value = 'foo-bar-token' - mocker.patch.object(conjur_variable, 'open_url', return_value=mock_response) - - response = conjur_variable._fetch_conjur_token('http://conjur', 'account', 'username', 'api_key') - assert response == 'foo-bar-token' - - def test_valid_fetch_conjur_variable(self, mocker): - mock_response = MagicMock(spec_set=http_client.HTTPResponse) - try: - mock_response.getcode.return_value = 200 - except Exception: - # HTTPResponse is a Python 3 only feature. This uses a generic mock for python 2.6 - mock_response = MagicMock() - mock_response.getcode.return_value = 200 - - mock_response.read.return_value = 'foo-bar' - mocker.patch.object(conjur_variable, 'open_url', return_value=mock_response) - - response = conjur_variable._fetch_conjur_token('super-secret', 'token', 'http://conjur', 'account') - assert response == 'foo-bar' - - def test_invalid_fetch_conjur_variable(self, mocker): - for code in [401, 403, 404]: - mock_response = MagicMock(spec_set=http_client.HTTPResponse) - try: - mock_response.getcode.return_value = code - except Exception: - # HTTPResponse is a Python 3 only feature. This uses a generic mock for python 2.6 - mock_response = MagicMock() - mock_response.getcode.return_value = code - - mocker.patch.object(conjur_variable, 'open_url', return_value=mock_response) - - with pytest.raises(AnsibleError): - response = conjur_variable._fetch_conjur_token('super-secret', 'token', 'http://conjur', 'account') diff --git a/test/units/plugins/lookup/test_lastpass.py b/test/units/plugins/lookup/test_lastpass.py deleted file mode 100644 index 7de54d0fc5..0000000000 --- a/test/units/plugins/lookup/test_lastpass.py +++ /dev/null @@ -1,187 +0,0 @@ -# (c)2016 Andrew Zenk <azenk@umn.edu> -# -# 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 argparse import ArgumentParser - -from units.compat import unittest -from units.compat.mock import patch - -from ansible.errors import AnsibleError -from ansible.module_utils import six -from ansible.plugins.lookup.lastpass import LookupModule, LPass, LPassException - - -MOCK_ENTRIES = [{'username': 'user', - 'name': 'Mock Entry', - 'password': 't0pS3cret passphrase entry!', - 'url': 'https://localhost/login', - 'notes': 'Test\nnote with multiple lines.\n', - 'id': '0123456789'}] - - -class MockLPass(LPass): - - _mock_logged_out = False - _mock_disconnected = False - - def _lookup_mock_entry(self, key): - for entry in MOCK_ENTRIES: - if key == entry['id'] or key == entry['name']: - return entry - - def _run(self, args, stdin=None, expected_rc=0): - # Mock behavior of lpass executable - base_options = ArgumentParser(add_help=False) - base_options.add_argument('--color', default="auto", choices=['auto', 'always', 'never']) - - p = ArgumentParser() - sp = p.add_subparsers(help='command', dest='subparser_name') - - logout_p = sp.add_parser('logout', parents=[base_options], help='logout') - show_p = sp.add_parser('show', parents=[base_options], help='show entry details') - - field_group = show_p.add_mutually_exclusive_group(required=True) - for field in MOCK_ENTRIES[0].keys(): - field_group.add_argument("--{0}".format(field), default=False, action='store_true') - field_group.add_argument('--field', default=None) - show_p.add_argument('selector', help='Unique Name or ID') - - args = p.parse_args(args) - - def mock_exit(output='', error='', rc=0): - if rc != expected_rc: - raise LPassException(error) - return output, error - - if args.color != 'never': - return mock_exit(error='Error: Mock only supports --color=never', rc=1) - - if args.subparser_name == 'logout': - if self._mock_logged_out: - return mock_exit(error='Error: Not currently logged in', rc=1) - - logged_in_error = 'Are you sure you would like to log out? [Y/n]' - if stdin and stdin.lower() == 'n\n': - return mock_exit(output='Log out: aborted.', error=logged_in_error, rc=1) - elif stdin and stdin.lower() == 'y\n': - return mock_exit(output='Log out: complete.', error=logged_in_error, rc=0) - else: - return mock_exit(error='Error: aborted response', rc=1) - - if args.subparser_name == 'show': - if self._mock_logged_out: - return mock_exit(error='Error: Could not find decryption key.' + - ' Perhaps you need to login with `lpass login`.', rc=1) - - if self._mock_disconnected: - return mock_exit(error='Error: Couldn\'t resolve host name.', rc=1) - - mock_entry = self._lookup_mock_entry(args.selector) - - if args.field: - return mock_exit(output=mock_entry.get(args.field, '')) - elif args.password: - return mock_exit(output=mock_entry.get('password', '')) - elif args.username: - return mock_exit(output=mock_entry.get('username', '')) - elif args.url: - return mock_exit(output=mock_entry.get('url', '')) - elif args.name: - return mock_exit(output=mock_entry.get('name', '')) - elif args.id: - return mock_exit(output=mock_entry.get('id', '')) - elif args.notes: - return mock_exit(output=mock_entry.get('notes', '')) - - raise LPassException('We should never get here') - - -class DisconnectedMockLPass(MockLPass): - - _mock_disconnected = True - - -class LoggedOutMockLPass(MockLPass): - - _mock_logged_out = True - - -class TestLPass(unittest.TestCase): - - def test_lastpass_cli_path(self): - lp = MockLPass(path='/dev/null') - self.assertEqual('/dev/null', lp.cli_path) - - def test_lastpass_build_args_logout(self): - lp = MockLPass() - self.assertEqual(['logout', '--color=never'], lp._build_args("logout")) - - def test_lastpass_logged_in_true(self): - lp = MockLPass() - self.assertTrue(lp.logged_in) - - def test_lastpass_logged_in_false(self): - lp = LoggedOutMockLPass() - self.assertFalse(lp.logged_in) - - def test_lastpass_show_disconnected(self): - lp = DisconnectedMockLPass() - - with self.assertRaises(LPassException): - lp.get_field('0123456789', 'username') - - def test_lastpass_show(self): - lp = MockLPass() - for entry in MOCK_ENTRIES: - entry_id = entry.get('id') - for k, v in six.iteritems(entry): - self.assertEqual(v.strip(), lp.get_field(entry_id, k)) - - -class TestLastpassPlugin(unittest.TestCase): - - @patch('ansible.plugins.lookup.lastpass.LPass', new=MockLPass) - def test_lastpass_plugin_normal(self): - lookup_plugin = LookupModule() - - for entry in MOCK_ENTRIES: - entry_id = entry.get('id') - for k, v in six.iteritems(entry): - self.assertEqual(v.strip(), - lookup_plugin.run([entry_id], field=k)[0]) - - @patch('ansible.plugins.lookup.lastpass.LPass', LoggedOutMockLPass) - def test_lastpass_plugin_logged_out(self): - lookup_plugin = LookupModule() - - entry = MOCK_ENTRIES[0] - entry_id = entry.get('id') - with self.assertRaises(AnsibleError): - lookup_plugin.run([entry_id], field='password') - - @patch('ansible.plugins.lookup.lastpass.LPass', DisconnectedMockLPass) - def test_lastpass_plugin_disconnected(self): - lookup_plugin = LookupModule() - - entry = MOCK_ENTRIES[0] - entry_id = entry.get('id') - with self.assertRaises(AnsibleError): - lookup_plugin.run([entry_id], field='password') diff --git a/test/units/plugins/lookup/test_manifold.py b/test/units/plugins/lookup/test_manifold.py deleted file mode 100644 index 1cc1b68967..0000000000 --- a/test/units/plugins/lookup/test_manifold.py +++ /dev/null @@ -1,536 +0,0 @@ -# (c) 2018, Arigato Machine Inc. -# (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 - -from units.compat import unittest -from units.compat.mock import patch, call -from ansible.errors import AnsibleError -from ansible.module_utils.urls import ConnectionError, SSLValidationError -from ansible.module_utils.six.moves.urllib.error import HTTPError, URLError -from ansible.module_utils import six -from ansible.plugins.lookup.manifold import ManifoldApiClient, LookupModule, ApiError -import json - - -API_FIXTURES = { - 'https://api.marketplace.manifold.co/v1/resources': - [ - { - "body": { - "label": "resource-1", - "name": "Resource 1" - }, - "id": "rid-1" - }, - { - "body": { - "label": "resource-2", - "name": "Resource 2" - }, - "id": "rid-2" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?label=resource-1': - [ - { - "body": { - "label": "resource-1", - "name": "Resource 1" - }, - "id": "rid-1" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?label=resource-2': - [ - { - "body": { - "label": "resource-2", - "name": "Resource 2" - }, - "id": "rid-2" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1': - [ - { - "body": { - "label": "resource-1", - "name": "Resource 1" - }, - "id": "rid-1" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?project_id=pid-1': - [ - { - "body": { - "label": "resource-2", - "name": "Resource 2" - }, - "id": "rid-2" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?project_id=pid-2': - [ - { - "body": { - "label": "resource-1", - "name": "Resource 1" - }, - "id": "rid-1" - }, - { - "body": { - "label": "resource-3", - "name": "Resource 3" - }, - "id": "rid-3" - } - ], - 'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1': - [ - { - "body": { - "label": "resource-1", - "name": "Resource 1" - }, - "id": "rid-1" - } - ], - 'https://api.marketplace.manifold.co/v1/projects': - [ - { - "body": { - "label": "project-1", - "name": "Project 1", - }, - "id": "pid-1", - }, - { - "body": { - "label": "project-2", - "name": "Project 2", - }, - "id": "pid-2", - } - ], - 'https://api.marketplace.manifold.co/v1/projects?label=project-2': - [ - { - "body": { - "label": "project-2", - "name": "Project 2", - }, - "id": "pid-2", - } - ], - 'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-1': - [ - { - "body": { - "resource_id": "rid-1", - "values": { - "RESOURCE_TOKEN_1": "token-1", - "RESOURCE_TOKEN_2": "token-2" - } - }, - "id": "cid-1", - } - ], - 'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-2': - [ - { - "body": { - "resource_id": "rid-2", - "values": { - "RESOURCE_TOKEN_3": "token-3", - "RESOURCE_TOKEN_4": "token-4" - } - }, - "id": "cid-2", - } - ], - 'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-3': - [ - { - "body": { - "resource_id": "rid-3", - "values": { - "RESOURCE_TOKEN_1": "token-5", - "RESOURCE_TOKEN_2": "token-6" - } - }, - "id": "cid-3", - } - ], - 'https://api.identity.manifold.co/v1/teams': - [ - { - "id": "tid-1", - "body": { - "name": "Team 1", - "label": "team-1" - } - }, - { - "id": "tid-2", - "body": { - "name": "Team 2", - "label": "team-2" - } - } - ] -} - - -def mock_fixture(open_url_mock, fixture=None, data=None, headers=None): - if not headers: - headers = {} - if fixture: - data = json.dumps(API_FIXTURES[fixture]) - if 'content-type' not in headers: - headers['content-type'] = 'application/json' - - open_url_mock.return_value.read.return_value = data - open_url_mock.return_value.headers = headers - - -class TestManifoldApiClient(unittest.TestCase): - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_sends_default_headers(self, open_url_mock): - mock_fixture(open_url_mock, data='hello') - client = ManifoldApiClient('token-123') - client.request('test', 'endpoint') - open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint', - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_decodes_json(self, open_url_mock): - mock_fixture(open_url_mock, fixture='https://api.marketplace.manifold.co/v1/resources') - client = ManifoldApiClient('token-123') - self.assertIsInstance(client.request('marketplace', 'resources'), list) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_streams_text(self, open_url_mock): - mock_fixture(open_url_mock, data='hello', headers={'content-type': "text/plain"}) - client = ManifoldApiClient('token-123') - self.assertEqual('hello', client.request('test', 'endpoint')) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_processes_parameterized_headers(self, open_url_mock): - mock_fixture(open_url_mock, data='hello') - client = ManifoldApiClient('token-123') - client.request('test', 'endpoint', headers={'X-HEADER': 'MANIFOLD'}) - open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint', - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123', - 'X-HEADER': 'MANIFOLD'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_passes_arbitrary_parameters(self, open_url_mock): - mock_fixture(open_url_mock, data='hello') - client = ManifoldApiClient('token-123') - client.request('test', 'endpoint', use_proxy=False, timeout=5) - open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint', - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0', - use_proxy=False, timeout=5) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_raises_on_incorrect_json(self, open_url_mock): - mock_fixture(open_url_mock, data='noJson', headers={'content-type': "application/json"}) - client = ManifoldApiClient('token-123') - with self.assertRaises(ApiError) as context: - client.request('test', 'endpoint') - self.assertEqual('JSON response can\'t be parsed while requesting https://api.test.manifold.co/v1/endpoint:\n' - 'noJson', - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_raises_on_status_500(self, open_url_mock): - open_url_mock.side_effect = HTTPError('https://api.test.manifold.co/v1/endpoint', - 500, 'Server error', {}, six.StringIO('ERROR')) - client = ManifoldApiClient('token-123') - with self.assertRaises(ApiError) as context: - client.request('test', 'endpoint') - self.assertEqual('Server returned: HTTP Error 500: Server error while requesting ' - 'https://api.test.manifold.co/v1/endpoint:\nERROR', - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_raises_on_bad_url(self, open_url_mock): - open_url_mock.side_effect = URLError('URL is invalid') - client = ManifoldApiClient('token-123') - with self.assertRaises(ApiError) as context: - client.request('test', 'endpoint') - self.assertEqual('Failed lookup url for https://api.test.manifold.co/v1/endpoint : <url' - 'open error URL is invalid>', - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_raises_on_ssl_error(self, open_url_mock): - open_url_mock.side_effect = SSLValidationError('SSL Error') - client = ManifoldApiClient('token-123') - with self.assertRaises(ApiError) as context: - client.request('test', 'endpoint') - self.assertEqual('Error validating the server\'s certificate for https://api.test.manifold.co/v1/endpoint: ' - 'SSL Error', - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_request_raises_on_connection_error(self, open_url_mock): - open_url_mock.side_effect = ConnectionError('Unknown connection error') - client = ManifoldApiClient('token-123') - with self.assertRaises(ApiError) as context: - client.request('test', 'endpoint') - self.assertEqual('Error connecting to https://api.test.manifold.co/v1/endpoint: Unknown connection error', - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_resources_get_all(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/resources' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_resources()) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_resources_filter_label(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/resources?label=resource-1' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_resources(label='resource-1')) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_resources_filter_team_and_project(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_resources(team_id='tid-1', project_id='pid-1')) - args, kwargs = open_url_mock.call_args - url_called = args[0] - # Dict order is not guaranteed, so an url may have querystring parameters order randomized - self.assertIn('team_id=tid-1', url_called) - self.assertIn('project_id=pid-1', url_called) - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_teams_get_all(self, open_url_mock): - url = 'https://api.identity.manifold.co/v1/teams' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_teams()) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_teams_filter_label(self, open_url_mock): - url = 'https://api.identity.manifold.co/v1/teams' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url][1:2], client.get_teams(label='team-2')) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_projects_get_all(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/projects' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_projects()) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_projects_filter_label(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/projects?label=project-2' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_projects(label='project-2')) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - @patch('ansible.plugins.lookup.manifold.open_url') - def test_get_credentials(self, open_url_mock): - url = 'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-1' - mock_fixture(open_url_mock, fixture=url) - client = ManifoldApiClient('token-123') - self.assertListEqual(API_FIXTURES[url], client.get_credentials(resource_id='rid-1')) - open_url_mock.assert_called_with(url, - headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'}, - http_agent='python-manifold-ansible-1.0.0') - - -class TestLookupModule(unittest.TestCase): - def setUp(self): - self.lookup = LookupModule() - self.lookup._load_name = "manifold" - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_get_all(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_1': 'token-1', - 'RESOURCE_TOKEN_2': 'token-2', - 'RESOURCE_TOKEN_3': 'token-3', - 'RESOURCE_TOKEN_4': 'token-4' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources'] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123')) - client_mock.assert_called_with('token-123') - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_get_one_resource(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_3': 'token-3', - 'RESOURCE_TOKEN_4': 'token-4' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?label=resource-2'] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run(['resource-2'], api_token='token-123')) - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None, label='resource-2') - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_get_two_resources(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_1': 'token-1', - 'RESOURCE_TOKEN_2': 'token-2', - 'RESOURCE_TOKEN_3': 'token-3', - 'RESOURCE_TOKEN_4': 'token-4' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources'] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run(['resource-1', 'resource-2'], api_token='token-123')) - client_mock.assert_called_with('token-123') - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None) - - @patch('ansible.plugins.lookup.manifold.display') - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_get_resources_with_same_credential_names(self, client_mock, display_mock): - expected_result = [{'RESOURCE_TOKEN_1': 'token-5', - 'RESOURCE_TOKEN_2': 'token-6' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?project_id=pid-2'] - client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects?label=project-2'] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-2')) - client_mock.assert_called_with('token-123') - display_mock.warning.assert_has_calls([ - call("'RESOURCE_TOKEN_1' with label 'resource-1' was replaced by resource data with label 'resource-3'"), - call("'RESOURCE_TOKEN_2' with label 'resource-1' was replaced by resource data with label 'resource-3'")], - any_order=True - ) - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-2') - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_filter_by_team(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_1': 'token-1', - 'RESOURCE_TOKEN_2': 'token-2' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?team_id=tid-1'] - client_mock.return_value.get_teams.return_value = API_FIXTURES['https://api.identity.manifold.co/v1/teams'][0:1] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', team='team-1')) - client_mock.assert_called_with('token-123') - client_mock.return_value.get_resources.assert_called_with(team_id='tid-1', project_id=None) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_filter_by_project(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_3': 'token-3', - 'RESOURCE_TOKEN_4': 'token-4' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?project_id=pid-1'] - client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects'][0:1] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-1')) - client_mock.assert_called_with('token-123') - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-1') - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_filter_by_team_and_project(self, client_mock): - expected_result = [{'RESOURCE_TOKEN_1': 'token-1', - 'RESOURCE_TOKEN_2': 'token-2' - }] - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1'] - client_mock.return_value.get_teams.return_value = API_FIXTURES['https://api.identity.manifold.co/v1/teams'][0:1] - client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects'][0:1] - client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/' - 'credentials?resource_id={0}'.format(x)] - self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-1')) - client_mock.assert_called_with('token-123') - client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-1') - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_raise_team_doesnt_exist(self, client_mock): - client_mock.return_value.get_teams.return_value = [] - with self.assertRaises(AnsibleError) as context: - self.lookup.run([], api_token='token-123', team='no-team') - self.assertEqual("Team 'no-team' does not exist", - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_raise_project_doesnt_exist(self, client_mock): - client_mock.return_value.get_projects.return_value = [] - with self.assertRaises(AnsibleError) as context: - self.lookup.run([], api_token='token-123', project='no-project') - self.assertEqual("Project 'no-project' does not exist", - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_raise_resource_doesnt_exist(self, client_mock): - client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources'] - with self.assertRaises(AnsibleError) as context: - self.lookup.run(['resource-1', 'no-resource-1', 'no-resource-2'], api_token='token-123') - self.assertEqual("Resource(s) no-resource-1, no-resource-2 do not exist", - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_catch_api_error(self, client_mock): - client_mock.side_effect = ApiError('Generic error') - with self.assertRaises(AnsibleError) as context: - self.lookup.run([], api_token='token-123') - self.assertEqual("API Error: Generic error", - str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_catch_unhandled_exception(self, client_mock): - client_mock.side_effect = Exception('Unknown error') - with self.assertRaises(AnsibleError) as context: - self.lookup.run([], api_token='token-123') - self.assertTrue('Exception: Unknown error' in str(context.exception)) - - @patch('ansible.plugins.lookup.manifold.os.getenv') - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_falls_back_to_env_var(self, client_mock, getenv_mock): - getenv_mock.return_value = 'token-321' - client_mock.return_value.get_resources.return_value = [] - client_mock.return_value.get_credentials.return_value = [] - self.lookup.run([]) - getenv_mock.assert_called_with('MANIFOLD_API_TOKEN') - client_mock.assert_called_with('token-321') - - @patch('ansible.plugins.lookup.manifold.os.getenv') - @patch('ansible.plugins.lookup.manifold.ManifoldApiClient') - def test_falls_raises_on_no_token(self, client_mock, getenv_mock): - getenv_mock.return_value = None - client_mock.return_value.get_resources.return_value = [] - client_mock.return_value.get_credentials.return_value = [] - with self.assertRaises(AnsibleError) as context: - self.lookup.run([]) - self.assertEqual('API token is required. Please set api_token parameter or MANIFOLD_API_TOKEN env var', - str(context.exception)) diff --git a/test/units/plugins/lookup/test_onepassword.py b/test/units/plugins/lookup/test_onepassword.py deleted file mode 100644 index dee53b1b01..0000000000 --- a/test/units/plugins/lookup/test_onepassword.py +++ /dev/null @@ -1,321 +0,0 @@ -# (c) 2018, Scott Buchanan <sbuchanan@ri.pn> -# (c) 2016, Andrew Zenk <azenk@umn.edu> (test_lastpass.py used as starting point) -# 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 json -import datetime - -try: - from urllib.parse import urlparse -except ImportError: - from urlparse import urlparse - -from argparse import ArgumentParser - - -from units.compat import unittest -from units.compat.mock import patch -from ansible.errors import AnsibleError -from ansible.plugins.lookup.onepassword import OnePass, LookupModule -from ansible.plugins.lookup.onepassword_raw import LookupModule as OnePasswordRawLookup - - -# Intentionally excludes metadata leaf nodes that would exist in real output if not relevant. -MOCK_ENTRIES = [ - { - 'vault_name': 'Acme "Quot\'d" Servers', - 'queries': [ - '0123456789', - 'Mock "Quot\'d" Server' - ], - 'output': { - 'uuid': '0123456789', - 'vaultUuid': '2468', - 'overview': { - 'title': 'Mock "Quot\'d" Server' - }, - 'details': { - 'sections': [{ - 'title': '', - 'fields': [ - {'t': 'username', 'v': 'jamesbond'}, - {'t': 'password', 'v': 't0pS3cret'}, - {'t': 'notes', 'v': 'Test note with\nmultiple lines and trailing space.\n\n'}, - {'t': 'tricksy "quot\'d" field\\', 'v': '"quot\'d" value'} - ] - }] - } - } - }, - { - 'vault_name': 'Acme Logins', - 'queries': [ - '9876543210', - 'Mock Website', - 'acme.com' - ], - 'output': { - 'uuid': '9876543210', - 'vaultUuid': '1357', - 'overview': { - 'title': 'Mock Website', - 'URLs': [ - {'l': 'website', 'u': 'https://acme.com/login'} - ] - }, - 'details': { - 'sections': [{ - 'title': '', - 'fields': [ - {'t': 'password', 'v': 't0pS3cret'} - ] - }] - } - } - }, - { - 'vault_name': 'Acme Logins', - 'queries': [ - '864201357' - ], - 'output': { - 'uuid': '864201357', - 'vaultUuid': '1357', - 'overview': { - 'title': 'Mock Something' - }, - 'details': { - 'fields': [ - { - 'value': 'jbond@mi6.gov.uk', - 'name': 'emailAddress' - }, - { - 'name': 'password', - 'value': 'vauxhall' - }, - {}, - ] - } - } - }, -] - - -def get_mock_query_generator(require_field=None): - def _process_field(field, section_title=None): - field_name = field.get('name', field.get('t', '')) - field_value = field.get('value', field.get('v', '')) - - if require_field is None or field_name == require_field: - return entry, query, section_title, field_name, field_value - - for entry in MOCK_ENTRIES: - for query in entry['queries']: - for field in entry['output']['details'].get('fields', []): - fixture = _process_field(field) - if fixture: - yield fixture - for section in entry['output']['details'].get('sections', []): - for field in section['fields']: - fixture = _process_field(field, section['title']) - if fixture: - yield fixture - - -def get_one_mock_query(require_field=None): - generator = get_mock_query_generator(require_field) - return next(generator) - - -class MockOnePass(OnePass): - - _mock_logged_out = False - _mock_timed_out = False - - def _lookup_mock_entry(self, key, vault=None): - for entry in MOCK_ENTRIES: - if vault is not None and vault.lower() != entry['vault_name'].lower() and vault.lower() != entry['output']['vaultUuid'].lower(): - continue - - match_fields = [ - entry['output']['uuid'], - entry['output']['overview']['title'] - ] - - # Note that exactly how 1Password matches on domains in non-trivial cases is neither documented - # nor obvious, so this may not precisely match the real behavior. - urls = entry['output']['overview'].get('URLs') - if urls is not None: - match_fields += [urlparse(url['u']).netloc for url in urls] - - if key in match_fields: - return entry['output'] - - def _run(self, args, expected_rc=0, command_input=None, ignore_errors=False): - parser = ArgumentParser() - - command_parser = parser.add_subparsers(dest='command') - - get_parser = command_parser.add_parser('get') - get_options = ArgumentParser(add_help=False) - get_options.add_argument('--vault') - get_type_parser = get_parser.add_subparsers(dest='object_type') - get_type_parser.add_parser('account', parents=[get_options]) - get_item_parser = get_type_parser.add_parser('item', parents=[get_options]) - get_item_parser.add_argument('item_id') - - args = parser.parse_args(args) - - def mock_exit(output='', error='', rc=0): - if rc != expected_rc: - raise AnsibleError(error) - if error != '': - now = datetime.date.today() - error = '[LOG] {0} (ERROR) {1}'.format(now.strftime('%Y/%m/%d %H:$M:$S'), error) - return rc, output, error - - if args.command == 'get': - if self._mock_logged_out: - return mock_exit(error='You are not currently signed in. Please run `op signin --help` for instructions', rc=1) - - if self._mock_timed_out: - return mock_exit(error='401: Authentication required.', rc=1) - - if args.object_type == 'item': - mock_entry = self._lookup_mock_entry(args.item_id, args.vault) - - if mock_entry is None: - return mock_exit(error='Item {0} not found'.format(args.item_id)) - - return mock_exit(output=json.dumps(mock_entry)) - - if args.object_type == 'account': - # Since we don't actually ever use this output, don't bother mocking output. - return mock_exit() - - raise AnsibleError('Unsupported command string passed to OnePass mock: {0}'.format(args)) - - -class LoggedOutMockOnePass(MockOnePass): - - _mock_logged_out = True - - -class TimedOutMockOnePass(MockOnePass): - - _mock_timed_out = True - - -class TestOnePass(unittest.TestCase): - - def test_onepassword_cli_path(self): - op = MockOnePass(path='/dev/null') - self.assertEqual('/dev/null', op.cli_path) - - def test_onepassword_logged_in(self): - op = MockOnePass() - try: - op.assert_logged_in() - except Exception: - self.fail() - - def test_onepassword_logged_out(self): - op = LoggedOutMockOnePass() - with self.assertRaises(AnsibleError): - op.assert_logged_in() - - def test_onepassword_timed_out(self): - op = TimedOutMockOnePass() - with self.assertRaises(AnsibleError): - op.assert_logged_in() - - def test_onepassword_get(self): - op = MockOnePass() - op.logged_in = True - query_generator = get_mock_query_generator() - for dummy, query, dummy, field_name, field_value in query_generator: - self.assertEqual(field_value, op.get_field(query, field_name)) - - def test_onepassword_get_raw(self): - op = MockOnePass() - op.logged_in = True - for entry in MOCK_ENTRIES: - for query in entry['queries']: - self.assertEqual(json.dumps(entry['output']), op.get_raw(query)) - - def test_onepassword_get_not_found(self): - op = MockOnePass() - op.logged_in = True - self.assertEqual('', op.get_field('a fake query', 'a fake field')) - - def test_onepassword_get_with_section(self): - op = MockOnePass() - op.logged_in = True - dummy, query, section_title, field_name, field_value = get_one_mock_query() - self.assertEqual(field_value, op.get_field(query, field_name, section=section_title)) - - def test_onepassword_get_with_vault(self): - op = MockOnePass() - op.logged_in = True - entry, query, dummy, field_name, field_value = get_one_mock_query() - for vault_query in [entry['vault_name'], entry['output']['vaultUuid']]: - self.assertEqual(field_value, op.get_field(query, field_name, vault=vault_query)) - - def test_onepassword_get_with_wrong_vault(self): - op = MockOnePass() - op.logged_in = True - dummy, query, dummy, field_name, dummy = get_one_mock_query() - self.assertEqual('', op.get_field(query, field_name, vault='a fake vault')) - - def test_onepassword_get_diff_case(self): - op = MockOnePass() - op.logged_in = True - entry, query, section_title, field_name, field_value = get_one_mock_query() - self.assertEqual( - field_value, - op.get_field( - query, - field_name.upper(), - vault=entry['vault_name'].upper(), - section=section_title.upper() - ) - ) - - -@patch('ansible.plugins.lookup.onepassword.OnePass', MockOnePass) -class TestLookupModule(unittest.TestCase): - - def test_onepassword_plugin_multiple(self): - lookup_plugin = LookupModule() - - entry = MOCK_ENTRIES[0] - field = entry['output']['details']['sections'][0]['fields'][0] - - self.assertEqual( - [field['v']] * len(entry['queries']), - lookup_plugin.run(entry['queries'], field=field['t']) - ) - - def test_onepassword_plugin_default_field(self): - lookup_plugin = LookupModule() - - dummy, query, dummy, dummy, field_value = get_one_mock_query('password') - self.assertEqual([field_value], lookup_plugin.run([query])) - - -@patch('ansible.plugins.lookup.onepassword_raw.OnePass', MockOnePass) -class TestOnePasswordRawLookup(unittest.TestCase): - - def test_onepassword_raw_plugin_multiple(self): - raw_lookup_plugin = OnePasswordRawLookup() - - entry = MOCK_ENTRIES[0] - raw_value = entry['output'] - - self.assertEqual( - [raw_value] * len(entry['queries']), - raw_lookup_plugin.run(entry['queries']) - ) diff --git a/test/units/plugins/terminal/test_slxos.py b/test/units/plugins/terminal/test_slxos.py deleted file mode 100644 index d886fe06ff..0000000000 --- a/test/units/plugins/terminal/test_slxos.py +++ /dev/null @@ -1,53 +0,0 @@ -# -# (c) 2018 Extreme Networks Inc. -# -# 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/>. -# -from __future__ import (absolute_import, division, print_function) -__metaclass__ = type - -from mock import MagicMock - -from units.compat import unittest -from ansible.plugins.terminal import slxos -from ansible.errors import AnsibleConnectionFailure - - -class TestPluginTerminalSLXOS(unittest.TestCase): - """ Test class for SLX-OS Terminal Module - """ - def setUp(self): - self._mock_connection = MagicMock() - self._terminal = slxos.TerminalModule(self._mock_connection) - - def test_on_open_shell(self): - """ Test on_open_shell - """ - self._mock_connection.exec_command.side_effect = [ - b'Looking out my window I see a brick building, and people. Cool.', - ] - self._terminal.on_open_shell() - self._mock_connection.exec_command.assert_called_with(u'terminal length 0') - - def test_on_open_shell_error(self): - """ Test on_open_shell with error - """ - self._mock_connection.exec_command.side_effect = [ - AnsibleConnectionFailure - ] - - with self.assertRaises(AnsibleConnectionFailure): - self._terminal.on_open_shell() |