diff options
Diffstat (limited to 'test/units/modules/net_tools')
20 files changed, 0 insertions, 4918 deletions
diff --git a/test/units/modules/net_tools/nios/fixtures/nios_result.txt b/test/units/modules/net_tools/nios/fixtures/nios_result.txt deleted file mode 100644 index e69de29bb2..0000000000 --- a/test/units/modules/net_tools/nios/fixtures/nios_result.txt +++ /dev/null diff --git a/test/units/modules/net_tools/nios/test_nios_a_record.py b/test/units/modules/net_tools/nios/test_nios_a_record.py deleted file mode 100644 index 72101da7df..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_a_record.py +++ /dev/null @@ -1,159 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_a_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosARecordModule(TestNiosModule): - - module = nios_a_record - - def setUp(self): - super(TestNiosARecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_a_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_a_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_a_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosARecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_a_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'a.ansible.com', - 'ipv4': '192.168.10.1', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "ipv4": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi.__dict__) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'ipv4': '192.168.10.1'}) - - def test_nios_a_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'a.ansible.com', 'ipv4': '192.168.10.1', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "arecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "a.ansible.com", - "ipv4": "192.168.10.1", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "ipv4": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_a_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'a.ansible.com', 'ipv4': '192.168.10.1', - 'comment': None, 'extattrs': None} - - ref = "arecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "a.ansible.com", - "ipv4": "192.168.10.1", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "ipv4": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_a_record_update_record_name(self): - self.module.params = {'provider': None, 'state': 'present', 'name': {'new_name': 'a_new.ansible.com', 'old_name': 'a.ansible.com'}, - 'comment': 'comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "arecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "a_new.ansible.com", - "old_name": "a.ansible.com", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) diff --git a/test/units/modules/net_tools/nios/test_nios_aaaa_record.py b/test/units/modules/net_tools/nios/test_nios_aaaa_record.py deleted file mode 100644 index dc5dedb80b..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_aaaa_record.py +++ /dev/null @@ -1,159 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_aaaa_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosAAAARecordModule(TestNiosModule): - - module = nios_aaaa_record - - def setUp(self): - super(TestNiosAAAARecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_aaaa_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_aaaa_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_aaaa_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosAAAARecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_aaaa_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'aaaa.ansible.com', - 'ipv6': '2001:0db8:85a3:0000:0000:8a2e:0370:7334', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "ipv6": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi.__dict__) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'ipv6': '2001:0db8:85a3:0000:0000:8a2e:0370:7334'}) - - def test_nios_aaaa_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'aaaa.ansible.com', - 'ipv6': '2001:0db8:85a3:0000:0000:8a2e:0370:7334', 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "aaaarecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "aaaa.ansible.com", - "ipv6": "2001:0db8:85a3:0000:0000:8a2e:0370:7334", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "ipv6": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_aaaa_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'aaaa.ansible.com', - 'ipv6': '2001:0db8:85a3:0000:0000:8a2e:0370:7334', 'comment': None, 'extattrs': None} - - ref = "aaaarecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "aaaa.ansible.com", - "ipv6": "2001:0db8:85a3:0000:0000:8a2e:0370:7334", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "ipv6": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_aaaa_record_update_record_name(self): - self.module.params = {'provider': None, 'state': 'present', 'name': {'new_name': 'aaaa_new.ansible.com', 'old_name': 'aaaa.ansible.com'}, - 'comment': 'comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "aaaarecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "aaaa_new.ansible.com", - "old_name": "aaaa.ansible.com", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) diff --git a/test/units/modules/net_tools/nios/test_nios_cname_record.py b/test/units/modules/net_tools/nios/test_nios_cname_record.py deleted file mode 100644 index ecf8dfcba0..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_cname_record.py +++ /dev/null @@ -1,133 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_cname_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosCNameRecordModule(TestNiosModule): - - module = nios_cname_record - - def setUp(self): - super(TestNiosCNameRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_cname_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_cname_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_cname_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosCNameRecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_a_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'cname.ansible.com', - 'canonical': 'realhost.ansible.com', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "canonical": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi.__dict__) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'canonical': 'realhost.ansible.com'}) - - def test_nios_a_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'cname.ansible.com', - 'canonical': 'realhost.ansible.com', 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "cnamerecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "cname.ansible.com", - "canonical": "realhost.ansible.com", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "canonical": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_a_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'cname.ansible.com', - 'canonical': 'realhost.ansible.com', 'comment': None, 'extattrs': None} - - ref = "cnamerecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "cname.ansible.com", - "canonical": "realhost.ansible.com", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "canonical": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_dns_view.py b/test/units/modules/net_tools/nios/test_nios_dns_view.py deleted file mode 100644 index d5e7656cb2..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_dns_view.py +++ /dev/null @@ -1,127 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_dns_view -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosDnsViewModule(TestNiosModule): - - module = nios_dns_view - - def setUp(self): - super(TestNiosDnsViewModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_dns_view.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_dns_view.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_dns_view.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosDnsViewModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_dns_view_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible-dns', - 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__()}) - - def test_nios_dns_view_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible-dns', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "dnsview/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "ansible-dns", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_dns_view_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'ansible-dns', - 'comment': None, 'extattrs': None} - - ref = "dnsview/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "ansible-dns", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_fixed_address.py b/test/units/modules/net_tools/nios/test_nios_fixed_address.py deleted file mode 100644 index 9b21c2d320..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_fixed_address.py +++ /dev/null @@ -1,201 +0,0 @@ -# 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 ansible.module_utils.net_tools.nios import api -from ansible.modules.net_tools.nios import nios_fixed_address -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosFixedAddressModule(TestNiosModule): - - module = nios_fixed_address - - def setUp(self): - super(TestNiosFixedAddressModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_fixed_address.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_fixed_address.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_fixed_address.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosFixedAddressModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def test_nios_fixed_address_ipv4_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'test_fa', 'ipaddr': '192.168.10.1', 'mac': '08:6d:41:e8:fd:e8', - 'network': '192.168.10.0/24', 'network_view': 'default', 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "name": {}, - "ipaddr": {"ib_req": True}, - "mac": {"ib_req": True}, - "network": {"ib_req": True}, - "network_view": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': 'test_fa', 'ipaddr': '192.168.10.1', 'mac': '08:6d:41:e8:fd:e8', - 'network': '192.168.10.0/24', 'network_view': 'default'}) - - def test_nios_fixed_address_ipv4_dhcp_update(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'test_fa', 'ipaddr': '192.168.10.1', 'mac': '08:6d:41:e8:fd:e8', - 'network': '192.168.10.0/24', 'network_view': 'default', 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "name": "test_fa", - "_ref": "network/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "ipaddr": "192.168.10.1", - "mac": "08:6d:41:e8:fd:e8", - "network": "192.168.10.0/24", - "network_view": "default", - "extattrs": {'options': {'name': 'test', 'value': 'ansible.com'}} - } - ] - - test_spec = { - "name": {}, - "ipaddr": {"ib_req": True}, - "mac": {"ib_req": True}, - "network": {"ib_req": True}, - "network_view": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_fixed_address_ipv4_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'test_fa', 'ipaddr': '192.168.10.1', 'mac': '08:6d:41:e8:fd:e8', - 'network': '192.168.10.0/24', 'network_view': 'default', 'comment': None, 'extattrs': None} - - ref = "fixedaddress/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "name": "test_fa", - "_ref": ref, - "ipaddr": "192.168.10.1", - "mac": "08:6d:41:e8:fd:e8", - "network": "192.168.10.0/24", - "network_view": "default", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {}, - "ipaddr": {"ib_req": True}, - "mac": {"ib_req": True}, - "network": {"ib_req": True}, - "network_view": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_fixed_address_ipv6_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'test_fa', 'ipaddr': 'fe80::1/10', 'mac': '08:6d:41:e8:fd:e8', - 'network': 'fe80::/64', 'network_view': 'default', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {}, - "ipaddr": {"ib_req": True}, - "mac": {"ib_req": True}, - "network": {"ib_req": True}, - "network_view": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': 'test_fa', 'ipaddr': 'fe80::1/10', 'mac': '08:6d:41:e8:fd:e8', - 'network': 'fe80::/64', 'network_view': 'default'}) - - def test_nios_fixed_address_ipv6_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'test_fa', 'ipaddr': 'fe80::1/10', 'mac': '08:6d:41:e8:fd:e8', - 'network': 'fe80::/64', 'network_view': 'default', 'comment': None, 'extattrs': None} - - ref = "ipv6fixedaddress/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "name": "test_fa", - "_ref": ref, - "ipaddr": "fe80::1/10", - "mac": "08:6d:41:e8:fd:e8", - "network": "fe80::/64", - "network_view": "default", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {}, - "ipaddr": {"ib_req": True}, - "mac": {"ib_req": True}, - "network": {"ib_req": True}, - "network_view": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_host_record.py b/test/units/modules/net_tools/nios/test_nios_host_record.py deleted file mode 100644 index 0a317288d1..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_host_record.py +++ /dev/null @@ -1,152 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_host_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosHostRecordModule(TestNiosModule): - - module = nios_host_record - - def setUp(self): - - super(TestNiosHostRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_host_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_host_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_host_record.WapiModule.run') - self.mock_wapi_run.start() - - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosHostRecordModule, self).tearDown() - self.mock_wapi.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_host_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible', - 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__()}) - - def test_nios_host_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'ansible', - 'comment': None, 'extattrs': None} - - ref = "record:host/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "ansible", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_host_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'default', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "record:host/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "default", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) - - def test_nios_host_record_update_record_name(self): - self.module.params = {'provider': None, 'state': 'present', 'name': {'new_name': 'default', 'old_name': 'old_default'}, - 'comment': 'comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "record:host/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "default", - "old_name": "old_default", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) diff --git a/test/units/modules/net_tools/nios/test_nios_member.py b/test/units/modules/net_tools/nios/test_nios_member.py deleted file mode 100644 index f81263cdc0..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_member.py +++ /dev/null @@ -1,162 +0,0 @@ -# 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 ansible.module_utils.net_tools.nios import api -from ansible.modules.net_tools.nios import nios_member -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosMemberModule(TestNiosModule): - - module = nios_member - - def setUp(self): - super(TestNiosMemberModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_member.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_member.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_member.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosMemberModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def test_nios_member_create(self): - self.module.params = {'provider': None, 'state': 'present', 'host_name': 'test_member', - 'vip_setting': {'address': '192.168.1.110', 'subnet_mask': '255.255.255.0', 'gateway': '192.168.1.1'}, - 'config_addr_type': 'IPV4', 'platform': 'VNIOS', 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "host_name": {"ib_req": True}, - "vip_setting": {}, - "config_addr_type": {}, - "platform": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'host_name': 'test_member', - 'vip_setting': {'address': '192.168.1.110', 'subnet_mask': '255.255.255.0', - 'gateway': '192.168.1.1'}, - 'config_addr_type': 'IPV4', 'platform': 'VNIOS'}) - - def test_nios_member_update(self): - self.module.params = {'provider': None, 'state': 'present', 'host_name': 'test_member', - 'vip_setting': {'address': '192.168.1.110', 'subnet_mask': '255.255.255.0', 'gateway': '192.168.1.1'}, - 'config_addr_type': 'IPV4', 'platform': 'VNIOS', 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "Created with Ansible", - "_ref": "member/b25lLnZpcnR1YWxfbm9kZSQ3:member01.ansible-dev.com", - "config_addr_type": "IPV4", - "host_name": "member01.ansible-dev.com", - "platform": "VNIOS", - "service_type_configuration": "ALL_V4", - "vip_setting": - { - "address": "192.168.1.100", - "dscp": 0, - "gateway": "192.168.1.1", - "primary": True, - "subnet_mask": "255.255.255.0", - "use_dscp": False - } - } - ] - - test_spec = { - "host_name": {"ib_req": True}, - "vip_setting": {}, - "config_addr_type": {}, - "platform": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_member_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'host_name': 'test_member', - 'vip_setting': {'address': '192.168.1.110', 'subnet_mask': '255.255.255.0', 'gateway': '192.168.1.1'}, - 'config_addr_type': 'IPV4', 'platform': 'VNIOS', 'comment': 'updated comment', 'extattrs': None} - - ref = "member/b25lLnZpcnR1YWxfbm9kZSQ3:member01.ansible-dev.com" - - test_object = [ - { - "comment": "Created with Ansible", - "_ref": "member/b25lLnZpcnR1YWxfbm9kZSQ3:member01.ansible-dev.com", - "config_addr_type": "IPV4", - "host_name": "member01.ansible-dev.com", - "platform": "VNIOS", - "service_type_configuration": "ALL_V4", - "vip_setting": - { - "address": "192.168.1.100", - "dscp": 0, - "gateway": "192.168.1.1", - "primary": True, - "subnet_mask": "255.255.255.0", - "use_dscp": False - } - } - ] - - test_spec = { - "host_name": {"ib_req": True}, - "vip_setting": {}, - "config_addr_type": {}, - "platform": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_module.py b/test/units/modules/net_tools/nios/test_nios_module.py deleted file mode 100644 index dbf5885173..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_module.py +++ /dev/null @@ -1,88 +0,0 @@ -# (c) 2018 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 - -import os -import json - -from units.modules.utils import AnsibleExitJson, AnsibleFailJson, ModuleTestCase - - -fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures') -fixture_data = {} - - -def load_fixture(name): - path = os.path.join(fixture_path, name) - - if path in fixture_data: - return fixture_data[path] - - with open(path) as f: - data = f.read() - - try: - data = json.loads(data) - except Exception: - pass - - fixture_data[path] = data - return data - - -class TestNiosModule(ModuleTestCase): - - def execute_module(self, failed=False, changed=False, commands=None, sort=True, defaults=False): - - self.load_fixtures(commands) - - if failed: - result = self.failed() - self.assertTrue(result['failed'], result) - else: - result = self.changed(changed) - self.assertEqual(result['changed'], changed, result) - - if commands is not None: - if sort: - self.assertEqual(sorted(commands), sorted(result['commands']), result['commands']) - else: - self.assertEqual(commands, result['commands'], result['commands']) - - return result - - def failed(self): - with self.assertRaises(AnsibleFailJson) as exc: - self.module.main() - - result = exc.exception.args[0] - self.assertTrue(result['failed'], result) - return result - - def changed(self, changed=False): - with self.assertRaises(AnsibleExitJson) as exc: - self.module.main() - - result = exc.exception.args[0] - self.assertEqual(result['changed'], changed, result) - return result - - def load_fixtures(self, commands=None): - pass diff --git a/test/units/modules/net_tools/nios/test_nios_mx_record.py b/test/units/modules/net_tools/nios/test_nios_mx_record.py deleted file mode 100644 index 21ac7b2b9f..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_mx_record.py +++ /dev/null @@ -1,137 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_mx_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosMXRecordModule(TestNiosModule): - - module = nios_mx_record - - def setUp(self): - super(TestNiosMXRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_mx_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_mx_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_mx_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosMXRecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_mx_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible.com', - 'mx': 'mailhost.ansible.com', 'preference': 0, 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "mx": {"ib_req": True}, - "preference": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'mx': 'mailhost.ansible.com', 'preference': 0}) - - def test_nios_mx_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible.com', 'mx': 'mailhost.ansible.com', - 'preference': 0, 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "mxrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "ansible.com", - "mx": "mailhost.ansible.com", - "preference": 0, - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "mx": {"ib_req": True}, - "preference": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_mx_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'ansible.com', 'mx': 'mailhost.ansible.com', - 'preference': 0, 'comment': None, 'extattrs': None} - - ref = "mxrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "ansible.com", - "mx": "mailhost.ansible.com", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "mx": {"ib_req": True}, - "preference": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_naptr_record.py b/test/units/modules/net_tools/nios/test_nios_naptr_record.py deleted file mode 100644 index 2f1547c29f..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_naptr_record.py +++ /dev/null @@ -1,147 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_naptr_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosNAPTRRecordModule(TestNiosModule): - - module = nios_naptr_record - - def setUp(self): - super(TestNiosNAPTRRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_naptr_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_naptr_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_naptr_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosNAPTRRecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_naptr_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': '*.subscriber-100.ansiblezone.com', - 'order': '1000', 'preference': '10', 'replacement': 'replacement1.network.ansiblezone.com', - 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "order": {"ib_req": True}, - "preference": {"ib_req": True}, - "replacement": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi.__dict__) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'order': '1000', 'preference': '10', - 'replacement': 'replacement1.network.ansiblezone.com'}) - - def test_nios_naptr_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': '*.subscriber-100.ansiblezone.com', - 'order': '1000', 'preference': '10', 'replacement': 'replacement1.network.ansiblezone.com', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "naptrrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "*.subscriber-100.ansiblezone.com", - "order": "1000", - "preference": "10", - "replacement": "replacement1.network.ansiblezone.com", - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "order": {"ib_req": True}, - "preference": {"ib_req": True}, - "replacement": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_naptr_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': '*.subscriber-100.ansiblezone.com', - 'order': '1000', 'preference': '10', 'replacement': 'replacement1.network.ansiblezone.com', - 'comment': None, 'extattrs': None} - - ref = "naptrrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "*.subscriber-100.ansiblezone.com", - "order": "1000", - "preference": "10", - "replacement": "replacement1.network.ansiblezone.com", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "order": {"ib_req": True}, - "preference": {"ib_req": True}, - "replacement": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_network.py b/test/units/modules/net_tools/nios/test_nios_network.py deleted file mode 100644 index 859e24d0a6..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_network.py +++ /dev/null @@ -1,248 +0,0 @@ -# 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 ansible.module_utils.net_tools.nios import api -from ansible.modules.net_tools.nios import nios_network -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosNetworkModule(TestNiosModule): - - module = nios_network - - def setUp(self): - super(TestNiosNetworkModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_network.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_network.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_network.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosNetworkModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def test_nios_network_ipv4_create(self): - self.module.params = {'provider': None, 'state': 'present', 'network': '192.168.10.0/24', - 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'network': '192.168.10.0/24'}) - - def test_nios_network_ipv4_dhcp_update(self): - self.module.params = {'provider': None, 'state': 'present', 'network': '192.168.10.0/24', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "network/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "network": "192.168.10.0/24", - "extattrs": {'options': {'name': 'test', 'value': 'ansible.com'}} - } - ] - - test_spec = { - "network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_network_ipv6_dhcp_update(self): - self.module.params = {'provider': None, 'state': 'present', 'ipv6network': 'fe80::/64', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "ipv6network/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "ipv6network": "fe80::/64", - "extattrs": {'options': {'name': 'test', 'value': 'ansible.com'}} - } - ] - - test_spec = { - "ipv6network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - self.assertTrue(res['changed']) - - def test_nios_network_ipv4_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'network': '192.168.10.0/24', - 'comment': None, 'extattrs': None} - - ref = "network/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "network": "192.168.10.0/24", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_network_ipv6_create(self): - self.module.params = {'provider': None, 'state': 'present', 'ipv6network': 'fe80::/64', - 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "ipv6network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'ipv6network': 'fe80::/64'}) - - def test_nios_network_ipv6_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'ipv6network': 'fe80::/64', - 'comment': None, 'extattrs': None} - - ref = "ipv6network/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "ipv6network": "fe80::/64", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "ipv6network": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_networkcontainer_ipv4_create(self): - self.module.params = {'provider': None, 'state': 'present', 'networkcontainer': '192.168.10.0/24', - 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "networkcontainer": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'networkcontainer': '192.168.10.0/24'}) - - def test_nios_networkcontainer_ipv4_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'networkcontainer': '192.168.10.0/24', - 'comment': None, 'extattrs': None} - - ref = "networkcontainer/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "networkcontainer": "192.168.10.0/24" - }] - - test_spec = { - "networkcontainer": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_networkcontainer_ipv6_create(self): - self.module.params = {'provider': None, 'state': 'present', 'ipv6networkcontainer': 'fe80::/64', - 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "ipv6networkcontainer": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'ipv6networkcontainer': 'fe80::/64'}) diff --git a/test/units/modules/net_tools/nios/test_nios_network_view.py b/test/units/modules/net_tools/nios/test_nios_network_view.py deleted file mode 100644 index 4a4be1244b..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_network_view.py +++ /dev/null @@ -1,156 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_network_view -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosNetworkViewModule(TestNiosModule): - - module = nios_network_view - - def setUp(self): - super(TestNiosNetworkViewModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_network_view.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_network_view.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_network_view.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosNetworkViewModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_network_view_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'ansible', - 'comment': None, 'extattrs': None} - - test_object = None - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__()}) - - def test_nios_network_view_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'default', - 'comment': 'updated comment', 'extattrs': None, 'network_view': 'default'} - - test_object = [ - { - "comment": "test comment", - "_ref": "networkview/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "default", - "extattrs": {}, - "network_view": "default" - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) - - def test_nios_network_view_update_name(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'default', 'old_name': 'old_default', - 'comment': 'updated comment', 'extattrs': None, 'network_view': 'default'} - - test_object = [ - { - "comment": "test comment", - "_ref": "networkview/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "default", - "old_name": "old_default", - "extattrs": {}, - "network_view": "default" - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) - - def test_nios_network_view_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'ansible', - 'comment': None, 'extattrs': None} - - ref = "networkview/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "ansible", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_nsgroup.py b/test/units/modules/net_tools/nios/test_nios_nsgroup.py deleted file mode 100644 index 463fc24687..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_nsgroup.py +++ /dev/null @@ -1,125 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_nsgroup -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosNSGroupModule(TestNiosModule): - - module = nios_nsgroup - - def setUp(self): - - super(TestNiosNSGroupModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_nsgroup.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_nsgroup.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_nsgroup.WapiModule.run') - self.mock_wapi_run.start() - - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosNSGroupModule, self).tearDown() - self.mock_wapi.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_nsgroup_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'my-simple-group', - 'comment': None, 'grid_primary': None} - - test_object = None - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "grid_primary": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__()}) - - def test_nios_nsgroup_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': 'my-simple-group', - 'comment': None, 'grid_primary': None} - - ref = "nsgroup/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "name": "my-simple-group", - "grid_primary": {'name': 'infoblox-test.example.com'} - }] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "grid_primary": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_nsgroup_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': 'default', - 'comment': 'updated comment', 'grid_primary': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "nsgroup/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "default", - "grid_primary": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "comment": {}, - "grid_primary": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) diff --git a/test/units/modules/net_tools/nios/test_nios_ptr_record.py b/test/units/modules/net_tools/nios/test_nios_ptr_record.py deleted file mode 100644 index f41132d292..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_ptr_record.py +++ /dev/null @@ -1,184 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_ptr_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosPTRRecordModule(TestNiosModule): - - module = nios_ptr_record - - def setUp(self): - - super(TestNiosPTRRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_ptr_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_ptr_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_ptr_record.WapiModule.run') - self.mock_wapi_run.start() - - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosPTRRecordModule, self).tearDown() - self.mock_wapi.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_ptr_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'ptrdname': 'ansible.test.com', - 'ipv4addr': '10.36.241.14', 'comment': None, 'extattrs': None, 'view': 'default'} - - test_object = None - test_spec = { - "ipv4addr": {"ib_req": True}, - "ptrdname": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - "view": {"ib_req": True} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'ipv4addr': '10.36.241.14', 'ptrdname': 'ansible.test.com', 'view': 'default'}) - - def test_nios_ptr_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'ptrdname': 'ansible.test.com', - 'ipv4addr': '10.36.241.14', 'comment': None, 'extattrs': None, 'view': 'default'} - - ref = "record:ptr/ZG5zLm5ldHdvcmtfdmlldyQw:14.241.36.10.in-addr.arpa/default" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "ptrdname": "ansible.test.com", - "ipv4addr": "10.36.241.14", - "view": "default", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "ipv4addr": {"ib_req": True}, - "ptrdname": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - "view": {"ib_req": True} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_ptr_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'ptrdname': 'ansible.test.com', - 'ipv4addr': '10.36.241.14', 'comment': 'updated comment', 'extattrs': None, 'view': 'default'} - - test_object = [ - { - "comment": "test comment", - "_ref": "record:ptr/ZG5zLm5ldHdvcmtfdmlldyQw:14.241.36.10.in-addr.arpa/default", - "ptrdname": "ansible.test.com", - "ipv4addr": "10.36.241.14", - "extattrs": {}, - "view": "default" - } - ] - - test_spec = { - "ipv4addr": {"ib_req": True}, - "ptrdname": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - "view": {"ib_req": True} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) - - def test_nios_ptr_record_update_record_ptrdname(self): - self.module.params = {'provider': None, 'state': 'present', 'ptrdname': 'ansible.test.org', - 'ipv4addr': '10.36.241.14', 'comment': 'comment', 'extattrs': None, 'view': 'default'} - - test_object = [ - { - "comment": "test comment", - "_ref": "record:ptr/ZG5zLm5ldHdvcmtfdmlldyQw:14.241.36.10.in-addr.arpa/default", - "ptrdname": "ansible.test.com", - "ipv4addr": "10.36.241.14", - "extattrs": {}, - "view": "default" - } - ] - - test_spec = { - "ipv4addr": {"ib_req": True}, - "ptrdname": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - "view": {"ib_req": True} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.update_object.called_once_with(test_object) - - def test_nios_ptr6_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'ptrdname': 'ansible6.test.com', - 'ipv6addr': '2002:8ac3:802d:1242:20d:60ff:fe38:6d16', 'comment': None, 'extattrs': None, 'view': 'default'} - - test_object = None - test_spec = {"ipv6addr": {"ib_req": True}, - "ptrdname": {"ib_req": True}, - "comment": {}, - "extattrs": {}, - "view": {"ib_req": True}} - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'ipv6addr': '2002:8ac3:802d:1242:20d:60ff:fe38:6d16', - 'ptrdname': 'ansible6.test.com', 'view': 'default'}) diff --git a/test/units/modules/net_tools/nios/test_nios_srv_record.py b/test/units/modules/net_tools/nios/test_nios_srv_record.py deleted file mode 100644 index 2c0b27388c..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_srv_record.py +++ /dev/null @@ -1,153 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_srv_record -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosSRVRecordModule(TestNiosModule): - - module = nios_srv_record - - def setUp(self): - super(TestNiosSRVRecordModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_srv_record.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_srv_record.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_srv_record.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosSRVRecordModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_srv_record_create(self): - self.module.params = {'provider': None, 'state': 'present', 'name': '_sip._tcp.service.ansible.com', - 'port': 5080, 'target': 'service1.ansible.com', 'priority': 10, 'weight': 10, - 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "name": {"ib_req": True}, - "port": {"ib_req": True}, - "target": {"ib_req": True}, - "priority": {"ib_req": True}, - "weight": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'name': self.module._check_type_dict().__getitem__(), - 'port': 5080, 'target': 'service1.ansible.com', 'priority': 10, 'weight': 10}) - - def test_nios_srv_record_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'name': '_sip._tcp.service.ansible.com', - 'port': 5080, 'target': 'service1.ansible.com', 'priority': 10, 'weight': 10, - 'comment': None, 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "srvrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "name": "_sip._tcp.service.ansible.com", - 'port': 5080, - "target": "mailhost.ansible.com", - "priority": 10, - 'weight': 10, - "extattrs": {} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "port": {"ib_req": True}, - "target": {"ib_req": True}, - "priority": {"ib_req": True}, - "weight": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_srv_record_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'name': '_sip._tcp.service.ansible.com', - 'port': 5080, 'target': 'service1.ansible.com', 'priority': 10, 'weight': 10, - 'comment': None, 'extattrs': None} - - ref = "srvrecord/ZG5zLm5ldHdvcmtfdmlldyQw:default/false" - - test_object = [ - { - "comment": "test comment", - "_ref": ref, - "name": "_sip._tcp.service.ansible.com", - "port": 5080, - "target": "mailhost.ansible.com", - "priority": 10, - "weight": 10, - "extattrs": {'Site': {'value': 'test'}} - } - ] - - test_spec = { - "name": {"ib_req": True}, - "port": {"ib_req": True}, - "target": {"ib_req": True}, - "priority": {"ib_req": True}, - "weight": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/nios/test_nios_zone.py b/test/units/modules/net_tools/nios/test_nios_zone.py deleted file mode 100644 index 39cd09dbda..0000000000 --- a/test/units/modules/net_tools/nios/test_nios_zone.py +++ /dev/null @@ -1,287 +0,0 @@ -# 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 ansible.modules.net_tools.nios import nios_zone -from ansible.module_utils.net_tools.nios import api -from units.compat.mock import patch, MagicMock, Mock -from .test_nios_module import TestNiosModule, load_fixture - - -class TestNiosZoneModule(TestNiosModule): - - module = nios_zone - - def setUp(self): - super(TestNiosZoneModule, self).setUp() - self.module = MagicMock(name='ansible.modules.net_tools.nios.nios_zone.WapiModule') - self.module.check_mode = False - self.module.params = {'provider': None} - self.mock_wapi = patch('ansible.modules.net_tools.nios.nios_zone.WapiModule') - self.exec_command = self.mock_wapi.start() - self.mock_wapi_run = patch('ansible.modules.net_tools.nios.nios_zone.WapiModule.run') - self.mock_wapi_run.start() - self.load_config = self.mock_wapi_run.start() - - def tearDown(self): - super(TestNiosZoneModule, self).tearDown() - self.mock_wapi.stop() - self.mock_wapi_run.stop() - - def _get_wapi(self, test_object): - wapi = api.WapiModule(self.module) - wapi.get_object = Mock(name='get_object', return_value=test_object) - wapi.create_object = Mock(name='create_object') - wapi.update_object = Mock(name='update_object') - wapi.delete_object = Mock(name='delete_object') - return wapi - - def load_fixtures(self, commands=None): - self.exec_command.return_value = (0, load_fixture('nios_result.txt').strip(), None) - self.load_config.return_value = dict(diff=None, session='session') - - def test_nios_zone_create(self): - self.module.params = {'provider': None, 'state': 'present', 'fqdn': 'ansible.com', - 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "fqdn": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'fqdn': 'ansible.com'}) - - def test_nios_zone_remove(self): - self.module.params = {'provider': None, 'state': 'absent', 'fqdn': 'ansible.com', - 'comment': None, 'extattrs': None} - - ref = "zone/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "fqdn": "ansible.com", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "fqdn": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_zone_update_comment(self): - self.module.params = {'provider': None, 'state': 'present', 'fqdn': 'ansible.com', - 'comment': 'updated comment', 'extattrs': None} - - test_object = [ - { - "comment": "test comment", - "_ref": "zone/ZG5zLm5ldHdvcmtfdmlldyQw:default/true", - "fqdn": "ansible.com", - "extattrs": {'Site': {'value': 'test'}} - } - ] - - test_spec = { - "fqdn": {"ib_req": True}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - - def test_nios_zone_create_using_grid_primary_secondaries(self): - self.module.params = {'provider': None, 'state': 'present', 'fqdn': 'ansible.com', - 'grid_primary': [{"name": "gridprimary.grid.com"}], - 'grid_secondaries': [{"name": "gridsecondary1.grid.com"}, - {"name": "gridsecondary2.grid.com"}], - 'restart_if_needed': True, - 'comment': None, 'extattrs': None} - - test_object = None - grid_spec = dict( - name=dict(required=True), - ) - test_spec = { - "fqdn": {"ib_req": True}, - "grid_primary": {}, - "grid_secondaries": {}, - "restart_if_needed": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'fqdn': 'ansible.com', - "grid_primary": [{"name": "gridprimary.grid.com"}], - "grid_secondaries": [{"name": "gridsecondary1.grid.com"}, - {"name": "gridsecondary2.grid.com"}], - "restart_if_needed": True - }) - - def test_nios_zone_remove_using_grid_primary_secondaries(self): - self.module.params = {'provider': None, 'state': 'absent', 'fqdn': 'ansible.com', - 'grid_primary': [{"name": "gridprimary.grid.com"}], - 'grid_secondaries': [{"name": "gridsecondary1.grid.com"}, - {"name": "gridsecondary2.grid.com"}], - 'restart_if_needed': True, - 'comment': None, 'extattrs': None} - - ref = "zone/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "fqdn": "ansible.com", - "grid_primary": [{"name": "gridprimary.grid.com"}], - "grid_secondaries": [{"name": "gridsecondary1.grid.com"}, {"name": "gridsecondary2.grid.com"}], - "restart_if_needed": True, - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "fqdn": {"ib_req": True}, - "grid_primary": {}, - "grid_secondaries": {}, - "restart_if_needed": {}, - "comment": {}, - "extattrs": {} - } - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_zone_create_using_name_server_group(self): - self.module.params = {'provider': None, 'state': 'present', 'fqdn': 'ansible.com', - 'ns_group': 'examplensg', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "fqdn": {"ib_req": True}, - "ns_group": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'fqdn': 'ansible.com', - 'ns_group': 'examplensg'}) - - def test_nios_zone_remove_using_name_server_group(self): - self.module.params = {'provider': None, 'state': 'absent', 'fqdn': 'ansible.com', - 'ns_group': 'examplensg', 'comment': None, 'extattrs': None} - - ref = "zone/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "fqdn": "ansible.com", - "ns_group": "examplensg", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "fqdn": {"ib_req": True}, - "ns_group": {}, - "comment": {}, - "extattrs": {} - } - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) - - def test_nios_zone_create_using_zone_format(self): - self.module.params = {'provider': None, 'state': 'present', 'fqdn': '10.10.10.in-addr.arpa', - 'zone_format': 'IPV4', 'comment': None, 'extattrs': None} - - test_object = None - - test_spec = { - "fqdn": {"ib_req": True}, - "zone_format": {}, - "comment": {}, - "extattrs": {} - } - - wapi = self._get_wapi(test_object) - print("WAPI: ", wapi) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.create_object.assert_called_once_with('testobject', {'fqdn': '10.10.10.in-addr.arpa', - 'zone_format': 'IPV4'}) - - def test_nios_zone_remove_using_using_zone_format(self): - self.module.params = {'provider': None, 'state': 'absent', 'fqdn': 'ansible.com', - 'zone_format': 'IPV4', 'comment': None, 'extattrs': None} - - ref = "zone/ZG5zLm5ldHdvcmtfdmlldyQw:ansible/false" - - test_object = [{ - "comment": "test comment", - "_ref": ref, - "fqdn": "ansible.com", - "zone_format": "IPV4", - "extattrs": {'Site': {'value': 'test'}} - }] - - test_spec = { - "fqdn": {"ib_req": True}, - "zone_format": {}, - "comment": {}, - "extattrs": {} - } - wapi = self._get_wapi(test_object) - res = wapi.run('testobject', test_spec) - - self.assertTrue(res['changed']) - wapi.delete_object.assert_called_once_with(ref) diff --git a/test/units/modules/net_tools/test_hetzner_firewall.py b/test/units/modules/net_tools/test_hetzner_firewall.py deleted file mode 100644 index 870d8e3967..0000000000 --- a/test/units/modules/net_tools/test_hetzner_firewall.py +++ /dev/null @@ -1,1405 +0,0 @@ -# (c) 2019 Felix Fontein <felix@fontein.de> -# 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.module_utils.hetzner import BASE_URL -from ansible.modules.net_tools import hetzner_firewall - - -# ########################################################## -# ## General test framework - -import json - -from mock import MagicMock -from units.modules.utils import set_module_args -from ansible.module_utils.six.moves.urllib.parse import parse_qs - - -class FetchUrlCall: - def __init__(self, method, status): - assert method == method.upper(), \ - 'HTTP method names are case-sensitive and should be upper-case (RFCs 7230 and 7231)' - self.method = method - self.status = status - self.body = None - self.headers = {} - self.error_data = {} - self.expected_url = None - self.expected_headers = {} - self.form_parse = False - self.form_present = set() - self.form_values = {} - self.form_values_one = {} - - def result(self, body): - self.body = body - assert self.error_data.get('body') is None, 'Error body must not be given' - return self - - def result_str(self, str_body): - return self.result(str_body.encode('utf-8')) - - def result_json(self, json_body): - return self.result(json.dumps(json_body).encode('utf-8')) - - def result_error(self, msg, body=None): - self.error_data['msg'] = msg - if body is not None: - self.error_data['body'] = body - assert self.body is None, 'Result must not be given if error body is provided' - return self - - def expect_url(self, url): - self.expected_url = url - return self - - def return_header(self, name, value): - assert value is not None - self.headers[name] = value - return self - - def expect_header(self, name, value): - self.expected_headers[name] = value - return self - - def expect_header_unset(self, name): - self.expected_headers[name] = None - return self - - def expect_form_present(self, key): - self.form_parse = True - self.form_present.append(key) - return self - - def expect_form_value(self, key, value): - self.form_parse = True - self.form_values[key] = [value] - return self - - def expect_form_value_absent(self, key): - self.form_parse = True - self.form_values[key] = [] - return self - - def expect_form_value_one_of(self, key, value): - self.form_parse = True - if key not in self.form_values_subset: - self.form_values_subset[key] = set() - self.form_values_subset[key].add(value) - return self - - -class FetchUrlProxy: - def __init__(self, calls): - self.calls = calls - self.index = 0 - - def _validate_form(self, call, data): - form = {} - if data is not None: - form = parse_qs(data, keep_blank_values=True) - for k in call.form_present: - assert k in form - for k, v in call.form_values.items(): - if len(v) == 0: - assert k not in form - else: - assert form[k] == v - for k, v in call.form_values_one.items(): - assert v <= set(form[k]) - - def _validate_headers(self, call, headers): - given_headers = {} - if headers is not None: - for k, v in headers.items(): - given_headers[k.lower()] = v - for k, v in call.expected_headers: - if v is None: - assert k.lower() not in given_headers, \ - 'Header "{0}" specified for fetch_url call, but should not be'.format(k) - else: - assert given_headers.get(k.lower()) == v, \ - 'Header "{0}" specified for fetch_url call, but with wrong value'.format(k) - - def __call__(self, module, url, data=None, headers=None, method=None, - use_proxy=True, force=False, last_mod_time=None, timeout=10, - use_gssapi=False, unix_socket=None, ca_path=None, cookies=None): - assert self.index < len(self.calls), 'Got more fetch_url calls than expected' - call = self.calls[self.index] - self.index += 1 - - # Validate call - assert method == call.method - if call.expected_url is not None: - assert url == call.expected_url, \ - 'Exepected URL does not match for fetch_url call' - if call.expected_headers: - self._validate_headers(call, headers) - if call.form_parse: - self._validate_form(call, data) - - # Compose result - info = dict(status=call.status) - for k, v in call.headers.items(): - info[k.lower()] = v - info.update(call.error_data) - res = object() - if call.body is not None: - res = MagicMock() - res.read = MagicMock(return_value=call.body) - return (res, info) - - def assert_is_done(self): - assert self.index == len(self.calls), 'Got less fetch_url calls than expected' - - -class ModuleExitException(Exception): - def __init__(self, kwargs): - self.kwargs = kwargs - - -class ModuleFailException(Exception): - def __init__(self, kwargs): - self.kwargs = kwargs - - -def run_module(mocker, module, arguments, fetch_url): - def exit_json(module, **kwargs): - module._return_formatted(kwargs) - raise ModuleExitException(kwargs) - - def fail_json(module, **kwargs): - module._return_formatted(kwargs) - raise ModuleFailException(kwargs) - - mocker.patch('ansible.module_utils.hetzner.fetch_url', fetch_url) - mocker.patch('ansible.module_utils.hetzner.time.sleep', lambda duration: None) - mocker.patch('ansible.modules.net_tools.hetzner_firewall.AnsibleModule.exit_json', exit_json) - mocker.patch('ansible.modules.net_tools.hetzner_firewall.AnsibleModule.fail_json', fail_json) - set_module_args(arguments) - module.main() - - -def run_module_success(mocker, module, arguments, fetch_url_calls): - fetch_url = FetchUrlProxy(fetch_url_calls or []) - with pytest.raises(ModuleExitException) as e: - run_module(mocker, module, arguments, fetch_url) - fetch_url.assert_is_done() - return e.value.kwargs - - -def run_module_failed(mocker, module, arguments, fetch_url_calls): - fetch_url = FetchUrlProxy(fetch_url_calls or []) - with pytest.raises(ModuleFailException) as e: - run_module(mocker, module, arguments, fetch_url) - fetch_url.assert_is_done() - return e.value.kwargs - - -# ########################################################## -# ## Hetzner firewall tests - - -# Tests for state (absent and present) - - -def test_absent_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_absent_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'disabled'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Tests for state (absent and present) with check mode - - -def test_absent_idempotency_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_absent_changed_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'absent', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'disabled' - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_idempotency_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present_changed_check(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - '_ansible_check_mode': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Tests for port - - -def test_port_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'port': 'main', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['port'] == 'main' - assert result['diff']['after']['port'] == 'main' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['port'] == 'main' - - -def test_port_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'port': 'main', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': True, - 'port': 'kvm', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('port', 'main'), - ]) - assert result['changed'] is True - assert result['diff']['before']['port'] == 'kvm' - assert result['diff']['after']['port'] == 'main' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['port'] == 'main' - - -# Tests for whitelist_hos - - -def test_whitelist_hos_idempotency(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'whitelist_hos': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['whitelist_hos'] is True - assert result['diff']['after']['whitelist_hos'] is True - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['whitelist_hos'] is True - - -def test_whitelist_hos_changed(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'whitelist_hos': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('whitelist_hos', 'true'), - ]) - assert result['changed'] is True - assert result['diff']['before']['whitelist_hos'] is False - assert result['diff']['after']['whitelist_hos'] is True - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert result['firewall']['whitelist_hos'] is True - - -# Tests for wait_for_configured in getting status - - -def test_wait_get(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_wait_get_timeout(mocker): - result = run_module_failed(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Timeout while waiting for firewall to be configured.' - - -def test_nowait_get(mocker): - result = run_module_failed(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Firewall configuration cannot be read as it is not configured.' - - -# Tests for wait_for_configured in setting status - - -def test_wait_update(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - 'state': 'present', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_wait_update_timeout(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert 'Timeout while waiting for firewall to be configured.' in result['warnings'] - - -def test_nowait_update(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'disabled' - assert result['diff']['after']['status'] == 'active' - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -# Idempotency checks: different amount of input rules - -def test_input_rule_len_change_0_1(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [ - { - 'ip_version': 'ipv4', - 'action': 'discard', - }, - ], - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][0][name]') - .expect_form_value('rules[input][0][ip_version]', 'ipv4') - .expect_form_value_absent('rules[input][0][dst_ip]') - .expect_form_value_absent('rules[input][0][dst_port]') - .expect_form_value_absent('rules[input][0][src_ip]') - .expect_form_value_absent('rules[input][0][src_port]') - .expect_form_value_absent('rules[input][0][protocol]') - .expect_form_value_absent('rules[input][0][tcp_flags]') - .expect_form_value('rules[input][0][action]', 'discard') - .expect_form_value_absent('rules[input][1][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 0 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - - -def test_input_rule_len_change_1_0(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][0][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 0 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 0 - - -def test_input_rule_len_change_1_2(mocker): - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [ - { - 'ip_version': 'ipv4', - 'dst_port': 80, - 'protocol': 'tcp', - 'action': 'accept', - }, - { - 'ip_version': 'ipv4', - 'action': 'discard', - }, - ], - }, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': '80', - 'src_ip': None, - 'src_port': None, - 'protocol': 'tcp', - 'tcp_flags': None, - 'action': 'accept', - }, - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - }, - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value('rules[input][0][action]', 'accept') - .expect_form_value('rules[input][1][action]', 'discard') - .expect_form_value_absent('rules[input][2][action]'), - ]) - assert result['changed'] is True - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 2 - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 2 - - -# Idempotency checks: change one value - - -def create_params(parameter, *values): - assert len(values) > 1 - result = [] - for i in range(1, len(values)): - result.append((parameter, values[i - 1], values[i])) - return result - - -def flatten(list_of_lists): - result = [] - for l in list_of_lists: - result.extend(l) - return result - - -@pytest.mark.parametrize("parameter, before, after", flatten([ - create_params('name', None, '', 'Test', 'Test', 'foo', '', None), - create_params('ip_version', 'ipv4', 'ipv4', 'ipv6', 'ipv6'), - create_params('dst_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), - create_params('dst_port', None, '80', '80-443', '80-443', None), - create_params('src_ip', None, '1.2.3.4/24', '1.2.3.4/32', '1.2.3.4/32', None), - create_params('src_port', None, '80', '80-443', '80-443', None), - create_params('protocol', None, 'tcp', 'tcp', 'udp', 'udp', None), - create_params('tcp_flags', None, 'syn', 'syn|fin', 'syn|fin', 'syn&fin', '', None), - create_params('action', 'accept', 'accept', 'discard', 'discard'), -])) -def test_input_rule_value_change(mocker, parameter, before, after): - input_call = { - 'ip_version': 'ipv4', - 'action': 'discard', - } - input_before = { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - input_after = { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - if after is not None: - input_call[parameter] = after - input_before[parameter] = before - input_after[parameter] = after - - calls = [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [input_before], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ] - - changed = (before != after) - if changed: - after_call = ( - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [input_after], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][1][action]') - ) - if parameter != 'ip_version': - after_call.expect_form_value('rules[input][0][ip_version]', 'ipv4') - if parameter != 'action': - after_call.expect_form_value('rules[input][0][action]', 'discard') - if after is not None: - after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after) - else: - after_call.expect_form_value_absent('rules[input][0][{0}]'.format(parameter)) - calls.append(after_call) - - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [input_call], - }, - }, calls) - assert result['changed'] == changed - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['diff']['before']['rules']['input'][0][parameter] == before - assert result['diff']['after']['rules']['input'][0][parameter] == after - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - assert result['firewall']['rules']['input'][0][parameter] == after - - -# Idempotency checks: IP address normalization - - -@pytest.mark.parametrize("ip_version, parameter, before_normalized, after_normalized, after", [ - ('ipv4', 'src_ip', '1.2.3.4/32', '1.2.3.4/32', '1.2.3.4'), - ('ipv6', 'src_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3::4'), - ('ipv6', 'dst_ip', '1:2:3::4/128', '1:2:3::4/128', '1:2:3:0::4'), - ('ipv6', 'dst_ip', '::/0', '::/0', '0:0::0/0'), -]) -def test_input_rule_ip_normalization(mocker, ip_version, parameter, before_normalized, after_normalized, after): - assert ip_version in ('ipv4', 'ipv6') - assert parameter in ('src_ip', 'dst_ip') - input_call = { - 'ip_version': ip_version, - 'action': 'discard', - } - input_before = { - 'name': None, - 'ip_version': ip_version, - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - input_after = { - 'name': None, - 'ip_version': ip_version, - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - if after is not None: - input_call[parameter] = after - input_before[parameter] = before_normalized - input_after[parameter] = after_normalized - - calls = [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': True, - 'port': 'main', - 'rules': { - 'input': [input_before], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ] - - changed = (before_normalized != after_normalized) - if changed: - after_call = ( - FetchUrlCall('POST', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [input_after], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)) - .expect_form_value('status', 'active') - .expect_form_value_absent('rules[input][1][action]') - ) - after_call.expect_form_value('rules[input][0][ip_version]', ip_version) - after_call.expect_form_value('rules[input][0][action]', 'discard') - after_call.expect_form_value('rules[input][0][{0}]'.format(parameter), after_normalized) - calls.append(after_call) - - result = run_module_success(mocker, hetzner_firewall, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'state': 'present', - 'rules': { - 'input': [input_call], - }, - }, calls) - assert result['changed'] == changed - assert result['diff']['before']['status'] == 'active' - assert result['diff']['after']['status'] == 'active' - assert len(result['diff']['before']['rules']['input']) == 1 - assert len(result['diff']['after']['rules']['input']) == 1 - assert result['diff']['before']['rules']['input'][0][parameter] == before_normalized - assert result['diff']['after']['rules']['input'][0][parameter] == after_normalized - assert result['firewall']['status'] == 'active' - assert len(result['firewall']['rules']['input']) == 1 - assert result['firewall']['rules']['input'][0][parameter] == after_normalized diff --git a/test/units/modules/net_tools/test_hetzner_firewall_info.py b/test/units/modules/net_tools/test_hetzner_firewall_info.py deleted file mode 100644 index a21d2d8056..0000000000 --- a/test/units/modules/net_tools/test_hetzner_firewall_info.py +++ /dev/null @@ -1,239 +0,0 @@ -# (c) 2019 Felix Fontein <felix@fontein.de> -# 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.module_utils.hetzner import BASE_URL -from ansible.modules.net_tools import hetzner_firewall_info -from .test_hetzner_firewall import FetchUrlCall, run_module_success, run_module_failed - - -# Tests for state (absent and present) - - -def test_absent(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'disabled', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'disabled' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_present(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert len(result['firewall']['rules']['input']) == 0 - - -def test_present_w_rules(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [ - { - 'name': 'Accept HTTPS traffic', - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': '443', - 'src_ip': None, - 'src_port': None, - 'protocol': 'tcp', - 'tcp_flags': None, - 'action': 'accept', - }, - { - 'name': None, - 'ip_version': 'ipv4', - 'dst_ip': None, - 'dst_port': None, - 'src_ip': None, - 'src_port': None, - 'protocol': None, - 'tcp_flags': None, - 'action': 'discard', - } - ], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - assert len(result['firewall']['rules']['input']) == 2 - assert result['firewall']['rules']['input'][0]['name'] == 'Accept HTTPS traffic' - assert result['firewall']['rules']['input'][0]['dst_port'] == '443' - assert result['firewall']['rules']['input'][0]['action'] == 'accept' - assert result['firewall']['rules']['input'][1]['dst_port'] is None - assert result['firewall']['rules']['input'][1]['action'] == 'discard' - - -# Tests for wait_for_configured in getting status - - -def test_wait_get(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'active', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'active' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 - - -def test_wait_get_timeout(mocker): - result = run_module_failed(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': True, - 'timeout': 0, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['msg'] == 'Timeout while waiting for firewall to be configured.' - - -def test_nowait_get(mocker): - result = run_module_success(mocker, hetzner_firewall_info, { - 'hetzner_user': '', - 'hetzner_password': '', - 'server_ip': '1.2.3.4', - 'wait_for_configured': False, - }, [ - FetchUrlCall('GET', 200) - .result_json({ - 'firewall': { - 'server_ip': '1.2.3.4', - 'server_number': 1, - 'status': 'in process', - 'whitelist_hos': False, - 'port': 'main', - 'rules': { - 'input': [], - }, - }, - }) - .expect_url('{0}/firewall/1.2.3.4'.format(BASE_URL)), - ]) - assert result['changed'] is False - assert result['firewall']['status'] == 'in process' - assert result['firewall']['server_ip'] == '1.2.3.4' - assert result['firewall']['server_number'] == 1 diff --git a/test/units/modules/net_tools/test_nmcli.py b/test/units/modules/net_tools/test_nmcli.py deleted file mode 100644 index 94ee5f2a82..0000000000 --- a/test/units/modules/net_tools/test_nmcli.py +++ /dev/null @@ -1,656 +0,0 @@ -# Copyright: (c) 2017 Ansible Project -# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) - -import json - -import pytest - -from ansible.modules.net_tools import nmcli - -pytestmark = pytest.mark.usefixtures('patch_ansible_module') - -TESTCASE_CONNECTION = [ - { - 'type': 'ethernet', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'generic', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'team', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'bond', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'bond-slave', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'bridge', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'vlan', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'vxlan', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'ipip', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, - { - 'type': 'sit', - 'conn_name': 'non_existent_nw_device', - 'state': 'absent', - '_ansible_check_mode': True, - }, -] - -TESTCASE_GENERIC = [ - { - 'type': 'generic', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'generic_non_existant', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'state': 'present', - '_ansible_check_mode': False, - }, -] - -TESTCASE_GENERIC_DNS4_SEARCH = [ - { - 'type': 'generic', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'generic_non_existant', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'state': 'present', - 'dns4_search': 'search.redhat.com', - 'dns6_search': 'search6.redhat.com', - '_ansible_check_mode': False, - } -] - -TESTCASE_BOND = [ - { - 'type': 'bond', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'bond_non_existant', - 'mode': 'active-backup', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'state': 'present', - 'primary': 'non_existent_primary', - '_ansible_check_mode': False, - } -] - -TESTCASE_BRIDGE = [ - { - 'type': 'bridge', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'br0_non_existant', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'maxage': 100, - 'stp': True, - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_BRIDGE_SLAVE = [ - { - 'type': 'bridge-slave', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'br0_non_existant', - 'path_cost': 100, - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_VLAN = [ - { - 'type': 'vlan', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'vlan_not_exists', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'vlanid': 10, - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_VXLAN = [ - { - 'type': 'vxlan', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'vxlan-existent_nw_device', - 'vxlan_id': 11, - 'vxlan_local': '192.168.225.5', - 'vxlan_remote': '192.168.225.6', - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_IPIP = [ - { - 'type': 'ipip', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'ipip-existent_nw_device', - 'ip_tunnel_dev': 'non_existent_ipip_device', - 'ip_tunnel_local': '192.168.225.5', - 'ip_tunnel_remote': '192.168.225.6', - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_SIT = [ - { - 'type': 'sit', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'sit-existent_nw_device', - 'ip_tunnel_dev': 'non_existent_sit_device', - 'ip_tunnel_local': '192.168.225.5', - 'ip_tunnel_remote': '192.168.225.6', - 'state': 'present', - '_ansible_check_mode': False, - } -] - -TESTCASE_ETHERNET_DHCP = [ - { - 'type': 'ethernet', - 'conn_name': 'non_existent_nw_device', - 'ifname': 'ethernet_non_existant', - 'ip4': '10.10.10.10', - 'gw4': '10.10.10.1', - 'state': 'present', - '_ansible_check_mode': False, - 'dhcp_client_id': '00:11:22:AA:BB:CC:DD', - } -] - - -def mocker_set(mocker, connection_exists=False): - """ - Common mocker object - """ - mocker.patch('ansible.modules.net_tools.nmcli.HAVE_DBUS', True) - mocker.patch('ansible.modules.net_tools.nmcli.HAVE_NM_CLIENT', True) - get_bin_path = mocker.patch('ansible.module_utils.basic.AnsibleModule.get_bin_path') - get_bin_path.return_value = '/usr/bin/nmcli' - connection = mocker.patch.object(nmcli.Nmcli, 'connection_exists') - connection.return_value = connection_exists - return connection - - -@pytest.fixture -def mocked_generic_connection_create(mocker): - mocker_set(mocker) - command_result = mocker.patch.object(nmcli.Nmcli, 'execute_command') - command_result.return_value = {"rc": 100, "out": "aaa", "err": "none"} - return command_result - - -@pytest.fixture -def mocked_generic_connection_modify(mocker): - mocker_set(mocker, connection_exists=True) - command_result = mocker.patch.object(nmcli.Nmcli, 'execute_command') - command_result.return_value = {"rc": 100, "out": "aaa", "err": "none"} - return command_result - - -@pytest.fixture -def mocked_connection_exists(mocker): - connection = mocker_set(mocker, connection_exists=True) - return connection - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_BOND, indirect=['patch_ansible_module']) -def test_bond_connection_create(mocked_generic_connection_create): - """ - Test : Bond connection created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'bond' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - assert args[0][7] == 'ifname' - assert args[0][8] == 'bond_non_existant' - - for param in ['gw4', 'primary', 'autoconnect', 'mode', 'active-backup', 'ip4']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_GENERIC, indirect=['patch_ansible_module']) -def test_generic_connection_create(mocked_generic_connection_create): - """ - Test : Generic connection created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'generic' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - - for param in ['autoconnect', 'gw4', 'ip4']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_GENERIC, indirect=['patch_ansible_module']) -def test_generic_connection_modify(mocked_generic_connection_modify): - """ - Test : Generic connection modify - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['ipv4.gateway', 'ipv4.address']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_GENERIC_DNS4_SEARCH, indirect=['patch_ansible_module']) -def test_generic_connection_create_dns_search(mocked_generic_connection_create): - """ - Test : Generic connection created with dns search - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert 'ipv4.dns-search' in args[0] - assert 'ipv6.dns-search' in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_GENERIC_DNS4_SEARCH, indirect=['patch_ansible_module']) -def test_generic_connection_modify_dns_search(mocked_generic_connection_create): - """ - Test : Generic connection modified with dns search - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert 'ipv4.dns-search' in args[0] - assert 'ipv6.dns-search' in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_CONNECTION, indirect=['patch_ansible_module']) -def test_dns4_none(mocked_connection_exists, capfd): - """ - Test if DNS4 param is None - """ - with pytest.raises(SystemExit): - nmcli.main() - - out, err = capfd.readouterr() - results = json.loads(out) - assert results['changed'] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_BRIDGE, indirect=['patch_ansible_module']) -def test_create_bridge(mocked_generic_connection_create): - """ - Test if Bridge created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'bridge' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - - for param in ['ip4', '10.10.10.10', 'gw4', '10.10.10.1', 'bridge.max-age', 100, 'bridge.stp', 'yes']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_BRIDGE, indirect=['patch_ansible_module']) -def test_mod_bridge(mocked_generic_connection_modify): - """ - Test if Bridge modified - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - for param in ['ipv4.address', '10.10.10.10', 'ipv4.gateway', '10.10.10.1', 'bridge.max-age', 100, 'bridge.stp', 'yes']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_BRIDGE_SLAVE, indirect=['patch_ansible_module']) -def test_create_bridge_slave(mocked_generic_connection_create): - """ - Test if Bridge_slave created - """ - - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'bridge-slave' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - - for param in ['bridge-port.path-cost', 100]: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_BRIDGE_SLAVE, indirect=['patch_ansible_module']) -def test_mod_bridge_slave(mocked_generic_connection_modify): - """ - Test if Bridge_slave modified - """ - - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['bridge-port.path-cost', 100]: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_VLAN, indirect=['patch_ansible_module']) -def test_create_vlan_con(mocked_generic_connection_create): - """ - Test if VLAN created - """ - - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'vlan' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - - for param in ['ip4', '10.10.10.10', 'gw4', '10.10.10.1', 'id', '10']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_VLAN, indirect=['patch_ansible_module']) -def test_mod_vlan_conn(mocked_generic_connection_modify): - """ - Test if VLAN modified - """ - - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['ipv4.address', '10.10.10.10', 'ipv4.gateway', '10.10.10.1', 'vlan.id', '10']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_VXLAN, indirect=['patch_ansible_module']) -def test_create_vxlan(mocked_generic_connection_create): - """ - Test if vxlan created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'vxlan' - assert args[0][5] == 'con-name' - assert args[0][6] == 'non_existent_nw_device' - assert args[0][7] == 'ifname' - - for param in ['vxlan.local', '192.168.225.5', 'vxlan.remote', '192.168.225.6', 'vxlan.id', 11]: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_VXLAN, indirect=['patch_ansible_module']) -def test_vxlan_mod(mocked_generic_connection_modify): - """ - Test if vxlan modified - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['vxlan.local', '192.168.225.5', 'vxlan.remote', '192.168.225.6', 'vxlan.id', 11]: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_IPIP, indirect=['patch_ansible_module']) -def test_create_ipip(mocked_generic_connection_create): - """ - Test if ipip created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'ip-tunnel' - assert args[0][5] == 'mode' - assert args[0][6] == 'ipip' - assert args[0][7] == 'con-name' - assert args[0][8] == 'non_existent_nw_device' - assert args[0][9] == 'ifname' - assert args[0][10] == 'ipip-existent_nw_device' - assert args[0][11] == 'dev' - assert args[0][12] == 'non_existent_ipip_device' - - for param in ['ip-tunnel.local', '192.168.225.5', 'ip-tunnel.remote', '192.168.225.6']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_IPIP, indirect=['patch_ansible_module']) -def test_ipip_mod(mocked_generic_connection_modify): - """ - Test if ipip modified - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['ip-tunnel.local', '192.168.225.5', 'ip-tunnel.remote', '192.168.225.6']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_SIT, indirect=['patch_ansible_module']) -def test_create_sit(mocked_generic_connection_create): - """ - Test if sit created - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'add' - assert args[0][3] == 'type' - assert args[0][4] == 'ip-tunnel' - assert args[0][5] == 'mode' - assert args[0][6] == 'sit' - assert args[0][7] == 'con-name' - assert args[0][8] == 'non_existent_nw_device' - assert args[0][9] == 'ifname' - assert args[0][10] == 'sit-existent_nw_device' - assert args[0][11] == 'dev' - assert args[0][12] == 'non_existent_sit_device' - - for param in ['ip-tunnel.local', '192.168.225.5', 'ip-tunnel.remote', '192.168.225.6']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_SIT, indirect=['patch_ansible_module']) -def test_sit_mod(mocked_generic_connection_modify): - """ - Test if sit modified - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert args[0][0] == '/usr/bin/nmcli' - assert args[0][1] == 'con' - assert args[0][2] == 'mod' - assert args[0][3] == 'non_existent_nw_device' - - for param in ['ip-tunnel.local', '192.168.225.5', 'ip-tunnel.remote', '192.168.225.6']: - assert param in args[0] - - -@pytest.mark.parametrize('patch_ansible_module', TESTCASE_ETHERNET_DHCP, indirect=['patch_ansible_module']) -def test_eth_dhcp_client_id_con_create(mocked_generic_connection_create): - """ - Test : Ethernet connection created with DHCP_CLIENT_ID - """ - with pytest.raises(SystemExit): - nmcli.main() - - assert nmcli.Nmcli.execute_command.call_count == 1 - arg_list = nmcli.Nmcli.execute_command.call_args_list - args, kwargs = arg_list[0] - - assert 'ipv4.dhcp-client-id' in args[0] |