diff --git a/tests/unit/compat/__init__.py b/tests/unit/compat/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/unit/compat/builtins.py b/tests/unit/compat/builtins.py deleted file mode 100644 index bfc8adf..0000000 --- a/tests/unit/compat/builtins.py +++ /dev/null @@ -1,34 +0,0 @@ -# (c) 2014, Toshio Kuratomi -# -# 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 . - -# Make coding more python3-ish -from __future__ import absolute_import, division, print_function - -__metaclass__ = type - -# -# Compat for python2.7 -# - -# One unittest needs to import builtins via __import__() so we need to have -# the string that represents it -try: - import __builtin__ -except ImportError: - BUILTINS = "builtins" -else: - BUILTINS = "__builtin__" diff --git a/tests/unit/compat/mock.py b/tests/unit/compat/mock.py deleted file mode 100644 index 2c62c9e..0000000 --- a/tests/unit/compat/mock.py +++ /dev/null @@ -1,125 +0,0 @@ -# (c) 2014, Toshio Kuratomi -# -# 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 . - -# Make coding more python3-ish -from __future__ import absolute_import, division, print_function - -__metaclass__ = type - -""" -Compat module for Python3.x's unittest.mock module -""" -import sys - -import _io - -# Python 2.7 - -# Note: Could use the pypi mock library on python3.x as well as python2.x. It -# is the same as the python3 stdlib mock library - -try: - # Allow wildcard import because we really do want to import all of mock's - # symbols into this compat shim - # pylint: disable=wildcard-import,unused-wildcard-import - from unittest.mock import * -except ImportError: - # Python 2 - # pylint: disable=wildcard-import,unused-wildcard-import - try: - from mock import * - except ImportError: - print("You need the mock library installed on python2.x to run tests") - - -# Prior to 3.4.4, mock_open cannot handle binary read_data -if sys.version_info >= (3,) and sys.version_info < (3, 4, 4): - file_spec = None - - def _iterate_read_data(read_data): - # Helper for mock_open: - # Retrieve lines from read_data via a generator so that separate calls to - # readline, read, and readlines are properly interleaved - sep = b"\n" if isinstance(read_data, bytes) else "\n" - data_as_list = [l + sep for l in read_data.split(sep)] - - if data_as_list[-1] == sep: - # If the last line ended in a newline, the list comprehension will have an - # extra entry that's just a newline. Remove this. - data_as_list = data_as_list[:-1] - else: - # If there wasn't an extra newline by itself, then the file being - # emulated doesn't have a newline to end the last line remove the - # newline that our naive format() added - data_as_list[-1] = data_as_list[-1][:-1] - - for line in data_as_list: - yield line - - def mock_open(mock=None, read_data=""): - """ - A helper function to create a mock to replace the use of `open`. It works - for `open` called directly or used as a context manager. - - The `mock` argument is the mock object to configure. If `None` (the - default) then a `MagicMock` will be created for you, with the API limited - to methods or attributes available on standard file handles. - - `read_data` is a string for the `read` methoddline`, and `readlines` of the - file handle to return. This is an empty string by default. - """ - - def _readlines_side_effect(*args, **kwargs): - if handle.readlines.return_value is not None: - return handle.readlines.return_value - return list(_data) - - def _read_side_effect(*args, **kwargs): - if handle.read.return_value is not None: - return handle.read.return_value - return type(read_data)().join(_data) - - def _readline_side_effect(): - if handle.readline.return_value is not None: - while True: - yield handle.readline.return_value - for line in _data: - yield line - - global file_spec - if file_spec is None: - file_spec = list(set(dir(_io.TextIOWrapper)).union(set(dir(_io.BytesIO)))) - - if mock is None: - mock = MagicMock(name="open", spec=open) - - handle = MagicMock(spec=file_spec) - handle.__enter__.return_value = handle - - _data = _iterate_read_data(read_data) - - handle.write.return_value = None - handle.read.return_value = None - handle.readline.return_value = None - handle.readlines.return_value = None - - handle.read.side_effect = _read_side_effect - handle.readline.side_effect = _readline_side_effect() - handle.readlines.side_effect = _readlines_side_effect - - mock.return_value = handle - return mock diff --git a/tests/unit/compat/unittest.py b/tests/unit/compat/unittest.py deleted file mode 100644 index df3379b..0000000 --- a/tests/unit/compat/unittest.py +++ /dev/null @@ -1,39 +0,0 @@ -# (c) 2014, Toshio Kuratomi -# -# 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 . - -# Make coding more python3-ish -from __future__ import absolute_import, division, print_function - -__metaclass__ = type - -""" -Compat module for Python2.7's unittest module -""" - -import sys - -# Allow wildcard import because we really do want to import all of -# unittests's symbols into this compat shim -# pylint: disable=wildcard-import,unused-wildcard-import -if sys.version_info < (2, 7): - try: - # Need unittest2 on python2.6 - from unittest2 import * - except ImportError: - print("You need unittest2 installed on python2.6.x to run tests") -else: - from unittest import * diff --git a/tests/unit/modules/network/vyos/test_vyos_banner.py b/tests/unit/modules/network/vyos/test_vyos_banner.py index 1d6738c..f054ee0 100644 --- a/tests/unit/modules/network/vyos/test_vyos_banner.py +++ b/tests/unit/modules/network/vyos/test_vyos_banner.py @@ -1,59 +1,60 @@ # 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_banner -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule class TestVyosBannerModule(TestVyosModule): module = vyos_banner def setUp(self): super(TestVyosBannerModule, self).setUp() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_banner.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_banner.load_config" ) self.load_config = self.mock_load_config.start() def tearDown(self): super(TestVyosBannerModule, self).tearDown() self.mock_get_config.stop() self.mock_load_config.stop() def load_fixtures(self, commands=None, filename=None): self.load_config.return_value = dict(diff=None, session="session") def test_vyos_banner_create(self): set_module_args(dict(banner="pre-login", text="test\nbanner\nstring")) commands = ["set system login banner pre-login 'test\\nbanner\\nstring'"] self.execute_module(changed=True, commands=commands) def test_vyos_banner_remove(self): set_module_args(dict(banner="pre-login", state="absent")) self.execute_module(changed=False, commands=[]) diff --git a/tests/unit/modules/network/vyos/test_vyos_bgp_address_family.py b/tests/unit/modules/network/vyos/test_vyos_bgp_address_family.py index 0bbf74a..9ef28aa 100644 --- a/tests/unit/modules/network/vyos/test_vyos_bgp_address_family.py +++ b/tests/unit/modules/network/vyos/test_vyos_bgp_address_family.py @@ -1,600 +1,601 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_bgp_address_family -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosBgpafModule(TestVyosModule): module = vyos_bgp_address_family def setUp(self): super(TestVyosBgpafModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts." + "bgp_address_family.bgp_address_family.Bgp_address_familyFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosBgpafModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_bgp_address_family_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_vyos_bgp_address_family_merged_idempotent(self): set_module_args( dict( config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", as_set=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), dict(prefix="192.2.13.0/24", backdoor=True), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ripng", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="export", route_map="map01")], soft_reconfiguration=True, ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv6", attribute_unchanged=dict(next_hop=True), ) ], ), ], ) ) ) self.execute_module(changed=False, commands=[]) def test_vyos_bgp_address_family_merged(self): set_module_args( dict( config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", summary_only=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ospfv3", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.10.21.25", address_family=[ dict( afi="ipv6", distribute_list=[dict(action="export", acl=10)], route_server_client=True, ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv4", filter_list=[ dict(action="export", path_list="list01"), ], capability=dict(orf="send"), ) ], ), ], ) ) ) commands = [ "set protocols bgp 65536 address-family ipv4-unicast aggregate-address 192.0.2.0/24 as-setipv4-unicast aggregate-address 192.0.2.0/24 summary-only", "set protocols bgp 65536 address-family ipv6-unicast redistribute ospfv3 metric 20", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv4-unicast filter-list export list01", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv4-unicast capability prefix-list send", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast distribute-list export 10", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast route-server-client", ] self.execute_module(changed=True, commands=commands) def test_vyos_bgp_address_family_replaced_idempotent(self): set_module_args( dict( state="replaced", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", as_set=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), dict(prefix="192.2.13.0/24", backdoor=True), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ripng", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="export", route_map="map01")], soft_reconfiguration=True, ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv6", attribute_unchanged=dict(next_hop=True), ) ], ), ], ), ) ) self.execute_module(changed=False, commands=[]) def test_vyos_bgp_address_family_replaced(self): set_module_args( dict( state="replaced", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", summary_only=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ospfv3", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.10.21.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="import", route_map="map01")], ), dict( afi="ipv6", distribute_list=[dict(action="export", acl=10)], route_server_client=True, ), ], ), dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="export", route_map="map01")], ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv4", filter_list=[ dict(action="export", path_list="list01"), ], capability=dict(orf="send"), ) ], ), ], ), ) ) commands = [ "delete protocols bgp 65536 neighbor 203.0.113.5 address-family ipv6-unicast attribute-unchanged", "delete protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast soft-reconfiguration", "delete protocols bgp 65536 address-family ipv6-unicast redistribute ripng", "delete protocols bgp 65536 address-family ipv4-unicast network 192.2.13.0/24", "set protocols bgp 65536 address-family ipv4-unicast aggregate-address 192.0.2.0/24 summary-only", "set protocols bgp 65536 address-family ipv6-unicast redistribute ospfv3 metric 20", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv4-unicast route-map import map01", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast distribute-list export 10", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast route-server-client", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv4-unicast filter-list export list01", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv4-unicast capability prefix-list send", ] self.execute_module(changed=True, commands=commands) def test_vyos_bgp_address_family_overridden_idempotent(self): set_module_args( dict( state="overridden", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", as_set=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), dict(prefix="192.2.13.0/24", backdoor=True), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ripng", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="export", route_map="map01")], soft_reconfiguration=True, ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv6", attribute_unchanged=dict(next_hop=True), ) ], ), ], ), ) ) self.execute_module(changed=False, commands=[]) def test_vyos_bgp_address_family_overridden(self): set_module_args( dict( state="overridden", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ospfv3", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.10.21.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="import", route_map="map01")], ), dict( afi="ipv6", distribute_list=[dict(action="export", acl=10)], route_server_client=True, ), ], ), ], ), ) ) commands = [ "delete protocols bgp 65536 neighbor 203.0.113.5 address-family", "delete protocols bgp 65536 neighbor 192.0.2.25 address-family", "delete protocols bgp 65536 address-family ipv6-unicast redistribute ripng", "delete protocols bgp 65536 address-family ipv4 aggregate-address", "delete protocols bgp 65536 address-family ipv4-unicast network 192.2.13.0/24", "set protocols bgp 65536 address-family ipv6-unicast redistribute ospfv3 metric 20", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv4-unicast route-map import map01", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast distribute-list export 10", "set protocols bgp 65536 neighbor 192.10.21.25 address-family ipv6-unicast route-server-client", ] self.execute_module(changed=True, commands=commands) def test_vyos_bgp_address_family_deleted(self): set_module_args( dict( state="deleted", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", ), ], neighbors=[ dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", ), ], ), dict( neighbor_address="203.0.113.5", ), ], ), ) ) commands = [ "delete protocols bgp 65536 address-family ipv4-unicast", "delete protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast", "delete protocols bgp 65536 neighbor 203.0.113.5 address-family", ] self.execute_module(changed=True, commands=commands) def test_vyos_bgp_address_family_incorrect_instance(self): set_module_args( dict( state="overridden", config=dict( as_number=100, address_family=[ dict( afi="ipv4", networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ospfv3", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.10.21.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="import", route_map="map01")], ), dict( afi="ipv6", distribute_list=[dict(action="export", acl=10)], route_server_client=True, ), ], ), ], ), ) ) result = self.execute_module(failed=True) self.assertIn("Only one bgp instance is allowed per device", result["msg"]) def test_vyos_bgp_address_family_rendered(self): set_module_args( dict( state="rendered", config=dict( as_number=65536, address_family=[ dict( afi="ipv4", aggregate_address=[dict(prefix="192.0.2.0/24", as_set=True)], networks=[ dict(prefix="192.1.13.0/24", route_map="map01"), dict(prefix="192.2.13.0/24", backdoor=True), ], ), dict( afi="ipv6", redistribute=[dict(protocol="ripng", metric=20)], ), ], neighbors=[ dict( neighbor_address="192.0.2.25", address_family=[ dict( afi="ipv4", route_map=[dict(action="export", route_map="map01")], soft_reconfiguration=True, ), ], ), dict( neighbor_address="203.0.113.5", address_family=[ dict( afi="ipv6", attribute_unchanged=dict(next_hop=True), ) ], ), ], ), ) ) rendered_cmds = [ "set protocols bgp 65536 address-family ipv4-unicast network 192.1.13.0/24 route-map map01", "set protocols bgp 65536 address-family ipv4-unicast network 192.2.13.0/24 backdoor", "set protocols bgp 65536 address-family ipv4-unicast aggregate-address 192.0.2.0/24 as-set", "set protocols bgp 65536 address-family ipv6-unicast redistribute ripng metric 20", "set protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast route-map export map01", "set protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast soft-reconfiguration inbound", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv6-unicast attribute-unchanged next-hop", ] result = self.execute_module(changed=False) self.assertEqual( sorted(result["rendered"]), sorted(rendered_cmds), result["rendered"], ) def test_vyos_bgp_address_family_parsed(self): commands = [ "set protocols bgp 65536 address-family ipv4-unicast network 192.1.13.0/24 route-map map01", "set protocols bgp 65536 address-family ipv4-unicast network 192.2.13.0/24 backdoor", "set protocols bgp 65536 address-family ipv4-unicast aggregate-address 192.0.2.0/24 as-set", "set protocols bgp 65536 address-family ipv6-unicast redistribute ripng metric 20", "set protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast route-map export map01", "set protocols bgp 65536 neighbor 192.0.2.25 address-family ipv4-unicast soft-reconfiguration inbound", "set protocols bgp 65536 neighbor 203.0.113.5 address-family ipv6-unicast attribute-unchanged next-hop", ] parsed_str = "\n".join(commands) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = { "as_number": 65536, "address_family": [ { "afi": "ipv4", "networks": [ {"prefix": "192.1.13.0/24", "route_map": "map01"}, {"prefix": "192.2.13.0/24", "backdoor": True}, ], "aggregate_address": [{"prefix": "192.0.2.0/24", "as_set": True}], }, { "afi": "ipv6", "redistribute": [{"protocol": "ripng", "metric": 20}], }, ], "neighbors": [ { "neighbor_address": "192.0.2.25", "address_family": [ {"afi": "ipv4", "soft_reconfiguration": True}, ], }, { "neighbor_address": "203.0.113.5", "address_family": [ { "afi": "ipv6", "attribute_unchanged": {"next_hop": True}, } ], }, ], } self.assertEqual(sorted(parsed_list), sorted(result["parsed"])) def test_vyos_bgp_address_family_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False) gather_list = { "as_number": 65536, "address_family": [ { "afi": "ipv4", "networks": [ {"prefix": "192.1.13.0/24", "route_map": "map01"}, {"prefix": "192.2.13.0/24", "backdoor": True}, ], "aggregate_address": [{"prefix": "192.0.2.0/24", "as_set": True}], }, { "afi": "ipv6", "redistribute": [{"protocol": "ripng", "metric": 20}], }, ], "neighbors": [ { "neighbor_address": "192.0.2.25", "address_family": [ {"afi": "ipv4", "soft_reconfiguration": True}, ], }, { "neighbor_address": "203.0.113.5", "address_family": [ { "afi": "ipv6", "attribute_unchanged": {"next_hop": True}, } ], }, ], } self.assertEqual(sorted(gather_list), sorted(result["gathered"])) diff --git a/tests/unit/modules/network/vyos/test_vyos_bgp_global.py b/tests/unit/modules/network/vyos/test_vyos_bgp_global.py index d0167de..830fd38 100644 --- a/tests/unit/modules/network/vyos/test_vyos_bgp_global.py +++ b/tests/unit/modules/network/vyos/test_vyos_bgp_global.py @@ -1,549 +1,550 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_bgp_global -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosBgpglobalModule(TestVyosModule): module = vyos_bgp_global def setUp(self): super(TestVyosBgpglobalModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_execute_show_command_config = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.config.bgp_global.bgp_global.Bgp_global._get_config" ) self.execute_show_command_config = self.mock_execute_show_command_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.bgp_global.bgp_global.Bgp_globalFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosBgpglobalModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() self.mock_execute_show_command_config.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_bgp_global_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file self.execute_show_command_config.side_effect = load_from_file def test_vyos_bgp_global_merged_idempotent(self): set_module_args( dict( config=dict( as_number="65536", neighbor=[ dict( address="10.0.0.4", disable_connected_check=True, timers=dict(holdtime=30, keepalive=10), capability=dict(orf="receive"), ), dict( address="192.168.0.2", attribute_unchanged=dict(as_path=True, med=True), ebgp_multihop=2, remote_as="65535", soft_reconfiguration=True, update_source="192.168.0.1", ), dict( address="2001:db8::2", ebgp_multihop=2, remote_as="65535", maximum_prefix=34, update_source="2001:db8::1", ), ], network=[ dict(address="172.16.42.32/27", backdoor=True), dict(address="172.16.42.251/32", route_map="map01"), ], bgp_params=dict( bestpath=dict(as_path="confed", compare_routerid=True), default=dict(no_ipv4_unicast=True), router_id="10.1.1.1", ), redistribute=[ dict(protocol="kernel", route_map="map01"), dict(protocol="static", metric=20), dict(protocol="static", route_map="map01"), ], ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_bgp_global_merged(self): set_module_args( dict( config=dict( as_number="65536", maximum_paths=[ dict(path="ebgp", count=20), dict(path="ibgp", count=45), ], neighbor=[ dict( address="2001:db8::2", ebgp_multihop=2, remote_as="65535", maximum_prefix=34, update_source="2001:db8::1", distribute_list=[ dict(action="export", acl=31), dict(action="import", acl=9), ], ) ], bgp_params=dict( confederation=[dict(peers=20), dict(identifier=66)], router_id="10.1.1.1", ), ), state="merged", ) ) commands = [ "set protocols bgp 65536 neighbor 2001:db8::2 distribute-list export 31", "set protocols bgp 65536 neighbor 2001:db8::2 distribute-list import 9", "set protocols bgp 65536 parameters confederation peers 20", "set protocols bgp 65536 parameters confederation identifier 66", "set protocols bgp 65536 maximum-paths ebgp 20", "set protocols bgp 65536 maximum-paths ibgp 45", ] self.execute_module(changed=True, commands=commands) def test_vyos_bgp_global_replaced_idempotent(self): set_module_args( dict( config=dict( as_number="65536", neighbor=[ dict( address="10.0.0.4", disable_connected_check=True, timers=dict(holdtime=30, keepalive=10), capability=dict(orf="receive"), ), dict( address="192.168.0.2", attribute_unchanged=dict(as_path=True, med=True), ebgp_multihop=2, remote_as="65535", soft_reconfiguration=True, update_source="192.168.0.1", ), dict( address="2001:db8::2", ebgp_multihop=2, remote_as="65535", maximum_prefix=34, update_source="2001:db8::1", ), ], network=[ dict(address="172.16.42.32/27", backdoor=True), dict(address="172.16.42.251/32", route_map="map01"), ], bgp_params=dict( bestpath=dict(as_path="confed", compare_routerid=True), default=dict(no_ipv4_unicast=True), router_id="10.1.1.1", ), redistribute=[ dict(protocol="kernel", route_map="map01"), dict(protocol="static", metric=20), dict(protocol="static", route_map="map01"), ], ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) # def test_vyos_bgp_global_replaced(self): set_module_args( dict( config=dict( as_number="65536", timers=dict(holdtime=30, keepalive=10), neighbor=[ dict( address="200.11.155.3", prefix_list=[ dict(action="export", prefix_list=10), ], allowas_in=10, ), dict( address="2001:db8::2", remote_as="65535", as_override=True, default_originate="map01", route_map=[ dict(action="export", route_map="map01"), ], ), ], bgp_params=dict( log_neighbor_changes=True, no_client_to_client_reflection=True, confederation=[dict(peers=20), dict(identifier=66)], router_id="10.1.1.1", ), ), state="replaced", ) ) commands = [ "delete protocols bgp 65536 parameters default", "delete protocols bgp 65536 parameters bestpath compare-routerid", "delete protocols bgp 65536 parameters bestpath as-path confed", "delete protocols bgp 65536 network", "delete protocols bgp 65536 redistribute", "delete protocols bgp 65536 neighbor 2001:db8::2 update-source 2001:db8::1", "delete protocols bgp 65536 neighbor 2001:db8::2 maximum-prefix 34", "delete protocols bgp 65536 neighbor 2001:db8::2 ebgp-multihop 2", "delete protocols bgp 65536 neighbor 192.168.0.2", "delete protocols bgp 65536 neighbor 10.0.0.4", "set protocols bgp 65536 neighbor 200.11.155.3 prefix-list export 10", "set protocols bgp 65536 neighbor 200.11.155.3 allowas-in number 10", "set protocols bgp 65536 neighbor 2001:db8::2 as-override", "set protocols bgp 65536 neighbor 2001:db8::2 route-map export map01", "set protocols bgp 65536 parameters log-neighbor-changes", "set protocols bgp 65536 parameters no-client-to-client-reflection", "set protocols bgp 65536 parameters confederation peers 20", "set protocols bgp 65536 parameters confederation identifier 66", "set protocols bgp 65536 timers holdtime 30", "set protocols bgp 65536 timers keepalive 10", ] self.execute_module(changed=True, commands=commands) # def test_vyos_bgp_global_purged(self): set_module_args(dict(config=dict(as_number="65536"), state="purged")) # commands = ["delete protocols bgp 65536"] self.execute_module(changed=True, commands=commands) # def test_vyos_bgp_global_incorrect_instance(self): set_module_args( dict( config=dict( as_number="100", timers=dict(holdtime=30, keepalive=10), neighbor=[ dict( address="200.11.155.3", prefix_list=[ dict(action="export", prefix_list=10), ], allowas_in=10, ), dict( address="2001:db8::2", remote_as="65535", as_override=True, default_originate="map01", route_map=[ dict(action="export", route_map="map01"), ], ), ], bgp_params=dict( log_neighbor_changes=True, no_client_to_client_reflection=True, confederation=[dict(peers=20), dict(identifier=66)], router_id="10.1.1.1", ), ), state="replaced", ) ) result = self.execute_module(failed=True) self.assertIn("Only one bgp instance is allowed per device", result["msg"]) def test_vyos_bgp_global_replaced_af(self): set_module_args( dict( config=dict( as_number="65536", timers=dict(holdtime=30, keepalive=10), neighbor=[ dict( address="200.11.155.3", prefix_list=[ dict(action="export", prefix_list=10), ], allowas_in=10, ), dict( address="2001:db8::2", remote_as="65535", as_override=True, default_originate="map01", route_map=[ dict(action="export", route_map="map01"), ], ), ], bgp_params=dict( log_neighbor_changes=True, no_client_to_client_reflection=True, confederation=[dict(peers=20), dict(identifier=66)], router_id="10.1.1.1", ), ), state="replaced", ) ) result = self.execute_module(failed=True, filename="vyos_bgp_global_af_config.cfg") self.assertIn( "Use the _bgp_address_family module to delete the address_family under neighbor 5001::64, before replacing/deleting the neighbor.", result["msg"], ) def test_vyos_bgp_global_rendered(self): set_module_args( dict( config=dict( as_number="65536", neighbor=[ dict( address="10.0.0.4", disable_connected_check=True, timers=dict(holdtime=30, keepalive=10), capability=dict(orf="receive"), ), dict( address="192.168.0.2", attribute_unchanged=dict(as_path=True, med=True), ebgp_multihop=2, remote_as="65535", soft_reconfiguration=True, update_source="192.168.0.1", ), dict( address="2001:db8::2", ebgp_multihop=2, remote_as="65535", maximum_prefix=34, update_source="2001:db8::1", ), ], network=[ dict(address="172.16.42.32/27", backdoor=True), dict(address="172.16.42.251/32", route_map="map01"), ], bgp_params=dict( bestpath=dict(as_path="confed", compare_routerid=True), default=dict(no_ipv4_unicast=True), router_id="10.1.1.1", ), redistribute=[ dict(protocol="kernel", route_map="map01"), dict(protocol="static", metric=20), dict(protocol="static", route_map="map01"), ], ), state="rendered", ) ) rendered_cmds = [ "set protocols bgp 65536 neighbor 10.0.0.4 disable-connected-check", "set protocols bgp 65536 neighbor 10.0.0.4 timers holdtime 30", "set protocols bgp 65536 neighbor 10.0.0.4 timers keepalive 10", "set protocols bgp 65536 neighbor 10.0.0.4 capability orf prefix-list receive", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged as-path", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged med", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged next-hop", "set protocols bgp 65536 neighbor 192.168.0.2 ebgp-multihop 2", "set protocols bgp 65536 neighbor 192.168.0.2 remote-as 65535", "set protocols bgp 65536 neighbor 192.168.0.2 soft-reconfiguration", "set protocols bgp 65536 neighbor 192.168.0.2 update-source 192.168.0.1", "set protocols bgp 65536 neighbor 2001:db8::2 ebgp-multihop 2", "set protocols bgp 65536 neighbor 2001:db8::2 remote-as 65535", "set protocols bgp 65536 neighbor 2001:db8::2 maximum-prefix 34", "set protocols bgp 65536 neighbor 2001:db8::2 update-source 2001:db8::1", "set protocols bgp 65536 redistribute kernel route-map map01", "set protocols bgp 65536 redistribute static route-map map01", "set protocols bgp 65536 network 172.16.42.32/27 backdoor", "set protocols bgp 65536 network 172.16.42.251/32 route-map map01", "set protocols bgp 65536 parameters bestpath as-path confed", "set protocols bgp 65536 parameters bestpath compare-routerid", "set protocols bgp 65536 parameters default no-ipv4-unicast", "set protocols bgp 65536 parameters router-id 10.1.1.1", ] result = self.execute_module(changed=False) self.assertEqual( sorted(result["rendered"]), sorted(rendered_cmds), result["rendered"], ) def test_vyos_bgp_global_parsed(self): commands = [ "set protocols bgp 65536 neighbor 10.0.0.4 disable-connected-check", "set protocols bgp 65536 neighbor 10.0.0.4 timers holdtime 30", "set protocols bgp 65536 neighbor 10.0.0.4 timers keepalive 10", "set protocols bgp 65536 neighbor 10.0.0.4 capability orf prefix-list receive", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged as-path", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged med", "set protocols bgp 65536 neighbor 192.168.0.2 attribute-unchanged next-hop", "set protocols bgp 65536 neighbor 192.168.0.2 ebgp-multihop 2", "set protocols bgp 65536 neighbor 192.168.0.2 remote-as 65535", "set protocols bgp 65536 neighbor 192.168.0.2 soft-reconfiguration", "set protocols bgp 65536 neighbor 192.168.0.2 update-source 192.168.0.1", "set protocols bgp 65536 neighbor 2001:db8::2 ebgp-multihop 2", "set protocols bgp 65536 neighbor 2001:db8::2 remote-as 65535", "set protocols bgp 65536 neighbor 2001:db8::2 maximum-prefix 34", "set protocols bgp 65536 neighbor 2001:db8::2 update-source 2001:db8::1", "set protocols bgp 65536 redistribute kernel route-map map01", "set protocols bgp 65536 redistribute static route-map map01", "set protocols bgp 65536 network 172.16.42.32/27 backdoor", "set protocols bgp 65536 network 172.16.42.251/32 route-map map01", "set protocols bgp 65536 parameters bestpath as-path confed", "set protocols bgp 65536 parameters bestpath compare-routerid", "set protocols bgp 65536 parameters default no-ipv4-unicast", "set protocols bgp 65536 parameters router-id 10.1.1.1", ] parsed_str = "\n".join(commands) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = { "as_number": 65536, "bgp_params": { "bestpath": {"as_path": "confed", "compare_routerid": True}, "default": {"no_ipv4_unicast": True}, "router_id": "10.1.1.1", }, "neighbor": [ { "address": "10.0.0.4", "capability": {"orf": "receive"}, "disable_connected_check": True, "timers": {"holdtime": 30, "keepalive": 10}, }, { "address": "192.168.0.2", "attribute_unchanged": { "as_path": True, "med": True, "next_hop": True, }, "ebgp_multihop": 2, "remote_as": 65535, "update_source": "192.168.0.1", }, { "address": "2001:db8::2", "ebgp_multihop": 2, "maximum_prefix": 34, "remote_as": 65535, "update_source": "2001:db8::1", }, ], "network": [ {"address": "172.16.42.32/27", "backdoor": True}, {"address": "172.16.42.251/32", "route_map": "map01"}, ], "redistribute": [ {"protocol": "kernel", "route_map": "map01"}, {"protocol": "static", "route_map": "map01"}, ], } self.assertEqual(sorted(parsed_list), sorted(result["parsed"])) def test_vyos_bgp_global_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False) gather_list = { "as_number": 65536, "bgp_params": { "bestpath": {"as_path": "confed", "compare_routerid": True}, "default": {"no_ipv4_unicast": True}, "router_id": "10.1.1.1", }, "neighbor": [ { "address": "10.0.0.4", "capability": {"orf": "receive"}, "disable_connected_check": True, "timers": {"holdtime": 30, "keepalive": 10}, }, { "address": "192.168.0.2", "attribute_unchanged": {"as_path": True, "med": True}, "ebgp_multihop": 2, "remote_as": 65535, "soft_reconfiguration": True, "update_source": "192.168.0.1", }, { "address": "2001:db8::2", "ebgp_multihop": 2, "maximum_prefix": 34, "remote_as": 65535, "update_source": "2001:db8::1", }, ], "network": [ {"address": "172.16.42.32/27", "backdoor": True}, {"address": "172.16.42.251/32", "route_map": "map01"}, ], "redistribute": [ {"protocol": "kernel", "route_map": "map01"}, {"metric": 20, "protocol": "static"}, {"protocol": "static", "route_map": "map01"}, ], } self.assertEqual(sorted(gather_list), sorted(result["gathered"])) diff --git a/tests/unit/modules/network/vyos/test_vyos_command.py b/tests/unit/modules/network/vyos/test_vyos_command.py index 5479bbb..40db158 100644 --- a/tests/unit/modules/network/vyos/test_vyos_command.py +++ b/tests/unit/modules/network/vyos/test_vyos_command.py @@ -1,118 +1,119 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_command -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosCommandModule(TestVyosModule): module = vyos_command def setUp(self): super(TestVyosCommandModule, self).setUp() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_command.run_commands" ) self.run_commands = self.mock_run_commands.start() def tearDown(self): super(TestVyosCommandModule, self).tearDown() self.mock_run_commands.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): module, commands = args output = list() for item in commands: try: command = item["command"] except ValueError: command = item filename = str(command).replace(" ", "_") output.append(load_fixture(filename)) return output self.run_commands.side_effect = load_from_file def test_vyos_command_simple(self): set_module_args(dict(commands=["show version"])) result = self.execute_module() self.assertEqual(len(result["stdout"]), 1) self.assertTrue(result["stdout"][0].startswith("Version: VyOS")) def test_vyos_command_multiple(self): set_module_args(dict(commands=["show version", "show version"])) result = self.execute_module() self.assertEqual(len(result["stdout"]), 2) self.assertTrue(result["stdout"][0].startswith("Version: VyOS")) def test_vyos_command_wait_for(self): wait_for = 'result[0] contains "VyOS maintainers"' set_module_args(dict(commands=["show version"], wait_for=wait_for)) self.execute_module() def test_vyos_command_wait_for_fails(self): wait_for = 'result[0] contains "test string"' set_module_args(dict(commands=["show version"], wait_for=wait_for)) self.execute_module(failed=True) self.assertEqual(self.run_commands.call_count, 10) def test_vyos_command_retries(self): wait_for = 'result[0] contains "test string"' set_module_args(dict(commands=["show version"], wait_for=wait_for, retries=2)) self.execute_module(failed=True) self.assertEqual(self.run_commands.call_count, 3) def test_vyos_command_no_retries(self): wait_for = 'result[0] contains "test string"' set_module_args(dict(commands=["show version"], wait_for=wait_for, retries=0)) self.execute_module(failed=True) self.assertEqual(self.run_commands.call_count, 1) def test_vyos_command_match_any(self): wait_for = [ 'result[0] contains "VyOS maintainers"', 'result[0] contains "test string"', ] set_module_args(dict(commands=["show version"], wait_for=wait_for, match="any")) self.execute_module() def test_vyos_command_match_all(self): wait_for = [ 'result[0] contains "VyOS maintainers"', 'result[0] contains "maintainers@vyos.net"', ] set_module_args(dict(commands=["show version"], wait_for=wait_for, match="all")) self.execute_module() def test_vyos_command_match_all_failure(self): wait_for = [ 'result[0] contains "VyOS maintainers"', 'result[0] contains "test string"', ] commands = ["show version", "show version"] set_module_args(dict(commands=commands, wait_for=wait_for, match="all")) self.execute_module(failed=True) diff --git a/tests/unit/modules/network/vyos/test_vyos_config.py b/tests/unit/modules/network/vyos/test_vyos_config.py index ca949c6..743acab 100644 --- a/tests/unit/modules/network/vyos/test_vyos_config.py +++ b/tests/unit/modules/network/vyos/test_vyos_config.py @@ -1,140 +1,141 @@ # # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import MagicMock, patch + from ansible_collections.vyos.vyos.plugins.cliconf.vyos import Cliconf from ansible_collections.vyos.vyos.plugins.modules import vyos_config -from ansible_collections.vyos.vyos.tests.unit.compat.mock import MagicMock, patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosConfigModule(TestVyosModule): module = vyos_config def setUp(self): super(TestVyosConfigModule, self).setUp() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.run_commands" ) self.run_commands = self.mock_run_commands.start() self.mock_get_connection = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_config.get_connection" ) self.get_connection = self.mock_get_connection.start() self.cliconf_obj = Cliconf(MagicMock()) self.running_config = load_fixture("vyos_config_config.cfg") self.conn = self.get_connection() self.conn.edit_config = MagicMock() self.running_config = load_fixture("vyos_config_config.cfg") def tearDown(self): super(TestVyosConfigModule, self).tearDown() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_run_commands.stop() self.mock_get_connection.stop() def load_fixtures(self, commands=None, filename=None): config_file = "vyos_config_config.cfg" self.get_config.return_value = load_fixture(config_file) self.load_config.return_value = None def test_vyos_config_unchanged(self): src = load_fixture("vyos_config_config.cfg") self.conn.get_diff = MagicMock(return_value=self.cliconf_obj.get_diff(src, src)) set_module_args(dict(src=src)) self.execute_module() def test_vyos_config_src(self): src = load_fixture("vyos_config_src.cfg") set_module_args(dict(src=src)) candidate = "\n".join(self.module.format_commands(src.splitlines())) commands = [ "set system host-name foo", "delete interfaces ethernet eth0 address", ] self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config) ) self.execute_module(changed=True, commands=commands) def test_vyos_config_src_brackets(self): src = load_fixture("vyos_config_src_brackets.cfg") set_module_args(dict(src=src)) commands = [ "set interfaces ethernet eth0 address 10.10.10.10/24", "set policy route testroute rule 1 set table 10", "set system host-name foo", ] self.conn.get_diff = MagicMock(side_effect=self.cliconf_obj.get_diff) self.execute_module(changed=True, commands=commands) def test_vyos_config_backup(self): set_module_args(dict(backup=True)) result = self.execute_module() self.assertIn("__backup__", result) def test_vyos_config_lines(self): commands = ["set system host-name foo"] set_module_args(dict(lines=commands)) candidate = "\n".join(commands) self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, self.running_config) ) self.execute_module(changed=True, commands=commands) def test_vyos_config_config(self): config = "set system host-name localhost" new_config = ["set system host-name router"] set_module_args(dict(lines=new_config, config=config)) candidate = "\n".join(new_config) self.conn.get_diff = MagicMock(return_value=self.cliconf_obj.get_diff(candidate, config)) self.execute_module(changed=True, commands=new_config) def test_vyos_config_match_none(self): lines = [ "set system interfaces ethernet eth0 address 1.2.3.4/24", "set system interfaces ethernet eth0 description test string", ] set_module_args(dict(lines=lines, match="none")) candidate = "\n".join(lines) self.conn.get_diff = MagicMock( return_value=self.cliconf_obj.get_diff(candidate, None, diff_match="none") ) self.execute_module(changed=True, commands=lines, sort=False) diff --git a/tests/unit/modules/network/vyos/test_vyos_facts.py b/tests/unit/modules/network/vyos/test_vyos_facts.py index 727f69b..691d7cf 100644 --- a/tests/unit/modules/network/vyos/test_vyos_facts.py +++ b/tests/unit/modules/network/vyos/test_vyos_facts.py @@ -1,107 +1,107 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type import json +from unittest.mock import patch from ansible_collections.vyos.vyos.plugins.modules import vyos_facts -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosFactsModule(TestVyosModule): module = vyos_facts def setUp(self): super(TestVyosFactsModule, self).setUp() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.legacy.base.run_commands" ) self.run_commands = self.mock_run_commands.start() self.mock_get_resource_connection = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection = self.mock_get_resource_connection.start() self.mock_get_capabilities = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.legacy.base.get_capabilities" ) self.get_capabilities = self.mock_get_capabilities.start() self.get_capabilities.return_value = { "device_info": { "network_os": "vyos", "network_os_hostname": "vyos01", "network_os_model": "VMware", "network_os_version": "VyOS 1.1.7", }, "network_api": "cliconf", } def tearDown(self): super(TestVyosFactsModule, self).tearDown() self.mock_run_commands.stop() self.mock_get_capabilities.stop() self.mock_get_resource_connection.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): module, commands = args output = list() for item in commands: try: obj = json.loads(item) command = obj["command"] except ValueError: command = item filename = str(command).replace(" ", "_") output.append(load_fixture(filename)) return output self.run_commands.side_effect = load_from_file def test_vyos_facts_default(self): set_module_args(dict(gather_subset="default")) result = self.execute_module() facts = result.get("ansible_facts") self.assertEqual(len(facts), 10) self.assertEqual(facts["ansible_net_hostname"].strip(), "vyos01") self.assertEqual(facts["ansible_net_version"], "VyOS 1.1.7") def test_vyos_facts_not_all(self): set_module_args(dict(gather_subset="!all")) result = self.execute_module() facts = result.get("ansible_facts") self.assertEqual(len(facts), 10) self.assertEqual(facts["ansible_net_hostname"].strip(), "vyos01") self.assertEqual(facts["ansible_net_version"], "VyOS 1.1.7") def test_vyos_facts_exclude_most(self): set_module_args(dict(gather_subset=["!neighbors", "!config"])) result = self.execute_module() facts = result.get("ansible_facts") self.assertEqual(len(facts), 10) self.assertEqual(facts["ansible_net_hostname"].strip(), "vyos01") self.assertEqual(facts["ansible_net_version"], "VyOS 1.1.7") def test_vyos_facts_invalid_subset(self): set_module_args(dict(gather_subset="cereal")) self.execute_module(failed=True) diff --git a/tests/unit/modules/network/vyos/test_vyos_firewall_global.py b/tests/unit/modules/network/vyos/test_vyos_firewall_global.py index 8164768..71b3a18 100644 --- a/tests/unit/modules/network/vyos/test_vyos_firewall_global.py +++ b/tests/unit/modules/network/vyos/test_vyos_firewall_global.py @@ -1,360 +1,361 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_firewall_global -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosFirewallRulesModule(TestVyosModule): module = vyos_firewall_global def setUp(self): super(TestVyosFirewallRulesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.firewall_global.firewall_global.Firewall_globalFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosFirewallRulesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_firewall_global_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_firewall_global_set_01_merged(self): set_module_args( dict( config=dict( validation="strict", config_trap=True, log_martians=True, syn_cookies=True, twa_hazards_protection=True, ping=dict(all=True, broadcast=True), state_policy=[ dict( connection_type="established", action="accept", log=True, ), dict(connection_type="invalid", action="reject"), ], route_redirects=[ dict( afi="ipv4", ip_src_route=True, icmp_redirects=dict(send=True, receive=False), ) ], group=dict( address_group=[ dict( afi="ipv4", name="MGMT-HOSTS", description="This group has the Management hosts address lists", members=[ dict(address="192.0.1.1"), dict(address="192.0.1.3"), dict(address="192.0.1.5"), ], ), dict( afi="ipv6", name="GOOGLE-DNS-v6", members=[ dict(address="2001:4860:4860::8888"), dict(address="2001:4860:4860::8844"), ], ), ], network_group=[ dict( afi="ipv4", name="MGMT", description="This group has the Management network addresses", members=[dict(address="192.0.1.0/24")], ), dict( afi="ipv6", name="DOCUMENTATION-v6", description="IPv6 Addresses reserved for documentation per RFC 3849", members=[ dict(address="2001:0DB8::/32"), dict(address="3FFF:FFFF::/32"), ], ), ], port_group=[ dict( name="TELNET", description="This group has the telnet ports", members=[dict(port="23")], ) ], ), ), state="merged", ) ) commands = [ "set firewall group address-group MGMT-HOSTS address 192.0.1.1", "set firewall group address-group MGMT-HOSTS address 192.0.1.3", "set firewall group address-group MGMT-HOSTS address 192.0.1.5", "set firewall group address-group MGMT-HOSTS description 'This group has the Management hosts address lists'", "set firewall group address-group MGMT-HOSTS", "set firewall group ipv6-address-group GOOGLE-DNS-v6 address 2001:4860:4860::8888", "set firewall group ipv6-address-group GOOGLE-DNS-v6 address 2001:4860:4860::8844", "set firewall group ipv6-address-group GOOGLE-DNS-v6", "set firewall group network-group MGMT network 192.0.1.0/24", "set firewall group network-group MGMT description 'This group has the Management network addresses'", "set firewall group network-group MGMT", "set firewall group ipv6-network-group DOCUMENTATION-v6 network 2001:0DB8::/32", "set firewall group ipv6-network-group DOCUMENTATION-v6 network 3FFF:FFFF::/32", "set firewall group ipv6-network-group DOCUMENTATION-v6 description 'IPv6 Addresses reserved for documentation per RFC 3849'", "set firewall group ipv6-network-group DOCUMENTATION-v6", "set firewall group port-group TELNET port 23", "set firewall group port-group TELNET description 'This group has the telnet ports'", "set firewall group port-group TELNET", "set firewall ip-src-route 'enable'", "set firewall receive-redirects 'disable'", "set firewall send-redirects 'enable'", "set firewall config-trap 'enable'", "set firewall state-policy established action 'accept'", "set firewall state-policy established log 'enable'", "set firewall state-policy invalid action 'reject'", "set firewall broadcast-ping 'enable'", "set firewall all-ping 'enable'", "set firewall log-martians 'enable'", "set firewall twa-hazards-protection 'enable'", "set firewall syn-cookies 'enable'", "set firewall source-validation 'strict'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_global_set_01_merged_idem(self): set_module_args( dict( config=dict( group=dict( address_group=[ dict( afi="ipv4", name="RND-HOSTS", description="This group has the Management hosts address lists", members=[ dict(address="192.0.2.1"), dict(address="192.0.2.3"), dict(address="192.0.2.5"), ], ), dict( afi="ipv6", name="LOCAL-v6", description="This group has the hosts address lists of this machine", members=[ dict(address="::1"), dict(address="fdec:2503:89d6:59b3::1"), ], ), ], network_group=[ dict( afi="ipv4", name="RND", description="This group has the Management network addresses", members=[dict(address="192.0.2.0/24")], ), dict( afi="ipv6", name="UNIQUE-LOCAL-v6", description="This group encompasses the ULA address space in IPv6", members=[dict(address="fc00::/7")], ), ], port_group=[ dict( name="SSH", description="This group has the ssh ports", members=[dict(port="22")], ) ], ) ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_global_set_01_replaced(self): set_module_args( dict( config=dict( group=dict( address_group=[ dict( afi="ipv4", name="RND-HOSTS", description="This group has the Management hosts address lists", members=[ dict(address="192.0.2.1"), dict(address="192.0.2.7"), dict(address="192.0.2.9"), ], ), dict( afi="ipv6", name="LOCAL-v6", description="This group has the hosts address lists of this machine", members=[ dict(address="::1"), dict(address="fdec:2503:89d6:59b3::2"), ], ), ], network_group=[ dict( afi="ipv4", name="RND", description="This group has the Management network addresses", members=[dict(address="192.0.2.0/24")], ), dict( afi="ipv6", name="UNIQUE-LOCAL-v6", description="This group encompasses the ULA address space in IPv6", members=[dict(address="fc00::/7")], ), ], port_group=[ dict( name="SSH", description="This group has the ssh ports", members=[dict(port="2222")], ) ], ) ), state="replaced", ) ) commands = [ "delete firewall group address-group RND-HOSTS address 192.0.2.3", "delete firewall group address-group RND-HOSTS address 192.0.2.5", "set firewall group address-group RND-HOSTS address 192.0.2.7", "set firewall group address-group RND-HOSTS address 192.0.2.9", "delete firewall group ipv6-address-group LOCAL-v6 address fdec:2503:89d6:59b3::1", "set firewall group ipv6-address-group LOCAL-v6 address fdec:2503:89d6:59b3::2", "delete firewall group port-group SSH port 22", "set firewall group port-group SSH port 2222", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_global_set_01_replaced_idem(self): set_module_args( dict( config=dict( group=dict( address_group=[ dict( afi="ipv4", name="RND-HOSTS", description="This group has the Management hosts address lists", members=[ dict(address="192.0.2.1"), dict(address="192.0.2.3"), dict(address="192.0.2.5"), ], ), dict( afi="ipv6", name="LOCAL-v6", description="This group has the hosts address lists of this machine", members=[ dict(address="::1"), dict(address="fdec:2503:89d6:59b3::1"), ], ), ], network_group=[ dict( afi="ipv4", name="RND", description="This group has the Management network addresses", members=[dict(address="192.0.2.0/24")], ), dict( afi="ipv6", name="UNIQUE-LOCAL-v6", description="This group encompasses the ULA address space in IPv6", members=[dict(address="fc00::/7")], ), ], port_group=[ dict( name="SSH", description="This group has the ssh ports", members=[dict(port="22")], ) ], ) ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_global_set_01_deleted(self): set_module_args(dict(config=dict(), state="deleted")) commands = ["delete firewall "] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_firewall_interfaces.py b/tests/unit/modules/network/vyos/test_vyos_firewall_interfaces.py index acb9894..10d93ca 100644 --- a/tests/unit/modules/network/vyos/test_vyos_firewall_interfaces.py +++ b/tests/unit/modules/network/vyos/test_vyos_firewall_interfaces.py @@ -1,388 +1,389 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_firewall_interfaces -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosFirewallInterfacesModule(TestVyosModule): module = vyos_firewall_interfaces def setUp(self): super(TestVyosFirewallInterfacesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos." "facts.firewall_interfaces.firewall_interfaces.Firewall_interfacesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosFirewallInterfacesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_firewall_interfaces_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_firewall_rule_set_01_merged(self): set_module_args( dict( config=[ dict( name="eth1", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth3", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), ], state="merged", ) ) commands = [ "set interfaces ethernet eth1 firewall in name 'INBOUND'", "set interfaces ethernet eth1 firewall out name 'OUTBOUND'", "set interfaces ethernet eth1 firewall local name 'LOCAL'", "set interfaces ethernet eth1 firewall local ipv6-name 'V6-LOCAL'", "set interfaces ethernet eth3 firewall in name 'INBOUND'", "set interfaces ethernet eth3 firewall out name 'OUTBOUND'", "set interfaces ethernet eth3 firewall local name 'LOCAL'", "set interfaces ethernet eth3 firewall local ipv6-name 'V6-LOCAL'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_02_merged_idem(self): set_module_args( dict( config=[ dict( name="eth0", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth2", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_rule_set_01_deleted_per_afi(self): set_module_args( dict( config=[ dict( name="eth0", access_rules=[dict(afi="ipv4"), dict(afi="ipv6")], ) ], state="deleted", ) ) commands = [ "delete interfaces ethernet eth0 firewall in name", "delete interfaces ethernet eth0 firewall local name", "delete interfaces ethernet eth0 firewall out name", "delete interfaces ethernet eth0 firewall local ipv6-name", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_03_deleted_per_interface(self): set_module_args(dict(config=[dict(name="eth0"), dict(name="eth2")], state="deleted")) commands = [ "delete interfaces ethernet eth0 firewall", "delete interfaces ethernet eth2 firewall", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_03_deleted_all(self): set_module_args(dict(config=[], state="deleted")) commands = [ "delete interfaces ethernet eth0 firewall", "delete interfaces ethernet eth2 firewall", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_03_deleted(self): set_module_args(dict(config=[dict(name="eth0"), dict(name="eth2")], state="deleted")) commands = [ "delete interfaces ethernet eth0 firewall", "delete interfaces ethernet eth2 firewall", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_04_deleted_interface_idem(self): set_module_args(dict(config=[dict(name="eth1"), dict(name="eth3")], state="deleted")) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_rule_set_02_replaced_idem(self): set_module_args( dict( config=[ dict( name="eth0", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth2", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_rule_set_01_replaced(self): set_module_args( dict( config=[ dict( name="eth0", access_rules=[ dict( afi="ipv4", rules=[dict(name="INBOUND", direction="in")], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth2", access_rules=[ dict( afi="ipv4", rules=[dict(name="LOCAL", direction="local")], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth3", access_rules=[ dict( afi="ipv4", rules=[dict(name="LOCAL", direction="local")], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), ], state="replaced", ) ) commands = [ "delete interfaces ethernet eth0 firewall out name", "delete interfaces ethernet eth0 firewall local name", "delete interfaces ethernet eth2 firewall in name", "delete interfaces ethernet eth2 firewall out name", "set interfaces ethernet eth3 firewall local name 'LOCAL'", "set interfaces ethernet eth3 firewall local ipv6-name 'V6-LOCAL'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_01_overridden(self): set_module_args( dict( config=[ dict( name="eth1", access_rules=[ dict( afi="ipv4", rules=[dict(name="INBOUND", direction="in")], ) ], ) ], state="overridden", ) ) commands = [ "delete interfaces ethernet eth0 firewall", "delete interfaces ethernet eth2 firewall", "set interfaces ethernet eth1 firewall in name 'INBOUND'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_02_overridden_idem(self): set_module_args( dict( config=[ dict( name="eth0", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), dict( name="eth2", access_rules=[ dict( afi="ipv4", rules=[ dict(name="INBOUND", direction="in"), dict(name="OUTBOUND", direction="out"), dict(name="LOCAL", direction="local"), ], ), dict( afi="ipv6", rules=[dict(name="V6-LOCAL", direction="local")], ), ], ), ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) diff --git a/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py b/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py index 7952b47..18da678 100644 --- a/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py +++ b/tests/unit/modules/network/vyos/test_vyos_firewall_rules.py @@ -1,1182 +1,1183 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_firewall_rules -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosFirewallRulesModule(TestVyosModule): module = vyos_firewall_rules def setUp(self): super(TestVyosFirewallRulesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.static_routes.static_routes.Static_routesFacts.get_device_data" ) self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.firewall_rules.firewall_rules.Firewall_rulesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() self.mock_get_os_version = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.config.firewall_rules.firewall_rules.Firewall_rules._get_os_version" ) self.get_os_version = self.mock_get_os_version.start() self.get_os_version.return_value = "Vyos 1.2" def tearDown(self): super(TestVyosFirewallRulesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() self.mock_get_os_version.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_firewall_rules_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_firewall_rule_set_01_merged(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="V6-INBOUND", description="This is IPv6 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V6-OUTBOUND", description="This is IPv6 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), dict( afi="ipv4", rule_sets=[ dict( name="V4-INBOUND", description="This is IPv4 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V4-OUTBOUND", description="This is IPv4 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), ], state="merged", ) ) commands = [ "set firewall ipv6-name V6-INBOUND default-action 'reject'", "set firewall ipv6-name V6-INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name V6-INBOUND enable-default-log", "set firewall ipv6-name V6-OUTBOUND default-action 'accept'", "set firewall ipv6-name V6-OUTBOUND description 'This is IPv6 OUTBOUND rule set'", "set firewall name V4-INBOUND default-action 'reject'", "set firewall name V4-INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name V4-INBOUND enable-default-log", "set firewall name V4-OUTBOUND default-action 'accept'", "set firewall name V4-OUTBOUND description 'This is IPv4 OUTBOUND rule set'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_rule_set_02_merged(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="V6-INBOUND", description="This is IPv6 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V6-OUTBOUND", description="This is IPv6 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), dict( afi="ipv4", rule_sets=[ dict( name="V4-INBOUND", description="This is IPv4 INBOUND rule set", default_action="reject", enable_default_log=True, rules=[], ), dict( name="V4-OUTBOUND", description="This is IPv4 OUTBOUND rule set", default_action="accept", enable_default_log=False, rules=[], ), ], ), ], state="merged", ) ) commands = [ "set firewall ipv6-name V6-INBOUND default-action 'reject'", "set firewall ipv6-name V6-INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name V6-INBOUND enable-default-log", "set firewall ipv6-name V6-OUTBOUND default-action 'accept'", "set firewall ipv6-name V6-OUTBOUND description 'This is IPv6 OUTBOUND rule set'", "set firewall name V4-INBOUND default-action 'reject'", "set firewall name V4-INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name V4-INBOUND enable-default-log", "set firewall name V4-OUTBOUND default-action 'accept'", "set firewall name V4-OUTBOUND description 'This is IPv4 OUTBOUND rule set'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", description="This is IPv4 INBOUND rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", log="disable", protocol="icmp", fragment="match-frag", disable=True, ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND default-action 'accept'", "set firewall name INBOUND description 'This is IPv4 INBOUND rule set'", "set firewall name INBOUND enable-default-log", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", "set firewall name INBOUND rule 101 fragment 'match-frag'", "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 disable", "set firewall name INBOUND rule 101 action 'accept'", "set firewall name INBOUND rule 101 ipsec 'match-ipsec'", "set firewall name INBOUND rule 101 log 'disable'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="tcp", source=dict( address="192.0.2.0", mac_address="38:00:25:19:76:0c", port=2127, ), destination=dict(address="192.0.1.0", port=2124), limit=dict( burst=10, rate=dict(number=20, unit="second"), ), recent=dict(count=10, time=20), state=dict( established=True, related=True, invalid=True, new=True, ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 protocol 'tcp'", "set firewall name INBOUND rule 101 destination address 192.0.1.0", "set firewall name INBOUND rule 101 destination port 2124", "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 source address 192.0.2.0", "set firewall name INBOUND rule 101 source mac-address 38:00:25:19:76:0c", "set firewall name INBOUND rule 101 source port 2127", "set firewall name INBOUND rule 101 state new enable", "set firewall name INBOUND rule 101 state invalid enable", "set firewall name INBOUND rule 101 state related enable", "set firewall name INBOUND rule 101 state established enable", "set firewall name INBOUND rule 101 limit burst 10", "set firewall name INBOUND rule 101 limit rate 20/second", "set firewall name INBOUND rule 101 recent count 10", "set firewall name INBOUND rule 101 recent time 20", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_03(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", destination=dict( group=dict( address_group="OUT-ADDR-GROUP", network_group="OUT-NET-GROUP", port_group="OUT-PORT-GROUP", ) ), source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 source group address-group IN-ADDR-GROUP", "set firewall name INBOUND rule 101 source group network-group IN-NET-GROUP", "set firewall name INBOUND rule 101 source group port-group IN-PORT-GROUP", "set firewall name INBOUND rule 101 destination group address-group OUT-ADDR-GROUP", "set firewall name INBOUND rule 101 destination group network-group OUT-NET-GROUP", "set firewall name INBOUND rule 101 destination group port-group OUT-PORT-GROUP", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_04(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", time=dict( monthdays="2", startdate="2020-01-24", starttime="13:20:00", stopdate="2020-01-28", stoptime="13:30:00", weekdays="!Sat,Sun", utc=True, ), tcp=dict(flags="ALL"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101", "set firewall name INBOUND rule 101 tcp flags ALL", "set firewall name INBOUND rule 101 time utc", "set firewall name INBOUND rule 101 time monthdays 2", "set firewall name INBOUND rule 101 time startdate 2020-01-24", "set firewall name INBOUND rule 101 time stopdate 2020-01-28", "set firewall name INBOUND rule 101 time weekdays !Sat,Sun", "set firewall name INBOUND rule 101 time stoptime 13:30:00", "set firewall name INBOUND rule 101 time starttime 13:20:00", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_01(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", description="This is IPv6 INBOUND rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", disabled=True, icmp=dict(type_name="echo-request"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND default-action 'accept'", "set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name INBOUND enable-default-log", "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", "set firewall ipv6-name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 disable", "set firewall ipv6-name INBOUND rule 101 action 'accept'", "set firewall ipv6-name INBOUND rule 101 ipsec 'match-ipsec'", "set firewall ipv6-name INBOUND rule 101 icmpv6 type echo-request", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_02(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="tcp", source=dict( address="2001:db8::12", mac_address="38:00:25:19:76:0c", port=2127, ), destination=dict(address="2001:db8::11", port=2124), limit=dict( burst=10, rate=dict(number=20, unit="second"), ), recent=dict(count=10, time=20), state=dict( established=True, related=True, invalid=True, new=True, ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 protocol 'tcp'", "set firewall ipv6-name INBOUND rule 101 destination address 2001:db8::11", "set firewall ipv6-name INBOUND rule 101 destination port 2124", "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 source address 2001:db8::12", "set firewall ipv6-name INBOUND rule 101 source mac-address 38:00:25:19:76:0c", "set firewall ipv6-name INBOUND rule 101 source port 2127", "set firewall ipv6-name INBOUND rule 101 state new enable", "set firewall ipv6-name INBOUND rule 101 state invalid enable", "set firewall ipv6-name INBOUND rule 101 state related enable", "set firewall ipv6-name INBOUND rule 101 state established enable", "set firewall ipv6-name INBOUND rule 101 limit burst 10", "set firewall ipv6-name INBOUND rule 101 recent count 10", "set firewall ipv6-name INBOUND rule 101 recent time 20", "set firewall ipv6-name INBOUND rule 101 limit rate 20/second", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_03(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", destination=dict( group=dict( address_group="OUT-ADDR-GROUP", network_group="OUT-NET-GROUP", port_group="OUT-PORT-GROUP", ) ), source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 source group address-group IN-ADDR-GROUP", "set firewall ipv6-name INBOUND rule 101 source group network-group IN-NET-GROUP", "set firewall ipv6-name INBOUND rule 101 source group port-group IN-PORT-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group address-group OUT-ADDR-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group network-group OUT-NET-GROUP", "set firewall ipv6-name INBOUND rule 101 destination group port-group OUT-PORT-GROUP", "set firewall ipv6-name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_04(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", time=dict( monthdays="2", startdate="2020-01-24", starttime="13:20:00", stopdate="2020-01-28", stoptime="13:30:00", weekdays="!Sat,Sun", utc=True, ), tcp=dict(flags="ALL"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 tcp flags ALL", "set firewall ipv6-name INBOUND rule 101 time utc", "set firewall ipv6-name INBOUND rule 101 time monthdays 2", "set firewall ipv6-name INBOUND rule 101 time startdate 2020-01-24", "set firewall ipv6-name INBOUND rule 101 time stopdate 2020-01-28", "set firewall ipv6-name INBOUND rule 101 time weekdays !Sat,Sun", "set firewall ipv6-name INBOUND rule 101 time stoptime 13:30:00", "set firewall ipv6-name INBOUND rule 101 time starttime 13:20:00", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v6_rule_sets_rule_merged_icmp_01(self): set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict(type_name="port-unreachable"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND rule 101 icmpv6 type port-unreachable", "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", "set firewall ipv6-name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_icmp_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict(type=1, code=1), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 icmp type 1", "set firewall name INBOUND rule 101 icmp code 1", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_rule_merged_icmp_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="INBOUND", rules=[ dict( number="101", protocol="icmp", icmp=dict(type_name="echo-request"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall name INBOUND rule 101 icmp type-name echo-request", "set firewall name INBOUND rule 101 protocol 'icmp'", "set firewall name INBOUND rule 101", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4_rule_sets_del_01(self): set_module_args( dict( config=[dict(afi="ipv4", rule_sets=[dict(name="V4-INGRESS")])], state="deleted", ) ) commands = ["delete firewall name V4-INGRESS"] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_02(self): set_module_args( dict( config=[ dict(afi="ipv4", rule_sets=[dict(name="V4-INGRESS")]), dict(afi="ipv6", rule_sets=[dict(name="V6-INGRESS")]), ], state="deleted", ) ) commands = [ "delete firewall name V4-INGRESS", "delete firewall ipv6-name V6-INGRESS", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_03(self): set_module_args(dict(config=[], state="deleted")) commands = ["delete firewall name", "delete firewall ipv6-name"] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_del_04(self): set_module_args( dict( config=[ dict(afi="ipv4", rule_sets=[dict(name="V4-ING")]), dict(afi="ipv6", rule_sets=[dict(name="V6-ING")]), ], state="deleted", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_rep_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="reject", description="Rule 101 is configured by Ansible RM", ipsec="match-ipsec", protocol="tcp", fragment="match-frag", disabled=False, ), dict( number="102", action="accept", description="Rule 102 is configured by Ansible RM", protocol="icmp", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", description="This rule-set is configured by Ansible RM", ), dict( name="EGRESS", default_action="reject", description="This rule-set is configured by Ansible RM", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) commands = [ "delete firewall name V4-INGRESS rule 101 disable", "set firewall name V4-INGRESS description 'This is IPv4 INGRESS rule set'", "set firewall name V4-INGRESS rule 101 protocol 'tcp'", "set firewall name V4-INGRESS rule 101 description 'Rule 101 is configured by Ansible RM'", "set firewall name V4-INGRESS rule 101 action 'reject'", "set firewall name V4-INGRESS rule 102 disable", "set firewall name V4-INGRESS rule 102 action 'accept'", "set firewall name V4-INGRESS rule 102 protocol 'icmp'", "set firewall name V4-INGRESS rule 102 description 'Rule 102 is configured by Ansible RM'", "set firewall name V4-INGRESS rule 102", "set firewall ipv6-name V6-INGRESS description 'This rule-set is configured by Ansible RM'", "set firewall ipv6-name EGRESS description 'This rule-set is configured by Ansible RM'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_rep_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=False, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) commands = [ "delete firewall name V4-INGRESS enable-default-log", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_rep_idem_02(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ), ], ), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_mer_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v4v6_rule_sets_rule_ovr_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-IN", description="This is IPv4 INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="1", action="reject", description="Rule 1 is configured by Ansible RM", ipsec="match-ipsec", log="enable", protocol="tcp", fragment="match-frag", disabled=False, source=dict( group=dict( address_group="IN-ADDR-GROUP", network_group="IN-NET-GROUP", port_group="IN-PORT-GROUP", ) ), ), dict( number="2", action="accept", description="Rule 102 is configured by Ansible RM", protocol="icmp", disabled=True, ), ], ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-IN", default_action="accept", description="This rule-set is configured by Ansible RM", ), dict( name="V6-EG", default_action="reject", description="This rule-set is configured by Ansible RM", ), ], ), ], state="overridden", ) ) commands = [ "delete firewall ipv6-name V6-INGRESS", "delete firewall ipv6-name EGRESS", "delete firewall name V4-INGRESS", "delete firewall name EGRESS", "set firewall name V4-IN default-action 'accept'", "set firewall name V4-IN description 'This is IPv4 INGRESS rule set'", "set firewall name V4-IN enable-default-log", "set firewall name V4-IN rule 1 protocol 'tcp'", "set firewall name V4-IN rule 1 log 'enable'", "set firewall name V4-IN rule 1 description 'Rule 1 is configured by Ansible RM'", "set firewall name V4-IN rule 1 fragment 'match-frag'", "set firewall name V4-IN rule 1 source group address-group IN-ADDR-GROUP", "set firewall name V4-IN rule 1 source group network-group IN-NET-GROUP", "set firewall name V4-IN rule 1 source group port-group IN-PORT-GROUP", "set firewall name V4-IN rule 1", "set firewall name V4-IN rule 1 action 'reject'", "set firewall name V4-IN rule 1 ipsec 'match-ipsec'", "set firewall name V4-IN rule 2 disable", "set firewall name V4-IN rule 2 action 'accept'", "set firewall name V4-IN rule 2 protocol 'icmp'", "set firewall name V4-IN rule 2 description 'Rule 102 is configured by Ansible RM'", "set firewall name V4-IN rule 2", "set firewall ipv6-name V6-IN default-action 'accept'", "set firewall ipv6-name V6-IN description 'This rule-set is configured by Ansible RM'", "set firewall ipv6-name V6-EG default-action 'reject'", "set firewall ipv6-name V6-EG description 'This rule-set is configured by Ansible RM'", ] self.execute_module(changed=True, commands=commands) def test_vyos_firewall_v4v6_rule_sets_rule_ovr_idem_01(self): set_module_args( dict( config=[ dict( afi="ipv4", rule_sets=[ dict( name="V4-INGRESS", description="This is IPv4 V4-INGRESS rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", fragment="match-frag", disabled=True, ) ], ), dict( name="EGRESS", default_action="reject", ), ], ), dict( afi="ipv6", rule_sets=[ dict( name="V6-INGRESS", default_action="accept", ), dict( name="EGRESS", default_action="reject", rules=[ dict( icmp=dict(type_name="echo-request"), number=20, ) ], ), ], ), ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_firewall_v6_rule_sets_rule_merged_01_version(self): self.get_os_version.return_value = "VyOS 1.4-rolling-202007010117" set_module_args( dict( config=[ dict( afi="ipv6", rule_sets=[ dict( name="INBOUND", description="This is IPv6 INBOUND rule set", default_action="accept", enable_default_log=True, rules=[ dict( number="101", action="accept", description="Rule 101 is configured by Ansible", ipsec="match-ipsec", protocol="icmp", disabled=True, icmp=dict(type_name="echo-request"), ) ], ), ], ) ], state="merged", ) ) commands = [ "set firewall ipv6-name INBOUND default-action 'accept'", "set firewall ipv6-name INBOUND description 'This is IPv6 INBOUND rule set'", "set firewall ipv6-name INBOUND enable-default-log", "set firewall ipv6-name INBOUND rule 101 protocol 'icmp'", "set firewall ipv6-name INBOUND rule 101 description 'Rule 101 is configured by Ansible'", "set firewall ipv6-name INBOUND rule 101", "set firewall ipv6-name INBOUND rule 101 disable", "set firewall ipv6-name INBOUND rule 101 action 'accept'", "set firewall ipv6-name INBOUND rule 101 ipsec 'match-ipsec'", "set firewall ipv6-name INBOUND rule 101 icmpv6 type-name echo-request", ] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_hostname.py b/tests/unit/modules/network/vyos/test_vyos_hostname.py index 3df9a17..c7edc26 100644 --- a/tests/unit/modules/network/vyos/test_vyos_hostname.py +++ b/tests/unit/modules/network/vyos/test_vyos_hostname.py @@ -1,112 +1,113 @@ # (c) 2021 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_hostname -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosHostnameModule(TestVyosModule): module = vyos_hostname def setUp(self): super(TestVyosHostnameModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.hostname.hostname.HostnameFacts.get_config" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosHostnameModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_hostname_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_vyos_hostname_merged_idempotent(self): set_module_args(dict(config=dict(hostname="vyos_test"))) self.execute_module(changed=False, commands=[]) def test_vyos_hostname_replaced_idempotent(self): set_module_args(dict(config=dict(hostname="vyos_test"), state="replaced")) self.execute_module(changed=False, commands=[]) def test_vyos_hostname_overridden_idempotent(self): set_module_args(dict(config=dict(hostname="vyos_test"), state="overridden")) self.execute_module(changed=False, commands=[]) def test_vyos_hostname_merged(self): set_module_args(dict(config=dict(hostname="vyos"))) self.execute_module(changed=True, commands=["set system host-name vyos"]) def test_vyos_hostname_replaced(self): set_module_args(dict(config=dict(hostname="vyos"), state="replaced")) self.execute_module(changed=True, commands=["set system host-name vyos"]) def test_vyos_hostname_overridden(self): set_module_args(dict(config=dict(hostname="vyos"), state="overridden")) def test_vyos_hostname_deleted(self): set_module_args(dict(state="deleted")) self.execute_module(changed=True, commands=["delete system host-name vyos_test"]) def test_vyos_hostname_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False, filename="vyos_hostname_config.cfg") gathered_list = {"hostname": "vyos_test"} self.assertEqual(sorted(gathered_list), sorted(result["gathered"])) def test_vyos_hostname_parsed(self): parsed_str = "set system host-name vyos_test" set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = {"hostname": "vyos_test"} self.assertEqual(sorted(parsed_list), sorted(result["parsed"])) def test_vyos_hostname_rendered(self): set_module_args(dict(state="rendered", config=dict(hostname="vyos_test"))) commands = ["set system host-name vyos_test"] result = self.execute_module(changed=False) self.assertEqual(sorted(result["rendered"]), sorted(commands), result["rendered"]) diff --git a/tests/unit/modules/network/vyos/test_vyos_interfaces.py b/tests/unit/modules/network/vyos/test_vyos_interfaces.py index ea6dbf5..06bbefa 100644 --- a/tests/unit/modules/network/vyos/test_vyos_interfaces.py +++ b/tests/unit/modules/network/vyos/test_vyos_interfaces.py @@ -1,185 +1,186 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_interfaces -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosFirewallInterfacesModule(TestVyosModule): module = vyos_interfaces def setUp(self): super(TestVyosFirewallInterfacesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos." "facts.interfaces.interfaces.InterfacesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosFirewallInterfacesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_interfaces_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_interfaces_merged(self): set_module_args( dict( config=[ dict(name="bond1", description="Bond - 1", enabled=True), dict(name="vtun1", description="vtun - 1", enabled=True), dict(name="wg01", description="wg - 1", enabled=True), ], state="merged", ) ) commands = [ "set interfaces bonding bond1 description 'Bond - 1'", "set interfaces openvpn vtun1 description 'vtun - 1'", "set interfaces wireguard wg01 description 'wg - 1'", ] self.execute_module(changed=True, commands=commands) def test_vyos_interfaces_merged_idempotent(self): set_module_args( dict( config=[ dict( name="wg02", description="wire guard int 2", enabled=True, ), ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_interfaces_merged_newinterface(self): set_module_args( dict( config=[ dict( name="eth4", description="Ethernet 4", enabled=True, speed="auto", duplex="auto", ), dict(name="eth1", description="Configured by Ansible"), ], state="merged", ) ) commands = [ "set interfaces ethernet eth1 description 'Configured by Ansible'", "set interfaces ethernet eth4 description 'Ethernet 4'", "set interfaces ethernet eth4 duplex 'auto'", "set interfaces ethernet eth4 speed 'auto'", ] self.execute_module(changed=True, commands=commands) def test_vyos_interfaces_replaced_newinterface(self): set_module_args( dict( config=[ dict( name="eth4", description="Ethernet 4", enabled=True, speed="auto", duplex="auto", ), dict(name="eth1", description="Configured by Ansible"), ], state="replaced", ) ) commands = [ "set interfaces ethernet eth1 description 'Configured by Ansible'", "set interfaces ethernet eth4 description 'Ethernet 4'", "set interfaces ethernet eth4 duplex 'auto'", "set interfaces ethernet eth4 speed 'auto'", ] self.execute_module(changed=True, commands=commands) def test_vyos_interfaces_overridden_newinterface(self): set_module_args( dict( config=[ dict( name="eth4", description="Ethernet 4", enabled=True, speed="auto", duplex="auto", ), dict(name="eth1", description="Configured by Ansible"), ], state="overridden", ) ) commands = [ "set interfaces ethernet eth1 description 'Configured by Ansible'", "set interfaces ethernet eth4 description 'Ethernet 4'", "set interfaces ethernet eth4 duplex 'auto'", "set interfaces ethernet eth4 speed 'auto'", "delete interfaces wireguard wg02 description", "delete interfaces ethernet eth3 description", ] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_logging_global.py b/tests/unit/modules/network/vyos/test_vyos_logging_global.py index f1ab00d..209844e 100644 --- a/tests/unit/modules/network/vyos/test_vyos_logging_global.py +++ b/tests/unit/modules/network/vyos/test_vyos_logging_global.py @@ -1,451 +1,451 @@ # # (c) 2021, Ansible by Red Hat, inc # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) # from __future__ import absolute_import, division, print_function __metaclass__ = type from textwrap import dedent +from unittest.mock import patch from ansible_collections.vyos.vyos.plugins.modules import vyos_logging_global -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule class TestVyosLoggingGlobalModule(TestVyosModule): module = vyos_logging_global def setUp(self): super(TestVyosLoggingGlobalModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.logging_global.logging_global.Logging_globalFacts.get_logging_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosLoggingGlobalModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() def test_vyos_logging_global_merged_idempotent(self): self.execute_show_command.return_value = dedent( """\ set system syslog console facility all set system syslog console facility local7 level 'err' set system syslog console facility news level 'debug' set system syslog file xyz set system syslog file abc archive size '125' set system syslog file def archive file '2' set system syslog file def facility local6 level 'emerg' set system syslog file def facility local7 level 'emerg' set system syslog global archive file '2' set system syslog global archive size '111' set system syslog global facility cron level 'debug' set system syslog global facility local7 level 'debug' set system syslog global marker interval '111' set system syslog global preserve-fqdn set system syslog host 10.0.2.12 facility all protocol 'udp' set system syslog host 10.0.2.15 facility all level 'all' set system syslog host 10.0.2.15 facility all protocol 'udp' set system syslog host 10.0.2.15 port '122' set system syslog user paul facility local7 level 'err' set system syslog user vyos facility local6 level 'alert' set system syslog user vyos facility local7 level 'debug' """ ) playbook = dict( config=dict( console=dict( facilities=[ dict(facility="all"), dict(facility="local7", severity="err"), dict(facility="news", severity="debug"), ] ), files=[ dict(path="xyz"), dict(path="abc", archive=dict(size=125)), dict( path="def", archive=dict(file_num=2), facilities=[ dict(facility="local6", severity="emerg"), dict(facility="local7", severity="emerg"), ], ), ], hosts=[ dict( hostname="10.0.2.15", port=122, facilities=[ dict(facility="all", severity="all"), dict(facility="all", protocol="udp"), ], ), dict( hostname="10.0.2.12", facilities=[dict(facility="all", protocol="udp")], ), ], users=[ dict( username="vyos", facilities=[ dict(facility="local7", severity="debug"), dict(facility="local6", severity="alert"), ], ), dict( username="paul", facilities=[dict(facility="local7", severity="err")], ), ], global_params=dict( archive=dict(size=111, file_num=2), marker_interval=111, preserve_fqdn="True", facilities=[ dict(facility="cron", severity="debug"), dict(facility="local7", severity="debug"), ], ), ) ) compare_cmds = [] playbook["state"] = "merged" set_module_args(playbook) result = self.execute_module() self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_merged(self): self.execute_show_command.return_value = dedent( """\ """ ) playbook = dict( config=dict( console=dict( facilities=[ dict(facility="all"), dict(facility="local7", severity="err"), dict(facility="news", severity="debug"), ] ), files=[ dict(path="xyz"), dict(path="abc", archive=dict(size=125)), dict( path="def", archive=dict(file_num=2), facilities=[ dict(facility="local6", severity="emerg"), dict(facility="local7", severity="emerg"), ], ), ], hosts=[ dict( hostname="10.0.2.15", port=122, facilities=[ dict(facility="all", severity="all"), dict(facility="all", protocol="udp"), ], ), dict( hostname="10.0.2.12", facilities=[dict(facility="all", protocol="udp")], ), ], users=[ dict( username="vyos", facilities=[ dict(facility="local7", severity="debug"), dict(facility="local6", severity="alert"), ], ), dict( username="paul", facilities=[dict(facility="local7", severity="err")], ), ], global_params=dict( archive=dict(size=111, file_num=2), marker_interval=111, preserve_fqdn="True", facilities=[ dict(facility="cron", severity="debug"), dict(facility="local7", severity="debug"), ], ), ) ) compare_cmds = [ "set system syslog user paul facility local7 level err", "set system syslog host 10.0.2.15 facility all protocol udp", "set system syslog global marker interval 111", "set system syslog file def archive file 2", "set system syslog file abc archive size 125", "set system syslog console facility local7 level err", "set system syslog host 10.0.2.12 facility all protocol udp", "set system syslog global facility local7 level debug", "set system syslog host 10.0.2.15 facility all level all", "set system syslog global preserve-fqdn", "set system syslog global archive file 2", "set system syslog console facility all", "set system syslog file xyz", "set system syslog file def facility local7 level emerg", "set system syslog console facility news level debug", "set system syslog user vyos facility local6 level alert", "set system syslog global facility cron level debug", "set system syslog file def facility local6 level emerg", "set system syslog global archive size 111", "set system syslog host 10.0.2.15 port 122", "set system syslog user vyos facility local7 level debug", ] playbook["state"] = "merged" set_module_args(playbook) result = self.execute_module(changed=True) self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_deleted(self): self.execute_show_command.return_value = dedent( """\ set system syslog console facility all set system syslog console facility local7 level 'err' set system syslog console facility news level 'debug' set system syslog file xyz set system syslog file abc archive size '125' set system syslog file def archive file '2' set system syslog file def facility local6 level 'emerg' set system syslog file def facility local7 level 'emerg' set system syslog global archive file '2' set system syslog global archive size '111' set system syslog global facility cron level 'debug' set system syslog global facility local7 level 'debug' set system syslog global marker interval '111' set system syslog global preserve-fqdn set system syslog host 10.0.2.12 facility all protocol 'udp' set system syslog host 10.0.2.15 facility all level 'all' set system syslog host 10.0.2.15 facility all protocol 'udp' set system syslog host 10.0.2.15 port '122' set system syslog user paul facility local7 level 'err' set system syslog user vyos facility local6 level 'alert' set system syslog user vyos facility local7 level 'debug' """ ) playbook = dict(config=dict()) compare_cmds = ["delete system syslog"] playbook["state"] = "deleted" set_module_args(playbook) result = self.execute_module(changed=True) self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_replaced(self): """ passing all commands as have and expecting [] commands """ self.execute_show_command.return_value = dedent( """\ set system syslog console facility all set system syslog console facility local7 level 'err' set system syslog console facility news level 'debug' set system syslog file xyz set system syslog file abc archive size '125' set system syslog file def archive file '2' set system syslog file def facility local6 level 'emerg' set system syslog file def facility local7 level 'emerg' set system syslog global archive file '2' set system syslog global archive size '111' set system syslog global facility cron level 'debug' set system syslog global facility local7 level 'debug' set system syslog global marker interval '111' set system syslog global preserve-fqdn set system syslog host 10.0.2.12 facility all protocol 'udp' set system syslog host 10.0.2.15 facility all level 'all' set system syslog host 10.0.2.15 facility all protocol 'udp' set system syslog host 10.0.2.15 port '122' set system syslog user paul facility local7 level 'err' set system syslog user vyos facility local6 level 'alert' set system syslog user vyos facility local7 level 'debug' """ ) playbook = dict( config=dict( console=dict(facilities=[dict(facility="local7", severity="emerg")]), files=[ dict( path="abc", archive=dict(file_num=2), facilities=[ dict(facility="local6", severity="err"), dict(facility="local7", severity="emerg"), ], ) ], ) ) compare_cmds = [ "delete system syslog console facility all", "delete system syslog console facility local7", "delete system syslog console facility news", "delete system syslog file def", "delete system syslog file xyz", "delete system syslog global facility cron", "delete system syslog global facility local7", "delete system syslog global archive file 2", "delete system syslog global archive size 111", "delete system syslog global marker", "delete system syslog global preserve-fqdn", "delete system syslog host 10.0.2.12", "delete system syslog host 10.0.2.15", "delete system syslog user paul", "delete system syslog user vyos", "set system syslog console facility local7 level emerg", "set system syslog file abc facility local6 level err", "set system syslog file abc facility local7 level emerg", "delete system syslog file abc archive size 125", "set system syslog file abc archive file 2", ] playbook["state"] = "replaced" set_module_args(playbook) result = self.execute_module(changed=True) self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_replaced_idempotent(self): """ passing all commands as have and expecting [] commands """ self.execute_show_command.return_value = dedent( """\ set system syslog console facility local6 """ ) playbook = dict(config=dict(console=dict(facilities=[dict(facility="local6")]))) compare_cmds = [] playbook["state"] = "replaced" set_module_args(playbook) result = self.execute_module(changed=False) self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_overridden(self): self.execute_show_command.return_value = dedent( """\ set system syslog console set system syslog global """ ) playbook = dict( config=dict( console=dict(facilities=[dict(facility="local7", severity="emerg")]), files=[ dict( path="abc", archive=dict(file_num=2), facilities=[ dict(facility="local6", severity="err"), dict(facility="local7", severity="emerg"), ], ) ], ) ) compare_cmds = [ "set system syslog console facility local7 level emerg", "set system syslog file abc facility local6 level err", "set system syslog file abc facility local7 level emerg", "set system syslog file abc archive file 2", ] playbook["state"] = "overridden" set_module_args(playbook) result = self.execute_module(changed=True) print(result["commands"]) self.maxDiff = None self.assertEqual(sorted(result["commands"]), sorted(compare_cmds)) def test_vyos_logging_global_rendered(self): playbook = dict( config=dict( console=dict(facilities=[dict(facility="all")]), hosts=[ dict( hostname="10.0.2.16", facilities=[dict(facility="local6")], ) ], users=[ dict(username="vyos"), dict(username="paul", facilities=[dict(facility="local7")]), ], ) ) compare_cmds = [ "set system syslog console facility all", "set system syslog host 10.0.2.16 facility local6", "set system syslog user vyos", "set system syslog user paul facility local7", ] playbook["state"] = "rendered" set_module_args(playbook) result = self.execute_module() self.maxDiff = None self.assertEqual(sorted(result["rendered"]), sorted(compare_cmds)) def test_vyos_logging_global_parsed(self): set_module_args( dict( running_config=dedent( """\ set system syslog console facility all set system syslog file xyz """ ), state="parsed", ) ) parsed = dict( console=dict(facilities=[dict(facility="all")]), files=[dict(path="xyz")], ) result = self.execute_module(changed=False) self.maxDiff = None self.assertEqual(sorted(result["parsed"]), sorted(parsed)) def test_vyos_logging_global_gathered(self): self.execute_show_command.return_value = dedent( """\ set system syslog console facility all """ ) set_module_args(dict(state="gathered")) gathered = dict(console=dict(facilities=[dict(facility="all")])) result = self.execute_module(changed=False) self.maxDiff = None self.assertEqual(sorted(result["gathered"]), sorted(gathered)) diff --git a/tests/unit/modules/network/vyos/test_vyos_ntp_global.py b/tests/unit/modules/network/vyos/test_vyos_ntp_global.py index 5da1056..37c851d 100644 --- a/tests/unit/modules/network/vyos/test_vyos_ntp_global.py +++ b/tests/unit/modules/network/vyos/test_vyos_ntp_global.py @@ -1,347 +1,348 @@ # (c) 2021 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_ntp_global -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosNTPModule(TestVyosModule): module = vyos_ntp_global def setUp(self): super(TestVyosNTPModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.ntp_global.ntp_global.Ntp_globalFacts.get_config" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosNTPModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_ntp_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_ntp_merged_idempotent(self): set_module_args( dict( config=dict( allow_clients=["10.1.1.0/24", "10.1.2.0/24"], listen_addresses=["10.2.3.1", "10.4.3.1"], servers=[ dict(server="server1"), dict(server="server3", options=["noselect", "dynamic"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_ntp_merged(self): set_module_args( dict( config=dict( allow_clients=["10.2.2.0/24", "10.3.3.0/24"], listen_addresses=["10.3.4.1", "10.4.5.1"], servers=[ dict(server="server4", options=["dynamic", "preempt"]), dict( server="server5", options=[ "noselect", "pool", "preempt", "prefer", ], ), ], ), state="merged", ) ) commands = [ "set system ntp allow-clients address 10.2.2.0/24", "set system ntp allow-clients address 10.3.3.0/24", "set system ntp listen-address 10.3.4.1", "set system ntp listen-address 10.4.5.1", "set system ntp server server4 dynamic", "set system ntp server server4 preempt", "set system ntp server server5 pool", "set system ntp server server5 noselect", "set system ntp server server5 preempt", "set system ntp server server5 prefer", ] self.execute_module(changed=True, commands=commands) def test_ntp_replaced(self): set_module_args( dict( config=dict( allow_clients=["10.3.4.0/24", "10.4.5.0/24"], listen_addresses=["10.3.3.1", "10.4.4.1"], servers=[ dict(server="server4", options=["noselect", "prefer"]), dict( server="server6", options=[ "noselect", "dynamic", "prefer", "preempt", ], ), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="replaced", ) ) commands = [ "delete system ntp allow-clients address 10.1.1.0/24", "delete system ntp allow-clients address 10.1.2.0/24", "delete system ntp listen-address 10.2.3.1", "delete system ntp listen-address 10.4.3.1", "delete system ntp server server1", "delete system ntp server server3", "set system ntp allow-clients address 10.3.4.0/24", "set system ntp allow-clients address 10.4.5.0/24", "set system ntp listen-address 10.3.3.1", "set system ntp listen-address 10.4.4.1", "set system ntp server server4 noselect", "set system ntp server server4 prefer", "set system ntp server server6 noselect", "set system ntp server server6 dynamic", "set system ntp server server6 prefer", "set system ntp server server6 preempt", ] self.execute_module(changed=True, commands=commands) def test_ntp_replaced_idempotent(self): set_module_args( dict( config=dict( allow_clients=["10.1.1.0/24", "10.1.2.0/24"], listen_addresses=["10.2.3.1", "10.4.3.1"], servers=[ dict(server="server1"), dict(server="server3", options=["noselect", "dynamic"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_ntp_overridden(self): set_module_args( dict( config=dict( allow_clients=["10.9.9.0/24"], listen_addresses=["10.9.9.1"], servers=[ dict(server="server9"), dict(server="server6", options=["noselect", "dynamic"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="overridden", ) ) commands = [ "delete system ntp allow-clients address 10.1.1.0/24", "delete system ntp allow-clients address 10.1.2.0/24", "delete system ntp listen-address 10.2.3.1", "delete system ntp listen-address 10.4.3.1", "delete system ntp server server1", "delete system ntp server server3", "set system ntp allow-clients address 10.9.9.0/24", "set system ntp listen-address 10.9.9.1", "set system ntp server server9", "set system ntp server server6 noselect", "set system ntp server server6 dynamic", ] self.execute_module(changed=True, commands=commands) def test_ntp_overridden_idempotent(self): set_module_args( dict( config=dict( allow_clients=["10.1.1.0/24", "10.1.2.0/24"], listen_addresses=["10.2.3.1", "10.4.3.1"], servers=[ dict(server="server1"), dict(server="server3", options=["noselect", "dynamic"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_ntp_rendered(self): set_module_args( dict( config=dict( allow_clients=["10.7.7.0/24", "10.8.8.0/24"], listen_addresses=["10.7.9.1"], servers=[ dict(server="server79"), dict(server="server46", options=["noselect", "dynamic"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="rendered", ) ) rendered_commands = [ "set system ntp allow-clients address 10.7.7.0/24", "set system ntp allow-clients address 10.8.8.0/24", "set system ntp listen-address 10.7.9.1", "set system ntp server server79", "set system ntp server server46 noselect", "set system ntp server server46 dynamic", "set system ntp server time1.vyos.net", "set system ntp server time2.vyos.net", "set system ntp server time3.vyos.net", ] result = self.execute_module(changed=False) self.assertEqual( sorted(result["rendered"]), sorted(rendered_commands), result["rendered"], ) def test_ntp_parsed(self): commands = ( "set system ntp allow-clients address 10.7.7.0/24", "set system ntp allow-clients address 10.6.7.0/24", "set system ntp listen-address 10.7.9.1", "set system ntp listen-address 10.7.7.1", "set system ntp server check", "set system ntp server server46 noselect", "set system ntp server server46 prefer", "set system ntp server time1.vyos.net", "set system ntp server time2.vyos.net", "set system ntp server time3.vyos.net", ) parsed_str = "\n".join(commands) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = { "allow_clients": ["10.6.7.0/24", "10.7.7.0/24"], "listen_addresses": ["10.7.7.1", "10.7.9.1"], "servers": [ {"server": "check"}, {"server": "server46", "options": ["noselect", "prefer"]}, {"server": "time1.vyos.net"}, {"server": "time2.vyos.net"}, {"server": "time3.vyos.net"}, ], } self.assertEqual(parsed_list, result["parsed"]) def test_ntp_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False) gathered_list = { "allow_clients": ["10.1.1.0/24", "10.1.2.0/24"], "listen_addresses": ["10.2.3.1", "10.4.3.1"], "servers": [ {"server": "server1"}, {"server": "server3", "options": ["dynamic", "noselect"]}, {"server": "time1.vyos.net"}, {"server": "time2.vyos.net"}, {"server": "time3.vyos.net"}, ], } self.assertEqual(gathered_list, result["gathered"]) def test_ntp_deleted(self): set_module_args( dict( config=dict( allow_clients=["10.1.1.0/24"], listen_addresses=["10.2.3.1"], servers=[ dict(server="server1"), dict(server="server3", options=["noselect"]), dict(server="time1.vyos.net"), dict(server="time2.vyos.net"), dict(server="time3.vyos.net"), ], ), state="deleted", ) ) commands = [ "delete system ntp allow-clients", "delete system ntp listen-address", "delete system ntp server server1", "delete system ntp server server3", "delete system ntp server time1.vyos.net", "delete system ntp server time2.vyos.net", "delete system ntp server time3.vyos.net", ] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_ospf_interfaces.py b/tests/unit/modules/network/vyos/test_vyos_ospf_interfaces.py index 74eaf7c..248b98e 100644 --- a/tests/unit/modules/network/vyos/test_vyos_ospf_interfaces.py +++ b/tests/unit/modules/network/vyos/test_vyos_ospf_interfaces.py @@ -1,443 +1,444 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_ospf_interfaces -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosOspfInterfacesModule(TestVyosModule): module = vyos_ospf_interfaces def setUp(self): super(TestVyosOspfInterfacesModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.ospf_interfaces.ospf_interfaces.Ospf_interfacesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosOspfInterfacesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_ospf_interfaces_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def sort_address_family(self, entry_list): for entry in entry_list: if entry.get("address_family"): entry["address_family"].sort(key=lambda i: i.get("afi")) def test_vyos_ospf_interfaces_merged_new_config(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict( afi="ipv4", cost=100, authentication=dict(plaintext_password="abcdefg!"), priority=55, ), dict(afi="ipv6", mtu_ignore=True, instance=20), ], ), dict( name="bond2", address_family=[ dict( afi="ipv4", transmit_delay=9, ), dict(afi="ipv6", passive=True), ], ), ], state="merged", ) ) commands = [ "set interfaces bonding bond2 ip ospf transmit-delay 9", "set interfaces bonding bond2 ipv6 ospfv3 passive", "set interfaces ethernet eth0 ip ospf cost 100", "set interfaces ethernet eth0 ip ospf priority 55", "set interfaces ethernet eth0 ip ospf authentication plaintext-password abcdefg!", "set interfaces ethernet eth0 ipv6 ospfv3 instance-id 20", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospf_interfaces_merged_idempotent(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict(afi="ipv6", mtu_ignore=True, instance=33), ], ), dict( name="eth1", address_family=[ dict( afi="ipv4", cost=100, ), dict(afi="ipv6", ifmtu=33), ], ), ], ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospf_interfaces_existing_config_merged(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict(afi="ipv6", cost=500), ], ), dict( name="eth1", address_family=[ dict( afi="ipv4", priority=100, ), dict(afi="ipv6", ifmtu=25), ], ), ], ) ) commands = [ "set interfaces ethernet eth0 ipv6 ospfv3 cost 500", "set interfaces ethernet eth1 ip ospf priority 100", "set interfaces ethernet eth1 ipv6 ospfv3 ifmtu 25", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospf_interfaces_replaced(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict( afi="ipv4", cost=100, authentication=dict(plaintext_password="abcdefg!"), priority=55, ), ], ), dict( name="bond2", address_family=[ dict( afi="ipv4", transmit_delay=9, ), dict(afi="ipv6", passive=True), ], ), ], state="replaced", ) ) commands = [ "set interfaces bonding bond2 ip ospf transmit-delay 9", "set interfaces bonding bond2 ipv6 ospfv3 passive", "set interfaces ethernet eth0 ip ospf cost 100", "set interfaces ethernet eth0 ip ospf priority 55", "set interfaces ethernet eth0 ip ospf authentication plaintext-password abcdefg!", "delete interfaces ethernet eth0 ipv6 ospfv3 instance-id 33", "delete interfaces ethernet eth0 ipv6 ospfv3 mtu-ignore", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospf_interfaces_replaced_idempotent(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict(afi="ipv6", mtu_ignore=True, instance=33), ], ), dict( name="eth1", address_family=[ dict( afi="ipv4", cost=100, ), dict(afi="ipv6", ifmtu=33), ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospf_interfaces_overridden(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict( afi="ipv4", cost=100, authentication=dict(plaintext_password="abcdefg!"), priority=55, ), ], ), dict( name="bond2", address_family=[ dict( afi="ipv4", transmit_delay=9, ), dict(afi="ipv6", passive=True), ], ), ], state="overridden", ) ) commands = [ "set interfaces bonding bond2 ip ospf transmit-delay 9", "set interfaces bonding bond2 ipv6 ospfv3 passive", "set interfaces ethernet eth0 ip ospf cost 100", "set interfaces ethernet eth0 ip ospf priority 55", "set interfaces ethernet eth0 ip ospf authentication plaintext-password abcdefg!", "delete interfaces ethernet eth1 ip ospf", "delete interfaces ethernet eth1 ipv6 ospfv3", "delete interfaces ethernet eth0 ipv6 ospfv3 mtu-ignore", "delete interfaces ethernet eth0 ipv6 ospfv3 instance-id 33", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospf_interfaces_overridden_idempotent(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict(afi="ipv6", mtu_ignore=True, instance=33), ], ), dict( name="eth1", address_family=[ dict( afi="ipv4", cost=100, ), dict(afi="ipv6", ifmtu=33), ], ), ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospf_interfaces_deleted(self): set_module_args( dict( config=[ dict( name="eth0", ), ], state="deleted", ) ) commands = ["delete interfaces ethernet eth0 ipv6 ospfv3"] self.execute_module(changed=True, commands=commands) def test_vyos_ospf_interfaces_notpresent_deleted(self): set_module_args( dict( config=[ dict( name="eth3", ), ], state="deleted", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospf_interfaces_rendered(self): set_module_args( dict( config=[ dict( name="eth0", address_family=[ dict( afi="ipv4", cost=100, authentication=dict(plaintext_password="abcdefg!"), priority=55, ), dict(afi="ipv6", mtu_ignore=True, instance=20), ], ), dict( name="bond2", address_family=[ dict( afi="ipv4", transmit_delay=9, ), dict(afi="ipv6", passive=True), ], ), ], state="rendered", ) ) commands = [ "set interfaces ethernet eth0 ip ospf cost 100", "set interfaces ethernet eth0 ip ospf authentication plaintext-password abcdefg!", "set interfaces ethernet eth0 ip ospf priority 55", "set interfaces ethernet eth0 ipv6 ospfv3 mtu-ignore", "set interfaces ethernet eth0 ipv6 ospfv3 instance-id 20", "set interfaces bonding bond2 ip ospf transmit-delay 9", "set interfaces bonding bond2 ipv6 ospfv3 passive", ] result = self.execute_module(changed=False) self.assertEqual(sorted(result["rendered"]), sorted(commands), result["rendered"]) def test_vyos_ospf_interfaces_parsed(self): commands = [ "set interfaces bonding bond2 ip ospf authentication md5 key-id 10 md5-key '1111111111232345'", "set interfaces bonding bond2 ip ospf bandwidth '70'", "set interfaces bonding bond2 ip ospf transmit-delay '45'", "set interfaces bonding bond2 ipv6 ospfv3 'passive'", "set interfaces ethernet eth0 ip ospf cost '50'", "set interfaces ethernet eth0 ip ospf priority '26'", "set interfaces ethernet eth0 ipv6 ospfv3 instance-id '33'", "set interfaces ethernet eth0 ipv6 ospfv3 'mtu-ignore'", "set interfaces ethernet eth1 ip ospf network 'point-to-point'", "set interfaces ethernet eth1 ip ospf priority '26'", "set interfaces ethernet eth1 ip ospf transmit-delay '50'", "set interfaces ethernet eth1 ipv6 ospfv3 dead-interval '39'", ] parsed_str = "\n".join(commands) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = [ { "address_family": [ { "afi": "ipv4", "authentication": { "md5_key": { "key": "1111111111232345", "key_id": 10, } }, "bandwidth": 70, "transmit_delay": 45, }, {"afi": "ipv6", "passive": True}, ], "name": "bond2", }, { "address_family": [ {"afi": "ipv4", "cost": 50, "priority": 26}, {"afi": "ipv6", "instance": "33", "mtu_ignore": True}, ], "name": "eth0", }, { "address_family": [ { "afi": "ipv4", "network": "point-to-point", "priority": 26, "transmit_delay": 50, }, {"afi": "ipv6", "dead_interval": 39}, ], "name": "eth1", }, ] result_list = self.sort_address_family(result["parsed"]) given_list = self.sort_address_family(parsed_list) self.assertEqual(result_list, given_list) def test_vyos_ospf_interfaces_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False, filename="vyos_ospf_interfaces_config.cfg") gathered_list = [ { "address_family": [{"afi": "ipv6", "instance": "33", "mtu_ignore": True}], "name": "eth0", }, { "address_family": [ {"afi": "ipv4", "cost": 100}, {"afi": "ipv6", "ifmtu": 33}, ], "name": "eth1", }, ] result_list = self.sort_address_family(result["gathered"]) given_list = self.sort_address_family(gathered_list) self.assertEqual(result_list, given_list) diff --git a/tests/unit/modules/network/vyos/test_vyos_ospfv2.py b/tests/unit/modules/network/vyos/test_vyos_ospfv2.py index 84b882d..d2ddb40 100644 --- a/tests/unit/modules/network/vyos/test_vyos_ospfv2.py +++ b/tests/unit/modules/network/vyos/test_vyos_ospfv2.py @@ -1,423 +1,424 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_ospfv2 -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosOspfv2Module(TestVyosModule): module = vyos_ospfv2 def setUp(self): super(TestVyosOspfv2Module, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.ospfv2.ospfv2.Ospfv2Facts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosOspfv2Module, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_ospfv2_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_vyos_ospfv2_merged_new_config(self): set_module_args( dict( config=dict( log_adjacency_changes="detail", mpls_te=dict(enabled=True, router_address="192.0.11.11"), auto_cost=dict(reference_bandwidth=2), areas=[ dict( area_id="2", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="4", area_type=dict(stub=dict(default_cost=10)), network=[dict(address="192.0.2.0/24")], range=[ dict(address="192.0.3.0/24", cost=10), dict(address="192.0.4.0/24", cost=12), ], ), ], ), state="merged", ) ) commands = [ "set protocols ospf mpls-te enable", "set protocols ospf mpls-te router-address '192.0.11.11'", "set protocols ospf auto-cost reference-bandwidth '2'", "set protocols ospf log-adjacency-changes 'detail'", "set protocols ospf area '2'", "set protocols ospf area 2 authentication plaintext-password", "set protocols ospf area 2 shortcut enable", "set protocols ospf area 2 area-type normal", "set protocols ospf area 4 range 192.0.3.0/24 cost 10", "set protocols ospf area 4 range 192.0.3.0/24", "set protocols ospf area 4 range 192.0.4.0/24 cost 12", "set protocols ospf area 4 range 192.0.4.0/24", "set protocols ospf area 4 area-type stub default-cost 10", "set protocols ospf area '4'", "set protocols ospf area 4 network 192.0.2.0/24", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv2_merged_idem(self): set_module_args( dict( config=dict( areas=[ dict( area_id="12", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="14", area_type=dict(stub=dict(default_cost=20)), network=[dict(address="192.0.12.0/24")], range=[ dict(address="192.0.13.0/24", cost=10), dict(address="192.0.14.0/24", cost=12), ], ), ], ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospfv2_merged_update_existing(self): set_module_args( dict( config=dict( areas=[ dict( area_id="12", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="14", area_type=dict(stub=dict(set=False)), network=[ dict(address="192.0.12.0/24"), dict(address="192.0.22.0/24"), ], range=[ dict(address="192.0.13.0/24", cost=10), dict(address="192.0.14.0/24", cost=12), ], ), ], ), state="merged", ) ) commands = [ "delete protocols ospf area 14 area-type stub", "set protocols ospf area 14 network 192.0.22.0/24", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv2_replaced(self): set_module_args( dict( config=dict( log_adjacency_changes="detail", mpls_te=dict(enabled=True, router_address="192.0.11.11"), auto_cost=dict(reference_bandwidth=2), areas=[ dict( area_id="12", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="15", area_type=dict(stub=dict(default_cost=10)), network=[dict(address="192.0.12.0/24")], range=[ dict(address="192.0.13.0/24", cost=10), dict(address="192.0.14.0/24", cost=12), dict(address="192.0.15.0/24", cost=14), ], ), ], ), state="replaced", ) ) commands = [ "set protocols ospf mpls-te enable", "set protocols ospf mpls-te router-address '192.0.11.11'", "set protocols ospf auto-cost reference-bandwidth '2'", "set protocols ospf log-adjacency-changes 'detail'", "delete protocols ospf area 14", "set protocols ospf area 15 range 192.0.13.0/24 cost 10", "set protocols ospf area 15 range 192.0.13.0/24", "set protocols ospf area 15 range 192.0.14.0/24 cost 12", "set protocols ospf area 15 range 192.0.14.0/24", "set protocols ospf area 15 range 192.0.15.0/24 cost 14", "set protocols ospf area 15 range 192.0.15.0/24", "set protocols ospf area 15 area-type stub default-cost 10", "set protocols ospf area '15'", "set protocols ospf area 15 network 192.0.12.0/24", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv2_replaced_idem(self): set_module_args( dict( config=dict( areas=[ dict( area_id="12", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="14", area_type=dict(stub=dict(default_cost=20)), network=[dict(address="192.0.12.0/24")], range=[ dict(address="192.0.13.0/24", cost=10), dict(address="192.0.14.0/24", cost=12), ], ), ], ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospfv2_deleted_no_config(self): set_module_args(dict(config=None, state="deleted")) commands = ["delete protocols ospf"] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv2_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False, filename="vyos_ospfv2_config.cfg") gather_dict = { "areas": [ { "area_id": "2", "area_type": {"normal": True}, "authentication": "plaintext-password", "shortcut": "enable", }, { "area_id": "14", "area_type": {"stub": {"default_cost": 20, "set": True}}, "network": [{"address": "192.0.12.0/24"}], "range": [ {"address": "192.0.13.0/24", "cost": 10}, {"address": "192.0.14.0/24", "cost": 12}, ], }, ], } self.assertEqual(sorted(gather_dict), sorted(result["gathered"])) def test_vyos_ospfv2_parsed(self): parsed_str = """set protocols ospf area 2 area-type 'normal' set protocols ospf area 2 authentication 'plaintext-password' set protocols ospf area 2 shortcut 'enable' set protocols ospf area 3 area-type 'nssa' set protocols ospf area 4 area-type stub default-cost '20' set protocols ospf area 4 network '192.0.2.0/24' set protocols ospf area 4 range 192.0.3.0/24 cost '10' set protocols ospf area 4 range 192.0.4.0/24 cost '12' set protocols ospf default-information originate 'always' set protocols ospf default-information originate metric '10' set protocols ospf default-information originate metric-type '2' set protocols ospf auto-cost reference-bandwidth '2' set protocols ospf default-information originate route-map 'ingress' set protocols ospf log-adjacency-changes 'detail' set protocols ospf max-metric router-lsa 'administrative' set protocols ospf max-metric router-lsa on-shutdown '10' set protocols ospf max-metric router-lsa on-startup '10' set protocols ospf mpls-te 'enable' set protocols ospf mpls-te router-address '192.0.11.11' set protocols ospf neighbor 192.0.11.12 poll-interval '10' set protocols ospf neighbor 192.0.11.12 priority '2' set protocols ospf parameters abr-type 'cisco' set protocols ospf parameters 'opaque-lsa' set protocols ospf parameters 'rfc1583-compatibility' set protocols ospf parameters router-id '192.0.1.1' set protocols ospf passive-interface 'eth1' set protocols ospf passive-interface 'eth2' set protocols ospf redistribute bgp metric '10' set protocols ospf redistribute bgp metric-type '2'""" set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = { "areas": [ { "area_id": "2", "area_type": {"normal": True}, "authentication": "plaintext-password", "shortcut": "enable", }, {"area_id": "3", "area_type": {"nssa": {"set": True}}}, { "area_id": "4", "area_type": {"stub": {"default_cost": 20, "set": True}}, "network": [{"address": "192.0.2.0/24"}], "range": [ {"address": "192.0.3.0/24", "cost": 10}, {"address": "192.0.4.0/24", "cost": 12}, ], }, ], "auto_cost": {"reference_bandwidth": 2}, "default_information": { "originate": { "always": True, "metric": 10, "metric_type": 2, "route_map": "ingress", } }, "log_adjacency_changes": "detail", "max_metric": { "router_lsa": { "administrative": True, "on_shutdown": 10, "on_startup": 10, } }, "mpls_te": {"enabled": True, "router_address": "192.0.11.11"}, "neighbor": [ { "neighbor_id": "192.0.11.12", "poll_interval": 10, "priority": 2, } ], "parameters": { "abr_type": "cisco", "opaque_lsa": True, "rfc1583_compatibility": True, "router_id": "192.0.1.1", }, "passive_interface": ["eth2", "eth1"], "redistribute": [{"metric": 10, "metric_type": 2, "route_type": "bgp"}], } self.assertEqual(sorted(parsed_list), sorted(result["parsed"])) def test_vyos_ospfv2_rendered(self): set_module_args( dict( config=dict( log_adjacency_changes="detail", mpls_te=dict(enabled=True, router_address="192.0.11.11"), auto_cost=dict(reference_bandwidth=2), areas=[ dict( area_id="2", area_type=dict(normal=True), authentication="plaintext-password", shortcut="enable", ), dict( area_id="4", area_type=dict(stub=dict(default_cost=10)), network=[dict(address="192.0.2.0/24")], range=[ dict(address="192.0.3.0/24", cost=10), dict(address="192.0.4.0/24", cost=12), ], ), ], ), state="rendered", ) ) commands = [ "set protocols ospf mpls-te enable", "set protocols ospf mpls-te router-address '192.0.11.11'", "set protocols ospf auto-cost reference-bandwidth '2'", "set protocols ospf log-adjacency-changes 'detail'", "set protocols ospf area '2'", "set protocols ospf area 2 authentication plaintext-password", "set protocols ospf area 2 shortcut enable", "set protocols ospf area 2 area-type normal", "set protocols ospf area 4 range 192.0.3.0/24 cost 10", "set protocols ospf area 4 range 192.0.3.0/24", "set protocols ospf area 4 range 192.0.4.0/24 cost 12", "set protocols ospf area 4 range 192.0.4.0/24", "set protocols ospf area 4 area-type stub default-cost 10", "set protocols ospf area '4'", "set protocols ospf area 4 network 192.0.2.0/24", ] result = self.execute_module(changed=False) self.assertEqual(sorted(result["rendered"]), sorted(commands), result["rendered"]) diff --git a/tests/unit/modules/network/vyos/test_vyos_ospfv3.py b/tests/unit/modules/network/vyos/test_vyos_ospfv3.py index 2231c0b..0002451 100644 --- a/tests/unit/modules/network/vyos/test_vyos_ospfv3.py +++ b/tests/unit/modules/network/vyos/test_vyos_ospfv3.py @@ -1,338 +1,339 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_ospfv3 -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosOspfv3Module(TestVyosModule): module = vyos_ospfv3 def setUp(self): super(TestVyosOspfv3Module, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.ospfv3.ospfv3.Ospfv3Facts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosOspfv3Module, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_ospfv3_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_vyos_ospfv3_merged_new_config(self): set_module_args( dict( config=dict( redistribute=[dict(route_type="bgp")], parameters=dict(router_id="192.0.2.10"), areas=[ dict( area_id="2", export_list="export1", import_list="import1", range=[ dict(address="2001:db10::/32"), dict(address="2001:db20::/32"), dict(address="2001:db30::/32"), ], ), dict( area_id="3", range=[dict(address="2001:db40::/32")], ), ], ), state="merged", ) ) commands = [ "set protocols ospfv3 redistribute bgp", "set protocols ospfv3 parameters router-id '192.0.2.10'", "set protocols ospfv3 area 2 range 2001:db10::/32", "set protocols ospfv3 area 2 range 2001:db20::/32", "set protocols ospfv3 area 2 range 2001:db30::/32", "set protocols ospfv3 area '2'", "set protocols ospfv3 area 2 export-list export1", "set protocols ospfv3 area 2 import-list import1", "set protocols ospfv3 area '3'", "set protocols ospfv3 area 3 range 2001:db40::/32", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv3_merged_idem(self): set_module_args( dict( config=dict( areas=[ dict( area_id="12", export_list="export1", import_list="import1", range=[ dict(address="2001:db11::/32"), dict(address="2001:db22::/32"), dict(address="2001:db33::/32"), ], ), dict( area_id="13", range=[dict(address="2001:db44::/32")], ), ], ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospfv3_merged_update_existing(self): set_module_args( dict( config=dict( redistribute=[dict(route_type="bgp")], parameters=dict(router_id="192.0.2.10"), areas=[ dict( area_id="12", export_list="export1", import_list="import1", range=[ dict(address="2001:db11::/32"), dict(address="2001:db22::/32"), dict(address="2001:db33::/32"), ], ), dict( area_id="13", range=[ dict(address="2001:db44::/32"), dict(address="2001:db55::/32"), ], ), ], ), state="merged", ) ) commands = [ "set protocols ospfv3 redistribute bgp", "set protocols ospfv3 parameters router-id '192.0.2.10'", "set protocols ospfv3 area 13 range 2001:db55::/32", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv3_replaced(self): set_module_args( dict( config=dict( redistribute=[dict(route_type="bgp")], parameters=dict(router_id="192.0.2.10"), areas=[ dict( area_id="12", export_list="export1", import_list="import1", range=[ dict(address="2001:db10::/32"), dict(address="2001:db22::/32"), dict(address="2001:db33::/32"), ], ), dict( area_id="14", range=[dict(address="2001:db40::/32")], ), ], ), state="replaced", ) ) commands = [ "set protocols ospfv3 redistribute bgp", "set protocols ospfv3 parameters router-id '192.0.2.10'", "delete protocols ospfv3 area 12 range 2001:db11::/32", "set protocols ospfv3 area 12 range 2001:db10::/32", "delete protocols ospfv3 area 13", "set protocols ospfv3 area '14'", "set protocols ospfv3 area 14 range 2001:db40::/32", ] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv3_replaced_idem(self): set_module_args( dict( config=dict( areas=[ dict( area_id="12", export_list="export1", import_list="import1", range=[ dict(address="2001:db11::/32"), dict(address="2001:db22::/32"), dict(address="2001:db33::/32"), ], ), dict( area_id="13", range=[dict(address="2001:db44::/32")], ), ], ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_ospfv3_deleted_no_config(self): set_module_args(dict(config=None, state="deleted")) commands = ["delete protocols ospfv3"] self.execute_module(changed=True, commands=commands) def test_vyos_ospfv3_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False, filename="vyos_ospfv3_config.cfg") gather_dict = { "areas": [ { "area_id": "12", "export_list": "export1", "import_list": "import1", "range": [ {"address": "2001:db11::/32"}, {"address": "2001:db22::/32"}, {"address": "2001:db33::/32"}, ], }, {"area_id": "13", "range": [{"address": "2001:db44::/32"}]}, ], } self.assertEqual(sorted(gather_dict), sorted(result["gathered"])) def test_vyos_ospfv3_parsed(self): parsed_str = """set protocols ospfv3 area 2 export-list 'export1' set protocols ospfv3 area 2 import-list 'import1' set protocols ospfv3 area 2 range '2001:db10::/32' set protocols ospfv3 area 2 range '2001:db20::/32' set protocols ospfv3 area 2 range '2001:db30::/32' set protocols ospfv3 area 3 range '2001:db40::/32' set protocols ospfv3 parameters router-id '192.0.2.10' set protocols ospfv3 redistribute 'bgp'""" set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_dict = { "areas": [ { "area_id": "2", "export_list": "export1", "import_list": "import1", "range": [ {"address": "2001:db10::/32"}, {"address": "2001:db20::/32"}, {"address": "2001:db30::/32"}, ], }, {"area_id": "3", "range": [{"address": "2001:db40::/32"}]}, ], "parameters": {"router_id": "192.0.2.10"}, "redistribute": [{"route_type": "bgp"}], } self.assertEqual(sorted(parsed_dict), sorted(result["parsed"])) def test_vyos_ospfv3_rendered(self): set_module_args( dict( config=dict( redistribute=[dict(route_type="bgp")], parameters=dict(router_id="192.0.2.10"), areas=[ dict( area_id="2", export_list="export1", import_list="import1", range=[ dict(address="2001:db10::/32"), dict(address="2001:db20::/32"), dict(address="2001:db30::/32"), ], ), dict( area_id="3", range=[dict(address="2001:db40::/32")], ), ], ), state="rendered", ) ) commands = [ "set protocols ospfv3 redistribute bgp", "set protocols ospfv3 parameters router-id '192.0.2.10'", "set protocols ospfv3 area 2 range 2001:db10::/32", "set protocols ospfv3 area 2 range 2001:db20::/32", "set protocols ospfv3 area 2 range 2001:db30::/32", "set protocols ospfv3 area '2'", "set protocols ospfv3 area 2 export-list export1", "set protocols ospfv3 area 2 import-list import1", "set protocols ospfv3 area '3'", "set protocols ospfv3 area 3 range 2001:db40::/32", ] result = self.execute_module(changed=False) self.assertEqual(sorted(result["rendered"]), sorted(commands), result["rendered"]) diff --git a/tests/unit/modules/network/vyos/test_vyos_ping.py b/tests/unit/modules/network/vyos/test_vyos_ping.py index 612e50c..25bb6cf 100644 --- a/tests/unit/modules/network/vyos/test_vyos_ping.py +++ b/tests/unit/modules/network/vyos/test_vyos_ping.py @@ -1,105 +1,106 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_ping -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosPingModule(TestVyosModule): module = vyos_ping def setUp(self): super(TestVyosPingModule, self).setUp() self.mock_run_commands = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_ping.run_commands" ) self.run_commands = self.mock_run_commands.start() def tearDown(self): super(TestVyosPingModule, self).tearDown() self.mock_run_commands.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): commands = kwargs["commands"] output = list() for command in commands: filename = str(command).split(" | ", 1)[0].replace(" ", "_") output.append(load_fixture("vyos_ping_%s" % filename)) return output self.run_commands.side_effect = load_from_file def test_vyos_ping_expected_success(self): """Test for successful pings when destination should be reachable""" set_module_args(dict(count=2, dest="10.10.10.10")) self.execute_module() def test_vyos_ping_expected_failure(self): """Test for unsuccessful pings when destination should not be reachable""" set_module_args(dict(count=4, dest="10.10.10.20", state="absent")) self.execute_module() def test_vyos_ping_unexpected_success(self): """Test for successful pings when destination should not be reachable - FAIL.""" set_module_args(dict(count=2, dest="10.10.10.10", state="absent")) self.execute_module(failed=True) def test_vyos_ping_unexpected_failure(self): """Test for unsuccessful pings when destination should be reachable - FAIL.""" set_module_args(dict(count=4, dest="10.10.10.20")) self.execute_module(failed=True) def test_vyos_ping_failure_stats(self): """Test for asserting stats when ping fails""" set_module_args(dict(count=4, dest="10.10.10.20")) result = self.execute_module(failed=True) self.assertEqual(result["packet_loss"], "100%") self.assertEqual(result["packets_rx"], 0) self.assertEqual(result["packets_tx"], 4) def test_vyos_ping_success_stats(self): """Test for asserting stats when ping passes""" set_module_args(dict(count=2, dest="10.10.10.10")) result = self.execute_module() self.assertEqual(result["packet_loss"], "0%") self.assertEqual(result["packets_rx"], 2) self.assertEqual(result["packets_tx"], 2) self.assertEqual(result["rtt"]["min"], 12) self.assertEqual(result["rtt"]["avg"], 17) self.assertEqual(result["rtt"]["max"], 22) self.assertEqual(result["rtt"]["mdev"], 10) def test_vyos_ping_success_stats_with_options(self): set_module_args(dict(count=10, ttl=128, size=512, dest="10.10.10.11")) result = self.execute_module() self.assertEqual(result["packet_loss"], "0%") self.assertEqual(result["packets_rx"], 10) self.assertEqual(result["packets_tx"], 10) self.assertEqual(result["rtt"]["min"], 1) self.assertEqual(result["rtt"]["avg"], 3) self.assertEqual(result["rtt"]["max"], 21) self.assertEqual(result["rtt"]["mdev"], 5) diff --git a/tests/unit/modules/network/vyos/test_vyos_prefix_lists.py b/tests/unit/modules/network/vyos/test_vyos_prefix_lists.py index dafba2a..5c488ec 100644 --- a/tests/unit/modules/network/vyos/test_vyos_prefix_lists.py +++ b/tests/unit/modules/network/vyos/test_vyos_prefix_lists.py @@ -1,1239 +1,1239 @@ # (c) 2021 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type from textwrap import dedent +from unittest.mock import patch from ansible_collections.vyos.vyos.plugins.modules import vyos_prefix_lists -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule class TestVyosPrefixListsModule(TestVyosModule): # Testing strategy # ------------------ # (a) The unit tests cover `merged` and `replaced` for every attribute. # Since `overridden` is essentially `replaced` but at a larger # scale, these indirectly cover `overridden` as well. # (b) For linear attributes replaced is not valid and hence, those tests # delete the attributes from the config subsection. # (c) The argspec for VRFs is same as the top-level spec and the config logic # is re-used. Hence, those attributes are not explictly covered. However, a # combination of VRF + top-level spec + AF is tested. module = vyos_prefix_lists def setUp(self): super(TestVyosPrefixListsModule, self).setUp() self.mock_get_resource_connection = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection = self.mock_get_resource_connection.start() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.prefix_lists.prefix_lists.Prefix_listsFacts.get_config" ) self.get_config = self.mock_get_config.start() def tearDown(self): super(TestVyosPrefixListsModule, self).tearDown() self.get_resource_connection.stop() self.get_config.stop() # test merged for linear attributes def test_vyos_prefix_lists_linear_merged(self): self.get_config.return_value = dedent( """\ """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", description="Test plist1", entries=[ dict( sequence=10, action="permit", description="Test rule 10", prefix="92.168.10.0/26", ), dict( sequence=20, action="deny", description="Test rule 20", prefix="72.168.2.0/24", ), ], ), dict( name="plist2", entries=[ dict( sequence=20, action="permit", prefix="82.168.10.0/26", le=32, ), dict( sequence=30, action="deny", prefix="62.168.2.0/24", ge=25, ), ], ), ], ), dict( afi="ipv6", prefix_lists=[ dict( name="plist3", description="Test plist3", entries=[ dict( sequence=10, action="deny", description="Test rule 10", prefix="2001:db8:1000::/36", le=36, ), dict( sequence=20, action="permit", description="Test rule 20", prefix="2001:db8:2000::/36", ), ], ), dict( name="plist4", entries=[ dict( sequence=20, action="permit", prefix="2001:db8:3000::/36", ), dict( sequence=50, action="deny", prefix="2001:db8:4000::/36", ), ], ), ], ), ], state="merged", ) ) commands = [ "set policy prefix-list plist1", "set policy prefix-list plist1 description 'Test plist1'", "set policy prefix-list plist1 rule 10", "set policy prefix-list plist1 rule 10 action 'permit'", "set policy prefix-list plist1 rule 10 description 'Test rule 10'", "set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26'", "set policy prefix-list plist1 rule 20", "set policy prefix-list plist1 rule 20 action 'deny'", "set policy prefix-list plist1 rule 20 description 'Test rule 20'", "set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24'", "set policy prefix-list plist2", "set policy prefix-list plist2 rule 20", "set policy prefix-list plist2 rule 20 action 'permit'", "set policy prefix-list plist2 rule 20 le '32'", "set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26'", "set policy prefix-list plist2 rule 30", "set policy prefix-list plist2 rule 30 action 'deny'", "set policy prefix-list plist2 rule 30 ge '25'", "set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24'", "set policy prefix-list6 plist3", "set policy prefix-list6 plist3 description 'Test plist3'", "set policy prefix-list6 plist3 rule 10", "set policy prefix-list6 plist3 rule 10 action 'deny'", "set policy prefix-list6 plist3 rule 10 description 'Test rule 10'", "set policy prefix-list6 plist3 rule 10 le '36'", "set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36'", "set policy prefix-list6 plist3 rule 20", "set policy prefix-list6 plist3 rule 20 action 'permit'", "set policy prefix-list6 plist3 rule 20 description 'Test rule 20'", "set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36'", "set policy prefix-list6 plist4", "set policy prefix-list6 plist4 rule 20", "set policy prefix-list6 plist4 rule 20 action 'permit'", "set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36'", "set policy prefix-list6 plist4 rule 50", "set policy prefix-list6 plist4 rule 50 action 'deny'", "set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36'", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test merged for linear attributes (idempotent) def test_vyos_prefix_lists_linear_merged_idempotent(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", description="Test plist1", entries=[ dict( sequence=10, action="permit", description="Test rule 10", prefix="92.168.10.0/26", ), dict( sequence=20, action="deny", description="Test rule 20", prefix="72.168.2.0/24", ), ], ), dict( name="plist2", entries=[ dict( sequence=20, action="permit", prefix="82.168.10.0/26", le=32, ), dict( sequence=30, action="deny", prefix="62.168.2.0/24", ge=25, ), ], ), ], ), dict( afi="ipv6", prefix_lists=[ dict( name="plist3", description="Test plist3", entries=[ dict( sequence=10, action="deny", description="Test rule 10", prefix="2001:db8:1000::/36", le=36, ), dict( sequence=20, action="permit", description="Test rule 20", prefix="2001:db8:2000::/36", ), ], ), dict( name="plist4", entries=[ dict( sequence=20, action="permit", prefix="2001:db8:3000::/36", ), dict( sequence=50, action="deny", prefix="2001:db8:4000::/36", ), ], ), ], ), ], state="merged", ) ) result = self.execute_module(changed=False) self.assertEqual(result["commands"], []) # test existing rule with replaced def test_vyos_prefix_lists_replaced_update(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", description="Test plist1", entries=[ dict( sequence=10, action="permit", prefix="82.168.10.0/26", ), dict( sequence=20, action="deny", description="Test rule 20", prefix="72.168.2.0/24", ), ], ) ], ) ], state="replaced", ) ) commands = [ "delete policy prefix-list plist1 rule 10 description 'Test rule 10'", "set policy prefix-list plist1 rule 10 prefix '82.168.10.0/26'", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test replaced def test_vyos_prefix_lists_replaced(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", entries=[ dict( sequence=10, action="permit", prefix="82.168.10.0/26", ) ], ) ], ) ], state="replaced", ) ) commands = [ "delete policy prefix-list plist1 description 'Test plist1'", "set policy prefix-list plist1 rule 10 prefix '82.168.10.0/26'", "delete policy prefix-list plist1 rule 20", "delete policy prefix-list plist1 rule 10 description 'Test rule 10'", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test update with overridden def test_vyos_prefix_lists_overridden_update(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", entries=[ dict( sequence=10, action="deny", prefix="102.168.10.0/26", ) ], ) ], ) ], state="overridden", ) ) commands = [ "delete policy prefix-list plist1 description 'Test plist1'", "delete policy prefix-list6 plist4", "delete policy prefix-list plist1 rule 10 description 'Test rule 10'", "set policy prefix-list plist1 rule 10 prefix '102.168.10.0/26'", "delete policy prefix-list6 plist3", "delete policy prefix-list plist1 rule 20", "set policy prefix-list plist1 rule 10 action 'deny'", "delete policy prefix-list plist2", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test overridden def test_vyos_prefix_lists_overridden(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist5", entries=[ dict( sequence=50, action="permit", prefix="102.168.10.0/26", ) ], ) ], ) ], state="overridden", ) ) commands = [ "set policy prefix-list plist5", "set policy prefix-list plist5 rule 50", "set policy prefix-list plist5 rule 50 action 'permit'", "set policy prefix-list plist5 rule 50 prefix '102.168.10.0/26'", "delete policy prefix-list plist1", "delete policy prefix-list plist2", "delete policy prefix-list6 plist3", "delete policy prefix-list6 plist4", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test deleted (all) def test_vyos_prefix_lists_deleted_all(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args(dict(state="deleted")) commands = [ "delete policy prefix-list plist1", "delete policy prefix-list plist2", "delete policy prefix-list6 plist3", "delete policy prefix-list6 plist4", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test deleted (AFI) def test_vyos_prefix_lists_deleted_afi(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args(dict(config=[dict(afi="ipv4")], state="deleted")) commands = [ "delete policy prefix-list plist1", "delete policy prefix-list plist2", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test deleted (one prefix-list) def test_vyos_prefix_lists_deleted_one(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[dict(afi="ipv6", prefix_lists=[dict(name="plist3")])], state="deleted", ) ) commands = ["delete policy prefix-list6 plist3"] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test deleted (one prefix-list from each AFI) def test_vyos_prefix_lists_deleted_one_from_each_afi(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args( dict( config=[ dict(afi="ipv4", prefix_lists=[dict(name="plist2")]), dict(afi="ipv6", prefix_lists=[dict(name="plist3")]), ], state="deleted", ) ) commands = [ "delete policy prefix-list plist2", "delete policy prefix-list6 plist3", ] result = self.execute_module(changed=True) self.assertEqual(set(result["commands"]), set(commands)) # test parsed def test_vyos_prefix_lists_parsed(self): cfg = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args(dict(running_config=cfg, state="parsed")) parsed = [ { "afi": "ipv4", "prefix_lists": [ { "description": "Test plist1", "name": "plist1", "entries": [ { "action": "permit", "description": "Test rule 10", "sequence": 10, "prefix": "92.168.10.0/26", }, { "action": "deny", "description": "Test rule 20", "sequence": 20, "prefix": "72.168.2.0/24", }, ], }, { "name": "plist2", "entries": [ { "action": "permit", "sequence": 20, "le": 32, "prefix": "82.168.10.0/26", }, { "action": "deny", "ge": 25, "sequence": 30, "prefix": "62.168.2.0/24", }, ], }, ], }, { "afi": "ipv6", "prefix_lists": [ { "description": "Test plist3", "name": "plist3", "entries": [ { "action": "deny", "description": "Test rule 10", "sequence": 10, "le": 36, "prefix": "2001:db8:1000::/36", }, { "action": "permit", "description": "Test rule 20", "sequence": 20, "prefix": "2001:db8:2000::/36", }, ], }, { "name": "plist4", "entries": [ { "action": "permit", "sequence": 20, "prefix": "2001:db8:3000::/36", }, { "action": "deny", "sequence": 50, "prefix": "2001:db8:4000::/36", }, ], }, ], }, ] result = self.execute_module(changed=False) self.assertEqual(result["parsed"], parsed) # test rendered def test_vyos_prefix_lists_rendered(self): set_module_args( dict( config=[ dict( afi="ipv4", prefix_lists=[ dict( name="plist1", description="Test plist1", entries=[ dict( sequence=10, action="permit", description="Test rule 10", prefix="92.168.10.0/26", ), dict( sequence=20, action="deny", description="Test rule 20", prefix="72.168.2.0/24", ), ], ), dict( name="plist2", entries=[ dict( sequence=20, action="permit", prefix="82.168.10.0/26", le=32, ), dict( sequence=30, action="deny", prefix="62.168.2.0/24", ge=25, ), ], ), ], ), dict( afi="ipv6", prefix_lists=[ dict( name="plist3", description="Test plist3", entries=[ dict( sequence=10, action="deny", description="Test rule 10", prefix="2001:db8:1000::/36", le=36, ), dict( sequence=20, action="permit", description="Test rule 20", prefix="2001:db8:2000::/36", ), ], ), dict( name="plist4", entries=[ dict( sequence=20, action="permit", prefix="2001:db8:3000::/36", ), dict( sequence=50, action="deny", prefix="2001:db8:4000::/36", ), ], ), ], ), ], state="rendered", ) ) rendered = [ "set policy prefix-list plist1", "set policy prefix-list plist1 description 'Test plist1'", "set policy prefix-list plist1 rule 10", "set policy prefix-list plist1 rule 10 action 'permit'", "set policy prefix-list plist1 rule 10 description 'Test rule 10'", "set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26'", "set policy prefix-list plist1 rule 20", "set policy prefix-list plist1 rule 20 action 'deny'", "set policy prefix-list plist1 rule 20 description 'Test rule 20'", "set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24'", "set policy prefix-list plist2", "set policy prefix-list plist2 rule 20", "set policy prefix-list plist2 rule 20 action 'permit'", "set policy prefix-list plist2 rule 20 le '32'", "set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26'", "set policy prefix-list plist2 rule 30", "set policy prefix-list plist2 rule 30 action 'deny'", "set policy prefix-list plist2 rule 30 ge '25'", "set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24'", "set policy prefix-list6 plist3", "set policy prefix-list6 plist3 description 'Test plist3'", "set policy prefix-list6 plist3 rule 10", "set policy prefix-list6 plist3 rule 10 action 'deny'", "set policy prefix-list6 plist3 rule 10 description 'Test rule 10'", "set policy prefix-list6 plist3 rule 10 le '36'", "set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36'", "set policy prefix-list6 plist3 rule 20", "set policy prefix-list6 plist3 rule 20 action 'permit'", "set policy prefix-list6 plist3 rule 20 description 'Test rule 20'", "set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36'", "set policy prefix-list6 plist4", "set policy prefix-list6 plist4 rule 20", "set policy prefix-list6 plist4 rule 20 action 'permit'", "set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36'", "set policy prefix-list6 plist4 rule 50", "set policy prefix-list6 plist4 rule 50 action 'deny'", "set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36'", ] result = self.execute_module(changed=False) self.assertEqual(set(result["rendered"]), set(rendered)) # test gathered def test_vyos_prefix_lists_gathered(self): self.get_config.return_value = dedent( """\ set policy prefix-list plist1 set policy prefix-list plist1 description 'Test plist1' set policy prefix-list plist1 rule 10 set policy prefix-list plist1 rule 10 action 'permit' set policy prefix-list plist1 rule 10 description 'Test rule 10' set policy prefix-list plist1 rule 10 prefix '92.168.10.0/26' set policy prefix-list plist1 rule 20 set policy prefix-list plist1 rule 20 action 'deny' set policy prefix-list plist1 rule 20 description 'Test rule 20' set policy prefix-list plist1 rule 20 prefix '72.168.2.0/24' set policy prefix-list plist2 set policy prefix-list plist2 rule 20 set policy prefix-list plist2 rule 20 action 'permit' set policy prefix-list plist2 rule 20 le '32' set policy prefix-list plist2 rule 20 prefix '82.168.10.0/26' set policy prefix-list plist2 rule 30 set policy prefix-list plist2 rule 30 action 'deny' set policy prefix-list plist2 rule 30 ge '25' set policy prefix-list plist2 rule 30 prefix '62.168.2.0/24' set policy prefix-list6 plist3 set policy prefix-list6 plist3 description 'Test plist3' set policy prefix-list6 plist3 rule 10 set policy prefix-list6 plist3 rule 10 action 'deny' set policy prefix-list6 plist3 rule 10 description 'Test rule 10' set policy prefix-list6 plist3 rule 10 le '36' set policy prefix-list6 plist3 rule 10 prefix '2001:db8:1000::/36' set policy prefix-list6 plist3 rule 20 set policy prefix-list6 plist3 rule 20 action 'permit' set policy prefix-list6 plist3 rule 20 description 'Test rule 20' set policy prefix-list6 plist3 rule 20 prefix '2001:db8:2000::/36' set policy prefix-list6 plist4 set policy prefix-list6 plist4 rule 20 set policy prefix-list6 plist4 rule 20 action 'permit' set policy prefix-list6 plist4 rule 20 prefix '2001:db8:3000::/36' set policy prefix-list6 plist4 rule 50 set policy prefix-list6 plist4 rule 50 action 'deny' set policy prefix-list6 plist4 rule 50 prefix '2001:db8:4000::/36' """ ) set_module_args(dict(state="gathered")) gathered = [ { "afi": "ipv4", "prefix_lists": [ { "description": "Test plist1", "name": "plist1", "entries": [ { "action": "permit", "description": "Test rule 10", "sequence": 10, "prefix": "92.168.10.0/26", }, { "action": "deny", "description": "Test rule 20", "sequence": 20, "prefix": "72.168.2.0/24", }, ], }, { "name": "plist2", "entries": [ { "action": "permit", "sequence": 20, "le": 32, "prefix": "82.168.10.0/26", }, { "action": "deny", "ge": 25, "sequence": 30, "prefix": "62.168.2.0/24", }, ], }, ], }, { "afi": "ipv6", "prefix_lists": [ { "description": "Test plist3", "name": "plist3", "entries": [ { "action": "deny", "description": "Test rule 10", "sequence": 10, "le": 36, "prefix": "2001:db8:1000::/36", }, { "action": "permit", "description": "Test rule 20", "sequence": 20, "prefix": "2001:db8:2000::/36", }, ], }, { "name": "plist4", "entries": [ { "action": "permit", "sequence": 20, "prefix": "2001:db8:3000::/36", }, { "action": "deny", "sequence": 50, "prefix": "2001:db8:4000::/36", }, ], }, ], }, ] result = self.execute_module(changed=False) self.assertEqual(result["gathered"], gathered) diff --git a/tests/unit/modules/network/vyos/test_vyos_route_maps.py b/tests/unit/modules/network/vyos/test_vyos_route_maps.py index 9d3c71a..adac6c3 100644 --- a/tests/unit/modules/network/vyos/test_vyos_route_maps.py +++ b/tests/unit/modules/network/vyos/test_vyos_route_maps.py @@ -1,582 +1,583 @@ # (c) 2021 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_route_maps -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosRouteMapsModule(TestVyosModule): module = vyos_route_maps def setUp(self): super(TestVyosRouteMapsModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.route_maps.route_maps.Route_mapsFacts.get_config" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosRouteMapsModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_route_maps_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_vyos_route_maps_merged_idempotent(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", metric=1, peer="1.1.1.2", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.20", local_preference=4, metric=5, metric_type="type-1", origin="egp", originator_id="10.0.2.3", src="10.0.2.15", tag=5, weight=4, ), ) ], ) ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_route_maps_merged(self): set_module_args( dict( config=[ dict( route_map="test2", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", metric=1, peer="1.1.1.3", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.22", large_community="10:20:21", local_preference=4, metric=5, metric_type="type-2", origin="egp", originator_id="10.0.2.2", src="10.0.2.15", tag=4, weight=4, ), ) ], ) ], state="merged", ) ) commands = [ "set policy route-map test2 rule 1 action permit", "set policy route-map test2 rule 1 set bgp-extcommunity-rt 22:11", "set policy route-map test2 rule 1 set ip-next-hop 10.20.10.22", "set policy route-map test2 rule 1 set ipv6-next-hop global fdda:5cc1:23:4::1f", "set policy route-map test2 rule 1 set large-community 10:20:21", "set policy route-map test2 rule 1 set local-preference 4", "set policy route-map test2 rule 1 set metric 5", "set policy route-map test2 rule 1 set metric-type type-2", "set policy route-map test2 rule 1 set origin egp", "set policy route-map test2 rule 1 set originator-id 10.0.2.2", "set policy route-map test2 rule 1 set src 10.0.2.15", "set policy route-map test2 rule 1 set tag 4", "set policy route-map test2 rule 1 set weight 4", "set policy route-map test2 rule 1 set community internet", "set policy route-map test2 rule 1 match interface eth2", "set policy route-map test2 rule 1 match metric 1", "set policy route-map test2 rule 1 match peer 1.1.1.3", "set policy route-map test2 rule 1 match ipv6 nexthop fdda:5cc1:23:4::1f", "set policy route-map test2 rule 1 match rpki invalid", ] self.execute_module(changed=True, commands=commands) def test_route_maps_replaced(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", metric=1, peer="1.1.1.3", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.22", large_community="10:20:21", local_preference=4, metric=5, metric_type="type-2", origin="egp", originator_id="10.0.2.2", src="10.0.2.15", tag=4, weight=4, ), ) ], ), ], state="replaced", ) ) commands = [ "delete policy route-map test3 rule 1 match interface eth2", "set policy route-map test3 rule 1 set ip-next-hop 10.20.10.22", "set policy route-map test3 rule 1 set large-community 10:20:21", "set policy route-map test3 rule 1 set metric-type type-2", "set policy route-map test3 rule 1 set originator-id 10.0.2.2", "set policy route-map test3 rule 1 set tag 4", "set policy route-map test3 rule 1 match peer 1.1.1.3", ] self.execute_module(changed=True, commands=commands) def test_vyos_route_maps_replaced_idempotent(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", metric=1, peer="1.1.1.2", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.20", local_preference=4, metric=5, metric_type="type-1", origin="egp", originator_id="10.0.2.3", src="10.0.2.15", tag=5, weight=4, ), ) ], ), ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_route_maps_overridden(self): set_module_args( dict( config=[ dict( route_map="test2", entries=[ dict( sequence=1, action="permit", match=dict(rpki="invalid", peer="1.1.1.3"), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.22", large_community="10:20:21", local_preference=4, metric=5, metric_type="type-2", origin="egp", originator_id="10.0.2.2", src="10.0.2.15", tag=4, weight=4, ), ) ], ) ], state="overridden", ) ) commands = [ "delete policy route-map test3", "set policy route-map test2 rule 1 action permit", "set policy route-map test2 rule 1 set bgp-extcommunity-rt 22:11", "set policy route-map test2 rule 1 set ip-next-hop 10.20.10.22", "set policy route-map test2 rule 1 set ipv6-next-hop global fdda:5cc1:23:4::1f", "set policy route-map test2 rule 1 set large-community 10:20:21", "set policy route-map test2 rule 1 set local-preference 4", "set policy route-map test2 rule 1 set metric 5", "set policy route-map test2 rule 1 set metric-type type-2", "set policy route-map test2 rule 1 set origin egp", "set policy route-map test2 rule 1 set originator-id 10.0.2.2", "set policy route-map test2 rule 1 set src 10.0.2.15", "set policy route-map test2 rule 1 set tag 4", "set policy route-map test2 rule 1 set weight 4", "set policy route-map test2 rule 1 set community internet", "set policy route-map test2 rule 1 match peer 1.1.1.3", "set policy route-map test2 rule 1 match rpki invalid", ] self.execute_module(changed=True, commands=commands) def test_vyos_route_maps_overridden_idempotent(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", metric=1, peer="1.1.1.2", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.20", local_preference=4, metric=5, metric_type="type-1", origin="egp", originator_id="10.0.2.3", src="10.0.2.15", tag=5, weight=4, ), ) ], ), ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_route_maps_rendered(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", metric=1, peer="1.1.1.2", ipv6=dict(next_hop="fdda:5cc1:23:4::1f"), ), set=dict( ipv6_next_hop=dict( ip_type="global", value="fdda:5cc1:23:4::1f", ), community=dict(value="internet"), bgp_extcommunity_rt="22:11", ip_next_hop="10.20.10.20", local_preference=4, metric=5, metric_type="type-1", origin="egp", originator_id="10.0.2.3", src="10.0.2.15", tag=5, weight=4, ), ) ], ), dict( route_map="test1", entries=[ dict( sequence=1, action="permit", description="test", on_match=dict(next=True), ), dict( sequence=2, action="permit", on_match=dict(goto=4), ), ], ), ], state="rendered", ) ) rendered_cmds = [ "set policy route-map test3 rule 1 action permit", "set policy route-map test3 rule 1 set bgp-extcommunity-rt 22:11", "set policy route-map test3 rule 1 set ip-next-hop 10.20.10.20", "set policy route-map test3 rule 1 set ipv6-next-hop global fdda:5cc1:23:4::1f", "set policy route-map test3 rule 1 set local-preference 4", "set policy route-map test3 rule 1 set metric 5", "set policy route-map test3 rule 1 set metric-type type-1", "set policy route-map test3 rule 1 set origin egp", "set policy route-map test3 rule 1 set originator-id 10.0.2.3", "set policy route-map test3 rule 1 set src 10.0.2.15", "set policy route-map test3 rule 1 set tag 5", "set policy route-map test3 rule 1 set weight 4", "set policy route-map test3 rule 1 set community internet", "set policy route-map test3 rule 1 match interface eth2", "set policy route-map test3 rule 1 match metric 1", "set policy route-map test3 rule 1 match peer 1.1.1.2", "set policy route-map test3 rule 1 match ipv6 nexthop fdda:5cc1:23:4::1f", "set policy route-map test3 rule 1 match rpki invalid", "set policy route-map test1 rule 1 description test", "set policy route-map test1 rule 1 action permit", "set policy route-map test1 rule 1 on-match next", "set policy route-map test1 rule 2 action permit", "set policy route-map test1 rule 2 on-match goto 4", ] result = self.execute_module(changed=False) self.assertEqual( sorted(result["rendered"]), sorted(rendered_cmds), result["rendered"], ) def test_yos_route_maps_parsed(self): parsed_str = ( "set policy route-map test3 rule 1 action 'permit'" "\nset policy route-map test3 rule 1 match interface 'eth2'\nset policy route-map test3 rule 1 match ipv6 nexthop" " 'fdda:5cc1:23:4::1f'\nset policy route-map test3 rule 1 match metric '1'\nset policy route-map test3 rule 1 match peer " "'1.1.1.2'\nset policy route-map test3 rule 1 match rpki 'invalid'\nset policy route-map test3 rule 1 set bgp-extcommunity-rt " "'22:11'\nset policy route-map test3 rule 1 set community 'internet'\nset policy route-map test3 rule 1 set ipv6-next-hop global" " 'fdda:5cc1:23:4::1f'\nset policy route-map test3 rule 1 set ip-next-hop '10.20.10.20'\nset policy route-map " "test3 rule 1 set local-preference '4'\nset policy route-map test3 rule 1 set metric '5'\nset policy route-map test3 " "rule 1 set metric-type 'type-1'\nset policy route-map test3 rule 1 set origin 'egp'\nset policy route-map test3 rule 1 set originator-id " "'10.0.2.3'\nset policy route-map test3 rule 1 set src '10.0.2.15'" "\nset policy route-map test3 rule 1 set tag '5'\nset policy route-map test3 rule 1 set weight '4'" ) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = [ { "entries": [ { "action": "permit", "match": { "interface": "eth2", "ipv6": {"next_hop": "fdda:5cc1:23:4::1f"}, "metric": 1, "peer": "1.1.1.2", "rpki": "invalid", }, "sequence": 1, "set": { "bgp_extcommunity_rt": "22:11", "community": {"value": "internet"}, "ip_next_hop": "10.20.10.20", "ipv6_next_hop": { "ip_type": "global", "value": "fdda:5cc1:23:4::1f", }, "local_preference": "4", "metric": "5", "metric_type": "type-1", "origin": "egp", "originator_id": "10.0.2.3", "src": "10.0.2.15", "tag": "5", "weight": "4", }, } ], "route_map": "test3", }, ] self.assertEqual(parsed_list, result["parsed"]) def test_vyos_route_maps_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False) gathered_list = [ { "entries": [ { "action": "permit", "match": { "interface": "eth2", "ipv6": {"next_hop": "fdda:5cc1:23:4::1f"}, "metric": 1, "peer": "1.1.1.2", "rpki": "invalid", }, "sequence": 1, "set": { "bgp_extcommunity_rt": "22:11", "community": {"value": "internet"}, "ip_next_hop": "10.20.10.20", "ipv6_next_hop": { "ip_type": "global", "value": "fdda:5cc1:23:4::1f", }, "local_preference": "4", "metric": "5", "metric_type": "type-1", "origin": "egp", "originator_id": "10.0.2.3", "src": "10.0.2.15", "tag": "5", "weight": "4", }, } ], "route_map": "test3", }, ] self.assertEqual(gathered_list, result["gathered"]) def test_vyos_route_maps_deleted(self): set_module_args( dict( config=[ dict( route_map="test3", entries=[ dict( sequence=1, action="permit", match=dict( rpki="invalid", interface="eth2", ), set=dict( origin="egp", originator_id="10.0.2.3", src="10.0.2.15", tag=5, weight=4, ), ) ], ), ], state="deleted", ) ) commands = ["delete policy route-map test3"] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_snmp_server.py b/tests/unit/modules/network/vyos/test_vyos_snmp_server.py index 73e29f6..b6d61b1 100644 --- a/tests/unit/modules/network/vyos/test_vyos_snmp_server.py +++ b/tests/unit/modules/network/vyos/test_vyos_snmp_server.py @@ -1,517 +1,518 @@ # (c) 2021 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_snmp_server -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosSnmpServerModule(TestVyosModule): module = vyos_snmp_server def setUp(self): super(TestVyosSnmpServerModule, self).setUp() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module_base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.snmp_server.snmp_server.Snmp_serverFacts.get_config" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosSnmpServerModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): if filename is None: filename = "vyos_snmp_server_config.cfg" def load_from_file(*args, **kwargs): output = load_fixture(filename) return output self.execute_show_command.side_effect = load_from_file def test_snmp_server_merged_idempotent(self): set_module_args( dict( config=dict( communities=[ dict( name="bridges", networks=["12.1.1.0/24", "1.1.1.0/24"], ) ], listen_addresses=[ dict(address="100.1.2.1", port=33), ], location="RDU, NC", snmp_v3=dict( users=[ dict( user="admin_user", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), dict( user="guest_user", authentication=dict(type="sha", plaintext_key="opq1234567"), privacy=dict(type="aes", plaintext_key="opq1234567"), ), ] ), ), state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_snmp_server_replaced_idempotent(self): set_module_args( dict( config=dict( communities=[ dict( name="bridges", networks=["12.1.1.0/24", "1.1.1.0/24"], ) ], listen_addresses=[ dict(address="100.1.2.1", port=33), ], location="RDU, NC", snmp_v3=dict( users=[ dict( user="admin_user", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), dict( user="guest_user", authentication=dict(type="sha", plaintext_key="opq1234567"), privacy=dict(type="aes", plaintext_key="opq1234567"), ), ] ), ), state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_snmp_server_overridden_idempotent(self): set_module_args( dict( config=dict( communities=[ dict( name="bridges", networks=["12.1.1.0/24", "1.1.1.0/24"], ) ], listen_addresses=[ dict(address="100.1.2.1", port=33), ], location="RDU, NC", snmp_v3=dict( users=[ dict( user="admin_user", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), dict( user="guest_user", authentication=dict(type="sha", plaintext_key="opq1234567"), privacy=dict(type="aes", plaintext_key="opq1234567"), ), ] ), ), state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_snmp_server_merged(self): set_module_args( dict( config=dict( communities=[ dict( name="routers", clients=["12.1.1.0/24", "1.1.1.0/24"], authorization_type="rw", ), dict(name="switches", authorization_type="ro"), ], contact="admin@example.com", description="snmp_config", smux_peer="peer1", trap_source="1.1.1.1", trap_target=dict(address="10.10.1.1", community="switches", port="80"), snmp_v3=dict( engine_id="34", groups=[ dict( group="default", mode="rw", seclevel="priv", view="view1", ) ], trap_targets=[ dict( address="20.12.1.1", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), ], ), ), state="merged", ) ) commands = [ "set service snmp community routers client 1.1.1.0/24", "set service snmp community routers client 12.1.1.0/24", "set service snmp community routers authorization rw", "set service snmp community switches authorization ro", "set service snmp v3 group default mode rw", "set service snmp v3 group default seclevel priv", "set service snmp v3 group default view view1", "set service snmp v3 engineid 34", "set service snmp contact admin@example.com", "set service snmp description snmp_config", "set service snmp smux-peer peer1", "set service snmp trap-source 1.1.1.1", "set service snmp trap-target 10.10.1.1", ] self.execute_module(changed=True, commands=commands) def test_snmp_server_replaced(self): set_module_args( dict( config=dict( communities=[ dict( name="routers", clients=["12.1.1.0/24", "1.1.1.0/24"], authorization_type="rw", ), dict(name="switches", authorization_type="ro"), ], contact="admin@example.com", description="snmp_config", smux_peer="peer1", trap_source="1.1.1.1", trap_target=dict(address="10.10.1.1", community="switches", port="80"), snmp_v3=dict( engine_id="34", groups=[ dict( group="default", mode="rw", seclevel="priv", view="view1", ) ], trap_targets=[ dict( address="20.12.1.1", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), ], ), ), state="replaced", ) ) commands = [ "set service snmp community routers client 1.1.1.0/24", "set service snmp community routers client 12.1.1.0/24", "set service snmp community routers authorization rw", "set service snmp community switches authorization ro", "delete service snmp community bridges network 1.1.1.0/24", "delete service snmp community bridges network 12.1.1.0/24", "delete service snmp listen-address 100.1.2.1 port 33", "set service snmp v3 group default mode rw", "set service snmp v3 group default seclevel priv", "set service snmp v3 group default view view1", "delete service snmp v3 user admin_user auth type sha", "delete service snmp v3 user admin_user auth plaintext-key abc1234567", "delete service snmp v3 user admin_user privacy type aes", "delete service snmp v3 user admin_user privacy plaintext-key abc1234567", "delete service snmp v3 user guest_user auth type sha", "delete service snmp v3 user guest_user auth plaintext-key opq1234567", "delete service snmp v3 user guest_user privacy type aes", "delete service snmp v3 user guest_user privacy plaintext-key opq1234567", "set service snmp v3 engineid 34", "set service snmp contact admin@example.com", "set service snmp description snmp_config", "set service snmp smux-peer peer1", "set service snmp trap-source 1.1.1.1", "set service snmp trap-target 10.10.1.1", "delete service snmp location 'RDU, NC'", ] self.execute_module(changed=True, commands=commands) def test_snmp_server_overridden(self): set_module_args( dict( config=dict( communities=[ dict( name="routers", clients=["12.1.1.0/24", "1.1.1.0/24"], authorization_type="rw", ), dict(name="switches", authorization_type="ro"), ], contact="admin@example.com", description="snmp_config", smux_peer="peer1", trap_source="1.1.1.1", trap_target=dict(address="10.10.1.1", community="switches", port="80"), snmp_v3=dict( engine_id="34", groups=[ dict( group="default", mode="rw", seclevel="priv", view="view1", ) ], trap_targets=[ dict( address="20.12.1.1", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), ], ), ), state="overridden", ) ) commands = [ "set service snmp community routers client 1.1.1.0/24", "set service snmp community routers client 12.1.1.0/24", "set service snmp community routers authorization rw", "set service snmp community switches authorization ro", "delete service snmp community bridges network 1.1.1.0/24", "delete service snmp community bridges network 12.1.1.0/24", "delete service snmp listen-address 100.1.2.1 port 33", "set service snmp v3 group default mode rw", "set service snmp v3 group default seclevel priv", "set service snmp v3 group default view view1", "delete service snmp v3 user admin_user auth type sha", "delete service snmp v3 user admin_user auth plaintext-key abc1234567", "delete service snmp v3 user admin_user privacy type aes", "delete service snmp v3 user admin_user privacy plaintext-key abc1234567", "delete service snmp v3 user guest_user auth type sha", "delete service snmp v3 user guest_user auth plaintext-key opq1234567", "delete service snmp v3 user guest_user privacy type aes", "delete service snmp v3 user guest_user privacy plaintext-key opq1234567", "set service snmp v3 engineid 34", "set service snmp contact admin@example.com", "set service snmp description snmp_config", "set service snmp smux-peer peer1", "set service snmp trap-source 1.1.1.1", "set service snmp trap-target 10.10.1.1", "delete service snmp location 'RDU, NC'", ] self.execute_module(changed=True, commands=commands) def test_snmp_server_deleted(self): set_module_args( dict( state="deleted", ) ) commands = ["delete service snmp"] self.execute_module(changed=True, commands=commands) def test_snmp_server_rendered(self): set_module_args( dict( config=dict( communities=[ dict( name="routers", clients=["12.1.1.0/24", "1.1.1.0/24"], authorization_type="rw", ), dict(name="switches", authorization_type="ro"), ], contact="admin@example.com", description="snmp_config", smux_peer="peer1", trap_source="1.1.1.1", trap_target=dict(address="10.10.1.1", community="switches", port="80"), snmp_v3=dict( engine_id="34", groups=[ dict( group="default", mode="rw", seclevel="priv", view="view1", ) ], trap_targets=[ dict( address="20.12.1.1", authentication=dict(type="sha", plaintext_key="abc1234567"), privacy=dict(type="aes", plaintext_key="abc1234567"), ), ], ), ), state="rendered", ) ) commands = [ "set service snmp community routers client 1.1.1.0/24", "set service snmp community routers client 12.1.1.0/24", "set service snmp community routers authorization rw", "set service snmp community switches authorization ro", "set service snmp v3 group default mode rw", "set service snmp v3 group default seclevel priv", "set service snmp v3 group default view view1", "set service snmp v3 engineid 34", "set service snmp contact admin@example.com", "set service snmp description snmp_config", "set service snmp smux-peer peer1", "set service snmp trap-source 1.1.1.1", "set service snmp trap-target 10.10.1.1", ] result = self.execute_module(changed=False) self.assertEqual( sorted(result["rendered"]), sorted(commands), result["rendered"], ) def test_snmp_server_parsed(self): commands = [ "set service snmp community routers client 1.1.1.0/24", "set service snmp community routers client 12.1.1.0/24", "set service snmp community routers authorization rw", "set service snmp community switches authorization ro", "set service snmp v3 group default mode rw", "set service snmp v3 group default seclevel priv", "set service snmp v3 group default view view1", "set service snmp v3 engineid 34", "set service snmp contact admin@example.com", "set service snmp description snmp_config", "set service snmp smux-peer peer1", "set service snmp trap-source 1.1.1.1", "set service snmp trap-target 10.10.1.1", ] parsed_str = "\n".join(commands) set_module_args(dict(running_config=parsed_str, state="parsed")) result = self.execute_module(changed=False) parsed_list = { "communities": [ { "authorization_type": "rw", "clients": ["1.1.1.0/24", "12.1.1.0/24"], "name": "routers", }, {"authorization_type": "ro", "name": "switches"}, ], "contact": "admin@example.com", "description": "snmp_config", "smux_peer": "peer1", "snmp_v3": { "engine_id": "34", "groups": [ { "group": "default", "mode": "rw", "seclevel": "priv", "view": "view1", } ], }, "trap_source": "1.1.1.1", "trap_target": {"address": "10.10.1.1"}, } self.assertEqual(parsed_list, result["parsed"]) def test_snmp_server_gathered(self): set_module_args(dict(state="gathered")) result = self.execute_module(changed=False) gathered_list = { "communities": [ {"name": "bridges", "networks": ["1.1.1.0/24", "12.1.1.0/24"]}, ], "listen_addresses": [{"address": "100.1.2.1", "port": 33}], "location": "RDU, NC", "snmp_v3": { "users": [ { "authentication": { "plaintext_key": "abc1234567", "type": "sha", }, "privacy": { "plaintext_key": "abc1234567", "type": "aes", }, "user": "admin_user", }, { "authentication": { "plaintext_key": "opq1234567", "type": "sha", }, "privacy": { "plaintext_key": "opq1234567", "type": "aes", }, "user": "guest_user", }, ] }, } self.assertEqual(gathered_list, result["gathered"]) diff --git a/tests/unit/modules/network/vyos/test_vyos_static_routes.py b/tests/unit/modules/network/vyos/test_vyos_static_routes.py index 3479009..96137bd 100644 --- a/tests/unit/modules/network/vyos/test_vyos_static_routes.py +++ b/tests/unit/modules/network/vyos/test_vyos_static_routes.py @@ -1,271 +1,272 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_static_routes -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosStaticRoutesModule(TestVyosModule): module = vyos_static_routes def setUp(self): super(TestVyosStaticRoutesModule, self).setUp() self.mock_get_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.network.Config.load_config" ) self.load_config = self.mock_load_config.start() self.mock_get_resource_connection_config = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base.get_resource_connection" ) self.get_resource_connection_config = self.mock_get_resource_connection_config.start() self.mock_get_resource_connection_facts = patch( "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.facts.facts.get_resource_connection" ) self.get_resource_connection_facts = self.mock_get_resource_connection_facts.start() self.mock_execute_show_command = patch( "ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.static_routes.static_routes.Static_routesFacts.get_device_data" ) self.execute_show_command = self.mock_execute_show_command.start() def tearDown(self): super(TestVyosStaticRoutesModule, self).tearDown() self.mock_get_resource_connection_config.stop() self.mock_get_resource_connection_facts.stop() self.mock_get_config.stop() self.mock_load_config.stop() self.mock_execute_show_command.stop() def load_fixtures(self, commands=None, filename=None): def load_from_file(*args, **kwargs): return load_fixture("vyos_static_routes_config.cfg") self.execute_show_command.side_effect = load_from_file def test_vyos_static_routes_merged(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.48/28", next_hops=[ dict( forward_router_address="192.0.2.9", admin_distance=10, ), dict( forward_router_address="192.0.2.10", interface="eth0", ), ], ) ], ) ] ) ], state="merged", ) ) commands = [ "set protocols static route 192.0.2.48/28", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.9'", "set protocols static route 192.0.2.48/28 next-hop 192.0.2.9 distance '10'", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.10'", "set protocols static route 192.0.2.48/28 next-hop 192.0.2.10 next-hop-interface 'eth0'", ] self.execute_module(changed=True, commands=commands) def test_vyos_static_routes_merged_idempotent(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.32/28", next_hops=[ dict(forward_router_address="192.0.2.9"), dict(forward_router_address="192.0.2.10"), ], ) ], ) ] ) ], state="merged", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_static_routes_replaced(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.48/28", next_hops=[ dict( forward_router_address="192.0.2.9", interface="eth0", ), dict( forward_router_address="192.0.2.10", admin_distance=10, ), ], ) ], ) ] ) ], state="replaced", ) ) commands = [ "set protocols static route 192.0.2.48/28", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.9'", "set protocols static route 192.0.2.48/28 next-hop 192.0.2.9 next-hop-interface 'eth0'", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.10'", "set protocols static route 192.0.2.48/28 next-hop 192.0.2.10 distance '10'", ] self.execute_module(changed=True, commands=commands) def test_vyos_static_routes_replaced_idempotent(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.32/28", next_hops=[ dict(forward_router_address="192.0.2.9"), dict(forward_router_address="192.0.2.10"), ], ) ], ) ] ) ], state="replaced", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_static_routes_overridden(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.48/28", next_hops=[ dict(forward_router_address="192.0.2.9"), dict(forward_router_address="192.0.2.10"), ], ) ], ) ] ) ], state="overridden", ) ) commands = [ "delete protocols static route 192.0.2.32/28", "set protocols static route 192.0.2.48/28", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.9'", "set protocols static route 192.0.2.48/28 next-hop '192.0.2.10'", ] self.execute_module(changed=True, commands=commands) def test_vyos_static_routes_overridden_idempotent(self): set_module_args( dict( config=[ dict( address_families=[ dict( afi="ipv4", routes=[ dict( dest="192.0.2.32/28", next_hops=[ dict(forward_router_address="192.0.2.9"), dict(forward_router_address="192.0.2.10"), ], ) ], ) ] ) ], state="overridden", ) ) self.execute_module(changed=False, commands=[]) def test_vyos_static_routes_deleted(self): set_module_args( dict( config=[dict(address_families=[dict(afi="ipv4")])], state="deleted", ) ) commands = ["delete protocols static route"] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_system.py b/tests/unit/modules/network/vyos/test_vyos_system.py index 03ef3df..252172d 100644 --- a/tests/unit/modules/network/vyos/test_vyos_system.py +++ b/tests/unit/modules/network/vyos/test_vyos_system.py @@ -1,112 +1,113 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_system -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosSystemModule(TestVyosModule): module = vyos_system def setUp(self): super(TestVyosSystemModule, self).setUp() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_system.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_system.load_config" ) self.load_config = self.mock_load_config.start() def tearDown(self): super(TestVyosSystemModule, self).tearDown() self.mock_get_config.stop() self.mock_load_config.stop() def load_fixtures(self, commands=None, filename=None): self.get_config.return_value = load_fixture("vyos_config_config.cfg") def test_vyos_system_hostname(self): set_module_args(dict(host_name="foo")) commands = ["set system host-name 'foo'"] self.execute_module(changed=True, commands=commands) def test_vyos_system_clear_hostname(self): set_module_args(dict(host_name="foo", state="absent")) commands = ["delete system host-name"] self.execute_module(changed=True, commands=commands) def test_vyos_remove_single_name_server(self): set_module_args(dict(name_server=["8.8.4.4"], state="absent")) commands = ["delete system name-server '8.8.4.4'"] self.execute_module(changed=True, commands=commands) def test_vyos_system_domain_name(self): set_module_args(dict(domain_name="example2.com")) commands = ["set system domain-name 'example2.com'"] self.execute_module(changed=True, commands=commands) def test_vyos_system_clear_domain_name(self): set_module_args(dict(domain_name="example.com", state="absent")) commands = ["delete system domain-name"] self.execute_module(changed=True, commands=commands) def test_vyos_system_domain_search(self): set_module_args(dict(domain_search=["foo.example.com", "bar.example.com"])) commands = [ "set system domain-search domain 'foo.example.com'", "set system domain-search domain 'bar.example.com'", ] self.execute_module(changed=True, commands=commands) def test_vyos_system_clear_domain_search(self): set_module_args(dict(domain_search=[])) commands = ["delete system domain-search domain"] self.execute_module(changed=True, commands=commands) def test_vyos_system_no_change(self): set_module_args( dict( host_name="router", domain_name="example.com", name_server=["8.8.8.8", "8.8.4.4"], ) ) result = self.execute_module() self.assertEqual([], result["commands"]) def test_vyos_system_clear_all(self): set_module_args(dict(state="absent")) commands = [ "delete system host-name", "delete system domain-search domain", "delete system domain-name", "delete system name-server", ] self.execute_module(changed=True, commands=commands) diff --git a/tests/unit/modules/network/vyos/test_vyos_user.py b/tests/unit/modules/network/vyos/test_vyos_user.py index 57639d1..2387296 100644 --- a/tests/unit/modules/network/vyos/test_vyos_user.py +++ b/tests/unit/modules/network/vyos/test_vyos_user.py @@ -1,129 +1,130 @@ # (c) 2016 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 . # Make coding more python3-ish from __future__ import absolute_import, division, print_function __metaclass__ = type +from unittest.mock import patch + from ansible_collections.vyos.vyos.plugins.modules import vyos_user -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch from ansible_collections.vyos.vyos.tests.unit.modules.utils import set_module_args from .vyos_module import TestVyosModule, load_fixture class TestVyosUserModule(TestVyosModule): module = vyos_user def setUp(self): super(TestVyosUserModule, self).setUp() self.mock_get_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_user.get_config" ) self.get_config = self.mock_get_config.start() self.mock_load_config = patch( "ansible_collections.vyos.vyos.plugins.modules.vyos_user.load_config" ) self.load_config = self.mock_load_config.start() def tearDown(self): super(TestVyosUserModule, self).tearDown() self.mock_get_config.stop() self.mock_load_config.stop() def load_fixtures(self, commands=None, filename=None): self.get_config.return_value = load_fixture("vyos_user_config.cfg") self.load_config.return_value = dict(diff=None, session="session") def test_vyos_user_password(self): set_module_args(dict(name="ansible", configured_password="test")) result = self.execute_module(changed=True) self.assertEqual( result["commands"], ["set system login user ansible authentication plaintext-password test"], ) def test_vyos_user_delete(self): set_module_args(dict(name="ansible", state="absent")) result = self.execute_module(changed=True) self.assertEqual(result["commands"], ["delete system login user ansible"]) def test_vyos_user_level(self): set_module_args(dict(name="ansible", level="operator")) result = self.execute_module(changed=True) self.assertEqual( result["commands"], ["set system login user ansible level operator"], ) def test_vyos_user_level_invalid(self): set_module_args(dict(name="ansible", level="sysadmin")) self.execute_module(failed=True) def test_vyos_user_purge(self): set_module_args(dict(purge=True)) result = self.execute_module(changed=True) self.assertEqual( sorted(result["commands"]), sorted( [ "delete system login user ansible", "delete system login user admin", ] ), ) def test_vyos_user_update_password_changed(self): set_module_args( dict( name="test", configured_password="test", update_password="on_create", ) ) result = self.execute_module(changed=True) self.assertEqual( result["commands"], ["set system login user test authentication plaintext-password test"], ) def test_vyos_user_update_password_on_create_ok(self): set_module_args( dict( name="ansible", configured_password="test", update_password="on_create", ) ) self.execute_module() def test_vyos_user_update_password_always(self): set_module_args( dict( name="ansible", configured_password="test", update_password="always", ) ) result = self.execute_module(changed=True) self.assertEqual( result["commands"], ["set system login user ansible authentication plaintext-password test"], ) diff --git a/tests/unit/modules/utils.py b/tests/unit/modules/utils.py index 779c68f..a7dd0b3 100644 --- a/tests/unit/modules/utils.py +++ b/tests/unit/modules/utils.py @@ -1,52 +1,51 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type import json +import unittest +from unittest.mock import patch from ansible.module_utils import basic from ansible.module_utils._text import to_bytes -from ansible_collections.vyos.vyos.tests.unit.compat import unittest -from ansible_collections.vyos.vyos.tests.unit.compat.mock import patch - def set_module_args(args): if "_ansible_remote_tmp" not in args: args["_ansible_remote_tmp"] = "/tmp" if "_ansible_keep_remote_files" not in args: args["_ansible_keep_remote_files"] = False args = json.dumps({"ANSIBLE_MODULE_ARGS": args}) basic._ANSIBLE_ARGS = to_bytes(args) class AnsibleExitJson(Exception): pass class AnsibleFailJson(Exception): pass def exit_json(*args, **kwargs): if "changed" not in kwargs: kwargs["changed"] = False raise AnsibleExitJson(kwargs) def fail_json(*args, **kwargs): kwargs["failed"] = True raise AnsibleFailJson(kwargs) class ModuleTestCase(unittest.TestCase): def setUp(self): self.mock_module = patch.multiple( basic.AnsibleModule, exit_json=exit_json, fail_json=fail_json ) self.mock_module.start() self.mock_sleep = patch("time.sleep") self.mock_sleep.start() set_module_args({}) self.addCleanup(self.mock_module.stop) self.addCleanup(self.mock_sleep.stop)