quality: validate knowledge-graph target existence

scripts/validate_entry.py

@@ -15,7 +15,17 @@
 from taxonomy import load_taxonomy, supported_address_categories, supported_relationship_types
 from url_safety import is_safe_https_url
 
-CORE_PREFIXES = ("feat/", "fix/", "docs/", "chore/", "ci/", "refactor/", "release/")
+CORE_PREFIXES = (
+    "feat/",
+    "fix/",
+    "docs/",
+    "chore/",
+    "ci/",
+    "refactor/",
+    "release/",
+    "quality/",
+    "test/",
+)
 AUTOMATION_PREFIXES = ("dependabot/",)
 NON_MODULE_PREFIXES = CORE_PREFIXES + AUTOMATION_PREFIXES
 REQUIRED_FILES = (

scripts/validate_relationships.py

@@ -71,13 +71,21 @@ def validate_entry_relationships(entry: dict, schema_version: int, taxonomy: dic
     return errors
 
 
-def validate_catalog(path: Path) -> list[str]:
+def _load_catalog(path: Path) -> tuple[dict | None, list[str]]:
     try:
         data = yaml.safe_load(path.read_text(encoding="utf-8"))
     except Exception as exc:
-        return [f"{path}: invalid YAML: {exc}"]
+        return None, [f"{path}: invalid YAML: {exc}"]
     if not isinstance(data, dict):
-        return [f"{path}: top level must be a mapping"]
+        return None, [f"{path}: top level must be a mapping"]
+    return data, []
+
+
+def validate_catalog(path: Path) -> list[str]:
+    data, load_errors = _load_catalog(path)
+    if load_errors:
+        return load_errors
+    assert data is not None
 
     schema_version = data.get("schema_version")
     if not isinstance(schema_version, int):
@@ -96,6 +104,50 @@ def validate_catalog(path: Path) -> list[str]:
     return [f"{path}: {error}" for error in errors]
 
 
+def validate_relationship_targets(paths: list[Path]) -> list[str]:
+    """Require valid relationship targets to resolve within the supplied catalog set."""
+    loaded: list[tuple[Path, dict]] = []
+    module_refs: set[str] = set()
+
+    for path in paths:
+        data, load_errors = _load_catalog(path)
+        if load_errors or data is None:
+            continue
+        entries = data.get("entries")
+        if not isinstance(entries, list):
+            continue
+        loaded.append((path, data))
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            category = entry.get("category")
+            module_id = entry.get("id")
+            if isinstance(category, str) and isinstance(module_id, str):
+                module_refs.add(f"{category}/{module_id}")
+
+    errors: list[str] = []
+    for path, data in loaded:
+        for entry in data.get("entries", []):
+            if not isinstance(entry, dict):
+                continue
+            relationships = entry.get("relationships")
+            if not isinstance(relationships, list):
+                continue
+            module_ref = f"{entry.get('category')}/{entry.get('id')}"
+            for number, relationship in enumerate(relationships, start=1):
+                if not isinstance(relationship, dict):
+                    continue
+                target = relationship.get("target")
+                if not isinstance(target, str) or target.count("/") != 1:
+                    continue
+                if target not in module_refs:
+                    errors.append(
+                        f"{path}: {module_ref}: relationship #{number} target {target!r} "
+                        "does not exist in the supplied catalogs"
+                    )
+    return errors
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("catalogs", nargs="+", type=Path)
@@ -104,6 +156,7 @@ def main() -> int:
     errors: list[str] = []
     for path in args.catalogs:
         errors.extend(validate_catalog(path))
+    errors.extend(validate_relationship_targets(args.catalogs))
 
     if errors:
         print("OpenDevIndex relationship validation failed:", file=sys.stderr)

tests/test_relationship_target_existence.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+from validate_relationships import validate_relationship_targets  # noqa: E402
+
+
+class RelationshipTargetExistenceTests(unittest.TestCase):
+    def write_catalog(self, root: Path, name: str, body: str) -> Path:
+        path = root / name
+        path.write_text(body, encoding="utf-8")
+        return path
+
+    def test_target_in_another_catalog_resolves(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            tools = self.write_catalog(
+                root,
+                "tools.yaml",
+                """schema_version: 3
+entries:
+  - category: tool
+    id: demo
+    relationships:
+      - type: integrates-with
+        target: cloud/example
+""",
+            )
+            cloud = self.write_catalog(
+                root,
+                "cloud.yaml",
+                """schema_version: 3
+entries:
+  - category: cloud
+    id: example
+""",
+            )
+            self.assertEqual(validate_relationship_targets([tools, cloud]), [])
+
+    def test_missing_target_is_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            tools = self.write_catalog(
+                root,
+                "tools.yaml",
+                """schema_version: 3
+entries:
+  - category: tool
+    id: demo
+    relationships:
+      - type: integrates-with
+        target: cloud/missing
+""",
+            )
+            errors = validate_relationship_targets([tools])
+            self.assertTrue(any("does not exist in the supplied catalogs" in error for error in errors), errors)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_validate_entry.py

@@ -24,6 +24,8 @@ def test_core_refs_are_not_treated_as_modules(self) -> None:
             "ci/validation",
             "refactor/index-builder",
             "release/v1.0",
+            "quality/relationship-targets",
+            "test/validator-regression",
         ):
             with self.subTest(ref=ref):
                 self.assertTrue(is_non_module_ref(ref))