Nick
24d14ff8e1
Migrate 22 malformed contract-discovery IDs to construction-surface, retain evidence/edges, keep candidates DECLARED, unify cold/warm language projection, and refresh the pinned aiohttp lock. Red regression tests precede the fix; local full suite 2301 passed, spine canaries 19/19 and 21/21.
219 lines
9.4 KiB
Python
219 lines
9.4 KiB
Python
"""test_structural_store.py — SCAFFOLD-TRANCHE-001 boundary tests.
|
|
|
|
Proves the shared structural-store primitive contains no Rune-specific
|
|
assumptions and behaves identically to the pre-extraction spine engine.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from rune import structural_store as ss # noqa: E402
|
|
|
|
|
|
class TestNoRuneConstants:
|
|
"""The shared core must contain zero Rune-specific assumptions."""
|
|
|
|
def test_no_rune_paths_in_source(self):
|
|
src = (ROOT / "src" / "rune" / "structural_store.py").read_text()
|
|
for forbidden in ("/src/rune", "/mnt/workdrive", "rune-code", "~/.rune", "~/.config/opencode", "tests/"):
|
|
assert forbidden not in src, f"Rune-specific reference: {forbidden}"
|
|
|
|
def test_no_rune_module_imports(self):
|
|
import ast
|
|
|
|
tree = ast.parse((ROOT / "src" / "rune" / "structural_store.py").read_text())
|
|
imported = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
imported.update(a.name for a in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
imported.add(node.module)
|
|
assert not any("rune" in m for m in imported), f"shared core must not import rune modules: {imported}"
|
|
|
|
def test_no_rune_vocabulary_in_schema(self):
|
|
"""Kinds like 'module'/'subsystem' are DATA, not schema. The shared
|
|
core must not hardcode them."""
|
|
src = (ROOT / "src" / "rune" / "structural_store.py").read_text()
|
|
for forbidden_kind in ('"module"', '"subsystem"', '"deployment_surface"'):
|
|
assert forbidden_kind not in src, f"Rune vocabulary hardcoded in shared core: {forbidden_kind}"
|
|
|
|
|
|
class TestStoreRoundtrip:
|
|
def _doc(self):
|
|
d = ss.new_doc()
|
|
ss.add_entity(d, id="project:alpha", kind="project", identities={"purpose": "test fixture"})
|
|
ss.add_entity(d, id="component:w", kind="widget_component", maturity="DECLARED")
|
|
ss.add_relationship(d, type="builds", from_id="project:alpha", to_id="component:w", evidence_refs=["ev"])
|
|
return d
|
|
|
|
def test_open_world_entity_kind_roundtrip(self, tmp_path):
|
|
"""An entity kind the store has never heard of round-trips intact."""
|
|
d = self._doc()
|
|
ss.add_entity(
|
|
d, id="capability:quantum_flip", kind="quantum_flipper", identities={"axis": "z"}, ext={"wibble": 42}
|
|
)
|
|
p = ss.save(d, tmp_path / "s.json")
|
|
loaded = ss.load(p)
|
|
e = ss.entity(loaded, "capability:quantum_flip")
|
|
assert e["kind"] == "quantum_flipper"
|
|
assert e["ext"]["wibble"] == 42
|
|
assert e["identities"]["axis"] == "z"
|
|
|
|
def test_unknown_relationship_type_roundtrip(self, tmp_path):
|
|
d = self._doc()
|
|
ss.add_relationship(
|
|
d, type="contextualises", from_id="project:alpha", to_id="component:w", evidence_refs=["ev"]
|
|
)
|
|
p = ss.save(d, tmp_path / "s.json")
|
|
loaded = ss.load(p)
|
|
rels = ss.relationships(loaded, type="contextualises")
|
|
assert len(rels) == 1 and rels[0]["to"] == "component:w"
|
|
|
|
def test_stable_identity_across_save_load(self, tmp_path):
|
|
d = self._doc()
|
|
uri = ss.spine_uri("project:alpha")
|
|
p = ss.save(d, tmp_path / "s.json")
|
|
loaded = ss.load(p)
|
|
assert uri == "spine://project/alpha"
|
|
assert ss.entity(loaded, ss.resolve_uri(uri))["kind"] == "project"
|
|
|
|
def test_alias_roundtrip(self, tmp_path):
|
|
d = self._doc()
|
|
ss.alias(d, "component:w", "component:widget_old_name")
|
|
p = ss.save(d, tmp_path / "s.json")
|
|
loaded = ss.load(p)
|
|
assert ss.entity(loaded, "component:widget_old_name")["id"] == "component:w"
|
|
|
|
def test_supersession_roundtrip(self, tmp_path):
|
|
d = self._doc()
|
|
ss.add_entity(d, id="component:w2", kind="widget_component", maturity="DECLARED")
|
|
ss.supersede(d, "component:w2", "component:w", "evidence:decision")
|
|
p = ss.save(d, tmp_path / "s.json")
|
|
loaded = ss.load(p)
|
|
old = ss.entity(loaded, "component:w")
|
|
assert old["superseded_by"] == "component:w2"
|
|
assert old["status"] == "SUPERSEDED"
|
|
|
|
def test_maturity_requires_evidence(self):
|
|
d = self._doc()
|
|
with pytest.raises(ValueError, match="requires evidence"):
|
|
ss.promote(d, "component:w", "PRESENT", "")
|
|
|
|
def test_no_evidence_no_live_proven(self):
|
|
d = self._doc()
|
|
assert ss.entity(d, "component:w")["maturity"] == "DECLARED"
|
|
|
|
def test_reconciliation_stales_unobserved(self, tmp_path):
|
|
d = self._doc()
|
|
rel = ss.relationships(d, type="builds")[0]
|
|
rel["authority"] = "test-projector"
|
|
rel["status"] = "CURRENT"
|
|
ss._doc_authorities(d)["test-projector"] = {"generation": 1, "observed": []}
|
|
staled = ss.reconcile_authorities(d)
|
|
assert len(staled) == 1
|
|
assert rel["status"] == "STALE"
|
|
|
|
def test_snapshot_deterministic(self, tmp_path):
|
|
d = self._doc()
|
|
d["generated_at"] = "FIXED"
|
|
p1 = ss.snapshot(d, tmp_path, "a", tag="deadbeef")
|
|
p2 = ss.snapshot(d, tmp_path, "a", tag="deadbeef")
|
|
assert p1.read_bytes() == p2.read_bytes()
|
|
|
|
def test_diff_categories(self, tmp_path):
|
|
old = self._doc()
|
|
new = json.loads(json.dumps(old))
|
|
ss.add_entity(new, id="component:new_thing", kind="widget_component", maturity="DECLARED")
|
|
ss.promote(new, "component:w", "WIRED", "new-imports")
|
|
d = ss.diff(old, new)
|
|
assert d["entities_added"] == ["component:new_thing"]
|
|
assert {"id": "component:w", "from": "DECLARED", "to": "WIRED"} in d["maturity_changed"]
|
|
|
|
def test_schema_version_enforced(self, tmp_path):
|
|
d = self._doc()
|
|
d["schema_version"] = 99
|
|
with pytest.raises(ValueError, match="schema_version"):
|
|
ss.save(d, tmp_path / "bad.json")
|
|
|
|
|
|
class TestSpineExtractionNeutrality:
|
|
"""Behavior-neutral proof: the shared core produces identical results
|
|
to the pre-extraction spine engine for identical inputs."""
|
|
|
|
def test_spine_imports_shared_core(self):
|
|
from rune import spine
|
|
|
|
assert hasattr(spine, "add_entity") and hasattr(spine, "impact_frontier")
|
|
|
|
def test_spine_rebuild_identical_structure(self, tmp_path):
|
|
"""Rebuild through the refactored spine produces identical structural
|
|
content to the pre-extraction snapshot (modulo observation timestamps
|
|
and the legitimate deployed-identity advance)."""
|
|
pre = json.loads((ROOT / "store" / "spine" / "current.json").read_text())
|
|
# run the actual spine build via CLI in a subprocess (real path)
|
|
r = subprocess.run(
|
|
[sys.executable, "-m", "rune.spine", "build"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
env={"PYTHONPATH": str(ROOT / "src"), "PATH": "/usr/bin:/bin", "HOME": "/tmp"},
|
|
cwd=str(ROOT),
|
|
)
|
|
assert r.returncode == 0, r.stderr[-300:]
|
|
post = json.loads((ROOT / "store" / "spine" / "current.json").read_text())
|
|
# structural content must be identical
|
|
pre_e = {e["id"]: {k: v for k, v in e.items() if k != "observed_at"} for e in pre["entities"]}
|
|
post_e = {e["id"]: {k: v for k, v in e.items() if k != "observed_at"} for e in post["entities"]}
|
|
# deployed identity advances legitimately (deploy moved 982a→e3b4)
|
|
pre_e.pop("deployment:current", None)
|
|
post_e.pop("deployment:current", None)
|
|
assert pre_e == post_e, "unexplained structural change in rebuild"
|
|
# relationships: structural content identical modulo observation
|
|
# timestamps AND the legitimately advancing deployed identity —
|
|
# deploy-identity-projector refreshes deployed_as edges (observed_at
|
|
# + evidence SHA) whenever the deployed tree moved between the
|
|
# committed store's build and this rebuild (observed live 2026-09-13:
|
|
# 0a9b0648 deploy landed between local build and CI build).
|
|
def _rel_norm(rels):
|
|
out = []
|
|
for r in rels:
|
|
if r["type"] == "deployed_as":
|
|
continue # advances legitimately with the deployment
|
|
out.append({k: v for k, v in r.items() if k != "observed_at"})
|
|
return out
|
|
|
|
assert _rel_norm(pre["relationships"]) == _rel_norm(post["relationships"])
|
|
|
|
def test_spine_store_schema_unchanged(self):
|
|
d = json.loads((ROOT / "store" / "spine" / "current.json").read_text())
|
|
assert d["schema_version"] == 1
|
|
kinds = {e["kind"] for e in d["entities"]}
|
|
assert kinds <= {
|
|
"module",
|
|
"subsystem",
|
|
"deployment_surface",
|
|
"project",
|
|
"contract",
|
|
"maintenance_item",
|
|
"prd_state",
|
|
# H-meta (contract BUILDER-2026-10-04T06:24:56Z-cli): meta-scaffold
|
|
# surface classes — crates, Go surfaces, configs, workflows. The
|
|
# spine is the meta-scaffold; these kinds are the vocabulary for
|
|
# every part of rune that is not a Python module.
|
|
"rust-crate",
|
|
"go-surface",
|
|
"config",
|
|
"workflow",
|
|
# Closed contracts may name files or validation surfaces that
|
|
# are not Python modules. Keep their provenance without lying
|
|
# about their module identity (issue #2465 CI regression).
|
|
"construction-surface",
|
|
}
|