#4681 Enable patch request on Cloudflare zone DNS settings (#4748)

Author: erinysong

src/registrar/services/cloudflare_service.py

@@ -116,6 +116,43 @@ def create_cf_zone(self, zone_name: str, x_account_id: str):
             raise
         return resp.json()
 
+    def update_zone_dns_settings(
+        self, zone_id: str, *, zone_mode: str = "dns_only", nameservers_type: str = "custom.tenant", ns_set: int = 1
+    ) -> CloudflareDnsSettingsUpdateResponse:
+        """PATCH /zones/{zone_id}/dns_settings
+        Required settings:
+        - zone_mode: "standard" | "cdn_only" | "dns_only"
+        - nameservers_type: "cloudflare.standard"
+                "cloudflare.standard.random"
+                "custom.account"
+                "custom.tenant"
+        - ns_set: Min 1, max 5. Default 1 when not passed as argument.
+        """
+        appended_url = f"/zones/{zone_id}/dns_settings"
+        data = {
+            "zone_mode": zone_mode,
+            "nameservers": {"ns_set": ns_set, "type": nameservers_type},
+        }
+
+        try:
+            resp = self.client.patch(appended_url, json=data)
+            resp.raise_for_status()
+            logger.info(
+                "Updated zone DNS settings for zone_id=%s (zone_mode=%s, nameservers.type=%s, namservers.ns_set=%s)",
+                zone_id,
+                zone_mode,
+                nameservers_type,
+                ns_set,
+            )
+        except RequestError as e:
+            logger.error(f"Failed to update dns settings for zone {zone_id}: {e}")
+            raise
+        except HTTPStatusError as e:
+            logger.error(f"Error {e.response.status_code} while updating dns settings: {e}")
+            raise
+
+        return CloudflareDnsSettingsUpdateResponse.from_json(resp.json())
+
     def create_dns_record(self, zone_id: str, record_data: dict[str, Any]):
         appended_url = f"/zones/{zone_id}/dns_records"
         try:

src/registrar/services/dns_host_service.py

@@ -34,6 +34,10 @@ def update_account_dns_settings(self, x_account_id: str) -> CloudflareDnsSetting
         """Ensure required Cloudflare DNS settings are applied for an account."""
         return self.dns_vendor_service.update_account_dns_settings(x_account_id)
 
+    def update_zone_dns_settings(self, x_zone_id: str) -> CloudflareDnsSettingsUpdateResponse:
+        """Ensure required Cloudflare DNS settings are applied for a zone."""
+        return self.dns_vendor_service.update_zone_dns_settings(x_zone_id)
+
     def _find_account_tag_by_pubname(self, items, name):
         """Find an item by name in a list of dictionaries."""
         return next((item.get("account_tag") for item in items if item.get("account_pubname") == name), None)

src/registrar/services/mock_cloudflare_service.py

@@ -115,6 +115,25 @@ def _register_zone_mocks(self):
             side_effect=self._mock_update_dns_record_response
         )
 
+        # PATCH account dns_settings
+        self._mock_context.patch(url__regex=r"/zones/[\w-]+/dns_settings").mock(
+            side_effect=self._mock_update_zone_dns_settings_response
+        )
+
+    def _mock_update_zone_dns_settings_response(self, request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "result": {
+                    "zone_mode": "dns_only",
+                    "nameservers": {"ns_set": 2, "type": "custom.tenant"},
+                },
+                "success": True,
+                "errors": [],
+                "messages": [],
+            },
+        )
+
     def _mock_get_page_accounts_response(self, request) -> httpx.Response:
         logger.debug("😎 Mocking accounts GET")
         # use exists.gov domain to simulate an account that already exists

src/registrar/tests/services/test_cloudflare_service.py

@@ -423,3 +423,41 @@ def test_update_account_dns_settings_failure(self):
                     self.service.update_account_dns_settings(account_id)
 
                 self.assertIn(error["message"], str(context.exception))
+
+    def test_update_zone_dns_settings_success(self):
+        """Test successful update_zone_dns_settings call"""
+        zone_id = "54321"
+        return_value = {
+            "success": True,
+            "result": {
+                "zone_mode": "dns_only",
+                "nameservers": {"ns_set": 2, "type": "custom.tenant"},
+            },
+            "errors": [],
+            "messages": [],
+        }
+        mock_response = self._setUpSuccessMockResponse(return_value)
+        self.service.client.patch.return_value = mock_response
+
+        resp = self.service.update_zone_dns_settings(zone_id)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(resp.result["zone_mode"], "dns_only")
+        self.assertEqual(resp.result["nameservers"]["ns_set"], 2)
+        self.assertEqual(resp.result["nameservers"]["type"], "custom.tenant")
+        self.assertEqual(resp.errors, [])
+
+    def test_update_zone_dns_settings_failure(self):
+        """Test update_zone_dns_settings when error results during call"""
+        zone_id = "54321"
+
+        for case in self.failure_cases:
+            with self.subTest(msg=case["test_name"], **case):
+                error = case["error"]
+                mock_response = self._setUpFailureMockResponse(error)
+                self.service.client.patch.return_value = mock_response
+
+                with self.assertRaises(error["exception"]) as context:
+                    self.service.update_zone_dns_settings(zone_id)
+
+                self.assertIn(error["message"], str(context.exception))

src/registrar/tests/services/test_dns_host_service.py

@@ -263,6 +263,69 @@ def test_create_cf_record_failure(self, mock_create_dns_record):
             self.service.create_and_save_record(zone_id, record_data)
         self.assertIn("Bad request: missing name", str(context.exception))
 
