diff options
Diffstat (limited to 'tests/unit')
| -rw-r--r-- | tests/unit/fixtures/l3_interfaces_running.json | 19 | ||||
| -rw-r--r-- | tests/unit/fixtures/lag_interfaces_running.json | 13 | ||||
| -rw-r--r-- | tests/unit/fixtures/ospf_interfaces_ipv4.json | 13 | ||||
| -rw-r--r-- | tests/unit/fixtures/ospf_interfaces_ipv6.json | 7 | ||||
| -rw-r--r-- | tests/unit/fixtures/ospfv2_running.json | 71 | ||||
| -rw-r--r-- | tests/unit/fixtures/ospfv3_running.json | 11 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_l3_interfaces.py | 448 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_lag_interfaces.py | 422 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospf_interfaces.py | 298 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospfv2.py | 408 | ||||
| -rw-r--r-- | tests/unit/modules/test_vyos_ospfv3.py | 260 | ||||
| -rw-r--r-- | tests/unit/test_module_utils_vyos.py | 88 |
12 files changed, 1167 insertions, 891 deletions
diff --git a/tests/unit/fixtures/l3_interfaces_running.json b/tests/unit/fixtures/l3_interfaces_running.json index 676055b..3115861 100644 --- a/tests/unit/fixtures/l3_interfaces_running.json +++ b/tests/unit/fixtures/l3_interfaces_running.json @@ -1,21 +1,12 @@ { "ethernet": { "eth0": { - "address": "dhcp", - "hw-id": "52:54:00:65:5a:24", - "vif": { - "100": { - "address": "192.0.2.100/24" - } - } - } + "address": ["192.0.2.1/24", "2001:db8::1/64"], + "vif": { "100": { "address": ["192.0.2.100/24"] } } + }, + "eth1": {} }, "loopback": { - "lo": { - "address": ["10.0.0.1/32", "10.0.0.2/32"] - } - }, - "dummy": { - "dummy0": {} + "lo": { "address": ["10.0.0.1/32"] } } } diff --git a/tests/unit/fixtures/lag_interfaces_running.json b/tests/unit/fixtures/lag_interfaces_running.json index 3477b1d..c3cc646 100644 --- a/tests/unit/fixtures/lag_interfaces_running.json +++ b/tests/unit/fixtures/lag_interfaces_running.json @@ -1,12 +1,15 @@ { "bond0": { + "mode": "802.3ad", + "hash-policy": "layer2", + "primary": "eth1", + "member": { "interface": { "eth1": {}, "eth2": {} } }, "arp-monitor": { "interval": "100", - "target": { - "192.0.2.1": {} - } - }, - "hash-policy": "layer2", + "target": { "192.0.2.1": {}, "192.0.2.2": {} } + } + }, + "bond1": { "mode": "active-backup" } } diff --git a/tests/unit/fixtures/ospf_interfaces_ipv4.json b/tests/unit/fixtures/ospf_interfaces_ipv4.json index 1f199a6..9b40f66 100644 --- a/tests/unit/fixtures/ospf_interfaces_ipv4.json +++ b/tests/unit/fixtures/ospf_interfaces_ipv4.json @@ -1,4 +1,13 @@ { - "eth1": { "cost": "100", "transmit-delay": "50", "priority": "26" }, - "eth2": { "cost": "200" } + "eth1": { + "cost": "100", + "transmit-delay": "50", + "priority": "26", + "dead-interval": "40", + "mtu-ignore": {}, + "authentication": { "md5": { "key-id": { "10": { "md5-key": "secret" } } } } + }, + "eth2": { + "cost": "200" + } } diff --git a/tests/unit/fixtures/ospf_interfaces_ipv6.json b/tests/unit/fixtures/ospf_interfaces_ipv6.json index 580cecf..5d96817 100644 --- a/tests/unit/fixtures/ospf_interfaces_ipv6.json +++ b/tests/unit/fixtures/ospf_interfaces_ipv6.json @@ -1,4 +1,9 @@ { - "eth1": { "dead-interval": "39", "passive": {} }, + "eth1": { + "dead-interval": "39", + "passive": {}, + "instance-id": "5", + "ifmtu": "1500" + }, "eth2": {} } diff --git a/tests/unit/fixtures/ospfv2_running.json b/tests/unit/fixtures/ospfv2_running.json index 9cc64a0..e5b5d72 100644 --- a/tests/unit/fixtures/ospfv2_running.json +++ b/tests/unit/fixtures/ospfv2_running.json @@ -2,41 +2,70 @@ "area": { "2": { "area-type": { "normal": {} }, - "network": "192.0.2.0/24" - }, - "3": { - "area-type": { "nssa": {} } + "authentication": "plaintext-password", + "shortcut": "enable" }, "4": { - "area-type": { "stub": { "default-cost": "20" } }, - "range": { - "192.0.3.0/24": { "cost": "10" }, - "192.0.4.0/24": {} + "area-type": { "stub": { "default-cost": "20", "no-summary": {} } }, + "network": { "192.0.2.0/24": {} }, + "range": { "192.0.3.0/24": { "cost": "10", "not-advertise": {} } }, + "virtual-link": { + "10.0.0.1": { + "authentication": { "plaintext-password": "secret" }, + "dead-interval": "40", + "hello-interval": "10" + } } + }, + "5": { + "area-type": { "nssa": { "default-cost": "5", "translate": "always" } } } }, "auto-cost": { "reference-bandwidth": "2" }, - "interface": { "eth1": { "passive": {} }, "eth2": { "passive": {} } }, + "default-information": { + "originate": { + "always": {}, + "metric": "10", + "metric-type": "2", + "route-map": "ingress" + } + }, + "default-metric": "5", + "distance": { + "global": "110", + "ospf": { "external": "150", "inter-area": "120", "intra-area": "100" } + }, "log-adjacency-changes": { "detail": {} }, - "neighbor": { - "192.0.11.12": { "priority": "2", "poll-interval": "10" } + "max-metric": { + "router-lsa": { + "administrative": {}, + "on-shutdown": "10", + "on-startup": "20" + } }, + "mpls-te": { "enable": {}, "router-address": "192.0.11.11" }, + "neighbor": { "192.0.11.12": { "priority": "2", "poll-interval": "30" } }, "parameters": { "router-id": "192.0.1.1", - "abr-type": "cisco", "opaque-lsa": {}, - "rfc1583-compatibility": {} + "rfc1583-compatibility": {}, + "abr-type": "cisco" + }, + "interface": { + "eth1": { "passive": {} }, + "eth2": { "passive": { "disable": {} } } }, "redistribute": { - "bgp": { "metric": "10", "metric-type": "2" }, - "connected": {} + "bgp": { "metric": "10", "metric-type": "2", "route-map": "redist-map" } }, - "default-information": { - "originate": { - "always": {}, - "metric": "10", - "metric-type": "2", - "route-map": "ingress" + "refresh": { "timers": "300" }, + "timers": { + "throttle": { + "spf": { + "delay": "200", + "initial-holdtime": "500", + "max-holdtime": "2000" + } } } } diff --git a/tests/unit/fixtures/ospfv3_running.json b/tests/unit/fixtures/ospfv3_running.json index 44a1bcd..6588a1d 100644 --- a/tests/unit/fixtures/ospfv3_running.json +++ b/tests/unit/fixtures/ospfv3_running.json @@ -1,6 +1,4 @@ { - "parameters": { "router-id": "192.0.2.10" }, - "redistribute": { "bgp": {}, "connected": { "route-map": "RM1" } }, "area": { "2": { "export-list": "export1", @@ -11,9 +9,12 @@ } }, "3": { - "range": { - "2001:db40::/32": {} - } + "range": { "2001:db40::/32": { "advertise": {} } } } + }, + "parameters": { "router-id": "192.0.2.10" }, + "redistribute": { + "bgp": { "route-map": "redist-map" }, + "static": {} } } diff --git a/tests/unit/modules/test_vyos_l3_interfaces.py b/tests/unit/modules/test_vyos_l3_interfaces.py index 66cd1fc..79fc8cf 100644 --- a/tests/unit/modules/test_vyos_l3_interfaces.py +++ b/tests/unit/modules/test_vyos_l3_interfaces.py @@ -4,300 +4,216 @@ 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, + _device_to_argspec, + _guess_iface_type, + _iface_base, + _resolve_iface_type, _split_addresses, + _vif_addr_cmds, build_commands, get_running_config, ) +from .base import load_fixture -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) + +_BASE = ["interfaces"] 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") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - 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 gather(self): + return _device_to_argspec(self.fixture) - 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({}) +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_config_directly(self): 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.assertIn("ethernet", result) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) + + def test_collapsed_response_normalized(self): + self.mock_vyos.get_config = MagicMock(return_value="eth0") + self.assertEqual(get_running_config(self.mock_vyos), {"eth0": {}}) + + +class TestTypeResolution(unittest.TestCase): + def test_resolves_from_have_when_present(self): + raw_have = {"loopback": {"lo": {}}} + self.assertEqual(_resolve_iface_type("lo", raw_have), "loopback") + + def test_falls_back_to_guess_for_new_interface(self): + self.assertEqual(_resolve_iface_type("eth5", {}), "ethernet") + + def test_guess_covers_all_11_types_aligned_with_vyos_interfaces(self): + """Confirmed original inconsistency: this module's own type + table was missing ppp/wlan compared to vyos_interfaces' table, + despite both modules operating on the same interface + namespace. Aligned here.""" + expected = { + "eth0": "ethernet", + "bond0": "bonding", + "lo": "loopback", + "tun0": "tunnel", + "wg0": "wireguard", + "vti0": "vti", + "dum0": "dummy", + "vtun0": "openvpn", + "ppp0": "pppoe", + "wlan0": "wireless", + "br0": "bridge", + } + for name, itype in expected.items(): + self.assertEqual(_guess_iface_type(name), itype) + + def test_iface_base_uses_resolved_type(self): + raw_have = {"bonding": {"bond0": {}}} + self.assertEqual(_iface_base("bond0", raw_have), ["interfaces", "bonding", "bond0"]) + + +class TestSplitAddresses(unittest.TestCase): + def test_ipv4_ipv6_dhcp_dhcpv6(self): + ipv4, ipv6 = _split_addresses(["192.0.2.1/24", "2001:db8::1/64", "dhcp", "dhcpv6"]) + self.assertEqual(ipv4, ["192.0.2.1/24", "dhcp"]) + self.assertEqual(ipv6, ["2001:db8::1/64", "dhcpv6"]) + + def test_auto_config(self): + ipv4, ipv6 = _split_addresses(["auto-config"]) + self.assertEqual(ipv6, ["auto-config"]) + + +class TestAddrCmds(unittest.TestCase): + """Confirmed scoped correctly: address commands only ever touch + the "address" leaf under the given base, regardless of whether + that base is an interface or a vif.""" + + def test_set_new_address(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], ["192.0.2.1/24"], [], "merged") + expected = ("set", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]) + self.assertEqual(cmds, [expected]) + + def test_merged_never_deletes(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], [], ["192.0.2.1/24"], "merged") 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_replaced_deletes_omitted(self): + cmds = _addr_cmds(["interfaces", "ethernet", "eth0"], [], ["192.0.2.1/24"], "replaced") + expected = ("delete", ["interfaces", "ethernet", "eth0", "address", "192.0.2.1/24"]) + self.assertEqual(cmds, [expected]) + + +class TestVifAddrCmdsNeverWholeSubtree(unittest.TestCase): + """Primary confirmed severe bug: the original generated a whole- + VIF delete (`delete ... vif <id>`) for an omitted VIF under + replaced/overridden/deleted. Since VIFs are shared with + vyos_interfaces (which owns description/mtu/disable on them), + this would destroy that module's own fields, not just this + module's addresses. Confirmed 4 separate occurrences in the + original build_commands. Fixed: every VIF operation here is + scoped to exactly the address leaf, never the VIF subtree.""" + + def test_omitted_vif_only_deletes_its_addresses(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "replaced") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + vif_addr_delete = ( + "delete", + ["interfaces", "ethernet", "eth0", "vif", "100", "address", "192.0.2.100/24"], ) - - 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): + self.assertIn(vif_addr_delete, cmds) + + def test_deleted_state_also_never_whole_subtree(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "deleted") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + + def test_overridden_also_never_whole_subtree(self): + have_vifs = {100: {"ipv4": ["192.0.2.100/24"], "ipv6": []}} + cmds = _vif_addr_cmds(["interfaces", "ethernet", "eth0"], {}, have_vifs, "overridden") + whole_vif_delete = ("delete", ["interfaces", "ethernet", "eth0", "vif", "100"]) + self.assertNotIn(whole_vif_delete, cmds) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_all_interfaces_with_addresses_present(self): + have = self.gather() + names = {e["name"] for e in have} + # eth1 has no addresses at all -- must not appear + self.assertEqual(names, {"eth0", "lo"}) + + def test_eth0_addresses_and_vif_parsed(self): + have = self.gather() + eth0 = next(e for e in have if e["name"] == "eth0") + self.assertEqual(eth0["ipv4"][0]["address"], "192.0.2.1/24") + self.assertEqual(eth0["ipv6"][0]["address"], "2001:db8::1/64") + self.assertEqual(eth0["vifs"][0]["vlan_id"], 100) + self.assertEqual(eth0["vifs"][0]["ipv4"][0]["address"], "192.0.2.100/24") + + def test_collapsed_single_address_string(self): + entry = _device_to_argspec({"ethernet": {"eth0": {"address": "192.0.2.1/24"}}}) + self.assertEqual(entry[0]["ipv4"][0]["address"], "192.0.2.1/24") + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_deleted_all_never_whole_interface_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + self.assertIn(("delete", _BASE + ["ethernet", "eth0", "address", "192.0.2.1/24"]), cmds) + + def test_deleted_named_never_whole_interface_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([{"name": "eth0"}], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + + def test_overridden_omitted_interface_never_whole_subtree(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + cmds = build_commands([], raw_have, "overridden") + self.assertNotIn(("delete", _BASE + ["ethernet", "eth0"]), cmds) + + def test_merged_adds_without_removing(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24"]}}} + config = [{"name": "eth0", "ipv4": [{"address": "192.0.2.2/24"}]}] + cmds = build_commands(config, raw_have, "merged") + self.assertIn(("set", _BASE + ["ethernet", "eth0", "address", "192.0.2.2/24"]), cmds) + self.assertFalse(any(c[0] == "delete" for c in cmds)) + + def test_replaced_removes_unlisted_address(self): + raw_have = {"ethernet": {"eth0": {"address": ["192.0.2.1/24", "192.0.2.2/24"]}}} + config = [{"name": "eth0", "ipv4": [{"address": "192.0.2.1/24"}]}] + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["ethernet", "eth0", "address", "192.0.2.2/24"]), cmds) + + def test_merged_new_vif_address(self): config = [ - { - "name": "eth0", - "vifs": [{"vlan_id": 100, "ipv4": [{"address": "192.0.2.100/24"}]}], - }, + {"name": "eth0", "vifs": [{"vlan_id": 200, "ipv4": [{"address": "192.0.2.200/24"}]}]}, ] - cmds = build_commands(config, [], "merged") + cmds = build_commands(config, {}, "merged") self.assertIn( - ("set", ["interfaces", "ethernet", "eth0", "vif", "100", "address", "192.0.2.100/24"]), + ("set", _BASE + ["ethernet", "eth0", "vif", "200", "address", "192.0.2.200/24"]), cmds, ) diff --git a/tests/unit/modules/test_vyos_lag_interfaces.py b/tests/unit/modules/test_vyos_lag_interfaces.py index 7c6df87..abb94a4 100644 --- a/tests/unit/modules/test_vyos_lag_interfaces.py +++ b/tests/unit/modules/test_vyos_lag_interfaces.py @@ -4,289 +4,193 @@ 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, + ARGUMENT_SPEC, + _bond_entry_from_device, + _bond_entry_to_device, + _device_to_argspec, build_commands, + cast_by_spec, get_running_config, ) +from .base import load_fixture -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) + +_BASE = ["interfaces", "bonding"] 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") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - 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") - + def gather(self): + have = _device_to_argspec(self.fixture) + for entry in have: + cast_by_spec(entry, ARGUMENT_SPEC["config"]["options"]) + return have -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": {}}}}, - }, - ) +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_config_directly(self): 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_unwraps_bonding_wrapper_key(self): + self.mock_vyos.get_config = MagicMock( + return_value={"bonding": {"bond0": {"mode": "802.3ad"}}}, ) - - 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"]}, + result = get_running_config(self.mock_vyos) + self.assertEqual(result, {"bond0": {"mode": "802.3ad"}}) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) + + def test_collapsed_response_normalized(self): + self.mock_vyos.get_config = MagicMock(return_value="bond0") + self.assertEqual(get_running_config(self.mock_vyos), {"bond0": {}}) + + +class TestScopeIsolation(unittest.TestCase): + """Primary confirmed severe bug, same class as vyos_interfaces/ + vyos_l3_interfaces: bond0's device path is shared with + vyos_interfaces (description/mtu/vrf) and vyos_l3_interfaces + (address). A blanket pass-through when parsing device data would + leak these into have, and a whole-subtree delete would destroy + them alongside this module's own fields.""" + + def test_from_device_does_not_leak_unmanaged_fields(self): + entry = _bond_entry_from_device( + {"description": "lag", "mtu": "1500", "address": ["10.0.0.1/24"], "mode": "802.3ad"}, + ) + self.assertNotIn("description", entry) + self.assertNotIn("mtu", entry) + self.assertNotIn("address", entry) + self.assertEqual(entry, {"mode": "802.3ad"}) + + def test_deleted_all_never_whole_bond_subtree(self): + raw_have = { + "bond0": { + "mode": "802.3ad", + "description": "lag interface", + "mtu": "1500", + "address": ["10.0.0.1/24"], }, - ] - - 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, - ) + } + cmds = build_commands([], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["bond0"]), cmds) + self.assertIn(("delete", _BASE + ["bond0", "mode"]), cmds) + + def test_deleted_named_never_whole_bond_subtree(self): + raw_have = {"bond0": {"mode": "802.3ad", "description": "important"}} + cmds = build_commands([{"name": "bond0"}], raw_have, "deleted") + self.assertNotIn(("delete", _BASE + ["bond0"]), cmds) + + def test_overridden_omitted_bond_never_whole_subtree(self): + raw_have = { + "bond0": {}, + "bond1": {"mode": "802.3ad", "description": "important", "address": ["10.0.0.1/24"]}, + } + cmds = build_commands([{"name": "bond0"}], raw_have, "overridden") + self.assertNotIn(("delete", _BASE + ["bond1"]), cmds) + self.assertIn(("delete", _BASE + ["bond1", "mode"]), cmds) + + def test_replaced_never_whole_bond_subtree(self): + raw_have = {"bond0": {"mode": "802.3ad", "description": "important"}} + cmds = build_commands([{"name": "bond0"}], raw_have, "replaced") + self.assertNotIn(("delete", _BASE + ["bond0"]), cmds) + + def test_allowlist_derived_from_argspec_not_a_separate_constant(self): + """Confirmed via scope_to_spec (module_utils/vyos.py), not a + second, manually-maintained field list that could drift out + of sync with the argspec -- already used by vyos_bgp_global + for exactly this cross-module scope problem.""" + from ansible_collections.vyos.rest.plugins.modules.vyos_lag_interfaces import ( + _ENTRY_OPTIONS, + ) + + entry = _bond_entry_from_device({k: "x" for k in _ENTRY_OPTIONS if k != "name"}) + # every simple scalar field in the argspec should survive + for field in ("mode", "primary"): + self.assertIn(field, entry) + + +class TestMemberAndArpMonitorPreservation(unittest.TestCase): + """Regression coverage for a bug caught during this module's own + build: autoclean recursively drops nested empty-dict values, + treating a presence-only leaf like {"eth1": {}} as "nothing to + see here" -- but that's a meaningful member-interface reference, + not nothing. Caught by testing the full round-trip, not assumed.""" + + def test_members_preserved_in_device_shape(self): + result = _bond_entry_to_device( + {"members": [{"member": "eth1"}, {"member": "eth2"}]}, + ) + self.assertEqual(result, {"member": {"interface": {"eth1": {}, "eth2": {}}}}) + + def test_arp_monitor_target_preserved_in_device_shape(self): + result = _bond_entry_to_device( + {"arp_monitor": {"interval": 100, "target": ["192.0.2.1"]}}, + ) + self.assertEqual(result, {"arp-monitor": {"interval": 100, "target": {"192.0.2.1": {}}}}) + + def test_partial_member_change_generates_per_interface_commands(self): + """Confirmed correct: dict_op generates per-interface set/ + delete for a partial member list change, not a whole-node + delete -- that only happens when members is entirely absent + from want.""" + raw_have = {"bond0": {"member": {"interface": {"eth1": {}, "eth2": {}}}}} + config = [{"name": "bond0", "members": [{"member": "eth1"}, {"member": "eth3"}]}] + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["bond0", "member", "interface", "eth2"]), cmds) + self.assertIn(("set", _BASE + ["bond0", "member", "interface", "eth3"]), cmds) + self.assertNotIn(("delete", _BASE + ["bond0", "member"]), cmds) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_both_bonds_present(self): + have = self.gather() + names = {e["name"] for e in have} + self.assertEqual(names, {"bond0", "bond1"}) + + def test_bond0_full_fields_parsed(self): + have = self.gather() + bond0 = next(e for e in have if e["name"] == "bond0") + self.assertEqual(bond0["mode"], "802.3ad") + self.assertEqual(bond0["hash_policy"], "layer2") + self.assertEqual({m["member"] for m in bond0["members"]}, {"eth1", "eth2"}) + self.assertEqual(bond0["arp_monitor"]["interval"], 100) + self.assertEqual(set(bond0["arp_monitor"]["target"]), {"192.0.2.1", "192.0.2.2"}) + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_clear_all_omitted_fields_on_replaced(self): + cmds = build_commands([{"name": "bond0", "mode": "802.3ad"}], self.fixture, "replaced") + self.assertIn(("delete", _BASE + ["bond0", "primary"]), cmds) + self.assertIn(("delete", _BASE + ["bond0", "hash-policy"]), cmds) + self.assertIn(("delete", _BASE + ["bond0", "member"]), cmds) + self.assertIn(("delete", _BASE + ["bond0", "arp-monitor"]), cmds) + + def test_merged_new_bond(self): + config = [{"name": "bond2", "mode": "802.3ad", "members": [{"member": "eth5"}]}] + cmds = build_commands(config, {}, "merged") + self.assertIn(("set", _BASE + ["bond2", "mode", "802.3ad"]), cmds) + self.assertIn(("set", _BASE + ["bond2", "member", "interface", "eth5"]), cmds) if __name__ == "__main__": diff --git a/tests/unit/modules/test_vyos_ospf_interfaces.py b/tests/unit/modules/test_vyos_ospf_interfaces.py index 4d6be0d..ee58f23 100644 --- a/tests/unit/modules/test_vyos_ospf_interfaces.py +++ b/tests/unit/modules/test_vyos_ospf_interfaces.py @@ -4,30 +4,34 @@ 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, + _AF_OPTIONS, + ARGUMENT_SPEC, + _af4_entry_from_device, + _af4_entry_to_device, + _af6_entry_from_device, + _af6_entry_to_device, + _auth_from_device, + _auth_to_device, + _derive_key_field, + _device_to_argspec, + _kebab_fields, build_commands, + cast_by_spec, get_running_config, ) +from .base import load_fixture + _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() @@ -37,111 +41,217 @@ class VyOSModuleTestCase(unittest.TestCase): side_effect=lambda path: (self.ipv4_fixture if path == _BASE4 else self.ipv6_fixture), ) + def gather(self): + raw4, raw6 = get_running_config(self.mock_vyos) + have = _device_to_argspec(raw4, raw6) + for iface in have: + for af in iface.get("address_family") or []: + cast_by_spec(af, _AF_OPTIONS) + return have + + +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_both_afi_trees_separately(self): + raw4, raw6 = get_running_config(self.mock_vyos) + self.assertIn("eth1", raw4) + self.assertIn("eth1", raw6) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + raw4, raw6 = get_running_config(self.mock_vyos) + self.assertEqual(raw4, {}) + self.assertEqual(raw6, {}) + + +class TestDeriveKeyField(unittest.TestCase): + def test_derives_name(self): + opts = ARGUMENT_SPEC["config"]["options"] + self.assertEqual(_derive_key_field(opts), "name") + + def test_derives_afi(self): + af_opts = ARGUMENT_SPEC["config"]["options"]["address_family"]["options"] + self.assertEqual(_derive_key_field(af_opts), "afi") + + +class TestKebabFields(unittest.TestCase): + """Regression coverage for the confirmed bug from vyos_ospfv2's + build, applied here from the start: dict_op requires have's keys + to already be genuine device kebab-case, but autoclean deliberately + leaves keys as given.""" + + def test_converts_multiword_keys(self): + result = _kebab_fields({"dead_interval": 40, "mtu_ignore": True}) + self.assertEqual(result, {"dead-interval": 40, "mtu-ignore": {}}) + + +class TestAuthentication(unittest.TestCase): + """IPv4-only. Confirmed against vyos-1x: md5 is a fixed "key-id" + node containing a single tagNode entry, not a list.""" + + def test_plaintext_to_device(self): + result = _auth_to_device({"plaintext_password": "pw"}) + self.assertEqual(result, {"plaintext-password": "pw"}) + + def test_md5_to_device(self): + result = _auth_to_device({"md5_key": {"key_id": 10, "key": "secret"}}) + self.assertEqual(result, {"md5": {"key-id": {"10": {"md5-key": "secret"}}}}) -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] + def test_md5_from_device(self): + entry = _auth_from_device({"md5": {"key-id": {"10": {"md5-key": "secret"}}}}) + self.assertEqual(entry["md5_key"], {"key_id": 10, "key": "secret"}) + + def test_empty(self): + self.assertEqual(_auth_to_device({}), {}) + self.assertIsNone(_auth_from_device({})) + + +class TestAf6InstanceRename(unittest.TestCase): + """Confirmed genuine rename: the argspec's "instance" (matching + the CLI collection's naming) maps to the device's actual leaf + name "instance-id".""" + + def test_to_device(self): + result = _af6_entry_to_device({"instance": "5"}) + self.assertEqual(result, {"instance-id": "5"}) + + def test_from_device(self): + entry = _af6_entry_from_device({"instance-id": "5"}) + self.assertEqual(entry["instance"], "5") + + +class TestAf4EntryToDeviceFromDevice(unittest.TestCase): + def test_generic_fields(self): + result = _af4_entry_to_device({"cost": 100, "priority": 26}) + self.assertEqual(result, {"cost": 100, "priority": 26}) + + def test_mtu_ignore_presence(self): + result = _af4_entry_to_device({"mtu_ignore": True}) + self.assertEqual(result, {"mtu-ignore": {}}) + + def test_from_device_with_authentication(self): + entry = _af4_entry_from_device( + {"cost": "100", "authentication": {"plaintext-password": "pw"}}, + ) + self.assertEqual(entry["authentication"]["plaintext_password"], "pw") + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_both_interfaces_present(self): + have = self.gather() + names = [i["name"] for i in have] 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, []) + def test_eth1_has_both_afis(self): + have = self.gather() + eth1 = next(i for i in have if i["name"] == "eth1") + afis = {af["afi"] for af in eth1["address_family"]} + self.assertEqual(afis, {"ipv4", "ipv6"}) + def test_eth2_ipv4_only_minimal(self): + """eth2's ipv6 fixture entry is empty ({}) -- confirms an + interface with config on only one AFI doesn't spuriously + appear on the other.""" + have = self.gather() + eth2 = next(i for i in have if i["name"] == "eth2") + afis = {af["afi"] for af in eth2["address_family"]} + self.assertEqual(afis, {"ipv4"}) -class TestVyOSOspfInterfacesBuildCommands(unittest.TestCase): + def test_ipv4_fields_parsed(self): + have = self.gather() + eth1 = next(i for i in have if i["name"] == "eth1") + af4 = next(af for af in eth1["address_family"] if af["afi"] == "ipv4") + self.assertEqual(af4["cost"], 100) + self.assertTrue(af4["mtu_ignore"]) + self.assertEqual(af4["authentication"]["md5_key"]["key_id"], 10) - 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_ipv6_fields_parsed(self): + have = self.gather() + eth1 = next(i for i in have if i["name"] == "eth1") + af6 = next(af for af in eth1["address_family"] if af["afi"] == "ipv6") + self.assertTrue(af6["passive"]) + self.assertEqual(af6["instance"], "5") + self.assertEqual(af6["ifmtu"], 1500) - 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_empty(self): + self.assertEqual(_device_to_argspec({}, {}), []) - 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, []) +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + raw4, raw6 = get_running_config(self.mock_vyos) + self.assertEqual(build_commands(have, (raw4, raw6), "merged"), []) - 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_replaced_idempotent_against_own_fixture(self): + have = self.gather() + raw4, raw6 = get_running_config(self.mock_vyos) + self.assertEqual(build_commands(have, (raw4, raw6), "replaced"), []) - def test_overridden_removes_extra_interface(self): - have = [ - {"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}, - {"name": "eth2", "address_family": [{"afi": "ipv4", "cost": 200}]}, - ] + def test_clear_omitted_attribute_on_replaced(self): + """Primary confirmed bug: the original's clear-on-omit logic + only fired for op="replace", which was never actually passed + anywhere in build_commands -- genuinely dead code, since + "replaced" already deleted and recreated the whole AFI block + before calling it with an empty have.""" + raw4 = {"eth1": {"cost": "100", "priority": "26"}} config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] - cmds = build_commands(config, have, "overridden") - self.assertIn(("delete", _BASE4 + ["eth2"]), cmds) + cmds = build_commands(config, (raw4, {}), "replaced") + self.assertIn(("delete", _BASE4 + ["eth1", "priority"]), 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") + def test_replaced_does_not_touch_unrelated_interfaces(self): + """Confirmed fix for the original's disruptive "delete + everything and recreate" replaced heuristic.""" + raw4 = {"eth1": {"cost": "100"}, "eth2": {"cost": "200"}} + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + cmds = build_commands(config, (raw4, {}), "replaced") self.assertEqual(cmds, []) - def test_replaced_rebuilds_on_change(self): - have = [ + def test_ipv4_and_ipv6_target_separate_device_trees(self): + config = [ { "name": "eth1", - "address_family": [{"afi": "ipv4", "cost": 100, "transmit_delay": 50}], + "address_family": [ + {"afi": "ipv4", "cost": 100}, + {"afi": "ipv6", "passive": True}, + ], }, ] - config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 200}]}] - cmds = build_commands(config, have, "replaced") + cmds = build_commands(config, ({}, {}), "merged") + self.assertIn(("set", _BASE4 + ["eth1", "cost", "100"]), cmds) + self.assertIn(("set", _BASE6 + ["eth1", "passive"]), cmds) + + def test_overridden_removes_omitted_interface(self): + raw4 = {"eth1": {"cost": "100"}, "eth3": {"cost": "50"}} + config = [{"name": "eth1", "address_family": [{"afi": "ipv4", "cost": 100}]}] + cmds = build_commands(config, (raw4, {}), "overridden") + self.assertIn(("delete", _BASE4 + ["eth3"]), cmds) + + def test_deleted_all_covers_both_trees(self): + cmds = build_commands([], ({"eth1": {}}, {"eth1": {}}), "deleted") self.assertIn(("delete", _BASE4 + ["eth1"]), cmds) - self.assertIn(("set", _BASE4 + ["eth1", "cost", "200"]), cmds) + self.assertIn(("delete", _BASE6 + ["eth1"]), cmds) + self.assertEqual(len(cmds), 2) + + def test_deleted_named_no_afi_removes_both(self): + cmds = build_commands([{"name": "eth1"}], ({"eth1": {}}, {"eth1": {}}), "deleted") + self.assertIn(("delete", _BASE4 + ["eth1"]), cmds) + self.assertIn(("delete", _BASE6 + ["eth1"]), cmds) + self.assertEqual(len(cmds), 2) + + def test_deleted_named_specific_afi_only(self): + config = [{"name": "eth1", "address_family": [{"afi": "ipv4"}]}] + cmds = build_commands(config, ({"eth1": {}}, {"eth1": {}}), "deleted") + self.assertEqual(cmds, [("delete", _BASE4 + ["eth1"])]) + + def test_deleted_named_nonexistent_is_noop(self): + cmds = build_commands([{"name": "eth99"}], ({"eth1": {}}, {}), "deleted") + self.assertEqual(cmds, []) + + 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) if __name__ == "__main__": diff --git a/tests/unit/modules/test_vyos_ospfv2.py b/tests/unit/modules/test_vyos_ospfv2.py index c05dd77..01738a3 100644 --- a/tests/unit/modules/test_vyos_ospfv2.py +++ b/tests/unit/modules/test_vyos_ospfv2.py @@ -4,30 +4,36 @@ 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, + _CONFIG_OPTIONS, + ARGUMENT_SPEC, + _area_type_from_device, + _area_type_to_device, + _derive_key_field, + _device_to_argspec, + _distance_from_device, + _distance_to_device, + _kebab_fields, + _passive_from_device, + _passive_to_device, + _timers_from_device, + _timers_to_device_refresh, + _timers_to_device_throttle, + _vlink_auth_from_device, + _vlink_auth_to_device, build_commands, + cast_by_spec, get_running_config, ) - -_BASE = ["protocols", "ospf"] +from .base import load_fixture -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) +_BASE = ["protocols", "ospf"] class VyOSModuleTestCase(unittest.TestCase): @@ -36,144 +42,278 @@ class VyOSModuleTestCase(unittest.TestCase): self.fixture = load_fixture("ospfv2_running.json") self.mock_vyos.get_config = MagicMock(return_value=self.fixture) + def gather(self): + have = _device_to_argspec(self.fixture) + cast_by_spec(have, _CONFIG_OPTIONS) + return have -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={}) + +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_config_directly(self): result = get_running_config(self.mock_vyos) + self.assertIn("area", result) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) + + +class TestDeriveKeyField(unittest.TestCase): + def test_derives_area_id(self): + area_opts = ARGUMENT_SPEC["config"]["options"]["areas"]["options"] + self.assertEqual(_derive_key_field(area_opts), "area_id") + + def test_derives_neighbor_id(self): + nb_opts = ARGUMENT_SPEC["config"]["options"]["neighbor"]["options"] + self.assertEqual(_derive_key_field(nb_opts), "neighbor_id") + + def test_raises_if_none_required(self): + with self.assertRaises(ValueError): + _derive_key_field({"a": {"type": "str"}}) + + +class TestKebabFields(unittest.TestCase): + """Regression tests for the primary confirmed bug found during + development: dict_op requires have's keys to already be genuine + device kebab-case (it only normalizes underscores for its own + lookup index, not for the output path). autoclean deliberately + leaves keys as given, so any multi-word field reconstructed via + this module's own entry-transforms (rather than coming straight + from the device) needs explicit conversion, or a generated delete + command uses the wrong (snake_case) path segment.""" + + def test_converts_multiword_keys(self): + result = _kebab_fields({"default_cost": 20, "no_summary": True}) + self.assertEqual(result, {"default-cost": 20, "no-summary": {}}) + + def test_drops_none_and_false(self): + result = _kebab_fields({"default_cost": None, "no_summary": False}) self.assertEqual(result, {}) + def test_single_word_keys_unaffected(self): + result = _kebab_fields({"cost": 10}) + self.assertEqual(result, {"cost": 10}) -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)]) +class TestAreaType(unittest.TestCase): + def test_nssa_set_flag_is_node_presence_not_a_device_leaf(self): + """Confirmed genuine structural exception: the argspec's + nssa.set/stub.set boolean doesn't exist as a device leaf -- + the node's own presence IS the set flag.""" + result = _area_type_to_device({"nssa": {"set": True, "default_cost": 5}}) + self.assertEqual(result, {"nssa": {"default-cost": 5}}) + self.assertNotIn("set", result["nssa"]) - def test_deleted_without_have(self): - cmds = build_commands({}, {}, "deleted") - self.assertEqual(cmds, []) + def test_stub_no_summary(self): + """Regression test for the confirmed original bug: stub's + no_summary was declared in ARGUMENT_SPEC but never checked + anywhere in the original hand-rolled command-building logic.""" + result = _area_type_to_device({"stub": {"no_summary": True}}) + self.assertEqual(result, {"stub": {"no-summary": {}}}) - 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_from_device_restores_set_flag(self): + entry = _area_type_from_device({"nssa": {"default-cost": "5"}}) + self.assertTrue(entry["nssa"]["set"]) - 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_normal_presence(self): + result = _area_type_to_device({"normal": True}) + self.assertEqual(result, {"normal": {}}) - 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_empty(self): + self.assertEqual(_area_type_to_device({}), {}) + self.assertIsNone(_area_type_from_device({})) - 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, - ) +class TestDistance(unittest.TestCase): + def test_global_and_ospf(self): + result = _distance_to_device({"global": 110, "ospf": {"inter_area": 120}}) + self.assertEqual(result, {"global": 110, "ospf": {"inter-area": 120}}) - 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_from_device(self): + entry = _distance_from_device({"global": "110", "ospf": {"external": "150"}}) + self.assertEqual(entry["global"], 110) + self.assertEqual(entry["ospf"]["external"], "150") # cast_by_spec's job downstream + + +class TestTimers(unittest.TestCase): + """Confirmed genuine structural exception: the argspec groups + "refresh" and "throttle" both under one "timers" parent, but the + device has them as two separate top-level nodes.""" + + def test_refresh_maps_to_separate_top_level_node(self): + result = _timers_to_device_refresh({"refresh": {"timers": 300}}) + self.assertEqual(result, {"timers": 300}) + + def test_throttle_maps_to_device_timers_node(self): + result = _timers_to_device_throttle({"throttle": {"spf": {"delay": 200}}}) + self.assertEqual(result, {"throttle": {"spf": {"delay": 200}}}) + + def test_from_device_recombines_both(self): + entry = _timers_from_device({"timers": "300"}, {"throttle": {"spf": {"delay": "200"}}}) + self.assertEqual(entry["refresh"]["timers"], 300) + self.assertEqual(entry["throttle"]["spf"]["delay"], "200") - def test_merged_neighbor(self): - config = {"neighbor": [{"neighbor_id": "192.0.11.12", "priority": 2}]} + +class TestPassiveInterface(unittest.TestCase): + """Confirmed genuine structural exception: passive_interface and + passive_interface_exclude both map onto the same per-interface + "interface <name> passive" device subtree -- presence alone means + enabled, "passive.disable" means explicitly excluded.""" + + def test_to_device_both(self): + result = _passive_to_device(["eth1"], ["eth2"]) + self.assertEqual(result, {"eth1": {"passive": {}}, "eth2": {"passive": {"disable": {}}}}) + + def test_from_device_both(self): + passive, excluded = _passive_from_device( + {"eth1": {"passive": {}}, "eth2": {"passive": {"disable": {}}}}, + ) + self.assertEqual(passive, ["eth1"]) + self.assertEqual(excluded, ["eth2"]) + + def test_from_device_ignores_non_passive_interfaces(self): + passive, excluded = _passive_from_device({"eth3": {"some-other-key": {}}}) + self.assertEqual(passive, []) + self.assertEqual(excluded, []) + + +class TestVirtualLinkAuth(unittest.TestCase): + def test_md5_to_device(self): + result = _vlink_auth_to_device({"md5": [{"key_id": 10, "md5_key": "secret"}]}) + self.assertEqual(result, {"md5": {"10": {"md5-key": "secret"}}}) + + def test_plaintext_to_device(self): + result = _vlink_auth_to_device({"plaintext_password": "pw"}) + self.assertEqual(result, {"plaintext-password": "pw"}) + + def test_md5_from_device(self): + entry = _vlink_auth_from_device({"md5": {"10": {"md5-key": "secret"}}}) + self.assertEqual(entry["md5"], [{"key_id": 10, "md5_key": "secret"}]) + + +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_areas_parsed(self): + have = self.gather() + area_ids = [a["area_id"] for a in have["areas"]] + self.assertEqual(set(area_ids), {"2", "4", "5"}) + + def test_stub_area_no_summary_parsed(self): + have = self.gather() + area4 = next(a for a in have["areas"] if a["area_id"] == "4") + self.assertTrue(area4["area_type"]["stub"]["no_summary"]) + + def test_virtual_link_parsed(self): + have = self.gather() + area4 = next(a for a in have["areas"] if a["area_id"] == "4") + vlink = area4["virtual_link"][0] + self.assertEqual(vlink["address"], "10.0.0.1") + self.assertEqual(vlink["authentication"]["plaintext_password"], "secret") + self.assertEqual(vlink["dead_interval"], 40) + + def test_passive_interface_and_exclude_parsed(self): + have = self.gather() + self.assertEqual(have["passive_interface"], ["eth1"]) + self.assertEqual(have["passive_interface_exclude"], ["eth2"]) + + def test_timers_parsed(self): + have = self.gather() + self.assertEqual(have["timers"]["refresh"]["timers"], 300) + self.assertEqual(have["timers"]["throttle"]["spf"]["delay"], 200) + + def test_max_metric_parsed(self): + have = self.gather() + self.assertTrue(have["max_metric"]["router_lsa"]["administrative"]) + self.assertEqual(have["max_metric"]["router_lsa"]["on_shutdown"], 10) + + def test_mpls_te_parsed(self): + have = self.gather() + self.assertTrue(have["mpls_te"]["enabled"]) + self.assertEqual(have["mpls_te"]["router_address"], "192.0.11.11") + + def test_empty_config(self): + self.assertEqual(_device_to_argspec({}), {}) + self.assertEqual(_device_to_argspec(None), {}) + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_clear_omitted_attribute_on_replaced(self): + """Primary confirmed bug from the original hand-rolled + implementation: clearing an omitted attribute never generated + a delete command. Also the exact scenario that caught the + kebab-key regression during development.""" + raw_have = {"area": {"4": {"area-type": {"stub": {"default-cost": "20"}}}}} + config = {"areas": [{"area_id": "4", "area_type": {"stub": {"set": True}}}]} + cmds = build_commands(config, raw_have, "replaced") + expected = ("delete", _BASE + ["area", "4", "area-type", "stub", "default-cost"]) + self.assertIn(expected, cmds) + + def test_stub_no_summary_now_works(self): + config = {"areas": [{"area_id": "4", "area_type": {"stub": {"no_summary": True}}}]} + cmds = build_commands(config, {}, "merged") + expected = ("set", _BASE + ["area", "4", "area-type", "stub", "no-summary"]) + self.assertIn(expected, cmds) + + def test_replaced_only_touches_what_changed(self): + """Confirmed fix for the original's disruptive "delete + everything and recreate" replaced heuristic -- a targeted + dict_op purge only touches the sections that actually + differ.""" + raw_have = { + "area": {"2": {"area-type": {"normal": {}}}}, + "parameters": {"router-id": "1.1.1.1"}, + } + config = { + "areas": [{"area_id": "2", "area_type": {"normal": True}}], + "parameters": {"router_id": "2.2.2.2"}, + } + cmds = build_commands(config, raw_have, "replaced") + self.assertFalse(any("area" in str(c) for c in cmds)) + expected = ("set", _BASE + ["parameters", "router-id", "2.2.2.2"]) + self.assertIn(expected, cmds) + + def test_passive_interface_and_exclude_together(self): + config = {"passive_interface": ["eth1"], "passive_interface_exclude": ["eth2"]} 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) + self.assertIn(("set", _BASE + ["interface", "eth1", "passive"]), cmds) + self.assertIn(("set", _BASE + ["interface", "eth2", "passive", "disable"]), cmds) + + def test_deleted_with_have(self): + cmds = build_commands({}, {"parameters": {"router-id": "1.1.1.1"}}, "deleted") + self.assertEqual(cmds, [("delete", _BASE)]) - def test_merged_default_information(self): - config = {"default_information": {"originate": {"always": True, "metric": 10}}} + def test_deleted_no_have_is_noop(self): + self.assertEqual(build_commands({}, {}, "deleted"), []) + + def test_merged_new_virtual_link(self): + config = { + "areas": [ + { + "area_id": "4", + "virtual_link": [ + {"address": "10.0.0.1", "authentication": {"plaintext_password": "pw"}}, + ], + }, + ], + } cmds = build_commands(config, {}, "merged") - self.assertIn(("set", _BASE + ["default-information", "originate", "always"]), cmds) - self.assertIn(("set", _BASE + ["default-information", "originate", "metric", "10"]), cmds) + vlink_path = _BASE + [ + "area", + "4", + "virtual-link", + "10.0.0.1", + "authentication", + "plaintext-password", + "pw", + ] + self.assertIn(("set", vlink_path), cmds) if __name__ == "__main__": diff --git a/tests/unit/modules/test_vyos_ospfv3.py b/tests/unit/modules/test_vyos_ospfv3.py index a84041b..a19b5fa 100644 --- a/tests/unit/modules/test_vyos_ospfv3.py +++ b/tests/unit/modules/test_vyos_ospfv3.py @@ -4,136 +4,218 @@ 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 ( + _CONFIG_OPTIONS, + ARGUMENT_SPEC, + _area_entry_from_device, + _area_entry_to_device, + _derive_key_field, + _device_to_argspec, + _kebab_fields, build_commands, + cast_by_spec, get_running_config, ) +from .base import load_fixture -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) + +_BASE = ["protocols", "ospfv3"] 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") + self.mock_vyos.get_config = MagicMock(return_value=self.fixture) - 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 gather(self): + have = _device_to_argspec(self.fixture) + cast_by_spec(have, _CONFIG_OPTIONS) + return have - def test_parses_redistribute(self): - self.set_running_config(self.fixture) + +class TestGetRunningConfig(VyOSModuleTestCase): + def test_returns_config_directly(self): 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) + self.assertIn("area", result) + + def test_empty_config(self): + self.mock_vyos.get_config = MagicMock(return_value=None) + self.assertEqual(get_running_config(self.mock_vyos), {}) + + def test_collapsed_response_normalized(self): + """Regression coverage for the confirmed failure mode from + vyos_ospf_interfaces's build: an unguarded collapsed response + (VyOS's single-value tag-node quirk) would otherwise be + iterated character-by-character downstream.""" + self.mock_vyos.get_config = MagicMock(return_value="eth1") 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"]) + self.assertEqual(result, {"eth1": {}}) - def test_empty_config_returns_empty(self): - self.set_running_config({}) - result = get_running_config(self.mock_vyos) - self.assertEqual(result, {}) +class TestDeriveKeyField(unittest.TestCase): + def test_derives_area_id(self): + area_opts = ARGUMENT_SPEC["config"]["options"]["areas"]["options"] + self.assertEqual(_derive_key_field(area_opts), "area_id") -class TestVyOSOspfv3BuildCommands(unittest.TestCase): + def test_raises_if_none_required(self): + with self.assertRaises(ValueError): + _derive_key_field({"a": {"type": "str"}}) - 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, []) +class TestKebabFields(unittest.TestCase): + def test_converts_multiword_keys(self): + result = _kebab_fields({"router_id": "1.1.1.1"}) + self.assertEqual(result, {"router-id": "1.1.1.1"}) - 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, + +class TestAreaEntry(unittest.TestCase): + def test_export_import_list_to_device(self): + result = _area_entry_to_device({"export_list": "el1", "import_list": "il1"}) + self.assertEqual(result, {"export-list": "el1", "import-list": "il1"}) + + def test_range_to_device(self): + result = _area_entry_to_device( + {"range": [{"address": "2001:db1::/32", "not_advertise": True}]}, ) + self.assertEqual(result, {"range": {"2001:db1::/32": {"not-advertise": {}}}}) - 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_from_device(self): + entry = _area_entry_from_device( + {"export-list": "el1", "range": {"2001:db1::/32": {"advertise": {}}}}, ) + self.assertEqual(entry["export_list"], "el1") + self.assertEqual(entry["range"][0]["address"], "2001:db1::/32") + self.assertTrue(entry["range"][0]["advertise"]) + + def test_empty(self): + self.assertEqual(_area_entry_to_device({}), {}) + self.assertEqual(_area_entry_from_device({}), {}) - 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): +class TestDeviceToArgspecFixture(VyOSModuleTestCase): + def test_areas_parsed(self): + have = self.gather() + area_ids = {a["area_id"] for a in have["areas"]} + self.assertEqual(area_ids, {"2", "3"}) + + def test_area_export_import_list_parsed(self): + have = self.gather() + area2 = next(a for a in have["areas"] if a["area_id"] == "2") + self.assertEqual(area2["export_list"], "export1") + self.assertEqual(area2["import_list"], "import1") + + def test_range_advertise_not_advertise_parsed(self): + have = self.gather() + area2 = next(a for a in have["areas"] if a["area_id"] == "2") + r1 = next(r for r in area2["range"] if r["address"] == "2001:db10::/32") + r2 = next(r for r in area2["range"] if r["address"] == "2001:db20::/32") + self.assertNotIn("advertise", r1) + self.assertNotIn("not_advertise", r1) + self.assertTrue(r2["not_advertise"]) + + area3 = next(a for a in have["areas"] if a["area_id"] == "3") + self.assertTrue(area3["range"][0]["advertise"]) + + def test_parameters_parsed(self): + have = self.gather() + self.assertEqual(have["parameters"]["router_id"], "192.0.2.10") + + def test_redistribute_parsed(self): + have = self.gather() + route_types = {r["route_type"] for r in have["redistribute"]} + self.assertEqual(route_types, {"bgp", "static"}) + bgp = next(r for r in have["redistribute"] if r["route_type"] == "bgp") + self.assertEqual(bgp["route_map"], "redist-map") + + def test_empty(self): + self.assertEqual(_device_to_argspec({}), {}) + self.assertEqual(_device_to_argspec(None), {}) + + +class TestBuildCommands(VyOSModuleTestCase): + def test_merged_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "merged"), []) + + def test_replaced_idempotent_against_own_fixture(self): + have = self.gather() + self.assertEqual(build_commands(have, self.fixture, "replaced"), []) + + def test_changing_existing_range_flag_via_replaced(self): + """Primary confirmed bug from the original hand-rolled + implementation: range advertise/not_advertise were only ever + set on initial range creation (an "if addr not in have_ranges" + guard) -- changing an existing range's flag generated no + command under either state. dict_op's purge (via "replaced") + correctly handles this generically.""" + raw_have = {"area": {"2": {"range": {"2001:db20::/32": {"not-advertise": {}}}}}} config = { - "areas": [{"area_id": "2", "range": [{"address": "2001:db10::/32"}]}], + "areas": [ + { + "area_id": "2", + "range": [ + {"address": "2001:db20::/32", "advertise": True}, + ], + }, + ], } - cmds = build_commands(config, {}, "merged") + cmds = build_commands(config, raw_have, "replaced") self.assertIn( - ("set", ["protocols", "ospfv3", "area", "2", "range", "2001:db10::/32"]), + ("delete", _BASE + ["area", "2", "range", "2001:db20::/32", "not-advertise"]), 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"]), + ("set", _BASE + ["area", "2", "range", "2001:db20::/32", "advertise"]), cmds, ) - def test_merged_area_export_list(self): - config = {"areas": [{"area_id": "2", "export_list": "export1"}]} + def test_merged_does_not_clear_conflicting_flag(self): + """merged never runs a purge pass, by design (established, + consistent semantic throughout this collection) -- confirms + it correctly does NOT clear not-advertise even when advertise + is set instead, unlike replaced above.""" + raw_have = {"area": {"2": {"range": {"2001:db20::/32": {"not-advertise": {}}}}}} + config = { + "areas": [ + { + "area_id": "2", + "range": [ + {"address": "2001:db20::/32", "advertise": True}, + ], + }, + ], + } + cmds = build_commands(config, raw_have, "merged") + self.assertFalse(any(c[0] == "delete" and "not-advertise" in str(c) for c in cmds)) + + def test_replaced_is_whole_resource_not_scoped(self): + """Confirmed matching this module's own documented contract + ("replaced replaces the module-managed OSPFv3 configuration") and + vyos_ospfv2's own established precedent -- unlike + vyos_static_routes/vyos_route_maps, which scope replaced per + named item.""" + raw_have = {"parameters": {"router-id": "1.1.1.1"}, "redistribute": {"bgp": {}}} + config = {"parameters": {"router_id": "1.1.1.1"}} + cmds = build_commands(config, raw_have, "replaced") + self.assertIn(("delete", _BASE + ["redistribute"]), cmds) + + def test_deleted_with_have(self): + cmds = build_commands({}, {"parameters": {"router-id": "1.1.1.1"}}, "deleted") + self.assertEqual(cmds, [("delete", _BASE)]) + + def test_deleted_no_have_is_noop(self): + self.assertEqual(build_commands({}, {}, "deleted"), []) + + def test_merged_new_area(self): + config = {"areas": [{"area_id": "5", "export_list": "el5"}]} cmds = build_commands(config, {}, "merged") - self.assertIn( - ("set", ["protocols", "ospfv3", "area", "2", "export-list", "export1"]), - cmds, - ) + self.assertIn(("set", _BASE + ["area", "5", "export-list", "el5"]), cmds) if __name__ == "__main__": diff --git a/tests/unit/test_module_utils_vyos.py b/tests/unit/test_module_utils_vyos.py index 920e9a5..213a2fb 100644 --- a/tests/unit/test_module_utils_vyos.py +++ b/tests/unit/test_module_utils_vyos.py @@ -14,7 +14,12 @@ __metaclass__ = type import unittest -from ansible_collections.vyos.rest.plugins.module_utils.vyos import cast_by_spec +from unittest.mock import MagicMock + +from ansible_collections.vyos.rest.plugins.module_utils.vyos import ( + VyOSModule, + cast_by_spec, +) class TestCastBySpecIntCollapse(unittest.TestCase): @@ -51,5 +56,86 @@ class TestCastBySpecIntCollapse(unittest.TestCase): self.assertIsNone(entry["distance"]) +class TestGetValue(unittest.TestCase): + """Regression tests for a confirmed bug (Copilot): get_value() + previously did `result.get("data") or ""`, which incorrectly + converts any falsy-but-valid scalar (0, False, an already-empty + string) into an empty string -- indistinguishable from the value + being genuinely unset. Fixed to only treat a missing "data" key + (None) as unset, leaving every other value -- including falsy + ones -- exactly as returned.""" + + def _vyos_with_data(self, data): + vyos = VyOSModule.__new__(VyOSModule) + vyos._client = MagicMock() + vyos._client.retrieve_return_value.return_value = {"data": data} + return vyos + + def test_zero_preserved_not_emptied(self): + vyos = self._vyos_with_data(0) + result = vyos.get_value(["some", "path"]) + self.assertEqual(result, 0) + self.assertIs(type(result), int) + + def test_false_preserved_not_emptied(self): + vyos = self._vyos_with_data(False) + self.assertIs(vyos.get_value(["some", "path"]), False) + + def test_genuine_string_value_passes_through(self): + vyos = self._vyos_with_data("vyos-core-01") + self.assertEqual(vyos.get_value(["some", "path"]), "vyos-core-01") + + def test_already_empty_string_stays_empty(self): + vyos = self._vyos_with_data("") + self.assertEqual(vyos.get_value(["some", "path"]), "") + + def test_missing_data_key_becomes_empty_string(self): + vyos = VyOSModule.__new__(VyOSModule) + vyos._client = MagicMock() + vyos._client.retrieve_return_value.return_value = {} + self.assertEqual(vyos.get_value(["some", "path"]), "") + + +class TestShow(unittest.TestCase): + """Regression tests for the same confirmed bug class as + TestGetValue, found independently in show(): `result.get("data") + or ""` incorrectly converts any falsy-but-valid scalar (0, False, + an already-empty string) from an operational show command into an + empty string -- indistinguishable from the command genuinely + returning nothing. Fixed to only treat a missing "data" key + (None) as empty, leaving every other value -- including falsy + ones -- exactly as returned.""" + + def _vyos_with_data(self, data): + vyos = VyOSModule.__new__(VyOSModule) + vyos._client = MagicMock() + vyos._client.show.return_value = {"data": data} + return vyos + + def test_zero_preserved_not_emptied(self): + vyos = self._vyos_with_data(0) + result = vyos.show(["some", "op", "path"]) + self.assertEqual(result, 0) + self.assertIs(type(result), int) + + def test_false_preserved_not_emptied(self): + vyos = self._vyos_with_data(False) + self.assertIs(vyos.show(["some", "op", "path"]), False) + + def test_genuine_output_passes_through(self): + vyos = self._vyos_with_data("interface eth0 up") + self.assertEqual(vyos.show(["some", "op", "path"]), "interface eth0 up") + + def test_already_empty_string_stays_empty(self): + vyos = self._vyos_with_data("") + self.assertEqual(vyos.show(["some", "op", "path"]), "") + + def test_missing_data_key_becomes_empty_string(self): + vyos = VyOSModule.__new__(VyOSModule) + vyos._client = MagicMock() + vyos._client.show.return_value = {} + self.assertEqual(vyos.show(["some", "op", "path"]), "") + + if __name__ == "__main__": unittest.main() |