fix: return 404 for get config on unknown subject

karapace/in_memory_database.py

@@ -137,13 +137,17 @@ def insert_subject(self, *, subject: Subject) -> None:
         self.subjects.setdefault(subject, SubjectData())
 
     def get_subject_compatibility(self, *, subject: Subject) -> Optional[str]:
-        return self.subjects[subject].compatibility
+        if subject in self.subjects:
+            return self.subjects[subject].compatibility
+        return None
 
     def delete_subject_compatibility(self, *, subject: Subject) -> None:
-        self.subjects[subject].compatibility = None
+        if subject in self.subjects:
+            self.subjects[subject].compatibility = None
 
     def set_subject_compatibility(self, *, subject: Subject, compatibility: str) -> None:
-        self.subjects[subject].compatibility = compatibility
+        if subject in self.subjects:
+            self.subjects[subject].compatibility = compatibility
 
     def find_schema(self, *, schema_id: SchemaId) -> Optional[TypedSchema]:
         return self.schemas[schema_id]

karapace/schema_registry_apis.py

@@ -533,8 +533,17 @@ async def config_subject_get(
 
         # Config for a subject can exist without schemas so no need to check for their existence
         assert self.schema_registry.schema_reader, "KarapaceSchemaRegistry not initialized. Missing call to _init"
-        compatibility = self.schema_registry.database.get_subject_compatibility(subject=subject)
+        if self.schema_registry.database.find_subject(subject=subject) is None:
+            self.r(
+                body={
+                    "error_code": SchemaErrorCodes.SUBJECT_NOT_FOUND.value,
+                    "message": SchemaErrorMessages.SUBJECT_NOT_FOUND_FMT.value.format(subject=subject),
+                },
+                content_type=content_type,
+                status=HTTPStatus.NOT_FOUND,
+            )
 
+        compatibility = self.schema_registry.database.get_subject_compatibility(subject=subject)
         default_to_global = request.query.get("defaultToGlobal", "false").lower() == "true"
         if not compatibility and default_to_global:
             compatibility = self.schema_registry.compatibility

tests/integration/test_schema.py

@@ -1997,6 +1997,19 @@ async def test_schema_version_number_existing_schema(registry_async_client: Clie
     assert schema_id_3 > schema_id_2
 
 
+@pytest.mark.parametrize("trail", ["", "/"])
+async def test_get_config_unknown_subject(registry_async_client: Client, trail: str) -> None:
+    res = await registry_async_client.get(f"config/unknown-subject{trail}")
+    assert res.status_code == 404, f"{res} - Should return 404 for unknown subject"
+
+    # Set global config, see that unknown subject is still returns correct 404 and does not fallback to global config
+    res = await registry_async_client.put(f"config{trail}", json={"compatibility": "FULL"})
+    assert res.status_code == 200
+
+    res = await registry_async_client.get(f"config/unknown-subject{trail}")
+    assert res.status_code == 404, f"{res} - Should return 404 for unknown subject also when global config set"
+
+
 @pytest.mark.parametrize("trail", ["", "/"])
 async def test_config(registry_async_client: Client, trail: str) -> None:
     subject_name_factory = create_subject_name_factory(f"test_config-{trail}")