T8989: wave4 vyos_command, dict_op refactorT8989_wave4

* T8989: vyos_command module

* T8989: vyos_command module UAT and SIT

* T8989: vyos_command changelog

* T8989: vyos_command linter

* T8989: vyos_config module

* T8989: vyos_config module changelog

* T8989: Wave 4 vyos_config module with integration and unit tests

* T8323: vyos_system module

* T8332: vyos_system SIT and UAT

* T8323: vyos_vlan module

* T8323: vyos_vlan module

* T8323: vyos_vlan module SIT and UAT

* T8323: vyos_system module

* T8989: Wave 4 vyos_vlan reworked with dict_op engine

* T8989: Fix dict_op single-value string list handling, add vyos_system integration tests

* T8989: logging_global refactor

* T8989: migrate ntp_global, logging_global, firewall_global to dict_op engine

* T8989: vyos_nat module for REST API collection

* T8989: vyos_nat module for REST API collection, linter fixes

* T8989: vyos_ha module for REST API collection

* T8989: vyos_ha module for REST API collection

* T8989: vyos_ha module sanity and linter fixes

* T8989: vyos_ha module sanity and linter fixes

* T8989: vyos_ha module linter fixes

* T8989: vyos.rest AI comment fixes

* T8323: vyos_nat AI comment fixes

* T8989 ai fixes

* T8989: vyos_bgp_address_family dict_op

* T8989: vyos_bgp_address_family vyos_bgp_global dict_op

* T8989: dict_op refactor for firewall_*, nat, user

* T8989: dict_op refactor for firewall_*, nat, user

* T8989: dict_op refactor for ntp_global, ha

* T8989: snmp_server dict_op refactor

* T8989: snmp_server dict_op refactor

* T8989: route_map dict_op refactor

1 files changed, 219 insertions, 122 deletions

diff --git a/tests/unit/modules/test_vyos_bgp_global.py b/tests/unit/modules/test_vyos_bgp_global.py
index f91516b..118cea8 100644
--- a/tests/unit/modules/test_vyos_bgp_global.py
+++ b/tests/unit/modules/test_vyos_bgp_global.py
@@ -4,25 +4,25 @@ 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_bgp_global import (
+    _device_to_argspec,
+    _neighbors_from_device,
+    _neighbors_to_device,
+    _peer_groups_from_device,
+    _peer_groups_to_device,
+    _want_to_device,
     build_commands,
     get_running_config,
 )
 
-
-_BASE = ["protocols", "bgp"]
+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", "bgp"]
 
 
 class VyOSModuleTestCase(unittest.TestCase):
@@ -32,156 +32,253 @@ class VyOSModuleTestCase(unittest.TestCase):
         self.mock_vyos.get_config = MagicMock(return_value=self.fixture)
 
 
-class TestVyOSBgpGlobalGetRunning(VyOSModuleTestCase):
-
-    def test_parses_as_number(self):
-        result = get_running_config(self.mock_vyos)
-        self.assertEqual(result["as_number"], 65000)
-
-    def test_parses_parameters(self):
-        result = get_running_config(self.mock_vyos)
-        self.assertEqual(result["parameters"]["router_id"], "192.0.1.1")
-        self.assertTrue(result["parameters"]["log_neighbor_changes"])
-
-    def test_parses_neighbors(self):
-        result = get_running_config(self.mock_vyos)
-        nb_addrs = [n["neighbor_address"] for n in result["neighbors"]]
-        self.assertIn("192.0.2.1", nb_addrs)
-        self.assertIn("192.0.2.2", nb_addrs)
-        nb1 = next(n for n in result["neighbors"] if n["neighbor_address"] == "192.0.2.1")
-        self.assertEqual(nb1["remote_as"], 65001)
-        self.assertEqual(nb1["description"], "peer1")
-        self.assertEqual(nb1["timers"]["holdtime"], 30)
-        self.assertEqual(nb1["timers"]["keepalive"], 10)
-        nb2 = next(n for n in result["neighbors"] if n["neighbor_address"] == "192.0.2.2")
-        self.assertEqual(nb2["ebgp_multihop"], 2)
-        self.assertEqual(nb2["update_source"], "eth0")
-
-    def test_parses_peer_groups(self):
-        result = get_running_config(self.mock_vyos)
-        self.assertEqual(len(result["peer_groups"]), 1)
-        self.assertEqual(result["peer_groups"][0]["peer_group"], "PG1")
-        self.assertEqual(result["peer_groups"][0]["remote_as"], 65003)
+class TestGetRunningConfig(VyOSModuleTestCase):
+    def test_returns_raw_device_dict(self):
+        self.assertEqual(get_running_config(self.mock_vyos), self.fixture)
 
     def test_empty_config(self):
