content-explorer/tests/test_quality.py

"""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()