diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py
index 09fb03a..162304c 100755
--- a/src/logic_network_generator.py
+++ b/src/logic_network_generator.py
@@ -2440,3 +2440,181 @@ def export_node_reaction_context(entity_uuid_registry: Dict[tuple, str],
cols = ["context_node", "reaction_id", "role"]
pd.DataFrame(rows, columns=cols).to_csv(output_file, index=False)
logger.info(f"Exported {len(rows)} node-reaction-context rows: {output_file}")
+
+
+def export_node_resolution(pathway_id: str,
+ pathway_logic_network: pd.DataFrame,
+ reaction_id_map: pd.DataFrame,
+ uuid_mapping: Dict[str, str],
+ output_file: str,
+ exclusions_file: str) -> None:
+ """Write node_resolution.csv and node_exclusions.csv.
+
+ One table answers both directions: grouped by ``stable_id`` it says which
+ nodes represent a Reactome entity, grouped by ``uuid`` it says what a node
+ stands for. See specs/005-node-identity-mapping in the deltasignal repo.
+
+ The motivating gap is set-valued readouts. EntitySets are split into their
+ member species, so a set has no node of its own and nothing recorded the
+ link back; that silently cost the benchmark 204 of 847 cases, every one of
+ the 20 blocked readouts being a set. ``set_member`` rows are that link.
+
+ Absence is declared rather than silent. ``node_exclusions.csv`` is NOT
+ expected to be empty: roughly 182 entries per catalog are the deliberately
+ atomic modifier sets (ubiquitin's UBB/UBC repeat units and the rest of
+ ``get_modifier_isoform_entity_set_ids``), which are a design decision, and
+ the rest are a real gap. The ``reason`` column is the only thing separating
+ those two in one list, so it is required and must be specific.
+ """
+ from src.neo4j_connector import (get_labels, get_pathway_participating_entities,
+ get_reactome_release, get_set_members,
+ get_modifier_isoform_entity_set_ids)
+ from src.set_resolution import make_neo4j_resolver
+
+ release = get_reactome_release()
+ release_str = str(release) if release is not None else ""
+ uuid_to_str = _uuid_to_stable_id_map(pathway_logic_network, uuid_mapping)
+ vr_to_reaction = dict(zip(reaction_id_map["uid"].astype(str),
+ reaction_id_map["reactome_id"].astype(str)))
+
+ rows: List[Dict[str, Any]] = []
+ seen: Set[tuple] = set()
+
+ def add(stable_id: str, node_uuid: str, relation: str, depth: int,
+ role: str = "", reaction_stid: str = "") -> None:
+ if not stable_id or not node_uuid:
+ return
+ key = (stable_id, node_uuid, relation, role, reaction_stid)
+ if key in seen:
+ return
+ seen.add(key)
+ rows.append({
+ "stable_id": stable_id, "uuid": node_uuid, "relation": relation,
+ "depth": depth, "role": role, "reaction_stid": reaction_stid,
+ # Populated by the diagram work (US3); the columns exist now so
+ # consumers do not have to branch on schema version later.
+ "glyph_id": "", "diagram_stid": "",
+ "release": release_str,
+ })
+
+ # 1. Reaction nodes.
+ for vr_uid, rid in vr_to_reaction.items():
+ if vr_uid in uuid_to_str:
+ continue # an entity node that happens to share the id space
+ add(rid, vr_uid, "reaction", 0, "", rid)
+
+ # 2. Entity nodes: what each node directly stands for.
+ stid_to_uuids: Dict[str, Set[str]] = {}
+ for node_uuid, node_str in uuid_to_str.items():
+ if "::variant::" in node_str:
+ parent = node_str.split("::variant::")[0]
+ add(parent, node_uuid, "variant", 0)
+ stid_to_uuids.setdefault(parent, set()).add(node_uuid)
+ for member in node_str.split("::variant::")[-1].split("_"):
+ if member.startswith("R-"):
+ add(member, node_uuid, "set_member", 1)
+ stid_to_uuids.setdefault(member, set()).add(node_uuid)
+ else:
+ add(node_str, node_uuid, "self", 0)
+ stid_to_uuids.setdefault(node_str, set()).add(node_uuid)
+ try:
+ labels = get_labels(node_str)
+ except Exception:
+ labels = []
+ if "Complex" in labels:
+ try:
+ for component in sorted(get_terminal_components(node_str)):
+ if component != node_str:
+ add(component, node_uuid, "complex_component", 1)
+ except Exception:
+ logger.warning(f"could not decompose complex {node_str}")
+
+ # 3. Reaction position, from the network itself rather than the fetch rows
+ # (see export_node_reaction_context and issue #67).
+ for _, e in pathway_logic_network.iterrows():
+ etype = str(e.get("edge_type") or "")
+ if etype in ("input", "catalyst", "regulator"):
+ node_uuid, rxn_uuid = str(e.get("source_id")), str(e.get("target_id"))
+ elif etype == "output":
+ rxn_uuid, node_uuid = str(e.get("source_id")), str(e.get("target_id"))
+ else:
+ continue
+ edge_rid = vr_to_reaction.get(rxn_uuid)
+ edge_node_str = uuid_to_str.get(node_uuid)
+ if not edge_rid or not edge_node_str:
+ continue
+ base = edge_node_str.split("::variant::")[0]
+ add(base, node_uuid,
+ "self" if "::variant::" not in edge_node_str else "variant",
+ 0, etype, edge_rid)
+
+ # 4. THE POINT: set -> the member nodes it was split into.
+ excluded: List[Dict[str, str]] = []
+ try:
+ participating = get_pathway_participating_entities(pathway_id)
+ except Exception:
+ logger.warning("could not list participating entities; set rows omitted")
+ participating = set()
+ try:
+ atomic = set(get_modifier_isoform_entity_set_ids())
+ except Exception:
+ atomic = set()
+ resolve = make_neo4j_resolver(get_set_members, get_labels)
+
+ for stable_id in sorted(participating):
+ try:
+ labels = get_labels(stable_id)
+ except Exception:
+ labels = []
+ if "EntitySet" not in labels:
+ if stable_id not in stid_to_uuids:
+ excluded.append({"stable_id": stable_id,
+ "reason": "participates but no node was generated",
+ "release": release_str})
+ continue
+ if stable_id in stid_to_uuids:
+ continue # the set itself is a node; nothing was split
+ resolution = resolve(stable_id)
+ hits = 0
+ for leaf in resolution.leaves:
+ for node_uuid in sorted(stid_to_uuids.get(leaf.stable_id, ())):
+ add(stable_id, node_uuid, "set_member", leaf.depth)
+ hits += 1
+ if hits == 0:
+ # Name the leaves. "none resolved" is true of a design decision and
+ # of a bug alike; the ids are what lets a reader tell them apart.
+ leaf_ids = ", ".join(leaf.stable_id for leaf in resolution.leaves[:4]) or "none"
+ reason = ("atomic modifier set, deliberately not expanded"
+ if stable_id in atomic else
+ f"set has no node and none of its {len(resolution.leaves)} "
+ f"leaves resolved ({leaf_ids})")
+ excluded.append({"stable_id": stable_id, "reason": reason,
+ "release": release_str})
+ elif resolution.truncated:
+ excluded.append({
+ "stable_id": stable_id,
+ "reason": f"partially resolved: depth bound hit at "
+ f"{resolution.max_depth_reached}",
+ "release": release_str})
+ else:
+ missing = [leaf.stable_id for leaf in resolution.leaves
+ if leaf.stable_id not in stid_to_uuids]
+ if missing:
+ # Reported, not silently combined over what did resolve.
+ reason = ("atomic modifier set, deliberately not expanded"
+ if stable_id in atomic else
+ f"partially resolved: {len(missing)} of "
+ f"{len(resolution.leaves)} leaves have no node "
+ f"({', '.join(sorted(missing)[:3])})")
+ excluded.append({"stable_id": stable_id, "reason": reason,
+ "release": release_str})
+
+ cols = ["stable_id", "uuid", "relation", "depth", "role", "reaction_stid",
+ "glyph_id", "diagram_stid", "release"]
+ pd.DataFrame(rows, columns=cols).to_csv(output_file, index=False)
+ pd.DataFrame(excluded, columns=["stable_id", "reason", "release"]).to_csv(
+ exclusions_file, index=False)
+ set_rows = sum(1 for r in rows if r["relation"] == "set_member")
+ logger.info(
+ f"Exported {len(rows)} node-resolution rows ({set_rows} set_member), "
+ f"{len(excluded)} exclusions: {output_file}")
diff --git a/src/pathway_generator.py b/src/pathway_generator.py
index ec0abf2..10e89f1 100755
--- a/src/pathway_generator.py
+++ b/src/pathway_generator.py
@@ -13,6 +13,7 @@
create_pathway_logic_network,
export_entity_reaction_proxy_mapping,
export_node_reaction_context,
+ export_node_resolution,
export_nodes,
export_uuid_to_reactome_mapping,
)
@@ -402,6 +403,14 @@ def generate_pathway_file(
str(pathway_output_dir / "node_reaction_context.csv"),
logic_network=result.logic_network,
)
+ export_node_resolution(
+ pathway_id,
+ result.logic_network,
+ result.reaction_id_map,
+ result.uuid_mapping,
+ str(pathway_output_dir / "node_resolution.csv"),
+ str(pathway_output_dir / "node_exclusions.csv"),
+ )
except Exception as e:
logger.error(f"Failed to write node provenance files: {e}", exc_info=True)
# Don't raise - supplementary
diff --git a/src/resolution_validation.py b/src/resolution_validation.py
new file mode 100644
index 0000000..b133d89
--- /dev/null
+++ b/src/resolution_validation.py
@@ -0,0 +1,151 @@
+"""Completeness checks for node_resolution.csv, in both directions.
+
+Adam's requirement: "we need to be able to map every node in the database to
+the LNG node perfectly. and same with the other way around." "Perfectly" is
+not checkable, so it is expressed here as: no absence goes undeclared.
+
+Pure functions over already-loaded rows, deliberately free of the filesystem
+and Neo4j, so the negative control in
+``tests/test_resolution_negative_control.py`` can corrupt the inputs directly
+and prove these checks FAIL. A completeness check that cannot fail is the
+specific defect this project has shipped — ``_decomposed_ids`` once passed
+11 of 11 while masking 18 dropped catalysts — so the negative control is the
+load-bearing half of this module, not its polish.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, Iterable, List, NamedTuple, Set
+
+
+class Violation(NamedTuple):
+ check: str
+ subject: str
+ detail: str
+
+
+def check_reverse_completeness(network_node_uuids: Set[str],
+ resolution_rows: Iterable[dict]) -> List[Violation]:
+ """Every node in the network must say what it stands for."""
+ mapped = {str(r["uuid"]) for r in resolution_rows}
+ return [
+ Violation("reverse_completeness", uuid,
+ "node appears in the logic network but in no resolution row")
+ for uuid in sorted(network_node_uuids - mapped)
+ ]
+
+
+def check_referential_integrity(network_node_uuids: Set[str],
+ resolution_rows: Iterable[dict]) -> List[Violation]:
+ """Every resolution row must point at a node that exists.
+
+ This is the shape of issue #67, where 100% of catalyst and regulator
+ context rows named the undecomposed parent entity — a node absent from
+ the network — and nothing noticed for as long as the export existed.
+ """
+ return [
+ Violation("referential_integrity", str(r["uuid"]),
+ f"resolution row for {r['stable_id']} names a node absent "
+ f"from the logic network")
+ for r in resolution_rows
+ if str(r["uuid"]) not in network_node_uuids
+ ]
+
+
+def check_forward_completeness(participating_entities: Set[str],
+ resolution_rows: Iterable[dict],
+ exclusion_rows: Iterable[dict]) -> List[Violation]:
+ """Every participating entity resolves, or is excluded with a reason."""
+ resolved = {str(r["stable_id"]) for r in resolution_rows}
+ excluded = {str(r["stable_id"]) for r in exclusion_rows}
+ return [
+ Violation("forward_completeness", stid,
+ "entity participates in a reaction but has no resolution "
+ "row and no exclusion")
+ for stid in sorted(participating_entities - resolved - excluded)
+ ]
+
+
+def check_exclusions_have_reasons(exclusion_rows: Iterable[dict]) -> List[Violation]:
+ """An exclusion without a reason is a silent absence wearing a hat.
+
+ The reason is load-bearing rather than bureaucratic: the same list holds
+ deliberately atomic modifier sets (a design decision) and entities that
+ should have been generated (a bug), and only this column separates them.
+ """
+ violations = []
+ for row in exclusion_rows:
+ reason = str(row.get("reason") or "").strip()
+ if not reason:
+ violations.append(Violation("exclusion_reason", str(row["stable_id"]),
+ "exclusion has no reason"))
+ elif len(reason) < 10:
+ violations.append(Violation("exclusion_reason", str(row["stable_id"]),
+ f"reason too vague to act on: {reason!r}"))
+ return violations
+
+
+def check_glyph_pairing(resolution_rows: Iterable[dict]) -> List[Violation]:
+ """A glyph id without its diagram is meaningless — ids are unique only
+ within a diagram, so one without the other cannot be resolved."""
+ violations = []
+ for row in resolution_rows:
+ glyph = str(row.get("glyph_id") or "").strip()
+ diagram = str(row.get("diagram_stid") or "").strip()
+ if bool(glyph) != bool(diagram):
+ violations.append(Violation(
+ "glyph_pairing", str(row["uuid"]),
+ f"glyph_id={glyph!r} and diagram_stid={diagram!r} must be "
+ f"present together or absent together"))
+ return violations
+
+
+def check_release_recorded(resolution_rows: Iterable[dict],
+ expected_release: str | None = None) -> List[Violation]:
+ """Every row carries its Reactome release.
+
+ Version skew has produced a false finding on this project; a mapping
+ without a release cannot be refused when it does not match.
+ """
+ violations = []
+ seen: Set[str] = set()
+ for row in resolution_rows:
+ release = str(row.get("release") or "").strip()
+ if not release:
+ violations.append(Violation("release_recorded", str(row["uuid"]),
+ "row carries no Reactome release"))
+ else:
+ seen.add(release)
+ if len(seen) > 1:
+ violations.append(Violation("release_recorded", "
",
+ f"mixed releases in one table: {sorted(seen)}"))
+ if expected_release and seen and seen != {str(expected_release)}:
+ violations.append(Violation("release_recorded", "",
+ f"expected release {expected_release}, found {sorted(seen)}"))
+ return violations
+
+
+def run_all(network_node_uuids: Set[str],
+ resolution_rows: List[dict],
+ exclusion_rows: List[dict],
+ participating_entities: Set[str] | None = None,
+ expected_release: str | None = None) -> List[Violation]:
+ """All checks. Empty result means the mapping is complete in both
+ directions; anything else names what is missing and why it matters."""
+ violations: List[Violation] = []
+ violations += check_reverse_completeness(network_node_uuids, resolution_rows)
+ violations += check_referential_integrity(network_node_uuids, resolution_rows)
+ violations += check_exclusions_have_reasons(exclusion_rows)
+ violations += check_glyph_pairing(resolution_rows)
+ violations += check_release_recorded(resolution_rows, expected_release)
+ if participating_entities is not None:
+ violations += check_forward_completeness(participating_entities,
+ resolution_rows, exclusion_rows)
+ return violations
+
+
+def summarise(violations: Iterable[Violation]) -> Dict[str, int]:
+ counts: Dict[str, int] = {}
+ for v in violations:
+ counts[v.check] = counts.get(v.check, 0) + 1
+ return counts
diff --git a/src/set_resolution.py b/src/set_resolution.py
new file mode 100644
index 0000000..8b00087
--- /dev/null
+++ b/src/set_resolution.py
@@ -0,0 +1,143 @@
+"""Resolve an EntitySet to the leaf members it was split into.
+
+The generator splits EntitySets into their member species, so a set has no
+node of its own. Nothing recorded the link back, which silently cost the
+benchmark 204 of 847 cases: every one of the 20 blocked readouts is a set,
+and they are the canonical set-shaped readouts of the best-known pathways —
+p-T,p-S-AKT, p-S9/21-GSK3, phospho-FOXO1/3/4/6, p-T,Y MAPK dimers, with 116
+of the 204 in PIP3 alone.
+
+**One hop is not enough.** 15 of those 20 resolve in a single hop, but the
+other 5 are sets whose members are themselves sets — "p-T,Y MAPK monomers
+and dimers" contains "p-T,Y MAPKs" and "p-T,Y MAPK dimers", neither of which
+is a leaf. Recursion to leaves resolves all 20.
+
+**On the cycle guard.** An earlier draft of the spec justified the visited
+set by claiming Reactome's membership graph contains cycles. Measured on
+Release97 that is false: zero self-membership, zero cycles at length 2, 3 or
+4, across 5,440 nested sets, with maximum nesting depth 5. The guard stays
+because a curation error would otherwise hang generation, not because
+anything observed needs it. `MAX_DEPTH` is a second guard on the same risk.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, NamedTuple, Set
+
+from src.argument_parser import logger
+
+# Observed maximum nesting depth is 5 on Release97. The bound is deliberately
+# loose: exceeding it means curation changed shape, which should be reported
+# rather than silently truncated.
+MAX_DEPTH = 10
+
+
+class SetLeaf(NamedTuple):
+ """One leaf member of a set, and how far down it was found."""
+ stable_id: str
+ depth: int
+
+
+class SetResolution(NamedTuple):
+ """The result of resolving one set.
+
+ ``truncated`` is the honest half: a caller must be able to tell a
+ complete resolution from one that hit the depth bound, because combining
+ member values over a partial set produces a plausible wrong number rather
+ than a visible failure.
+ """
+ stable_id: str
+ leaves: List[SetLeaf]
+ truncated: bool
+ max_depth_reached: int
+
+ @property
+ def leaf_ids(self) -> Set[str]:
+ return {leaf.stable_id for leaf in self.leaves}
+
+
+def resolve_set(
+ stable_id: str,
+ get_members,
+ is_set,
+ max_depth: int = MAX_DEPTH,
+) -> SetResolution:
+ """Expand ``stable_id`` through nested sets to its leaf members.
+
+ ``get_members(stid) -> set[str]`` and ``is_set(stid) -> bool`` are passed
+ in rather than imported so this is testable without Neo4j; production
+ callers hand it ``neo4j_connector.get_set_members`` and a label check.
+
+ A non-set input resolves to itself at depth 0, so callers do not need to
+ branch on whether they hold a set.
+ """
+ if not is_set(stable_id):
+ return SetResolution(stable_id, [SetLeaf(stable_id, 0)], False, 0)
+
+ leaves: Dict[str, int] = {}
+ visited: Set[str] = {stable_id}
+ truncated = False
+ deepest = 0
+ frontier = [(stable_id, 0)]
+
+ while frontier:
+ current, depth = frontier.pop()
+ if depth >= max_depth:
+ # Report rather than silently stopping: a set deeper than the
+ # bound means the graph changed shape.
+ truncated = True
+ logger.warning(
+ f"set resolution hit the depth bound ({max_depth}) at {current} "
+ f"while expanding {stable_id}; result is incomplete"
+ )
+ continue
+ members = get_members(current) or set()
+ if not members:
+ # A set with no members is a leaf in practice; recording it keeps
+ # the caller from seeing an empty resolution for a real entity.
+ if current != stable_id:
+ leaves.setdefault(current, depth)
+ deepest = max(deepest, depth)
+ continue
+ for member in sorted(members):
+ child_depth = depth + 1
+ deepest = max(deepest, child_depth)
+ if is_set(member):
+ if member in visited:
+ # Unreachable on Release97; see the module docstring.
+ logger.warning(
+ f"cycle in set membership: {member} revisited while "
+ f"expanding {stable_id}"
+ )
+ continue
+ visited.add(member)
+ frontier.append((member, child_depth))
+ else:
+ # Keep the SHALLOWEST depth for a leaf reachable by several
+ # routes; depth is "how far down this is", not "how we got here".
+ if member not in leaves or child_depth < leaves[member]:
+ leaves[member] = child_depth
+
+ resolved = [SetLeaf(sid, d) for sid, d in sorted(leaves.items())]
+ if not resolved:
+ logger.warning(f"set {stable_id} resolved to no leaves")
+ return SetResolution(stable_id, resolved, truncated, deepest)
+
+
+def make_neo4j_resolver(get_members_fn, get_labels_fn):
+ """Bind ``resolve_set`` to Neo4j accessors.
+
+ Kept separate so the recursion above stays free of the database.
+ """
+ def is_set(stable_id: str) -> bool:
+ try:
+ return "EntitySet" in (get_labels_fn(stable_id) or [])
+ except Exception:
+ # An unresolvable label means "not a set we can expand"; treating
+ # it as a set would recurse into nothing and report a false empty.
+ return False
+
+ def resolve(stable_id: str, max_depth: int = MAX_DEPTH) -> SetResolution:
+ return resolve_set(stable_id, get_members_fn, is_set, max_depth)
+
+ return resolve
diff --git a/tests/test_resolution_negative_control.py b/tests/test_resolution_negative_control.py
new file mode 100644
index 0000000..8b76b4d
--- /dev/null
+++ b/tests/test_resolution_negative_control.py
@@ -0,0 +1,149 @@
+"""The completeness checks must FAIL on a corrupted mapping.
+
+This file exists because of a specific past failure: `_decomposed_ids`
+reported 11 of 11 passing while masking 18 dropped catalysts, because the
+check had been relaxed until it could no longer fail. A completeness check
+that always passes is worse than no check, since it converts an unknown into
+a false assurance.
+
+So every check here is exercised twice — once against a correct mapping,
+where it must be silent, and once against a mapping broken in exactly the way
+that check exists to catch, where it must speak up.
+"""
+import pytest
+
+from src.resolution_validation import (
+ check_exclusions_have_reasons, check_forward_completeness,
+ check_glyph_pairing, check_referential_integrity, check_release_recorded,
+ check_reverse_completeness, run_all, summarise,
+)
+
+
+def _row(stable_id, uuid, relation="self", **kw):
+ row = {"stable_id": stable_id, "uuid": uuid, "relation": relation,
+ "depth": 0, "role": "", "reaction_stid": "", "glyph_id": "",
+ "diagram_stid": "", "release": "97"}
+ row.update(kw)
+ return row
+
+
+@pytest.fixture
+def good():
+ """A small but complete mapping: two entity nodes and one set over them."""
+ network = {"u1", "u2"}
+ resolution = [
+ _row("R-HSA-1", "u1"),
+ _row("R-HSA-2", "u2"),
+ _row("R-HSA-SET", "u1", "set_member", depth=1),
+ _row("R-HSA-SET", "u2", "set_member", depth=1),
+ ]
+ exclusions = [{"stable_id": "R-HSA-UB", "release": "97",
+ "reason": "atomic modifier set, deliberately not expanded"}]
+ participating = {"R-HSA-1", "R-HSA-2", "R-HSA-SET", "R-HSA-UB"}
+ return network, resolution, exclusions, participating
+
+
+def test_a_correct_mapping_passes(good):
+ network, resolution, exclusions, participating = good
+ assert run_all(network, resolution, exclusions, participating, "97") == []
+
+
+# --- each check, broken in the way it exists to catch -----------------------
+
+def test_reverse_completeness_fails_when_a_node_maps_to_nothing(good):
+ network, resolution, _, _ = good
+ network = network | {"u3_orphan"}
+ violations = check_reverse_completeness(network, resolution)
+ assert [v.subject for v in violations] == ["u3_orphan"]
+
+
+def test_referential_integrity_fails_on_a_row_naming_a_missing_node(good):
+ """This is issue #67's exact shape: the row named the undecomposed parent."""
+ network, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-9", "parent_uuid_not_in_network")]
+ violations = check_referential_integrity(network, resolution)
+ assert len(violations) == 1
+ assert violations[0].subject == "parent_uuid_not_in_network"
+
+
+def test_forward_completeness_fails_on_an_undeclared_absence(good):
+ _, resolution, exclusions, participating = good
+ participating = participating | {"R-HSA-NEVER-GENERATED"}
+ violations = check_forward_completeness(participating, resolution, exclusions)
+ assert [v.subject for v in violations] == ["R-HSA-NEVER-GENERATED"]
+
+
+def test_an_excluded_entity_is_not_a_forward_violation(good):
+ """Declaring an absence is the point — it must not also be reported."""
+ _, resolution, exclusions, participating = good
+ assert check_forward_completeness(participating, resolution, exclusions) == []
+
+
+def test_exclusion_without_a_reason_fails():
+ violations = check_exclusions_have_reasons([{"stable_id": "R-HSA-X", "reason": ""}])
+ assert len(violations) == 1
+
+
+def test_vague_exclusion_reason_fails():
+ """"missing" does not distinguish a design decision from a bug."""
+ violations = check_exclusions_have_reasons([{"stable_id": "R-HSA-X", "reason": "missing"}])
+ assert len(violations) == 1
+ assert "vague" in violations[0].detail
+
+
+def test_glyph_id_without_its_diagram_fails(good):
+ _, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-3", "u1", glyph_id="535")]
+ violations = check_glyph_pairing(resolution)
+ assert len(violations) == 1
+
+
+def test_diagram_without_its_glyph_also_fails(good):
+ _, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-3", "u1", diagram_stid="R-HSA-1257604")]
+ assert len(check_glyph_pairing(resolution)) == 1
+
+
+def test_both_glyph_fields_together_pass(good):
+ _, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-3", "u1", glyph_id="535",
+ diagram_stid="R-HSA-1257604")]
+ assert check_glyph_pairing(resolution) == []
+
+
+def test_missing_release_fails(good):
+ _, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-3", "u1", release="")]
+ assert len(check_release_recorded(resolution)) == 1
+
+
+def test_mixed_releases_in_one_table_fails(good):
+ _, resolution, _, _ = good
+ resolution = resolution + [_row("R-HSA-3", "u1", release="96")]
+ violations = check_release_recorded(resolution)
+ assert any("mixed releases" in v.detail for v in violations)
+
+
+def test_release_mismatch_against_the_networks_fails(good):
+ _, resolution, _, _ = good
+ assert len(check_release_recorded(resolution, expected_release="96")) == 1
+
+
+def test_run_all_reports_every_broken_check_at_once(good):
+ """One corrupted mapping, several independent failures — none masked."""
+ network, resolution, exclusions, participating = good
+ network = network | {"u_orphan"}
+ resolution = resolution + [
+ _row("R-HSA-9", "not_in_network"),
+ _row("R-HSA-3", "u1", glyph_id="535"),
+ ]
+ exclusions = exclusions + [{"stable_id": "R-HSA-Y", "reason": ""}]
+ participating = participating | {"R-HSA-NEVER"}
+ counts = summarise(run_all(network, resolution, exclusions, participating, "97"))
+ assert counts == {
+ "reverse_completeness": 1,
+ "referential_integrity": 1,
+ "exclusion_reason": 1,
+ "glyph_pairing": 1,
+ "forward_completeness": 1,
+ }
diff --git a/tests/test_set_resolution.py b/tests/test_set_resolution.py
new file mode 100644
index 0000000..27de0aa
--- /dev/null
+++ b/tests/test_set_resolution.py
@@ -0,0 +1,91 @@
+"""Unit tests for recursive EntitySet resolution. No Neo4j."""
+from src.set_resolution import MAX_DEPTH, resolve_set
+
+
+def _fixture():
+ """The real shape that broke one-hop resolution.
+
+ R-HSA-5674340 "p-T,Y MAPK monomers and dimers" contains two sets, neither
+ of which is a leaf. It is one of the 5 of 20 blocked readouts that a
+ single hop could not resolve.
+ """
+ members = {
+ "R-HSA-5674340": {"R-HSA-169289", "R-HSA-1268261"},
+ "R-HSA-169289": {"MAPK1", "MAPK3"},
+ "R-HSA-1268261": {"DIMER1", "DIMER2"},
+ "R-HSA-202072": {"AKT1p", "AKT2p", "AKT3p"},
+ }
+ sets = set(members)
+ return (lambda s: members.get(s, set()), lambda s: s in sets)
+
+
+def test_single_hop_set_resolves_to_members():
+ get, is_set = _fixture()
+ r = resolve_set("R-HSA-202072", get, is_set)
+ assert r.leaf_ids == {"AKT1p", "AKT2p", "AKT3p"}
+ assert all(leaf.depth == 1 for leaf in r.leaves)
+ assert not r.truncated
+
+
+def test_nested_set_recurses_to_leaves():
+ """One hop returns sets, not leaves — this is the 5-of-20 case."""
+ get, is_set = _fixture()
+ one_hop = get("R-HSA-5674340")
+ assert all(is_set(m) for m in one_hop), "fixture must nest, or it tests nothing"
+
+ r = resolve_set("R-HSA-5674340", get, is_set)
+ assert r.leaf_ids == {"MAPK1", "MAPK3", "DIMER1", "DIMER2"}
+ assert r.max_depth_reached == 2
+ assert not r.truncated
+
+
+def test_non_set_resolves_to_itself():
+ get, is_set = _fixture()
+ r = resolve_set("MAPK1", get, is_set)
+ assert r.leaves == [("MAPK1", 0)]
+ assert not r.truncated
+
+
+def test_single_member_set_behaves_like_its_member():
+ members = {"S": {"only"}}
+ r = resolve_set("S", lambda s: members.get(s, set()), lambda s: s in members)
+ assert r.leaf_ids == {"only"}
+
+
+def test_diamond_keeps_the_shallowest_depth():
+ """A leaf reachable two ways is reported once, at its shallowest depth.
+
+ depth answers "how far down is this", not "which route did we take".
+ """
+ members = {"top": {"mid", "leaf"}, "mid": {"leaf"}}
+ sets = {"top", "mid"}
+ r = resolve_set("top", lambda s: members.get(s, set()), lambda s: s in sets)
+ assert r.leaf_ids == {"leaf"}
+ assert [leaf.depth for leaf in r.leaves] == [1]
+
+
+def test_cycle_terminates_rather_than_hanging():
+ """Release97 has no membership cycles; this proves the guard works anyway.
+
+ Without the visited set this call does not return, so a curation error
+ would hang generation rather than fail it.
+ """
+ members = {"A": {"B"}, "B": {"A", "leaf"}}
+ sets = {"A", "B"}
+ r = resolve_set("A", lambda s: members.get(s, set()), lambda s: s in sets)
+ assert r.leaf_ids == {"leaf"}
+
+
+def test_exceeding_the_depth_bound_is_reported_not_silently_truncated():
+ """A partial resolution must be visibly partial.
+
+ Combining member values over a silently-truncated set yields a plausible
+ wrong number instead of a visible failure.
+ """
+ depth = MAX_DEPTH + 3
+ members = {f"S{i}": {f"S{i+1}"} for i in range(depth)}
+ members[f"S{depth}"] = {"leaf"}
+ sets = set(members)
+ r = resolve_set("S0", lambda s: members.get(s, set()), lambda s: s in sets)
+ assert r.truncated, "hitting the depth bound must set truncated"
+ assert "leaf" not in r.leaf_ids