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