coulomb/content-explorer
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Freeze the v0.1 contract and the local ingest connector.

Browse source 
The snapshot and manifest schemas, both example dataspaces, and the shared text renderer now pass the fixture validator and the negative contract tests. JSON, NDJSON, and Markdown frontmatter publish a snapshot only when rights, identity, and references allow it. CONTENT-WP-0001 and CONTENT-WP-0003 are marked finished against those artifacts. CONTENT-WP-0000-T04 stays open. SCOPE.md and the 19:05 scope-against-intent note record the stop.

Assistant: grok
Assistant-Session: 01a130f1-abfb-7d71-9cd6-a8b75ec7f95a
This commit is contained in:
tegwick 2026-10-12 19:07:32 +02:00
parent 2b1706559b
commit 072a7fa550
30 changed files with 1896 additions and 264 deletions
Download patch file Download diff file Expand all files Collapse all files

147
tests/test_contract.py Normal file
View file

@ -0,0 +1,147 @@
"""Contract checks against the shipped validator, not a copy of it."""
from __future__ import annotations
import copy
import json
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
import sys
sys.path.insert(0, str(ROOT))
from content_explorer.contract import ( # noqa: E402
resolve_claims,
validate_dataspace,
validate_documents,
)
from content_explorer.render import render_explorer # noqa: E402
EXAMPLES = ROOT / "examples" / "dataspaces"
RENDERER = ROOT / "content_explorer" / "render.py"
def load_pair(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 PublishedFixtures(unittest.TestCase):
def test_each_example_passes_the_file_entry_point(self) -> None:
names = sorted(path.name for path in EXAMPLES.iterdir() if (path / "snapshot.json").is_file())
self.assertGreaterEqual(len(names), 2)
for name in names:
errors = validate_dataspace(EXAMPLES / name, require_size=True)
self.assertEqual(errors, [], name)
def test_renderer_switches_with_the_manifest(self) -> None:
systems, systems_manifest = load_pair("systems-thinking")
software, software_manifest = load_pair("software-architecture")
systems_view = render_explorer(systems, systems_manifest)
software_view = render_explorer(software, software_manifest)
switched = render_explorer(systems, software_manifest)
self.assertIn(systems_manifest["home"]["heading"], systems_view)
self.assertIn(software_manifest["home"]["heading"], software_view)
self.assertIn(software_manifest["home"]["heading"], switched)
self.assertNotIn(systems_manifest["home"]["heading"], switched)
self.assertNotIn(software_manifest["home"]["heading"], systems_view)
def test_renderer_source_has_no_domain_vocabulary(self) -> None:
source = RENDERER.read_text(encoding="utf-8").lower()
for word in (
"cancer",
"oncolog",
"feedback",
"coupling",
"ashby",
"meadows",
"alexander",
"systems-thinking",
"software-architecture",
"causal",
"cohesion",
):
self.assertNotIn(word, source, word)
class NegativeGraph(unittest.TestCase):
def setUp(self) -> None:
self.snapshot, self.manifest = load_pair("systems-thinking")
def test_duplicate_entity_id(self) -> None:
snapshot = copy.deepcopy(self.snapshot)
snapshot["entities"].append(copy.deepcopy(snapshot["entities"][0]))
errors = validate_documents(snapshot, self.manifest)
self.assertTrue(any("duplicate entity id" in error for error in errors))
def test_dangling_relation(self) -> None:
snapshot = copy.deepcopy(self.snapshot)
snapshot["relations"][0]["to"] = "urn:ce:systems-thinking:concept:missing"
errors = validate_documents(snapshot, self.manifest)
self.assertTrue(any("dangling to" in error for error in errors))
def test_missing_source(self) -> None:
snapshot = copy.deepcopy(self.snapshot)
snapshot["entities"][0]["sourceRefs"] = ["src:missing"]
errors = validate_documents(snapshot, self.manifest)
self.assertTrue(any("missing source src:missing" in error for error in errors))
def test_taxonomic_cycle_fails_and_other_cycles_do_not(self) -> None:
snapshot = copy.deepcopy(self.snapshot)
left = snapshot["entities"][0]["id"]
right = snapshot["entities"][1]["id"]
source = snapshot["sources"][0]["id"]
snapshot["relations"].extend([
{
"id": "cycle-a",
"predicate": "related-to",
"from": left,
"to": right,
"sourceRefs": [source],
},
{
"id": "cycle-b",
"predicate": "related-to",
"from": right,
"to": left,
"sourceRefs": [source],
},
])
manifest = copy.deepcopy(self.manifest)
manifest["relationLabels"]["related-to"] = {"forward": "Related to", "inverse": "Related to"}
manifest["acyclicPredicates"] = []
open_cycle = validate_documents(snapshot, manifest)
self.assertFalse(any("taxonomic cycle" in error for error in open_cycle))
manifest["acyclicPredicates"] = ["related-to"]
closed = validate_documents(snapshot, manifest)
self.assertTrue(any("taxonomic cycle" in error for error in closed))
class FactPolicy(unittest.TestCase):
def test_missing_unknown_and_conflict(self) -> None:
claims = [
{"key": "measure", "status": "unknown", "value": None},
{"key": "name", "status": "asserted", "value": "one"},
{"key": "name", "status": "asserted", "value": "two"},
]
self.assertEqual(resolve_claims(claims, "absent", "show-all")["state"], "missing")
self.assertEqual(resolve_claims(claims, "measure", "show-all")["state"], "unknown")
shown = resolve_claims(claims, "name", "show-all")
self.assertEqual(shown["state"], "contradicted")
self.assertEqual(shown["values"], ["one", "two"])
withheld = resolve_claims(claims, "name", "withhold")
self.assertEqual(withheld["values"], [])
preferred_claims = [
{"key": "name", "status": "asserted", "value": "one", "preferred": True},
{"key": "name", "status": "asserted", "value": "two"},
]
chosen = resolve_claims(preferred_claims, "name", "preferred")
self.assertEqual(chosen, {"state": "present", "values": ["one"]})
if __name__ == "__main__":
unittest.main()

185
tests/test_ingest.py Normal file
View file

@ -0,0 +1,185 @@
"""Ingestion gates for JSON, NDJSON, and Markdown frontmatter."""
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from content_explorer.contract import validate_documents # noqa: E402
from content_explorer.ingest import ( # noqa: E402
load_json_records,
load_markdown_records,
load_ndjson_records,
project_records,
publish_directory,
)
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,
)
class IngestGates(unittest.TestCase):
def test_json_ndjson_and_markdown_match_and_repeat(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "records.json").write_text(json.dumps(RECORDS), encoding="utf-8")
(root / "records.ndjson").write_text(
"\n".join(json.dumps(record) for record in RECORDS) + "\n",
encoding="utf-8",
)
docs = root / "notes"
docs.mkdir()
for record in RECORDS:
body = "---\n" + _yaml(record) + "---\n\n" + record["summary"] + "\n"
(docs / f"{record['key']}.md").write_text(body, encoding="utf-8")
json_records, json_digest = load_json_records(root / "records.json")
ndjson_records, _ = load_ndjson_records(root / "records.ndjson")
markdown_records, _ = load_markdown_records(docs)
first = project_records(
json_records,
dataspace_id="demo-notes",
title="Demo notes",
description="A connector test collection.",
source=SOURCE,
kinds=KINDS,
observed_at="2026-10-12",
source_digest=json_digest,
relation_labels=LABELS,
)
self.assertEqual(first["report"]["sourceDigest"], json_digest)
second = project(ndjson_records)
third = project(markdown_records)
again = project(json_records)
self.assertTrue(first["report"]["published"])
self.assertEqual(first["report"]["contentHash"], second["report"]["contentHash"])
self.assertEqual(first["report"]["contentHash"], third["report"]["contentHash"])
self.assertEqual(first["report"]["contentHash"], again["report"]["contentHash"])
self.assertEqual(first["snapshot"], third["snapshot"])
for claim in first["snapshot"]["claims"]:
self.assertTrue(claim["evidenceRefs"])
errors = validate_documents(first["snapshot"], first["manifest"], require_size=False)
self.assertEqual(errors, [])
self.assertEqual(first["manifest"]["id"], "demo-notes")
self.assertIn("Represented by", first["manifest"]["relationLabels"]["represented-by"]["forward"])
def test_rights_and_dangling_refs_do_not_publish(self) -> None:
blocked = project(RECORDS, source={**SOURCE, "license": "CC-BY-NC-4.0"})
self.assertFalse(blocked["report"]["published"])
self.assertIsNone(blocked["snapshot"])
self.assertEqual(blocked["report"]["quarantined"][0]["reason"], "rights")
dangling = project([
{
**RECORDS[0],
"relations": [{"predicate": "represented-by", "to": "method:missing"}],
}
])
self.assertFalse(dangling["report"]["published"])
self.assertTrue(dangling["report"]["dangling"])
missing_credit = project(RECORDS, source={**SOURCE, "license": "CC-BY-4.0", "attribution": ""})
self.assertFalse(missing_credit["report"]["published"])
self.assertEqual(missing_credit["report"]["quarantined"][0]["detail"], "missing-attribution")
duplicate = project(RECORDS + [{**RECORDS[0], "title": "Other feedback"}])
self.assertFalse(duplicate["report"]["published"])
self.assertEqual(duplicate["report"]["conflicts"][0]["reason"], "duplicate-identity")
unlabeled = project(RECORDS, relation_labels={})
self.assertFalse(unlabeled["report"]["published"])
self.assertEqual(unlabeled["report"]["quarantined"][0]["reason"], "missing-relation-label")
def test_tombstone_is_omitted_and_publish_writes_both_files(self) -> None:
retired = project(RECORDS + [{
"kind": "concept",
"key": "retired",
"title": "Retired",
"tldr": "Gone.",
"deleted": True,
}])
self.assertEqual(retired["report"]["tombstoned"], ["retired"])
self.assertEqual(len(retired["snapshot"]["entities"]), 2)
with tempfile.TemporaryDirectory() as tmp:
destination = Path(tmp) / "out"
publish_directory(destination, retired)
self.assertTrue((destination / "snapshot.json").is_file())
self.assertTrue((destination / "manifest.json").is_file())
self.assertTrue((destination / "import-report.json").is_file())
human = (destination / "import-report.md").read_text(encoding="utf-8")
self.assertIn("Connector version: 0.1.0", human)
self.assertIn("License: CC0-1.0", human)
self.assertIn("retired", human)
published = json.loads((destination / "snapshot.json").read_text(encoding="utf-8"))
manifest = json.loads((destination / "manifest.json").read_text(encoding="utf-8"))
self.assertEqual(validate_documents(published, manifest, require_size=False), [])
with self.assertRaises(RuntimeError):
publish_directory(destination, project(RECORDS, source={**SOURCE, "license": ""}))
def _yaml(record: dict) -> str:
lines = [
f"kind: {record['kind']}",
f"key: {record['key']}",
f"title: {record['title']}",
f"tldr: {record['tldr']}",
f"summary: {record['summary']}",
"tags:",
*[f" - {tag}" for tag in record["tags"]],
"relations:",
]
if record["relations"]:
for link in record["relations"]:
lines.append(f" - predicate: {link['predicate']}")
lines.append(f" to: {link['to']}")
else:
lines.append(" []")
return "\n".join(lines) + "\n"
if __name__ == "__main__":
unittest.main()