Merge pull request 'fix(catalog): whole-system sources only report their own item types' (#5) from marioglealp/shikigami-types into main

Reviewed-on: #5

src/thoth/catalog/scopes.py

@@ -46,6 +46,13 @@ def _k8s_scan(scope: str, type_name: str, key: str, attributes: dict[str, Any])
 
 RULES: dict[str, Rule] = {Source.HOST: _host_scan, Source.K8S: _k8s_scan}
 
+# Sources that see a whole system (scope `all`): which item types each may report. Owning fields isn't enough to go
+# by - some types have none - and without a list a source could create any type, empty, under its name. A source
+# gets its list when it's built (D51); one without a list isn't limited yet.
+TYPES: dict[str, frozenset[str]] = {
+    Source.SHIKIGAMI: frozenset({"app", "route", "bot"}),  # what was requested; databases etc. join as it declares them
+}
+
 
 def scope_problem(source: str, scope: str) -> str | None:
     if not re.fullmatch(SCOPE_PATTERN, scope):
@@ -56,5 +63,7 @@ def scope_problem(source: str, scope: str) -> str | None:
 
 
 def in_scope(source: str, scope: str, type_name: str, key: str, attributes: dict[str, Any]) -> bool:
-    rule = RULES.get(source)
-    return rule is None or rule(scope, type_name, key, attributes)
+    if rule := RULES.get(source):
+        return rule(scope, type_name, key, attributes)
+    types = TYPES.get(source)
+    return types is None or type_name in types

tests/test_snapshots.py

@@ -103,6 +103,15 @@ def test_whole_system_sources_use_scope_all(client: TestClient) -> None:
     assert "its scope is 'all'" in response.json()["errors"][0]["msg"]
 
 
+def test_a_whole_system_source_only_reports_its_types(client: TestClient) -> None:
+    host = {"type": "host", "key": "made-up", "name": "made-up", "attributes": {}}
+    app = {"type": "app", "key": "echo", "name": "echo", "attributes": {"template": "nuwa/python"}}
+    response = put(client, "shikigami", "all", app, host)
+    assert response.status_code == 422  # all or nothing: the app isn't written either
+    assert [e["msg"] for e in response.json()["errors"]] == ["items[1] (host made-up): outside scope all"]
+    assert put(client, "shikigami", "all", app).status_code == 200
+
+
 def test_bad_keys_and_unknown_types_are_refused(client: TestClient) -> None:
     response = put(client, "host-scan", "clockmost-desktop", {**DESKTOP, "key": "Clockmost-Desktop"},
                    {"type": "toaster", "key": "x", "name": "x", "attributes": {}})  # fmt: skip