tegwick
55f6180b8e
All checks were successful
synthetic / suite (push) Successful in 1m2s
The connector maps native Q-ids to aliases under urn:ce ids, publishes only through the existing rights and reference gate, and leaves the real-user pilot open. Assistant: grok Assistant-Session: 01a15378-f601-7fe2-a143-8e1500a1d59c
195 lines
8.3 KiB
Python
195 lines
8.3 KiB
Python
"""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()
|