T8989: ospv2 dict_op refactor (#30)

* T8989: vyos_ospfv2 dict_op refactor

* T8989: vyos_ospfv2 dict_op refactor

* T8989: ospv2 AI comment fixes

---------

Co-authored-by: Daniil Baturin <daniil@vyos.io>

1 files changed, 274 insertions, 134 deletions

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__":