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

Finish the computing-basics pilot and the local OnCo reference slice.
All checks were successful
synthetic / suite (push) Successful in 56s
Details

Browse source 
The single-operator pilot keeps static projection. Local OnCo use requires a CC BY-NC origin notice, and the slice stays out of public CI.

Assistant: grok
Assistant-Session: 01a15378-f601-7fe2-a143-8e1500a1d59c
This commit is contained in:
tegwick 2026-10-19 13:04:55 +02:00
parent 55f6180b8e
commit f8c94d3272
33 changed files with 1278 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

221
tests/test_onco_reference.py Normal file
View file

@ -0,0 +1,221 @@
"""OnCo reference adapter. Public tests use a synthetic allowlist, not the corpus."""
from __future__ import annotations
import json
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from connectors.onco_reference import ( # noqa: E402
PIN,
assert_local_destination,
extract_allowlist,
find_object,
kinds_for,
load_authorization,
object_fields,
onco_source,
project_slice,
relation_labels,
upstream_sha,
write_projection,
)
from content_explorer.attribution import attribution_gaps # noqa: E402
def _has_simple_path(graph: dict[str, set[str]], start: str, hops: int) -> bool:
"""True when some walk visits `hops` new entities and does not repeat one."""
found = False
def walk(node: str, depth: int, path: set[str]) -> None:
nonlocal found
if found:
return
if depth == hops:
found = True
return
for neighbour in graph.get(node, ()):
if neighbour in path:
continue
walk(neighbour, depth + 1, path | {neighbour})
walk(start, 0, {start})
return found
ALLOWLIST = json.loads((ROOT / "tests" / "onco-reference" / "slice-allowlist.json").read_text(encoding="utf-8"))
UPSTREAM = ROOT / "upstream-onco"
SAMPLE = """
const asOf = "2026-09-04";
export const items = [
{
id: "alpha", kind: "concept", name: "Alpha", asOf,
tldr: "Alpha is a synthetic record.",
summary: "Alpha summary.",
related: ["beta", "outside"],
},
{ id: "beta", kind: "method", name: "Beta", tldr: "Beta is linked.", summary: "Beta summary.", asOf: "2026-10-01", related: ["alpha"] },
];
"""
def _source() -> dict:
return onco_source("2026-10-19")
class AttributionAndProjection(unittest.TestCase):
def test_cc_by_nc_gaps_block_projection(self) -> None:
source = _source()
self.assertEqual(attribution_gaps(source), [])
incomplete = dict(source)
incomplete["modificationNotice"] = ""
self.assertIn("modificationNotice", attribution_gaps(incomplete))
records = [
{"id": "alpha", "kind": "concept", "title": "Alpha", "tldr": "A.", "summary": "A.", "asOf": "2026-10-19", "relations": [("related", "beta")]},
{"id": "beta", "kind": "method", "title": "Beta", "tldr": "B.", "summary": "B.", "asOf": "2026-10-19", "relations": [("related", "alpha"), ("related", "missing")]},
]
with self.assertRaises(RuntimeError):
project_slice(
records,
source=incomplete,
kinds=[{"id": "concept", "label": "Concept", "plural": "Concepts"}, {"id": "method", "label": "Method", "plural": "Methods"}],
relation_labels=relation_labels(),
require_size=False,
)
result = project_slice(
records,
source=source,
kinds=[
{"id": "concept", "label": "Concept", "plural": "Concepts"},
{"id": "method", "label": "Method", "plural": "Methods"},
{"id": "term", "label": "Term", "plural": "Terms"},
] + [{"id": f"extra-{index}", "label": "Extra", "plural": "Extras"} for index in range(1)],
relation_labels=relation_labels(),
require_size=False,
)
stored = {(item["from"], item["to"]) for item in result["snapshot"]["relations"]}
self.assertIn("urn:ce:onco-reference:concept:alpha", {item[0] for item in stored})
self.assertTrue(any(item["to"] == "missing" for item in result["report"]["outsideAllowlist"]))
self.assertFalse(any("missing" in item["to"] for item in result["snapshot"]["relations"]))
self.assertEqual(
result["report"]["storedRelations"] + len(result["report"]["outsideAllowlist"]),
result["report"]["sourceEdges"],
)
self.assertEqual(result["snapshot"]["sources"][0]["attribution"], "Data from OnCo (onco.cc)")
self.assertEqual(result["snapshot"]["sources"][0]["license"], "CC-BY-NC-4.0")
alpha = next(entity for entity in result["snapshot"]["entities"] if entity["id"].endswith(":alpha"))
self.assertEqual(len(alpha["sourceRefs"]), 2)
self.assertTrue(all(attribution_gaps(source) == [] for source in result["snapshot"]["sources"]))
def test_parser_keeps_only_the_named_object(self) -> None:
obj = find_object(SAMPLE, "alpha")
fields = object_fields(obj, {"asOf": "2026-09-04"})
self.assertEqual(fields["name"], "Alpha")
self.assertEqual(fields["asOf"], "2026-09-04")
self.assertEqual(fields["related"], ["beta", "outside"])
self.assertNotIn('id: "beta"', obj)
def test_refuses_a_destination_outside_the_local_directory(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
with self.assertRaises(RuntimeError):
assert_local_destination(Path(tmp) / "examples" / "dataspaces" / "onco", ROOT)
with self.assertRaises(RuntimeError):
write_projection(ROOT / "examples" / "dataspaces" / "onco-reference", {"snapshot": {}, "manifest": {}, "report": {}}, ROOT)
def test_authorization_must_refuse_redistribution(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "authorization.json"
path.write_text(json.dumps({"granted": True, "distribution_allowed": True, "ci_allowed": False}), encoding="utf-8")
with self.assertRaises(RuntimeError):
load_authorization(path)
path.write_text((ROOT / "tests" / "onco-reference" / "authorization.example.json").read_text(encoding="utf-8"), encoding="utf-8")
self.assertTrue(load_authorization(path)["granted"])
def test_local_mutations_stay_explicit(self) -> None:
source = _source()
kinds = [
{"id": "concept", "label": "Concept", "plural": "Concepts"},
{"id": "method", "label": "Method", "plural": "Methods"},
]
alpha = {"id": "alpha", "kind": "concept", "title": "Alpha", "tldr": "A.", "summary": "A long.", "asOf": "2026-10-19", "relations": [("related", "beta")]}
beta = {"id": "beta", "kind": "method", "title": "Beta", "tldr": "B.", "summary": "B long.", "asOf": "2026-10-19", "relations": [("related", "alpha")]}
with self.assertRaises(RuntimeError):
project_slice([alpha, dict(alpha)], source=source, kinds=kinds, relation_labels=relation_labels(), require_size=False)
bare = dict(source)
del bare["license"]
with self.assertRaises(RuntimeError):
project_slice([alpha, beta], source=bare, kinds=kinds, relation_labels=relation_labels(), require_size=False)
removed = project_slice([alpha], source=source, kinds=kinds, relation_labels=relation_labels(), require_size=False)
self.assertEqual(removed["report"]["storedRelations"], 0)
self.assertEqual(removed["report"]["outsideAllowlist"], [{"from": "alpha", "predicate": "related", "to": "beta"}])
renamed = dict(alpha)
renamed["title"] = "Alpha renamed locally"
projected = project_slice([renamed, beta], source=source, kinds=kinds, relation_labels=relation_labels(), require_size=False)
titles = {entity["id"]: entity["title"] for entity in projected["snapshot"]["entities"]}
self.assertEqual(titles["urn:ce:onco-reference:concept:alpha"], "Alpha renamed locally")
self.assertIn("Projected", projected["snapshot"]["sources"][0]["modificationNotice"])
class LicensedSlice(unittest.TestCase):
def test_pinned_allowlist_projects_locally(self) -> None:
if not UPSTREAM.is_dir():
self.skipTest("upstream checkout is absent")
self.assertEqual(upstream_sha(UPSTREAM), PIN)
entries = ALLOWLIST["records"]
self.assertGreaterEqual(len(entries), 12)
self.assertLessEqual(len(entries), 25)
records = extract_allowlist(UPSTREAM, entries, ROOT / "tests" / "onco-reference" / "authorization.json")
self.assertEqual([record["id"] for record in records], [entry["id"] for entry in entries])
result = project_slice(
records,
source=_source(),
kinds=kinds_for(entries),
relation_labels=relation_labels(),
require_size=True,
)
self.assertGreaterEqual(len(result["snapshot"]["entities"]), 12)
self.assertEqual(
result["report"]["storedRelations"] + len(result["report"]["outsideAllowlist"]),
result["report"]["sourceEdges"],
)
self.assertGreater(result["report"]["sourceEdges"], result["report"]["storedRelations"])
destination = ROOT / "tests" / "onco-reference" / "local" / "dataspace"
write_projection(destination, result, ROOT)
written = json.loads((destination / "snapshot.json").read_text(encoding="utf-8"))
self.assertEqual(written["sources"][0]["originSha"], PIN)
by_id = {entity["id"]: entity for entity in written["entities"]}
tnbc = by_id["urn:ce:onco-reference:cancer:tnbc"]
self.assertTrue(tnbc["sourceRefs"])
neighbours = [
relation["to"] if relation["from"] == tnbc["id"] else relation["from"]
for relation in written["relations"]
if tnbc["id"] in (relation["from"], relation["to"])
]
self.assertGreaterEqual(len(neighbours), 1)
graph: dict[str, set[str]] = {}
for relation in written["relations"]:
graph.setdefault(relation["from"], set()).add(relation["to"])
graph.setdefault(relation["to"], set()).add(relation["from"])
self.assertTrue(_has_simple_path(graph, tnbc["id"], 3))
predicates = {relation["predicate"] for relation in written["relations"]}
self.assertGreaterEqual(len(predicates), 2)
labels = json.loads((destination / "manifest.json").read_text(encoding="utf-8"))["relationLabels"]
self.assertTrue(all(labels[predicate]["inverse"] for predicate in predicates))
grouped: dict[tuple[str, str], int] = {}
for relation in written["relations"]:
grouped[(relation["from"], relation["predicate"])] = grouped.get((relation["from"], relation["predicate"]), 0) + 1
self.assertTrue(any(count >= 2 for count in grouped.values()))
report = json.loads((destination / "import-report.json").read_text(encoding="utf-8"))
self.assertTrue(any("aka" in item["fields"] for item in report["missingOptional"]))
self.assertTrue(all(len(entity["sourceRefs"]) >= 2 for entity in written["entities"]))
self.assertTrue(all(attribution_gaps(source) == [] for source in written["sources"]))
self.assertTrue(any(entity["tldr"] != entity["summary"] and entity.get("asOf") for entity in written["entities"]))
self.assertEqual(labels and list(json.loads((destination / "manifest.json").read_text(encoding="utf-8"))["kindViews"].values())[0]["facets"], ["tags", "kind"])
if __name__ == "__main__":
unittest.main()