+    def test_update_account_dns_settings_success(self):
+        x_account_id = "12345"
+        expected_response = CloudflareDnsSettingsUpdateResponse(
+            success=True,
+            result={
+                "zone_defaults": {
+                    "zone_mode": "dns_only",
+                    "nameservers": {"type": "custom.tenant"},
+                }
+            },
+            errors=[],
+            messages=[],
+        )
+        self.service.dns_vendor_service.update_account_dns_settings = Mock(return_value=expected_response)
+
+        response = self.service.update_account_dns_settings(x_account_id)
+
+        self.service.dns_vendor_service.update_account_dns_settings.assert_called_once_with(x_account_id)
+        self.assertTrue(response.success)
+        self.assertEqual(response.result["zone_defaults"]["zone_mode"], "dns_only")
+        self.assertEqual(response.result["zone_defaults"]["nameservers"]["type"], "custom.tenant")
+        self.assertEqual(response.errors, [])
+
+    def test_update_account_dns_settings_failure(self):
+        x_account_id = "12345"
+        self.service.dns_vendor_service.update_account_dns_settings = Mock(
+            side_effect=HTTPStatusError(message="Error updating DNS settings", request=Mock(), response=Mock())
+        )
+
+        with self.assertRaises(HTTPStatusError):
+            self.service.update_account_dns_settings(x_account_id)
+
+    def test_update_zone_dns_settings_success(self):
+        x_zone_id = "54321"
+        expected_response = CloudflareDnsSettingsUpdateResponse(
+            success=True,
+            result={
+                "zone_mode": "dns_only",
+                "nameservers": {"ns_set": 2, "type": "custom.tenant"},
+            },
+            errors=[],
+            messages=[],
+        )
+        self.service.dns_vendor_service.update_zone_dns_settings = Mock(return_value=expected_response)
+
+        response = self.service.update_zone_dns_settings(x_zone_id)
+
+        self.service.dns_vendor_service.update_zone_dns_settings.assert_called_once_with(x_zone_id)
+        self.assertTrue(response.success)
+        self.assertEqual(response.result["zone_mode"], "dns_only")
+        self.assertEqual(response.result["nameservers"]["ns_set"], 2)
+        self.assertEqual(response.result["nameservers"]["type"], "custom.tenant")
+        self.assertEqual(response.errors, [])
+
+    def test_update_zone_dns_settings_failure(self):
+        x_zone_id = "54321"
+        self.service.dns_vendor_service.update_zone_dns_settings = Mock(
+            side_effect=HTTPStatusError(message="Error updating DNS settings", request=Mock(), response=Mock())
+        )
+
+        with self.assertRaises(HTTPStatusError):
+            self.service.update_zone_dns_settings(x_zone_id)
+
 
 class TestDnsHostServiceDB(TestCase):
     def setUp(self):
@@ -747,35 +810,3 @@ def test_update_db_record_with_bad_data_fails(self):
                     self.assertEqual(dns_record.content, self.vendor_record_data["result"].get("content"))
                     self.assertEqual(dns_record.ttl, self.vendor_record_data["result"].get("ttl"))
                     self.assertEqual(dns_record.comment, self.vendor_record_data["result"].get("comment"))
-
-    def test_update_account_dns_settings_success(self):
-        x_account_id = "12345"
-        expected_response = CloudflareDnsSettingsUpdateResponse(
-            success=True,
-            result={
-                "zone_defaults": {
-                    "zone_mode": "dns_only",
-                    "nameservers": {"type": "custom.tenant"},
-                }
-            },
-            errors=[],
-            messages=[],
-        )
-        self.service.dns_vendor_service.update_account_dns_settings = Mock(return_value=expected_response)
-
-        response = self.service.update_account_dns_settings(x_account_id)
-
-        self.service.dns_vendor_service.update_account_dns_settings.assert_called_once_with(x_account_id)
-        self.assertTrue(response.success)
-        self.assertEqual(response.result["zone_defaults"]["zone_mode"], "dns_only")
-        self.assertEqual(response.result["zone_defaults"]["nameservers"]["type"], "custom.tenant")
-        self.assertEqual(response.errors, [])
-
-    def test_update_account_dns_settings_failure(self):
-        x_account_id = "12345"
-        self.service.dns_vendor_service.update_account_dns_settings = Mock(
-            side_effect=HTTPStatusError(message="Error updating DNS settings", request=Mock(), response=Mock())
-        )
-
-        with self.assertRaises(HTTPStatusError):
-            self.service.update_account_dns_settings(x_account_id)

src/registrar/tests/services/test_mock_cloudflare_service.py

@@ -189,3 +189,15 @@ def test_mock_update_account_dns_settings_response(self):
         self.assertEqual(resp.result["zone_defaults"]["nameservers"]["type"], "custom.tenant")
         self.assertEqual(resp.errors, [])
         self.assertEqual(resp.messages, [])
+
+    def test_mock_update_zone_dns_settings_response(self):
+        zone_id = self.mock_api_service.fake_zone_id
+
+        resp = self.service.update_zone_dns_settings(zone_id)
+
+        self.assertTrue(resp.success)
+        self.assertEqual(resp.result["zone_mode"], "dns_only")
+        self.assertEqual(resp.result["nameservers"]["ns_set"], 2)
+        self.assertEqual(resp.result["nameservers"]["type"], "custom.tenant")
+        self.assertEqual(resp.errors, [])
+        self.assertEqual(resp.messages, [])