schema_registry: Wire up delete_config_subject

(cherry picked from commit df1a49d)

src/v/pandaproxy/schema_registry/service.cc

@@ -107,6 +107,10 @@ server::routes_t get_schema_registry_routes(ss::gate& gate, one_shot& es) {
       ss::httpd::schema_registry_json::put_config_subject,
       wrap(gate, es, put_config_subject)});
 
+    routes.routes.emplace_back(server::route_t{
+      ss::httpd::schema_registry_json::delete_config_subject,
+      wrap(gate, es, delete_config_subject)});
+
     routes.routes.emplace_back(server::route_t{
       ss::httpd::schema_registry_json::get_mode, wrap(gate, es, get_mode)});
 

tests/rptest/tests/schema_registry_test.py

@@ -304,6 +304,15 @@ def _set_config_subject(self,
                              data=data,
                              **kwargs)
 
+    def _delete_config_subject(self,
+                               subject,
+                               headers=HTTP_POST_HEADERS,
+                               **kwargs):
+        return self._request("DELETE",
+                             f"config/{subject}",
+                             headers=headers,
+                             **kwargs)
+
     def _get_mode(self, headers=HTTP_GET_HEADERS, **kwargs):
         return self._request("GET", "mode", headers=headers, **kwargs)
 
@@ -883,6 +892,54 @@ def test_config(self):
         result_raw = self._get_config_subject(subject=f"{topic}-key")
         assert result_raw.json()["compatibilityLevel"] == "BACKWARD_TRANSITIVE"
 
+        prev_compat = result_raw.json()["compatibilityLevel"]
+        global_config = self._get_config().json()
+
+        result_raw = self._delete_config_subject(subject=f"{topic}-key")
+        assert result_raw.json(
+        )["compatibilityLevel"] == prev_compat, f"{json.dumps(result_raw.json(), indent=1)}"
+
+        self.logger.debug("Second DELETE should return 40401")
+        result_raw = self._delete_config_subject(subject=f"{topic}-key")
+        assert result_raw.status_code == requests.codes.not_found, result_raw.status_code
+        assert result_raw.json(
+        )["error_code"] == 40401, f"Wrong err code: {result_raw.json()}"
+        assert result_raw.json(
+        )["message"] == f"Subject '{topic}-key' not found.", f"{json.dumps(result_raw.json(), indent=1)}"
+
+        self.logger.debug(
+            "GET config/{subject} should indicate missing subject-level config"
+        )
+        result_raw = self._get_config_subject(subject=f"{topic}-key")
+        assert result_raw.status_code == requests.codes.not_found
+        assert result_raw.json()["error_code"] == 40408
+        assert result_raw.json(
+        )["message"] == f"Subject '{topic}-key' does not have subject-level compatibility configured"
+
+        result_raw = self._get_config_subject(subject=f"{topic}-key",
+                                              fallback=True)
+        assert result_raw.json(
+        )["compatibilityLevel"] == global_config["compatibilityLevel"]
+
+        self.logger.debug(
+            "Subject compatibility should reflect the new global config")
+        global_config = self._set_config(
+            data=json.dumps({"compatibility": "NONE"}))
+        assert global_config.json()["compatibility"] == "NONE"
+
+        result_raw = self._get_config_subject(subject=f"{topic}-key",
+                                              fallback=True)
+        assert result_raw.json()["compatibilityLevel"] == global_config.json(
+        )["compatibility"]
+
+        self.logger.debug("DELETE on non-existant subject should 404")
+        result_raw = self._delete_config_subject(subject=f"foo-key")
+        assert result_raw.status_code == requests.codes.not_found, result_raw.status_code
+        assert result_raw.json(
+        )["error_code"] == 40401, f"Wrong err code: {result_raw.json()}"
+        assert result_raw.json(
+        )["message"] == f"Subject 'foo-key' not found.", f"{json.dumps(result_raw.json(), indent=1)}"
+
     @cluster(num_nodes=3)
     def test_mode(self):
         """
@@ -1920,6 +1977,25 @@ def test_config(self):
                                                     super_password))
         assert result_raw.json()["compatibilityLevel"] == "BACKWARD_TRANSITIVE"
 
+        global_config = self._get_config(auth=(super_username,
+                                               super_password)).json()
+
+        old_config = result_raw.json()
+
+        result_raw = self._delete_config_subject(subject=f"{topic}-key",
+                                                 auth=(super_username,
+                                                       super_password))
+        assert result_raw.json(
+        )["compatibilityLevel"] == old_config["compatibilityLevel"]
+        #, f"{json.dumps(result_raw.json(), indent=1)}, {json.dumps(global_config, indent=1)}"
+
+        result_raw = self._get_config_subject(subject=f"{topic}-key",
+                                              fallback=True,
+                                              auth=(super_username,
+                                                    super_password))
+        assert result_raw.json(
+        )["compatibilityLevel"] == global_config["compatibilityLevel"]
+
     @cluster(num_nodes=3)
     def test_mode(self):
         """