diff options
Diffstat (limited to 'tests')
| -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/modules/test_vyos_ospf_interfaces.py | 298 |
3 files changed, 221 insertions, 97 deletions
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/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__": |