diff options
Diffstat (limited to 'tests/unit/modules')
| -rw-r--r-- | tests/unit/modules/test_vyos_interfaces.py | 315 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_l3_interfaces.py | 306 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_lag_interfaces.py | 293 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_lldp_interfaces.py | 291 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_logging_global.py | 10 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospf_interfaces.py | 148 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospfv2.py | 180 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospfv3.py | 140 |
8 files changed, 1678 insertions, 5 deletions
diff --git a/tests/unit/modules/test_vyos_interfaces.py b/tests/unit/modules/test_vyos_interfaces.py new file mode 100644 index 0000000..d522c81 --- /dev/null +++ b/tests/unit/modules/test_vyos_interfaces.py @@ -0,0 +1,315 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_interfaces import ( + _delete_iface_config, + _iface_base, + _iface_cmds, + _iface_type, + build_commands, + get_running_config, +) + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + path = os.path.join(fixtures_dir, filename) + with open(path) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.mock_vyos.get_config = MagicMock(return_value={}) + + def set_running_config(self, data): + self.mock_vyos.get_config.return_value = data + + +class TestVyOSInterfacesIfaceType(unittest.TestCase): + + def test_eth_is_ethernet(self): + self.assertEqual(_iface_type("eth0"), "ethernet") + + def test_bond_is_bonding(self): + self.assertEqual(_iface_type("bond0"), "bonding") + + def test_lo_is_loopback(self): + self.assertEqual(_iface_type("lo"), "loopback") + + def test_wg_is_wireguard(self): + self.assertEqual(_iface_type("wg0"), "wireguard") + + def test_br_is_bridge(self): + self.assertEqual(_iface_type("br0"), "bridge") + + def test_unknown_defaults_to_ethernet(self): + self.assertEqual(_iface_type("xyz0"), "ethernet") + + def test_iface_base_ethernet(self): + self.assertEqual( + _iface_base("eth0"), + ["interfaces", "ethernet", "eth0"], + ) + + def test_iface_base_loopback(self): + self.assertEqual( + _iface_base("lo"), + ["interfaces", "loopback", "lo"], + ) + + +class TestVyOSInterfacesGetRunningFixture(VyOSModuleTestCase): + + def setUp(self): + super().setUp() + self.fixture = load_fixture("interfaces_running.json") + + def test_fixture_parses_eth0(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertEqual(eth0["description"], "Management") + self.assertEqual(eth0["mtu"], 1500) + self.assertEqual(eth0["duplex"], "auto") + self.assertEqual(eth0["speed"], "auto") + self.assertTrue(eth0["enabled"]) + + def test_fixture_parses_loopback(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + lo = next(e for e in result if e["name"] == "lo") + self.assertTrue(lo["enabled"]) + self.assertNotIn("description", lo) + + def test_fixture_disabled_interface(self): + fixture = dict(self.fixture) + fixture["ethernet"]["eth1"] = {"disable": {}, "description": "Unused"} + self.set_running_config(fixture) + result = get_running_config(self.mock_vyos) + eth1 = next(e for e in result if e["name"] == "eth1") + self.assertFalse(eth1["enabled"]) + + +class TestVyOSInterfacesGetRunning(VyOSModuleTestCase): + + def test_empty_returns_empty_list(self): + self.set_running_config({}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + def test_mtu_cast_to_int(self): + self.set_running_config( + { + "ethernet": { + "eth0": {"mtu": "1500"}, + }, + }, + ) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertIsInstance(eth0["mtu"], int) + self.assertEqual(eth0["mtu"], 1500) + + def test_hw_id_not_included(self): + self.set_running_config( + { + "ethernet": { + "eth0": {"hw-id": "52:54:00:65:5a:24"}, + }, + }, + ) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertNotIn("hw_id", eth0) + self.assertNotIn("hw-id", eth0) + + def test_enabled_true_when_no_disable(self): + self.set_running_config( + { + "ethernet": {"eth0": {}}, + }, + ) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertTrue(eth0["enabled"]) + + def test_enabled_false_when_disable_present(self): + self.set_running_config( + { + "ethernet": {"eth0": {"disable": {}}}, + }, + ) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertFalse(eth0["enabled"]) + + +class TestVyOSInterfacesIfaceCmds(unittest.TestCase): + + def test_set_description(self): + want = {"description": "WAN", "enabled": True} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "description", "WAN"]), + cmds, + ) + + def test_set_mtu(self): + want = {"mtu": 9000, "enabled": True} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "mtu", "9000"]), + cmds, + ) + + def test_set_duplex(self): + want = {"duplex": "full", "enabled": True} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "duplex", "full"]), + cmds, + ) + + def test_set_speed(self): + want = {"speed": "1000", "enabled": True} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "speed", "1000"]), + cmds, + ) + + def test_disable_interface(self): + want = {"enabled": False} + have = {"enabled": True} + cmds = _iface_cmds("eth0", want, have) + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "disable"]), + cmds, + ) + + def test_enable_interface(self): + want = {"enabled": True} + have = {"enabled": False} + cmds = _iface_cmds("eth0", want, have) + self.assertIn( + ("delete", ["interfaces", "ethernet", "eth0", "disable"]), + cmds, + ) + + def test_idempotent_description(self): + want = {"description": "WAN", "enabled": True} + have = {"description": "WAN", "enabled": True} + cmds = _iface_cmds("eth0", want, have) + self.assertEqual(cmds, []) + + def test_delete_description_when_none(self): + want = {"enabled": True} + have = {"description": "Old", "enabled": True} + cmds = _iface_cmds("eth0", want, have) + self.assertIn( + ("delete", ["interfaces", "ethernet", "eth0", "description"]), + cmds, + ) + + +class TestVyOSInterfacesDeleteIfaceConfig(unittest.TestCase): + + def test_deletes_all_l2_fields(self): + have = { + "description": "WAN", + "mtu": 1500, + "duplex": "auto", + "speed": "auto", + "enabled": True, + } + cmds = _delete_iface_config("eth0", have) + paths = [c[1] for c in cmds] + self.assertIn(["interfaces", "ethernet", "eth0", "description"], paths) + self.assertIn(["interfaces", "ethernet", "eth0", "mtu"], paths) + self.assertIn(["interfaces", "ethernet", "eth0", "duplex"], paths) + self.assertIn(["interfaces", "ethernet", "eth0", "speed"], paths) + + def test_deletes_disable_when_disabled(self): + have = {"enabled": False} + cmds = _delete_iface_config("eth0", have) + self.assertIn( + ("delete", ["interfaces", "ethernet", "eth0", "disable"]), + cmds, + ) + + def test_empty_have_produces_no_commands(self): + cmds = _delete_iface_config("eth0", {}) + self.assertEqual(cmds, []) + + +class TestVyOSInterfacesBuildCommands(unittest.TestCase): + + def _have_eth0(self): + return [ + { + "name": "eth0", + "description": "Management", + "mtu": 1500, + "enabled": True, + }, + ] + + def test_merged_adds_description(self): + config = [{"name": "eth0", "description": "WAN", "enabled": True}] + cmds = build_commands(config, [], "merged") + paths = [c[1] for c in cmds] + self.assertIn( + ["interfaces", "ethernet", "eth0", "description", "WAN"], + paths, + ) + + def test_merged_idempotent(self): + cmds = build_commands(self._have_eth0(), self._have_eth0(), "merged") + self.assertEqual(cmds, []) + + def test_deleted_removes_l2_fields(self): + config = [{"name": "eth0", "enabled": True}] + cmds = build_commands(config, self._have_eth0(), "deleted") + paths = [c[1] for c in cmds] + self.assertIn(["interfaces", "ethernet", "eth0", "description"], paths) + self.assertIn(["interfaces", "ethernet", "eth0", "mtu"], paths) + + def test_deleted_idempotent_when_no_l2(self): + have = [{"name": "eth0", "enabled": True}] + config = [{"name": "eth0", "enabled": True}] + cmds = build_commands(config, have, "deleted") + self.assertEqual(cmds, []) + + def test_replaced_idempotent(self): + cmds = build_commands(self._have_eth0(), self._have_eth0(), "replaced") + self.assertEqual(cmds, []) + + def test_replaced_updates_description(self): + config = [{"name": "eth0", "description": "NEW", "mtu": 1500, "enabled": True}] + cmds = build_commands(config, self._have_eth0(), "replaced") + self.assertTrue(len(cmds) > 0) + + def test_overridden_clears_interfaces_not_in_want(self): + have = [ + {"name": "eth0", "description": "Management", "enabled": True}, + {"name": "eth1", "description": "LAN", "enabled": True}, + ] + config = [{"name": "eth0", "description": "Management", "enabled": True}] + cmds = build_commands(config, have, "overridden") + paths = [c[1] for c in cmds] + self.assertIn(["interfaces", "ethernet", "eth1", "description"], paths) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_l3_interfaces.py b/tests/unit/modules/test_vyos_l3_interfaces.py new file mode 100644 index 0000000..66cd1fc --- /dev/null +++ b/tests/unit/modules/test_vyos_l3_interfaces.py @@ -0,0 +1,306 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_l3_interfaces import ( + _addr_cmds, + _addr_list, + _normalize, + _parse_iface, + _split_addresses, + build_commands, + get_running_config, +) + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + path = os.path.join(fixtures_dir, filename) + with open(path) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.mock_vyos.get_config = MagicMock(return_value={}) + + def set_running_config(self, data): + self.mock_vyos.get_config.return_value = data + + +class TestVyOSL3InterfacesAddrList(unittest.TestCase): + + def test_string_returns_list(self): + self.assertEqual(_addr_list("dhcp"), ["dhcp"]) + + def test_list_returned_sorted(self): + result = _addr_list(["10.0.0.2/32", "10.0.0.1/32"]) + self.assertEqual(result, ["10.0.0.1/32", "10.0.0.2/32"]) + + def test_none_returns_empty(self): + self.assertEqual(_addr_list(None), []) + + def test_empty_list_returns_empty(self): + self.assertEqual(_addr_list([]), []) + + +class TestVyOSL3InterfacesSplitAddresses(unittest.TestCase): + + def test_dhcp_goes_to_ipv4(self): + ipv4, ipv6 = _split_addresses(["dhcp"]) + self.assertIn("dhcp", ipv4) + self.assertEqual(ipv6, []) + + def test_dhcpv6_goes_to_ipv6(self): + ipv4, ipv6 = _split_addresses(["dhcpv6"]) + self.assertEqual(ipv4, []) + self.assertIn("dhcpv6", ipv6) + + def test_ipv4_cidr(self): + ipv4, ipv6 = _split_addresses(["192.0.2.1/24"]) + self.assertIn("192.0.2.1/24", ipv4) + self.assertEqual(ipv6, []) + + def test_ipv6_cidr(self): + ipv4, ipv6 = _split_addresses(["2001:db8::1/128"]) + self.assertEqual(ipv4, []) + self.assertIn("2001:db8::1/128", ipv6) + + def test_mixed(self): + ipv4, ipv6 = _split_addresses(["dhcp", "192.0.2.1/24", "2001:db8::1/128"]) + self.assertIn("dhcp", ipv4) + self.assertIn("192.0.2.1/24", ipv4) + self.assertIn("2001:db8::1/128", ipv6) + + +class TestVyOSL3InterfacesParseIface(unittest.TestCase): + + def test_parse_dhcp(self): + result = _parse_iface("eth0", {"address": "dhcp"}) + self.assertEqual(result["name"], "eth0") + self.assertEqual(result["ipv4"], [{"address": "dhcp"}]) + + def test_parse_multiple_addresses(self): + result = _parse_iface("lo", {"address": ["10.0.0.1/32", "10.0.0.2/32"]}) + addrs = [a["address"] for a in result["ipv4"]] + self.assertIn("10.0.0.1/32", addrs) + self.assertIn("10.0.0.2/32", addrs) + + def test_parse_vif(self): + result = _parse_iface( + "eth0", + { + "address": "dhcp", + "vif": {"100": {"address": "192.0.2.100/24"}}, + }, + ) + self.assertIn("vifs", result) + self.assertEqual(result["vifs"][0]["vlan_id"], 100) + self.assertEqual(result["vifs"][0]["ipv4"][0]["address"], "192.0.2.100/24") + + def test_parse_no_address(self): + result = _parse_iface("lo", {}) + self.assertNotIn("ipv4", result) + self.assertNotIn("ipv6", result) + + def test_hw_id_ignored(self): + result = _parse_iface("eth0", {"hw-id": "52:54:00:65:5a:24"}) + self.assertNotIn("hw_id", result) + self.assertNotIn("hw-id", result) + + +class TestVyOSL3InterfacesGetRunningFixture(VyOSModuleTestCase): + + def setUp(self): + super().setUp() + self.fixture = load_fixture("l3_interfaces_running.json") + + def test_fixture_parses_eth0_dhcp(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth0 = next((e for e in result if e["name"] == "eth0"), None) + self.assertIsNotNone(eth0) + addrs = [a["address"] for a in eth0.get("ipv4", [])] + self.assertIn("dhcp", addrs) + + def test_fixture_parses_loopback_addresses(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + lo = next((e for e in result if e["name"] == "lo"), None) + self.assertIsNotNone(lo) + addrs = [a["address"] for a in lo.get("ipv4", [])] + self.assertIn("10.0.0.1/32", addrs) + self.assertIn("10.0.0.2/32", addrs) + + def test_fixture_parses_vif(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertIn("vifs", eth0) + self.assertEqual(eth0["vifs"][0]["vlan_id"], 100) + + def test_empty_interface_not_included(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + names = [e["name"] for e in result] + # loopback with no addresses should not appear + self.assertNotIn("dummy0", names) + + +class TestVyOSL3InterfacesGetRunning(VyOSModuleTestCase): + + def test_empty_returns_empty_list(self): + self.set_running_config({}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + def test_interface_without_address_excluded(self): + self.set_running_config( + { + "ethernet": {"eth0": {"hw-id": "52:54:00:65:5a:24"}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + +class TestVyOSL3InterfacesNormalize(unittest.TestCase): + + def test_normalize_ipv4(self): + config = [ + { + "name": "lo", + "ipv4": [{"address": "10.0.0.1/32"}], + }, + ] + result = _normalize(config) + self.assertIn("lo", result) + self.assertIn("10.0.0.1/32", result["lo"]["ipv4"]) + + def test_normalize_vif(self): + config = [ + { + "name": "eth0", + "vifs": [{"vlan_id": 100, "ipv4": [{"address": "192.0.2.100/24"}]}], + }, + ] + result = _normalize(config) + self.assertIn(100, result["eth0"]["vifs"]) + self.assertIn("192.0.2.100/24", result["eth0"]["vifs"][100]["ipv4"]) + + def test_normalize_empty(self): + result = _normalize([]) + self.assertEqual(result, {}) + + +class TestVyOSL3InterfacesAddrCmds(unittest.TestCase): + + def test_adds_new_address(self): + base = ["interfaces", "loopback", "lo"] + cmds = _addr_cmds(base, ["10.0.0.1/32"], [], "merged") + self.assertIn(("set", base + ["address", "10.0.0.1/32"]), cmds) + + def test_idempotent(self): + base = ["interfaces", "loopback", "lo"] + cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32"], "merged") + self.assertEqual(cmds, []) + + def test_merged_does_not_delete_extra(self): + base = ["interfaces", "loopback", "lo"] + cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32", "10.0.0.2/32"], "merged") + self.assertEqual(cmds, []) + + def test_replaced_deletes_extra(self): + base = ["interfaces", "loopback", "lo"] + cmds = _addr_cmds(base, ["10.0.0.1/32"], ["10.0.0.1/32", "10.0.0.2/32"], "replaced") + self.assertIn(("delete", base + ["address", "10.0.0.2/32"]), cmds) + + +class TestVyOSL3InterfacesBuildCommands(unittest.TestCase): + + def _have_lo(self): + return [ + { + "name": "lo", + "ipv4": [ + {"address": "10.0.0.1/32"}, + {"address": "10.0.0.2/32"}, + ], + }, + ] + + def test_merged_adds_address(self): + config = [{"name": "lo", "ipv4": [{"address": "10.0.0.3/32"}]}] + cmds = build_commands(config, [], "merged") + self.assertIn( + ("set", ["interfaces", "loopback", "lo", "address", "10.0.0.3/32"]), + cmds, + ) + + def test_merged_idempotent(self): + cmds = build_commands(self._have_lo(), self._have_lo(), "merged") + self.assertEqual(cmds, []) + + def test_deleted_no_config_removes_all(self): + cmds = build_commands([], self._have_lo(), "deleted") + paths = [c[1] for c in cmds] + self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.1/32"], paths) + self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.2/32"], paths) + + def test_deleted_with_config_removes_interface_addresses(self): + config = [{"name": "lo"}] + cmds = build_commands(config, self._have_lo(), "deleted") + paths = [c[1] for c in cmds] + self.assertIn(["interfaces", "loopback", "lo", "address", "10.0.0.1/32"], paths) + + def test_deleted_idempotent_when_empty(self): + cmds = build_commands([], [], "deleted") + self.assertEqual(cmds, []) + + def test_replaced_removes_extra_address(self): + config = [{"name": "lo", "ipv4": [{"address": "10.0.0.1/32"}]}] + cmds = build_commands(config, self._have_lo(), "replaced") + self.assertIn( + ("delete", ["interfaces", "loopback", "lo", "address", "10.0.0.2/32"]), + cmds, + ) + + def test_overridden_removes_unlisted_interface(self): + config = [{"name": "lo", "ipv4": [{"address": "10.0.0.1/32"}]}] + have = self._have_lo() + [ + { + "name": "eth0", + "ipv4": [{"address": "192.0.2.1/24"}], + }, + ] + cmds = build_commands(config, have, "overridden") + self.assertIn( + ("delete", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]), + cmds, + ) + + def test_vif_added(self): + config = [ + { + "name": "eth0", + "vifs": [{"vlan_id": 100, "ipv4": [{"address": "192.0.2.100/24"}]}], + }, + ] + cmds = build_commands(config, [], "merged") + self.assertIn( + ("set", ["interfaces", "ethernet", "eth0", "vif", "100", "address", "192.0.2.100/24"]), + cmds, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_lag_interfaces.py b/tests/unit/modules/test_vyos_lag_interfaces.py new file mode 100644 index 0000000..7c6df87 --- /dev/null +++ b/tests/unit/modules/test_vyos_lag_interfaces.py @@ -0,0 +1,293 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_lag_interfaces import ( + _bond_base, + _bond_cmds, + _normalize, + build_commands, + get_running_config, +) + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + path = os.path.join(fixtures_dir, filename) + with open(path) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.mock_vyos.get_config = MagicMock(return_value={}) + + def set_running_config(self, data): + self.mock_vyos.get_config.return_value = data + + +class TestVyOSLagInterfacesBondBase(unittest.TestCase): + + def test_bond_base(self): + self.assertEqual( + _bond_base("bond0"), + ["interfaces", "bonding", "bond0"], + ) + + +class TestVyOSLagInterfacesGetRunningFixture(VyOSModuleTestCase): + + def setUp(self): + super().setUp() + self.fixture = load_fixture("lag_interfaces_running.json") + + def test_fixture_parses_bond0_mode(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + bond0 = next((e for e in result if e["name"] == "bond0"), None) + self.assertIsNotNone(bond0) + self.assertEqual(bond0["mode"], "active-backup") + + def test_fixture_parses_hash_policy(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + bond0 = next(e for e in result if e["name"] == "bond0") + self.assertEqual(bond0["hash_policy"], "layer2") + + def test_fixture_parses_arp_monitor(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + bond0 = next(e for e in result if e["name"] == "bond0") + self.assertIn("arp_monitor", bond0) + self.assertEqual(bond0["arp_monitor"]["interval"], 100) + self.assertIn("192.0.2.1", bond0["arp_monitor"]["target"]) + + def test_fixture_unwraps_bonding_key(self): + # wrap in extra "bonding" key as VyOS sometimes returns + wrapped = {"bonding": self.fixture} + self.set_running_config(wrapped) + result = get_running_config(self.mock_vyos) + self.assertTrue(len(result) > 0) + self.assertEqual(result[0]["name"], "bond0") + + +class TestVyOSLagInterfacesGetRunning(VyOSModuleTestCase): + + def test_empty_returns_empty_list(self): + self.set_running_config({}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + def test_parses_mode(self): + self.set_running_config( + { + "bond0": {"mode": "802.3ad"}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result[0]["mode"], "802.3ad") + + def test_parses_hash_policy(self): + self.set_running_config( + { + "bond0": {"hash-policy": "layer2+3"}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result[0]["hash_policy"], "layer2+3") + + def test_parses_members(self): + self.set_running_config( + { + "bond0": {"member": {"interface": {"eth1": {}, "eth2": {}}}}, + }, + ) + result = get_running_config(self.mock_vyos) + members = [m["member"] for m in result[0]["members"]] + self.assertIn("eth1", members) + self.assertIn("eth2", members) + + def test_parses_arp_monitor_interval(self): + self.set_running_config( + { + "bond0": {"arp-monitor": {"interval": "100"}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result[0]["arp_monitor"]["interval"], 100) + + def test_parses_arp_monitor_target_dict(self): + self.set_running_config( + { + "bond0": {"arp-monitor": {"target": {"192.0.2.1": {}}}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertIn("192.0.2.1", result[0]["arp_monitor"]["target"]) + + def test_skips_type_keys(self): + self.set_running_config( + { + "bonding": {"bond0": {"mode": "802.3ad"}}, + }, + ) + result = get_running_config(self.mock_vyos) + # "bonding" key should be skipped + names = [e["name"] for e in result] + self.assertNotIn("bonding", names) + + +class TestVyOSLagInterfacesNormalize(unittest.TestCase): + + def test_normalize_basic(self): + config = [{"name": "bond0", "mode": "802.3ad", "hash_policy": "layer2"}] + result = _normalize(config) + self.assertIn("bond0", result) + self.assertEqual(result["bond0"]["mode"], "802.3ad") + self.assertEqual(result["bond0"]["hash_policy"], "layer2") + + def test_normalize_members_sorted(self): + config = [ + { + "name": "bond0", + "members": [{"member": "eth2"}, {"member": "eth1"}], + }, + ] + result = _normalize(config) + self.assertEqual(result["bond0"]["members"], ["eth1", "eth2"]) + + def test_normalize_arp_targets_sorted(self): + config = [ + { + "name": "bond0", + "arp_monitor": {"interval": 100, "target": ["192.0.2.2", "192.0.2.1"]}, + }, + ] + result = _normalize(config) + self.assertEqual(result["bond0"]["arp_targets"], ["192.0.2.1", "192.0.2.2"]) + + def test_normalize_empty(self): + result = _normalize([]) + self.assertEqual(result, {}) + + +class TestVyOSLagInterfacesBondCmds(unittest.TestCase): + + def test_set_mode(self): + want = {"mode": "802.3ad"} + cmds = _bond_cmds("bond0", want, {}) + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "mode", "802.3ad"]), + cmds, + ) + + def test_set_hash_policy(self): + want = {"hash_policy": "layer2"} + cmds = _bond_cmds("bond0", want, {}) + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "hash-policy", "layer2"]), + cmds, + ) + + def test_set_member(self): + want = {"members": ["eth1"]} + cmds = _bond_cmds("bond0", want, {}) + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "member", "interface", "eth1"]), + cmds, + ) + + def test_set_arp_interval(self): + want = {"arp_interval": 100} + cmds = _bond_cmds("bond0", want, {}) + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "arp-monitor", "interval", "100"]), + cmds, + ) + + def test_set_arp_target(self): + want = {"arp_targets": ["192.0.2.1"]} + cmds = _bond_cmds("bond0", want, {}) + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "arp-monitor", "target", "192.0.2.1"]), + cmds, + ) + + def test_idempotent_mode(self): + want = {"mode": "802.3ad"} + have = {"mode": "802.3ad"} + cmds = _bond_cmds("bond0", want, have) + self.assertEqual(cmds, []) + + def test_no_commands_when_empty_want(self): + cmds = _bond_cmds("bond0", {}, {}) + self.assertEqual(cmds, []) + + +class TestVyOSLagInterfacesBuildCommands(unittest.TestCase): + + def _have_bond0(self): + return [ + { + "name": "bond0", + "mode": "active-backup", + "hash_policy": "layer2", + "arp_monitor": {"interval": 100, "target": ["192.0.2.1"]}, + }, + ] + + def test_merged_adds_bond(self): + config = [{"name": "bond0", "mode": "802.3ad"}] + cmds = build_commands(config, [], "merged") + self.assertIn( + ("set", ["interfaces", "bonding", "bond0", "mode", "802.3ad"]), + cmds, + ) + + def test_merged_idempotent(self): + cmds = build_commands(self._have_bond0(), self._have_bond0(), "merged") + self.assertEqual(cmds, []) + + def test_deleted_no_config_removes_all(self): + cmds = build_commands([], self._have_bond0(), "deleted") + self.assertIn( + ("delete", ["interfaces", "bonding", "bond0"]), + cmds, + ) + + def test_deleted_with_name_removes_bond(self): + config = [{"name": "bond0"}] + cmds = build_commands(config, self._have_bond0(), "deleted") + self.assertIn( + ("delete", ["interfaces", "bonding", "bond0"]), + cmds, + ) + + def test_deleted_idempotent_when_empty(self): + cmds = build_commands([], [], "deleted") + self.assertEqual(cmds, []) + + def test_replaced_idempotent(self): + cmds = build_commands(self._have_bond0(), self._have_bond0(), "replaced") + self.assertEqual(cmds, []) + + def test_overridden_removes_unlisted_bond(self): + config = [{"name": "bond1", "mode": "802.3ad"}] + cmds = build_commands(config, self._have_bond0(), "overridden") + self.assertIn( + ("delete", ["interfaces", "bonding", "bond0"]), + cmds, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_lldp_interfaces.py b/tests/unit/modules/test_vyos_lldp_interfaces.py new file mode 100644 index 0000000..0950991 --- /dev/null +++ b/tests/unit/modules/test_vyos_lldp_interfaces.py @@ -0,0 +1,291 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_lldp_interfaces import ( + _iface_base, + _iface_cmds, + _normalize, + build_commands, + get_running_config, +) + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + path = os.path.join(fixtures_dir, filename) + with open(path) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.mock_vyos.get_config = MagicMock(return_value={}) + + def set_running_config(self, data): + self.mock_vyos.get_config.return_value = data + + +class TestVyOSLldpInterfacesIfaceBase(unittest.TestCase): + + def test_iface_base(self): + self.assertEqual( + _iface_base("eth0"), + ["service", "lldp", "interface", "eth0"], + ) + + +class TestVyOSLldpInterfacesGetRunningFixture(VyOSModuleTestCase): + + def setUp(self): + super().setUp() + self.fixture = load_fixture("lldp_interfaces_running.json") + + def test_fixture_parses_eth0_mode(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth0 = next((e for e in result if e["name"] == "eth0"), None) + self.assertIsNotNone(eth0) + self.assertEqual(eth0["mode"], "disable") + + def test_fixture_parses_elin(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth0 = next(e for e in result if e["name"] == "eth0") + self.assertEqual(eth0["location"]["elin"], "1234567890") + + def test_fixture_parses_coordinate_based(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + eth1 = next((e for e in result if e["name"] == "eth1"), None) + self.assertIsNotNone(eth1) + cb = eth1["location"]["coordinate_based"] + self.assertEqual(cb["latitude"], "33.524449N") + self.assertEqual(cb["longitude"], "22.267255E") + self.assertEqual(cb["altitude"], 2200) + self.assertEqual(cb["datum"], "WGS84") + + +class TestVyOSLldpInterfacesGetRunning(VyOSModuleTestCase): + + def test_empty_returns_empty_list(self): + self.set_running_config({}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + def test_no_interface_returns_empty(self): + self.set_running_config({"snmp": "enable"}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + def test_parses_mode(self): + self.set_running_config( + { + "interface": {"eth0": {"mode": "rx-tx"}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result[0]["mode"], "rx-tx") + + def test_parses_elin(self): + self.set_running_config( + { + "interface": {"eth0": {"location": {"elin": "9876543210"}}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertEqual(result[0]["location"]["elin"], "9876543210") + + def test_parses_coordinate_based(self): + self.set_running_config( + { + "interface": { + "eth0": { + "location": { + "coordinate-based": { + "latitude": "33.524449N", + "longitude": "22.267255E", + "altitude": "2200", + "datum": "WGS84", + }, + }, + }, + }, + }, + ) + result = get_running_config(self.mock_vyos) + cb = result[0]["location"]["coordinate_based"] + self.assertEqual(cb["latitude"], "33.524449N") + self.assertEqual(cb["altitude"], 2200) + + def test_no_mode_not_in_entry(self): + self.set_running_config( + { + "interface": {"eth0": {"location": {"elin": "1234567890"}}}, + }, + ) + result = get_running_config(self.mock_vyos) + self.assertNotIn("mode", result[0]) + + +class TestVyOSLldpInterfacesNormalize(unittest.TestCase): + + def test_normalize_mode(self): + config = [{"name": "eth0", "mode": "disable"}] + result = _normalize(config) + self.assertEqual(result["eth0"]["mode"], "disable") + + def test_normalize_elin(self): + config = [{"name": "eth0", "location": {"elin": "1234567890"}}] + result = _normalize(config) + self.assertEqual(result["eth0"]["elin"], "1234567890") + + def test_normalize_coordinate_based(self): + config = [ + { + "name": "eth0", + "location": { + "coordinate_based": { + "latitude": "33.524449N", + "longitude": "22.267255E", + "altitude": 2200, + "datum": "WGS84", + }, + }, + }, + ] + result = _normalize(config) + self.assertEqual(result["eth0"]["latitude"], "33.524449N") + self.assertEqual(result["eth0"]["altitude"], 2200) + + def test_normalize_empty(self): + result = _normalize([]) + self.assertEqual(result, {}) + + +class TestVyOSLldpInterfacesIfaceCmds(unittest.TestCase): + + def test_set_mode(self): + want = {"mode": "disable"} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["service", "lldp", "interface", "eth0", "mode", "disable"]), + cmds, + ) + + def test_set_elin(self): + want = {"elin": "1234567890"} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ("set", ["service", "lldp", "interface", "eth0", "location", "elin", "1234567890"]), + cmds, + ) + + def test_set_latitude(self): + want = {"latitude": "33.524449N", "longitude": "22.267255E"} + cmds = _iface_cmds("eth0", want, {}) + self.assertIn( + ( + "set", + [ + "service", + "lldp", + "interface", + "eth0", + "location", + "coordinate-based", + "latitude", + "33.524449N", + ], + ), + cmds, + ) + + def test_idempotent_mode(self): + want = {"mode": "disable"} + have = {"mode": "disable"} + cmds = _iface_cmds("eth0", want, have) + self.assertEqual(cmds, []) + + def test_idempotent_elin(self): + want = {"elin": "1234567890"} + have = {"elin": "1234567890"} + cmds = _iface_cmds("eth0", want, have) + self.assertEqual(cmds, []) + + def test_delete_mode_when_none(self): + want = {} + have = {"mode": "disable"} + cmds = _iface_cmds("eth0", want, have) + self.assertIn( + ("delete", ["service", "lldp", "interface", "eth0", "mode"]), + cmds, + ) + + +class TestVyOSLldpInterfacesBuildCommands(unittest.TestCase): + + def _have_eth0(self): + return [ + { + "name": "eth0", + "mode": "disable", + "location": {"elin": "1234567890"}, + }, + ] + + def test_merged_adds_interface(self): + config = [{"name": "eth0", "mode": "disable"}] + cmds = build_commands(config, [], "merged") + self.assertIn( + ("set", ["service", "lldp", "interface", "eth0", "mode", "disable"]), + cmds, + ) + + def test_merged_idempotent(self): + cmds = build_commands(self._have_eth0(), self._have_eth0(), "merged") + self.assertEqual(cmds, []) + + def test_deleted_no_config_removes_all(self): + cmds = build_commands([], self._have_eth0(), "deleted") + self.assertIn( + ("delete", ["service", "lldp", "interface", "eth0"]), + cmds, + ) + + def test_deleted_with_config_removes_named(self): + config = [{"name": "eth0"}] + cmds = build_commands(config, self._have_eth0(), "deleted") + self.assertIn( + ("delete", ["service", "lldp", "interface", "eth0"]), + cmds, + ) + + def test_deleted_idempotent_when_empty(self): + cmds = build_commands([], [], "deleted") + self.assertEqual(cmds, []) + + def test_replaced_idempotent(self): + cmds = build_commands(self._have_eth0(), self._have_eth0(), "replaced") + self.assertEqual(cmds, []) + + def test_overridden_removes_unlisted(self): + config = [{"name": "eth1", "mode": "rx-tx"}] + cmds = build_commands(config, self._have_eth0(), "overridden") + self.assertIn( + ("delete", ["service", "lldp", "interface", "eth0"]), + cmds, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_logging_global.py b/tests/unit/modules/test_vyos_logging_global.py index 8593f1f..7d91182 100644 --- a/tests/unit/modules/test_vyos_logging_global.py +++ b/tests/unit/modules/test_vyos_logging_global.py @@ -98,7 +98,7 @@ class TestVyOSLoggingGlobalNormalize(unittest.TestCase): def test_normalize_running_host_port_not_cast(self): """Port is NOT cast to int — stored as-is from API response.""" raw = { - "host": { + "remote": { "172.16.0.1": { "port": "514", "facility": {}, @@ -112,7 +112,7 @@ class TestVyOSLoggingGlobalNormalize(unittest.TestCase): def test_normalize_running_global_archive_key(self): """Archive stored under 'archive' key — no file_num remapping.""" raw = { - "global": { + "local": { "archive": {"file": "2", "size": "111"}, "marker": {"interval": "111"}, "preserve-fqdn": {}, @@ -133,7 +133,7 @@ class TestVyOSLoggingGlobalNormalize(unittest.TestCase): def test_normalize_running_host_facilities(self): raw = { - "host": { + "remote": { "172.16.0.1": { "facility": { "local7": {"level": "all"}, @@ -196,14 +196,14 @@ class TestVyOSLoggingGlobalBuildCommands(unittest.TestCase): cmds = build_commands(want, self._empty_have(), "merged") paths = [c[1] for c in cmds] # diff_map only adds the host key, not per-facility details - self.assertIn(["system", "syslog", "host", "172.16.0.1"], paths) + self.assertIn(["system", "syslog", "remote", "172.16.0.1"], paths) def test_replaced_removes_extra_host(self): want = self._empty_have() have = self._empty_have() have["hosts"]["172.16.0.1"] = {"port": None, "facilities": {}} cmds = build_commands(want, have, "replaced") - self.assertIn(("delete", ["system", "syslog", "host", "172.16.0.1"]), cmds) + self.assertIn(("delete", ["system", "syslog", "remote", "172.16.0.1"]), cmds) def test_deleted_removes_per_field(self): """deleted state removes per-facility entries, not single subtree.""" diff --git a/tests/unit/modules/test_vyos_ospf_interfaces.py b/tests/unit/modules/test_vyos_ospf_interfaces.py new file mode 100644 index 0000000..4d6be0d --- /dev/null +++ b/tests/unit/modules/test_vyos_ospf_interfaces.py @@ -0,0 +1,148 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_ospf_interfaces import ( + _parse_ipv4_iface, + _parse_ipv6_iface, + build_commands, + get_running_config, +) + + +_BASE4 = ["protocols", "ospf", "interface"] +_BASE6 = ["protocols", "ospfv3", "interface"] + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + with open(os.path.join(fixtures_dir, filename)) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.ipv4_fixture = load_fixture("ospf_interfaces_ipv4.json") + self.ipv6_fixture = load_fixture("ospf_interfaces_ipv6.json") + self.mock_vyos.get_config = MagicMock( + side_effect=lambda path: (self.ipv4_fixture if path == _BASE4 else self.ipv6_fixture), + ) + + +class TestVyOSOspfInterfacesParse(VyOSModuleTestCase): + + def test_parse_ipv4_scalars(self): + raw = {"cost": "100", "transmit-delay": "50", "priority": "26"} + result = _parse_ipv4_iface(raw) + self.assertEqual(result["afi"], "ipv4") + self.assertEqual(result["cost"], 100) + self.assertEqual(result["transmit_delay"], 50) + self.assertEqual(result["priority"], 26) + + def test_parse_ipv4_md5_auth(self): + raw = {"authentication": {"md5": {"key-id": {"10": {"md5-key": "secret"}}}}} + result = _parse_ipv4_iface(raw) + self.assertEqual(result["authentication"]["md5_key"]["key_id"], 10) + self.assertEqual(result["authentication"]["md5_key"]["key"], "secret") + + def test_parse_ipv6_scalars(self): + raw = {"dead-interval": "39", "passive": {}} + result = _parse_ipv6_iface(raw) + self.assertEqual(result["afi"], "ipv6") + self.assertEqual(result["dead_interval"], 39) + self.assertTrue(result["passive"]) + + def test_get_running_config(self): + result = get_running_config(self.mock_vyos) + names = [e["name"] for e in result] + self.assertIn("eth1", names) + self.assertIn("eth2", names) + eth1 = next(e for e in result if e["name"] == "eth1") + afis = {af["afi"] for af in eth1["address_family"]} + self.assertIn("ipv4", afis) + self.assertIn("ipv6", afis) + + def test_get_running_config_empty(self): + self.mock_vyos.get_config = MagicMock(return_value={}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, []) + + +class TestVyOSOspfInterfacesBuildCommands(unittest.TestCase): + + def test_deleted_all(self): + have = [ + { + "name": "eth1", + "address_family": [{"afi": "ipv4", "cost": 100}, {"afi": "ipv6", "passive": True}], + }, + ] + cmds = build_commands([], have, "deleted") + self.assertIn(("delete", _BASE4 + ["eth1"]), cmds) + self.assertIn(("delete", _BASE6 + ["eth1"]), cmds) + + def test_deleted_specific(self): + have = [ + {"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}, + {"name": "eth2", "address_family": [{"afi": "ipv4", "cost": 200}]}, + ] + cmds = build_commands([{"name": "eth1"}], have, "deleted") + self.assertIn(("delete", _BASE4 + ["eth1"]), cmds) + paths = [c[1] for c in cmds] + self.assertNotIn(_BASE4 + ["eth2"], paths) + + def test_merged_adds_cost(self): + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + cmds = build_commands(config, [], "merged") + self.assertIn(("set", _BASE4 + ["eth1", "cost", "100"]), cmds) + + def test_merged_idempotent(self): + have = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + cmds = build_commands(config, have, "merged") + self.assertEqual(cmds, []) + + def test_merged_ipv6_passive(self): + config = [{"name": "eth1", "address_family": [{"afi": "ipv6", "passive": True}]}] + cmds = build_commands(config, [], "merged") + self.assertIn(("set", _BASE6 + ["eth1", "passive"]), cmds) + + def test_overridden_removes_extra_interface(self): + have = [ + {"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}, + {"name": "eth2", "address_family": [{"afi": "ipv4", "cost": 200}]}, + ] + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + cmds = build_commands(config, have, "overridden") + self.assertIn(("delete", _BASE4 + ["eth2"]), cmds) + + def test_replaced_idempotent(self): + have = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 200}]}] + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 200}]}] + cmds = build_commands(config, have, "replaced") + self.assertEqual(cmds, []) + + def test_replaced_rebuilds_on_change(self): + have = [ + { + "name": "eth1", + "address_family": [{"afi": "ipv4", "cost": 100, "transmit_delay": 50}], + }, + ] + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 200}]}] + cmds = build_commands(config, have, "replaced") + self.assertIn(("delete", _BASE4 + ["eth1"]), cmds) + self.assertIn(("set", _BASE4 + ["eth1", "cost", "200"]), cmds) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_ospfv2.py b/tests/unit/modules/test_vyos_ospfv2.py new file mode 100644 index 0000000..c05dd77 --- /dev/null +++ b/tests/unit/modules/test_vyos_ospfv2.py @@ -0,0 +1,180 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_ospfv2 import ( + _parse_areas, + _parse_default_information, + _parse_neighbor, + _parse_parameters, + _parse_redistribute, + build_commands, + get_running_config, +) + + +_BASE = ["protocols", "ospf"] + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + with open(os.path.join(fixtures_dir, filename)) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.fixture = load_fixture("ospfv2_running.json") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) + + +class TestVyOSOspfv2Parse(VyOSModuleTestCase): + + def test_parse_parameters(self): + result = _parse_parameters(self.fixture["parameters"]) + self.assertEqual(result["router_id"], "192.0.1.1") + self.assertEqual(result["abr_type"], "cisco") + self.assertTrue(result["opaque_lsa"]) + self.assertTrue(result["rfc1583_compatibility"]) + + def test_parse_redistribute(self): + result = _parse_redistribute(self.fixture["redistribute"]) + route_types = [r["route_type"] for r in result] + self.assertIn("bgp", route_types) + self.assertIn("connected", route_types) + bgp = next(r for r in result if r["route_type"] == "bgp") + self.assertEqual(bgp["metric"], 10) + self.assertEqual(bgp["metric_type"], 2) + + def test_parse_neighbor(self): + result = _parse_neighbor(self.fixture["neighbor"]) + self.assertEqual(len(result), 1) + nb = result[0] + self.assertEqual(nb["neighbor_id"], "192.0.11.12") + self.assertEqual(nb["priority"], 2) + self.assertEqual(nb["poll_interval"], 10) + + def test_parse_default_information(self): + result = _parse_default_information(self.fixture["default-information"]) + orig = result["originate"] + self.assertTrue(orig["always"]) + self.assertEqual(orig["metric"], 10) + self.assertEqual(orig["metric_type"], 2) + self.assertEqual(orig["route_map"], "ingress") + + def test_parse_areas(self): + result = _parse_areas(self.fixture["area"]) + self.assertEqual(len(result), 3) + area2 = next(a for a in result if a["area_id"] == "2") + self.assertTrue(area2["area_type"]["normal"]) + self.assertEqual(area2["network"][0]["address"], "192.0.2.0/24") + + area3 = next(a for a in result if a["area_id"] == "3") + self.assertTrue(area3["area_type"]["nssa"]["set"]) + + area4 = next(a for a in result if a["area_id"] == "4") + self.assertEqual(area4["area_type"]["stub"]["default_cost"], 20) + self.assertEqual(len(area4["range"]), 2) + r = next(r for r in area4["range"] if r["address"] == "192.0.3.0/24") + self.assertEqual(r["cost"], 10) + + def test_get_running_config(self): + result = get_running_config(self.mock_vyos) + self.assertEqual(result["parameters"]["router_id"], "192.0.1.1") + self.assertIn("eth1", result["passive_interface"]) + self.assertIn("eth2", result["passive_interface"]) + self.assertEqual(result["auto_cost"]["reference_bandwidth"], 2) + self.assertEqual(result["log_adjacency_changes"], "detail") + + def test_get_running_config_empty(self): + self.mock_vyos.get_config = MagicMock(return_value={}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, {}) + + +class TestVyOSOspfv2BuildCommands(unittest.TestCase): + + def test_deleted_with_have(self): + have = {"parameters": {"router_id": "192.0.1.1"}} + cmds = build_commands({}, have, "deleted") + self.assertEqual(cmds, [("delete", _BASE)]) + + def test_deleted_without_have(self): + cmds = build_commands({}, {}, "deleted") + self.assertEqual(cmds, []) + + def test_merged_parameters(self): + config = {"parameters": {"router_id": "192.0.1.1"}} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["parameters", "router-id", "192.0.1.1"]), cmds) + + def test_merged_redistribute(self): + config = {"redistribute": [{"route_type": "bgp", "metric": 10}]} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["redistribute", "bgp"]), cmds) + self.assertIn(("set", _BASE + ["redistribute", "bgp", "metric", "10"]), cmds) + + def test_merged_passive_interface(self): + config = {"passive_interface": ["eth1"]} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["interface", "eth1", "passive"]), cmds) + + def test_merged_area_normal(self): + config = {"areas": [{"area_id": "2", "area_type": {"normal": True}}]} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["area", "2", "area-type", "normal"]), cmds) + + def test_merged_area_stub_with_cost(self): + config = {"areas": [{"area_id": "4", "area_type": {"stub": {"default_cost": 20}}}]} + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", _BASE + ["area", "4", "area-type", "stub", "default-cost", "20"]), + cmds, + ) + + def test_merged_idempotent(self): + have = {"parameters": {"router_id": "192.0.1.1"}} + config = {"parameters": {"router_id": "192.0.1.1"}} + cmds = build_commands(config, have, "merged") + self.assertEqual(cmds, []) + + def test_replaced_idempotent(self): + have = {"parameters": {"router_id": "192.0.1.1"}} + config = {"parameters": {"router_id": "192.0.1.1"}} + cmds = build_commands(config, have, "replaced") + self.assertEqual(cmds, []) + + def test_replaced_rebuilds_on_change(self): + have = {"parameters": {"router_id": "192.0.1.1"}} + config = {"parameters": {"router_id": "192.0.1.2"}} + cmds = build_commands(config, have, "replaced") + self.assertEqual(cmds[0], ("delete", _BASE)) + self.assertIn( + ("set", _BASE + ["parameters", "router-id", "192.0.1.2"]), + cmds, + ) + + def test_merged_neighbor(self): + config = {"neighbor": [{"neighbor_id": "192.0.11.12", "priority": 2}]} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["neighbor", "192.0.11.12"]), cmds) + self.assertIn(("set", _BASE + ["neighbor", "192.0.11.12", "priority", "2"]), cmds) + + def test_merged_default_information(self): + config = {"default_information": {"originate": {"always": True, "metric": 10}}} + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["default-information", "originate", "always"]), cmds) + self.assertIn(("set", _BASE + ["default-information", "originate", "metric", "10"]), cmds) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/modules/test_vyos_ospfv3.py b/tests/unit/modules/test_vyos_ospfv3.py new file mode 100644 index 0000000..a84041b --- /dev/null +++ b/tests/unit/modules/test_vyos_ospfv3.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +from __future__ import absolute_import, division, print_function + + +__metaclass__ = type + +import json +import os +import unittest + +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.modules.vyos_ospfv3 import ( + build_commands, + get_running_config, +) + + +def load_fixture(filename): + fixtures_dir = os.path.join(os.path.dirname(__file__), "..", "fixtures") + with open(os.path.join(fixtures_dir, filename)) as f: + return json.load(f) + + +class VyOSModuleTestCase(unittest.TestCase): + def setUp(self): + self.mock_vyos = MagicMock() + self.mock_vyos.get_config = MagicMock(return_value={}) + + def set_running_config(self, data): + self.mock_vyos.get_config.return_value = data + + +class TestVyOSOspfv3Parse(VyOSModuleTestCase): + + def setUp(self): + super().setUp() + self.fixture = load_fixture("ospfv3_running.json") + + def test_parses_parameters(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + self.assertEqual(result["parameters"]["router_id"], "192.0.2.10") + + def test_parses_redistribute(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + route_types = [r["route_type"] for r in result["redistribute"]] + self.assertIn("bgp", route_types) + self.assertIn("connected", route_types) + connected = next(r for r in result["redistribute"] if r["route_type"] == "connected") + self.assertEqual(connected["route_map"], "RM1") + + def test_parses_areas(self): + self.set_running_config(self.fixture) + result = get_running_config(self.mock_vyos) + self.assertEqual(len(result["areas"]), 2) + area2 = next(a for a in result["areas"] if a["area_id"] == "2") + self.assertEqual(area2["export_list"], "export1") + self.assertEqual(area2["import_list"], "import1") + self.assertEqual(len(area2["range"]), 2) + not_adv = next(r for r in area2["range"] if r["address"] == "2001:db20::/32") + self.assertTrue(not_adv["not_advertise"]) + + def test_empty_config_returns_empty(self): + self.set_running_config({}) + result = get_running_config(self.mock_vyos) + self.assertEqual(result, {}) + + +class TestVyOSOspfv3BuildCommands(unittest.TestCase): + + def test_deleted_with_have(self): + have = {"parameters": {"router_id": "192.0.2.10"}} + cmds = build_commands({}, have, "deleted") + self.assertEqual(cmds, [("delete", ["protocols", "ospfv3"])]) + + def test_deleted_without_have(self): + cmds = build_commands({}, {}, "deleted") + self.assertEqual(cmds, []) + + def test_merged_parameters(self): + config = {"parameters": {"router_id": "192.0.2.10"}} + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.10"]), + cmds, + ) + + def test_merged_redistribute(self): + config = {"redistribute": [{"route_type": "bgp"}]} + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", ["protocols", "ospfv3", "redistribute", "bgp"]), + cmds, + ) + + def test_merged_idempotent(self): + config = {"parameters": {"router_id": "192.0.2.10"}} + have = {"parameters": {"router_id": "192.0.2.10"}} + cmds = build_commands(config, have, "merged") + self.assertEqual(cmds, []) + + def test_merged_area_range(self): + config = { + "areas": [{"area_id": "2", "range": [{"address": "2001:db10::/32"}]}], + } + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", ["protocols", "ospfv3", "area", "2", "range", "2001:db10::/32"]), + cmds, + ) + + def test_replaced_idempotent(self): + config = {"parameters": {"router_id": "192.0.2.10"}} + have = {"parameters": {"router_id": "192.0.2.10"}} + cmds = build_commands(config, have, "replaced") + self.assertEqual(cmds, []) + + def test_replaced_rebuilds_on_change(self): + config = {"parameters": {"router_id": "192.0.2.11"}} + have = {"parameters": {"router_id": "192.0.2.10"}} + cmds = build_commands(config, have, "replaced") + self.assertEqual(cmds[0], ("delete", ["protocols", "ospfv3"])) + self.assertIn( + ("set", ["protocols", "ospfv3", "parameters", "router-id", "192.0.2.11"]), + cmds, + ) + + def test_merged_area_export_list(self): + config = {"areas": [{"area_id": "2", "export_list": "export1"}]} + cmds = build_commands(config, {}, "merged") + self.assertIn( + ("set", ["protocols", "ospfv3", "area", "2", "export-list", "export1"]), + cmds, + ) + + +if __name__ == "__main__": + unittest.main() |