"""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.contract import graph_errors # noqa: E402 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), ["computing-basics", "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_entity_source_ref_is_unresolved_beside_a_licensed_source(self) -> None: snapshot, manifest = load_example("systems-thinking") entity = next( item for item in snapshot["entities"] if item["id"] == "urn:ce:systems-thinking:concept:feedback" ) licensed = entity["sourceRefs"][0] self.assertIn(licensed, {source["id"] for source in snapshot["sources"]}) entity["sourceRefs"] = list(entity["sourceRefs"]) + ["src:does-not-exist"] report = quality_report(snapshot, manifest, today="2026-10-19") self.assertTrue(any( item["id"] == entity["id"] and item["field"] == "sourceRefs" and item["target"] == "src:does-not-exist" for item in report["unresolvedReferences"] )) self.assertGreater(report["failureClasses"]["unresolved-reference"], 0) self.assertIn(entity["id"], report["provenance"]["incomplete"]) self.assertTrue(any( error == f"{entity['id']}: missing source src:does-not-exist" for error in graph_errors(snapshot, manifest) )) 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()