fix skip_ipv4 and skip_ipv6 for unifi records

test_unifi_update.py

@@ -367,3 +367,76 @@ class TestUpdateRecords:
         with patch("ddns.unifi_update._get_session", return_value=mock_session):
             with pytest.raises(Exception, match="Update failed"):
                 update_records(unifi_config, ipv4="1.2.3.4", ipv6=None)  # type: ignore[arg-type]
+
+    def test_update_records_skip_ipv4(self) -> None:
+        mock_session = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"data": []}
+        mock_response.raise_for_status = MagicMock()
+        mock_session.get.return_value = mock_response
+        mock_session.post.return_value = MagicMock()
+        mock_session.post.return_value.raise_for_status = MagicMock()
+        mock_session.verify = False
+
+        unifi_config = {
+            "host": "https://unifi.example.com",
+            "site_id": "site123",
+            "api_token": "my-token",
+            "verify_ssl": False,
+            "records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv4": True}],
+        }
+
+        with patch("ddns.unifi_update._get_session", return_value=mock_session):
+            update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1")  # type: ignore[arg-type]
+
+        mock_session.post.assert_called_once()
+        payload = mock_session.post.call_args[1]["json"]
+        assert payload["type"] == "AAAA_RECORD"
+        assert payload["ipv6Address"] == "2001:db8::1"
+
+    def test_update_records_skip_ipv6(self) -> None:
+        mock_session = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"data": []}
+        mock_response.raise_for_status = MagicMock()
+        mock_session.get.return_value = mock_response
+        mock_session.post.return_value = MagicMock()
+        mock_session.post.return_value.raise_for_status = MagicMock()
+        mock_session.verify = False
+
+        unifi_config = {
+            "host": "https://unifi.example.com",
+            "site_id": "site123",
+            "api_token": "my-token",
+            "verify_ssl": False,
+            "records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv6": True}],
+        }
+
+        with patch("ddns.unifi_update._get_session", return_value=mock_session):
+            update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1")  # type: ignore[arg-type]
+
+        mock_session.post.assert_called_once()
+        payload = mock_session.post.call_args[1]["json"]
+        assert payload["type"] == "A_RECORD"
+        assert payload["ipv4Address"] == "1.2.3.4"
+
+    def test_update_records_skip_both(self) -> None:
+        mock_session = MagicMock()
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"data": []}
+        mock_response.raise_for_status = MagicMock()
+        mock_session.get.return_value = mock_response
+        mock_session.verify = False
+
+        unifi_config = {
+            "host": "https://unifi.example.com",
+            "site_id": "site123",
+            "api_token": "my-token",
+            "verify_ssl": False,
+            "records": [{"record": "test.example.com", "ttl_seconds": 7200, "skip_ipv4": True, "skip_ipv6": True}],
+        }
+
+        with patch("ddns.unifi_update._get_session", return_value=mock_session):
+            update_records(unifi_config, ipv4="1.2.3.4", ipv6="2001:db8::1")  # type: ignore[arg-type]
+
+        mock_session.post.assert_not_called()

unifi_update.py

@@ -20,6 +20,8 @@ logger = logging.getLogger(__name__)
 class UnifiRecordType(TypedDict):
     record: str
     ttl_seconds: int
+    skip_ipv4: bool | None
+    skip_ipv6: bool | None
 
 
 class UnifiConfig(TypedDict):
@@ -147,7 +149,7 @@ def update_records(
         ttl = record.get("ttl_seconds", 14400)
         logger.info("=== Processing UniFi record: %s (ttl=%s) ===", domain, ttl)
 
-        if ipv4:
+        if ipv4 and not record.get("skip_ipv4"):
             existing = policy_map.get(f"{domain}:A_RECORD")
             if existing:
                 logger.debug("Found existing A_RECORD policy for %s: id=%s, ip=%s",
@@ -159,8 +161,10 @@ def update_records(
                 "A_RECORD", domain, ipv4, ttl,
                 existing,
             )
+        elif ipv4 and record.get("skip_ipv4"):
+            logger.info("Skipping IPv4 for %s (skip_ipv4=true)", domain)
 
-        if ipv6:
+        if ipv6 and not record.get("skip_ipv6"):
             existing = policy_map.get(f"{domain}:AAAA_RECORD")
             if existing:
                 logger.debug("Found existing AAAA_RECORD policy for %s: id=%s, ip=%s",
@@ -172,5 +176,7 @@ def update_records(
                 "AAAA_RECORD", domain, ipv6, ttl,
                 existing,
             )
+        elif ipv6 and record.get("skip_ipv6"):
+            logger.info("Skipping IPv6 for %s (skip_ipv6=true)", domain)
 
         logger.info("=== Done processing UniFi record: %s ===", domain)

update.py

@@ -84,6 +84,8 @@ class Route53RecordType(TypedDict):
 class UnifiRecordType(TypedDict):
     record: str
     provider: Literal["unifi"]
+    skip_ipv4: bool | None
+    skip_ipv6: bool | None
     ttl_seconds: int | None