Add a synthetic quality suite and upstream parity record.

Assistant: grok
Assistant-Session: 01a1531b-6698-7732-8587-f28ea604bcdf

tests/test_quality.py

@@ -0,0 +1,171 @@
+"""Quality report and public snapshot rejection."""
+
+from __future__ import annotations
+
+import copy
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from content_explorer.ingest import project_records, publish_directory  # noqa: E402
+from content_explorer.quality import FAILURE_CLASSES, quality_report, report_dataspaces  # noqa: E402
+
+EXAMPLES = ROOT / "examples" / "dataspaces"
+LABELS = {"represented-by": {"forward": "Represented by", "inverse": "Represents"}}
+KINDS = [
+    {"id": "concept", "label": "Concept", "plural": "Concepts"},
+    {"id": "method", "label": "Method", "plural": "Methods"},
+]
+SOURCE = {
+    "id": "src:demo:notes",
+    "uri": "https://example.invalid/content-explorer/ingest",
+    "title": "Demo notes",
+    "license": "CC0-1.0",
+    "attribution": "Content Explorer demo",
+}
+RECORDS = [
+    {
+        "kind": "concept",
+        "key": "feedback",
+        "title": "Feedback",
+        "tldr": "Results change the next action.",
+        "summary": "A loop carries an outcome back to the actor.",
+        "tags": ["regulation"],
+        "relations": [{"predicate": "represented-by", "to": "method:loop-diagram"}],
+    },
+    {
+        "kind": "method",
+        "key": "loop-diagram",
+        "title": "Loop diagram",
+        "tldr": "Arrows that show influence.",
+        "summary": "A drawing of influences among parts.",
+        "tags": ["diagram"],
+        "relations": [],
+    },
+]
+
+
+def project(records, source=SOURCE, relation_labels=None):
+    return project_records(
+        records,
+        dataspace_id="demo-notes",
+        title="Demo notes",
+        description="A connector test collection.",
+        source=source,
+        kinds=KINDS,
+        observed_at="2026-10-12",
+        relation_labels=LABELS if relation_labels is None else relation_labels,
+    )
+
+
+def load_example(name: str) -> tuple[dict, dict]:
+    directory = EXAMPLES / name
+    snapshot = json.loads((directory / "snapshot.json").read_text(encoding="utf-8"))
+    manifest = json.loads((directory / "manifest.json").read_text(encoding="utf-8"))
+    return snapshot, manifest
+
+
+class ExampleReports(unittest.TestCase):
+    def test_published_examples_are_complete(self) -> None:
+        reports = report_dataspaces(EXAMPLES, today="2026-10-19")
+        self.assertEqual(
+            sorted(report["dataspace"] for report in reports),
+            ["software-architecture", "systems-thinking"],
+        )
+        for report in reports:
+            self.assertGreater(report["provenance"]["records"], 0)
+            self.assertEqual(report["provenance"]["complete"], report["provenance"]["records"])
+            self.assertEqual(report["provenance"]["incomplete"], [])
+            self.assertEqual(report["stale"]["count"], 0)
+            self.assertEqual(report["unresolvedReferences"], [])
+            self.assertEqual(report["sources"]["missingLicense"], [])
+            self.assertEqual(report["sources"]["complete"], report["sources"]["count"])
+            self.assertGreater(report["sources"]["count"], 0)
+            self.assertIn("CC0-1.0", report["sources"]["licenses"])
+            for name in FAILURE_CLASSES:
+                self.assertEqual(report["failureClasses"][name], 0)
+
+    def test_missing_license_dangling_and_duplicate_are_failures(self) -> None:
+        snapshot, manifest = load_example("systems-thinking")
+        source_id = snapshot["sources"][0]["id"]
+        missing = copy.deepcopy(snapshot)
+        missing["sources"][0]["license"] = ""
+        missing_report = quality_report(missing, manifest)
+        self.assertIn(source_id, missing_report["sources"]["missingLicense"])
+        self.assertGreater(missing_report["failureClasses"]["missing-license"], 0)
+        self.assertTrue(any(
+            item["class"] == "missing-license" and item["id"] == source_id
+            for item in missing_report["failures"]
+        ))
+
+        dangling = copy.deepcopy(snapshot)
+        relation_id = dangling["relations"][0]["id"]
+        dangling["relations"][0]["to"] = "urn:ce:systems-thinking:concept:missing"
+        dangling_report = quality_report(dangling, manifest)
+        self.assertGreater(dangling_report["failureClasses"]["dangling-reference"], 0)
+        self.assertTrue(any(
+            item["id"] == relation_id and item["field"] == "to" for item in dangling_report["unresolvedReferences"]
+        ))
+
+        duplicate = copy.deepcopy(snapshot)
+        entity_id = duplicate["entities"][0]["id"]
+        duplicate["entities"].append(copy.deepcopy(duplicate["entities"][0]))
+        duplicate_report = quality_report(duplicate, manifest)
+        self.assertGreater(duplicate_report["failureClasses"]["duplicate-identity"], 0)
+        self.assertTrue(any(
+            item["class"] == "duplicate-identity" and item["id"] == entity_id
+            for item in duplicate_report["failures"]
+        ))
+
+        allowed = copy.deepcopy(snapshot)
+        allowed["sources"][0]["license"] = "MIT"
+        allowed_report = quality_report(allowed, manifest)
+        self.assertIn("MIT", allowed_report["sources"]["licenses"])
+        self.assertEqual(allowed_report["failureClasses"]["rights"], 0)
+        self.assertEqual(allowed_report["failureClasses"]["missing-license"], 0)
+
+    def test_stale_state_is_counted(self) -> None:
+        snapshot, manifest = load_example("software-architecture")
+        entity_id = snapshot["entities"][0]["id"]
+        snapshot["entities"][0]["state"] = "stale"
+        report = quality_report(snapshot, manifest, today="2026-10-19")
+        self.assertIn(entity_id, report["stale"]["ids"])
+        self.assertGreater(report["stale"]["count"], 0)
+        self.assertGreater(report["failureClasses"]["stale"], 0)
+
+
+class IngestFailures(unittest.TestCase):
+    def test_ingest_classes_and_publish_rejection(self) -> None:
+        rights = project(RECORDS, source={**SOURCE, "license": "CC-BY-NC-4.0"})
+        missing = project(RECORDS, source={**SOURCE, "license": ""})
+        dangling = project([{
+            **RECORDS[0],
+            "relations": [{"predicate": "represented-by", "to": "method:missing"}],
+        }])
+        duplicate = project(RECORDS + [{**RECORDS[0], "title": "Other feedback"}])
+        cases = {
+            "rights": (rights, "rights"),
+            "missing-license": (missing, "missing-license"),
+            "dangling-reference": (dangling, "dangling-reference"),
+            "duplicate-identity": (duplicate, "duplicate-identity"),
+        }
+        for result, kind in cases.values():
+            self.assertFalse(result["report"]["published"])
+            self.assertIsNone(result["snapshot"])
+            report = quality_report(ingest_report=result["report"])
+            self.assertGreater(report["failureClasses"][kind], 0)
+            self.assertEqual(report["provenance"]["records"], 0)
+            with tempfile.TemporaryDirectory() as tmp:
+                destination = Path(tmp) / "out"
+                with self.assertRaises(RuntimeError):
+                    publish_directory(destination, result)
+                self.assertFalse((destination / "snapshot.json").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()