Merge pull request 'fix: sync_dns_zone no longer copies an override made by hand' (#18) from fix/hand-made-override-copies into master

napalm_opnsense/opnsense.py

@@ -2933,8 +2933,17 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
         # Re-add all enabled A/AAAA records, one override per logical record.
         # Two rows for the same name is a state netOrk's own database can
         # legitimately be in; the device must still end up with one.
+        # An enabled override made by hand already answers for its record: the
+        # caller's records include what it read off this device, and adding
+        # those again under the tag left each one twice (NetOrk/netork#1123).
         added = 0
-        pushed: set[tuple[str, str, str]] = set()
+        pushed: set[tuple[str, str, str]] = {
+            (entry["hostname"].lower(), entry.get("record_type", "A"), entry["ip"])
+            for entry in existing
+            if entry["domain"].rstrip(".") == zone_clean
+            and not entry["description"].startswith(self._NETORK_TAG)
+            and entry.get("enabled", True)
+        }
         for rec in records:
             if not rec.get("enabled", True):
                 continue

tests/unit/test_driver.py

@@ -2596,6 +2596,70 @@ class TestSyncDnsZone:
         added = [d for p, d in calls if p.endswith("addhostoverride")]
         assert len(added) == 1
 
+    def test_does_not_copy_an_override_made_by_hand(self, driver):
+        """The records netOrk sends include what its poll read off this very
+        device, hand-made overrides among them. Adding those again under the
+        tag left every one of them twice on gw-01 and gw-02 (NetOrk/netork#1123)."""
+        rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")]
+        calls = self._record_calls(driver, rows)
+
+        driver.sync_dns_zone("office.local", [
+            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+        ])
+
+        assert [p for p, _ in calls if p.endswith("addhostoverride")] == []
+        assert [p for p, _ in calls if "delhostoverride" in p] == []
+
+    def test_clears_a_copy_it_once_made_of_a_hand_made_override(self, driver):
+        """Devices that already carry the copy lose it on the next sync."""
+        rows = [
+            _override_row("by-hand", host="NAS", ip="172.22.40.8", description="NAS"),
+            _override_row("copy", host="nas", ip="172.22.40.8"),
+        ]
+        calls = self._record_calls(driver, rows)
+
+        driver.sync_dns_zone("office.local", [
+            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+            {"hostname": "NAS", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+        ])
+
+        assert [p.rsplit("/", 1)[-1] for p, _ in calls if "delhostoverride" in p] == ["copy"]
+        assert [p for p, _ in calls if p.endswith("addhostoverride")] == []
+
+    def test_still_adds_a_record_that_differs_from_the_hand_made_one(self, driver):
+        rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="")]
+        calls = self._record_calls(driver, rows)
+
+        driver.sync_dns_zone("office.local", [
+            {"hostname": "nas", "ip": "172.22.40.9", "record_type": "A", "enabled": True},
+        ])
+
+        added = [d for p, d in calls if p.endswith("addhostoverride")]
+        assert [d["host"]["server"] for d in added] == ["172.22.40.9"]
+
+    def test_a_disabled_hand_made_override_does_not_stand_in(self, driver):
+        """It answers nothing, so the record still has to be added."""
+        rows = [_override_row("by-hand", host="nas", ip="172.22.40.8", description="",
+                              enabled="0")]
+        calls = self._record_calls(driver, rows)
+
+        driver.sync_dns_zone("office.local", [
+            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+        ])
+
+        assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1
+
+    def test_a_hand_made_override_in_another_zone_does_not_stand_in(self, driver):
+        rows = [_override_row("by-hand", host="nas", domain="other.local", ip="172.22.40.8",
+                              description="")]
+        calls = self._record_calls(driver, rows)
+
+        driver.sync_dns_zone("office.local", [
+            {"hostname": "nas", "ip": "172.22.40.8", "record_type": "A", "enabled": True},
+        ])
+
+        assert len([p for p, _ in calls if p.endswith("addhostoverride")]) == 1
+
 
 # ---------------------------------------------------------------------------
 # uninstall_package – says what it cannot do