-        self.mock_vyos.get_config = MagicMock(return_value={})
-        result = get_running_config(self.mock_vyos)
-        self.assertEqual(result, {})
+        self.mock_vyos.get_config = MagicMock(return_value=None)
+        self.assertEqual(get_running_config(self.mock_vyos), {})
 
 
-class TestVyOSBgpGlobalBuildCommands(unittest.TestCase):
+class TestNeighborsToDeviceFromDevice(unittest.TestCase):
+    def test_bare_neighbor_is_presence(self):
+        self.assertEqual(
+            _neighbors_to_device([{"neighbor_address": "192.0.2.1"}]),
+            {"192.0.2.1": {}},
+        )
 
-    def _have(self):
-        return {
-            "as_number": 65000,
-            "parameters": {"router_id": "192.0.1.1"},
-            "neighbors": [
+    def test_full_neighbor(self):
+        result = _neighbors_to_device(
+            [
                 {
                     "neighbor_address": "192.0.2.1",
                     "remote_as": 65001,
                     "description": "peer1",
+                    "shutdown": True,
+                    "timers": {"holdtime": 30, "keepalive": 10},
                 },
             ],
-            "peer_groups": [{"peer_group": "PG1", "remote_as": 65003}],
-        }
+        )
+        self.assertEqual(
+            result,
+            {
+                "192.0.2.1": {
+                    "remote_as": 65001,
+                    "description": "peer1",
+                    "shutdown": {},
+                    "timers": {"holdtime": 30, "keepalive": 10},
+                },
+            },
+        )
 
-    def test_deleted_with_have(self):
-        cmds = build_commands({}, self._have(), "deleted")
-        self.assertEqual(cmds, [("delete", _BASE)])
+    def test_from_device_ints_cast_via_argspec(self):
+        result = _neighbors_from_device(
+            {
+                "192.0.2.1": {
+                    "remote-as": "65001",
+                    "ebgp-multihop": "2",
+                    "timers": {"holdtime": "30", "keepalive": "10"},
+                },
+            },
+        )
+        entry = result[0]
+        self.assertEqual(entry["remote_as"], 65001)
+        self.assertEqual(entry["ebgp_multihop"], 2)
+        self.assertEqual(entry["timers"], {"holdtime": 30, "keepalive": 10})
+
+    def test_from_device_foreign_address_family_never_surfaces(self):
+        """Regression test: a neighbor's address-family subtree (owned by
+        vyos_bgp_address_family) must never appear in this module's have/
+        gathered output."""
+        result = _neighbors_from_device(
+            {
+                "192.0.2.1": {
+                    "remote-as": "65001",
+                    "address-family": {"ipv4-unicast": {"nexthop-self": {}}},
+                },
+            },
+        )
+        entry = result[0]
+        self.assertEqual(entry["remote_as"], 65001)
+        self.assertNotIn("address_family", entry)
 
-    def test_deleted_without_have(self):
-        cmds = build_commands({}, {}, "deleted")
-        self.assertEqual(cmds, [])
 
-    def test_merged_as_number(self):
-        config = {"as_number": 65000}
-        cmds = build_commands(config, {}, "merged")
-        self.assertIn(("set", _BASE + ["system-as", "65000"]), cmds)
+class TestPeerGroupsToDeviceFromDevice(unittest.TestCase):
+    def test_bare_peer_group_is_presence(self):
+        self.assertEqual(_peer_groups_to_device([{"peer_group": "PG1"}]), {"PG1": {}})
 
-    def test_merged_router_id(self):
-        config = {"as_number": 65000, "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_full_peer_group(self):
+        result = _peer_groups_to_device(
+            [{"peer_group": "PG1", "remote_as": 65002, "timers": {"holdtime": 30}}],
+        )
+        self.assertEqual(
+            result,
+            {"PG1": {"remote_as": 65002, "timers": {"holdtime": 30}}},
+        )
 
-    def test_merged_neighbor(self):
-        config = {
-            "as_number": 65000,
-            "neighbors": [
-                {"neighbor_address": "192.0.2.1", "remote_as": 65001},
-            ],
-        }
-        cmds = build_commands(config, {}, "merged")
-        self.assertIn(("set", _BASE + ["neighbor", "192.0.2.1", "remote-as", "65001"]), cmds)
+    def test_from_device_cast(self):
+        result = _peer_groups_from_device({"PG1": {"remote-as": "65002"}})
+        self.assertEqual(result, [{"peer_group": "PG1", "remote_as": 65002}])
 
-    def test_merged_neighbor_timers(self):
+
+class TestWantToDevice(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual(_want_to_device({}), {})
+        self.assertEqual(_want_to_device(None), {})
+
+    def test_confederation_peers_list_passes_through(self):
+        result = _want_to_device(
+            {
+                "as_number": 65000,
+                "parameters": {"confederation": {"identifier": 100, "peers": [65001, 65002]}},
+            },
+        )
+        self.assertEqual(
+            result,
+            {
+                "system_as": 65000,
+                "parameters": {"confederation": {"identifier": 100, "peers": [65001, 65002]}},
+            },
+        )
+
+    def test_full_config(self):
         config = {
             "as_number": 65000,
-            "neighbors": [
-                {
-                    "neighbor_address": "192.0.2.1",
-                    "remote_as": 65001,
-                    "timers": {"holdtime": 30, "keepalive": 10},
-                },
-            ],
+            "parameters": {"router_id": "192.0.1.1", "graceful_restart": True},
+            "neighbors": [{"neighbor_address": "192.0.2.1", "remote_as": 65001}],
+            "peer_groups": [{"peer_group": "PG1", "remote_as": 65002}],
         }
-        cmds = build_commands(config, {}, "merged")
+        result = _want_to_device(config)
+        self.assertEqual(
+            result,
+            {
+                "system_as": 65000,
+                "parameters": {"router_id": "192.0.1.1", "graceful_restart": {}},
+                "neighbor": {"192.0.2.1": {"remote_as": 65001}},
+                "peer_group": {"PG1": {"remote_as": 65002}},
+            },
+        )
+
+
+class TestDeviceToArgspecFixture(VyOSModuleTestCase):
+    def test_as_number_and_parameters(self):
+        have = _device_to_argspec(self.fixture)
+        self.assertEqual(have["as_number"], 65000)
+        self.assertEqual(have["parameters"]["router_id"], "192.0.1.1")
+        self.assertEqual(
+            have["parameters"]["confederation"],
+            {"identifier": 100, "peers": [65001, 65002]},
+        )
+
+    def test_neighbor_and_peer_group(self):
+        have = _device_to_argspec(self.fixture)
+        nb = next(n for n in have["neighbors"] if n["neighbor_address"] == "192.0.2.1")
+        self.assertEqual(nb["remote_as"], 65001)
+        self.assertEqual(nb["timers"], {"holdtime": 30, "keepalive": 10})
+        self.assertNotIn("address_family", nb)
+        pg = have["peer_groups"][0]
+        self.assertEqual(pg["peer_group"], "PG1")
+        self.assertEqual(pg["remote_as"], 65003)
+
+    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 = _device_to_argspec(self.fixture)
+        self.assertEqual(build_commands(have, self.fixture, "merged"), [])
+
+    def test_replaced_idempotent_against_own_fixture(self):
+        have = _device_to_argspec(self.fixture)
+        self.assertEqual(build_commands(have, self.fixture, "replaced"), [])
+
+    def test_replaced_purges_extra_confederation_peer(self):
+        """Regression test: dict_op's purge mode originally had no
+        handling for list-valued leaves at all (only dicts), so removing
+        a peer from confederation.peers under 'replaced' silently did
+        nothing. Fixed centrally in dict_op itself."""
+        have = _device_to_argspec(self.fixture)
+        have["parameters"]["confederation"]["peers"] = [65001]
+        cmds = build_commands(have, self.fixture, "replaced")
         self.assertIn(
-            ("set", _BASE + ["neighbor", "192.0.2.1", "timers", "holdtime", "30"]),
+            ("delete", _BASE + ["parameters", "confederation", "peers", "65002"]),
             cmds,
         )
+
+    def test_merged_new_neighbor_field(self):
+        have = _device_to_argspec(self.fixture)
+        have["neighbors"][0]["local_as"] = 65099
+        cmds = build_commands(have, self.fixture, "merged")
         self.assertIn(
-            ("set", _BASE + ["neighbor", "192.0.2.1", "timers", "keepalive", "10"]),
+            ("set", _BASE + ["neighbor", "192.0.2.1", "local-as", "65099"]),
             cmds,
         )
 
-    def test_merged_idempotent(self):
-        have = self._have()
-        config = {
-            "as_number": 65000,
-            "parameters": {"router_id": "192.0.1.1"},
-            "neighbors": [
-                {
-                    "neighbor_address": "192.0.2.1",
-                    "remote_as": 65001,
-                    "description": "peer1",
-                },
-            ],
-            "peer_groups": [{"peer_group": "PG1", "remote_as": 65003}],
-        }
-        cmds = build_commands(config, have, "merged")
+    def test_replaced_never_touches_address_family(self):
+        """Regression test: this module shares protocols.bgp with
+        vyos_bgp_address_family; replaced/deleted must never purge or
+        delete that sibling module's address-family subtree."""
+        cmds = build_commands({"as_number": 65000}, self.fixture, "replaced")
+        self.assertTrue(all("address-family" not in c[1] for c in cmds))
+
+    def test_deleted_removes_atomically_not_scoped(self):
+        """Regression test for the real device-model bug: VyOS rejects any
+        commit that removes system-as while other protocols.bgp content
+        (including a neighbor's address-family, owned by
+        vyos_bgp_address_family) still exists. A scoped/incremental
+        deletion here would leave an invalid intermediate state and hard-
+        fail at commit time -- deleted must delete the whole tree in one
+        atomic command whenever system-as is present."""
+        cmds = build_commands({}, self.fixture, "deleted")
+        self.assertEqual(cmds, [("delete", _BASE)])
+
+    def test_deleted_with_no_system_as_is_a_noop(self):
+        cmds = build_commands({}, {}, "deleted")
         self.assertEqual(cmds, [])
 
-    def test_replaced_idempotent(self):
-        have = self._have()
-        config = {
-            "as_number": 65000,
-            "parameters": {"router_id": "192.0.1.1"},
-            "neighbors": [
-                {
-                    "neighbor_address": "192.0.2.1",
-                    "remote_as": 65001,
-                    "description": "peer1",
-                },
-            ],
-            "peer_groups": [{"peer_group": "PG1", "remote_as": 65003}],
-        }
-        cmds = build_commands(config, have, "replaced")
+    def test_replaced_without_as_number_also_nukes_atomically(self):
+        """The same VyOS constraint applies to 'replaced' whenever the new
+        desired state omits as_number -- not just 'deleted'."""
+        cmds = build_commands({"neighbors": []}, self.fixture, "replaced")
+        self.assertEqual(cmds, [("delete", _BASE)])
+
+    def test_merged_with_empty_config_is_a_safe_noop(self):
+        """Regression test: merged must NEVER trigger the nuke short-
+        circuit just because as_number was omitted -- an omitted config
+        for merged means "nothing to change", not "delete everything"."""
+        cmds = build_commands({}, self.fixture, "merged")
+        self.assertEqual(cmds, [])
+
+    def test_replaced_keeping_as_number_still_scopes_normally(self):
+        """When as_number is retained, replaced must still use the normal
+        scoped purge/set flow, not the atomic nuke."""
+        have = _device_to_argspec(self.fixture)
+        cmds = build_commands(have, self.fixture, "replaced")
         self.assertEqual(cmds, [])
+        self.assertNotEqual(cmds, [("delete", _BASE)])
 
-    def test_replaced_rebuilds_on_change(self):
-        have = self._have()
-        config = {"as_number": 65000, "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_collapsed_single_neighbor_no_char_iteration_bug(self):
+        """A neighbor tag node collapsed to a bare address string by the
+        device (single neighbor, otherwise unconfigured) must not be
+        iterated character-by-character."""
+        raw_have = {"system-as": "65000", "neighbor": "192.0.2.1"}
+        config = {"as_number": 65000, "neighbors": [{"neighbor_address": "192.0.2.1"}]}
+        self.assertEqual(build_commands(config, raw_have, "merged"), [])
 
-    def test_merged_peer_group(self):
+    def test_fresh_merged_add(self):
         config = {
             "as_number": 65000,
-            "peer_groups": [{"peer_group": "PG1", "remote_as": 65003}],
+            "neighbors": [{"neighbor_address": "10.0.0.1", "remote_as": 65010}],
         }
         cmds = build_commands(config, {}, "merged")
-        self.assertIn(("set", _BASE + ["peer-group", "PG1", "remote-as", "65003"]), cmds)
+        self.assertIn(
+            ("set", _BASE + ["neighbor", "10.0.0.1", "remote-as", "65010"]),
+            cmds,
+        )
+        self.assertIn(("set", _BASE + ["system-as", "65000"]), cmds)
 
 
 if __name__ == "__main__":