Nick
f85c654d9d
14B: the out-of-lock publisher (publish_out_of_lock) no longer trusts the originating job's SHA. It re-reads CURRENT deployment authority immediately before publishing and constructs the note from the authority's identity. Adversarial race held-out: A promotes → B supersedes → B publishes → A resumes → final history says B, A's SHA never enters. Unreadable authority → no publication (never fabricate an identity). Body parsers (summary_from_body / sha_from_body) moved to release_notes.py as the single authority; release_history_cli.py delegates. 14C: .convergence-intent.json ELIMINATED (1 writer, 0 readers, non-atomic write, bogus recorded_at). Pending convergence is derived from the deployment authority itself (atomic symlink + build identity): restart- readable, corruption-immune, monotonic, idempotent by construction. T11 stale-refusal invariant preserved (CLI revalidation unchanged).
772 lines
31 KiB
Python
772 lines
31 KiB
Python
"""test_release_notes.py — SCAFFOLD-TRANCHE-009 test plans §24 (extraction),
|
|
§25 (publication), §26 (T8 correction integration)."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
|
|
|
|
from rune import structural_store as ss # noqa: E402
|
|
from rune import release_notes as rn # noqa: E402
|
|
|
|
|
|
def _governed_doc():
|
|
"""Scaffold doc carrying the governed record shapes release notes draw
|
|
from (closed contracts, resolved/open maintenance, quality records)."""
|
|
doc = ss.new_doc()
|
|
ss.add_entity(
|
|
doc,
|
|
id="contract:C1",
|
|
kind="construction_contract",
|
|
identities={"need": "add export capability"},
|
|
maturity="EXERCISED",
|
|
evidence_refs=["e"],
|
|
)
|
|
c = ss.entity(doc, "contract:C1")
|
|
c["status"] = "closed"
|
|
c.setdefault("ext", {})["contract_quality"] = {
|
|
"correctly_predicted": ["module:a", "module:b"],
|
|
"expected_new_observed": ["module:fresh"],
|
|
}
|
|
c["ext"]["release_evidence"] = {"final_deployed": "abc123 / 0.66.44"}
|
|
ss.add_entity(
|
|
doc,
|
|
id="contract:C2",
|
|
kind="construction_contract",
|
|
identities={"need": "release safety repair"},
|
|
maturity="LIVE_PROVEN",
|
|
evidence_refs=["e"],
|
|
)
|
|
c2 = ss.entity(doc, "contract:C2")
|
|
c2["status"] = "closed"
|
|
c2.setdefault("ext", {})["release_evidence"] = {"deployed_identity": "def456 / 0.66.32"}
|
|
ss.add_entity(
|
|
doc,
|
|
id="maintenance:m1",
|
|
kind="maintenance_item",
|
|
identities={"issue": "deploy-order race", "status": "resolved"},
|
|
evidence_refs=["e"],
|
|
)
|
|
ss.add_entity(
|
|
doc,
|
|
id="maintenance:m2",
|
|
kind="maintenance_item",
|
|
identities={"issue": "external repo proof pending user", "status": "open"},
|
|
evidence_refs=["e"],
|
|
)
|
|
ss.add_entity(
|
|
doc,
|
|
id="tui-projection:module:x",
|
|
kind="tui_projection_decision",
|
|
identities={"feature": "module:x", "projection": "NONE"},
|
|
evidence_refs=["e"],
|
|
)
|
|
return doc
|
|
|
|
|
|
# ── §24 evidence extraction ──────────────────────────────────────────────────
|
|
|
|
|
|
def test_24a_added_capability_extracted():
|
|
facts = rn.extract_evidence_facts(_governed_doc())
|
|
assert any("C1" in r for r in facts["references"])
|
|
|
|
|
|
def test_24b_fix_from_resolved_maintenance():
|
|
facts = rn.extract_evidence_facts(_governed_doc())
|
|
assert any("m1" in f and "deploy-order race" in f for f in facts["fixed"])
|
|
assert not any("m2" in f for f in facts["fixed"])
|
|
|
|
|
|
def test_24c_improvement_not_called_new():
|
|
"""Closed contracts with correctly-predicted effects = PROVEN evidence,
|
|
and the note builder places contract work under Improved/Proven, never
|
|
fabricates 'new'."""
|
|
facts = rn.extract_evidence_facts(_governed_doc())
|
|
assert any("C1" in p for p in facts["proven"])
|
|
note = rn.build_note(version="0.66.44", summary="s", scaffold_doc=_governed_doc())
|
|
# proven items never leak into added[]
|
|
assert not any("C1" in a or "C2" in a for a in note.added)
|
|
assert any("C1" in p for p in note.proven)
|
|
|
|
|
|
def test_24d_maturity_evidence_represented():
|
|
facts = rn.extract_evidence_facts(_governed_doc())
|
|
assert any("0.66.44" in p for p in facts["proven"])
|
|
assert any("0.66.32" in p for p in facts["proven"])
|
|
|
|
|
|
def test_24e_known_limitation_retained():
|
|
note = rn.build_note(version="0.66.44", summary="s", scaffold_doc=_governed_doc())
|
|
assert any("m2" in l and "pending user" in l for l in note.limitations)
|
|
|
|
|
|
def test_24f_unsupported_claim_rejected():
|
|
"""§24F/§12: the T8 close-refusal claim is excluded by the
|
|
exclude_unsupported surface — it cannot enter generated notes."""
|
|
doc = _governed_doc()
|
|
ss.add_entity(
|
|
doc,
|
|
id="contract:T8",
|
|
kind="construction_contract",
|
|
identities={"need": "close-from-verifying refusal reaction proof"},
|
|
maturity="WIRED",
|
|
evidence_refs=["e"],
|
|
)
|
|
c = ss.entity(doc, "contract:T8")
|
|
c["status"] = "closed"
|
|
c.setdefault("ext", {})["contract_quality"] = {"correctly_predicted": ["module:x"]}
|
|
note = rn.build_note(
|
|
version="0.66.44",
|
|
summary="s",
|
|
scaffold_doc=doc,
|
|
extra={"proven": ["bounded MiMo close-refusal reaction proven"]},
|
|
exclude_unsupported=["refusal", "reaction"],
|
|
)
|
|
blob = rn.render_markdown(note)
|
|
assert "refusal" not in blob.lower()
|
|
assert "reaction" not in blob.lower()
|
|
|
|
|
|
def test_24g_same_version_later_commit_represented():
|
|
note = rn.build_note(
|
|
version="0.66.44",
|
|
summary="s",
|
|
scaffold_doc=_governed_doc(),
|
|
release_sha="f86b2d1eecde",
|
|
deployed_sha="d84060cc027f",
|
|
)
|
|
md = rn.render_markdown(note)
|
|
assert "same-version convergence" in md
|
|
assert "f86b2d1eecde"[:12] in md and "d84060cc027f"[:12] in md
|
|
|
|
|
|
def test_24h_deployed_distinct_from_release_sha():
|
|
note = rn.build_note(
|
|
version="0.66.44",
|
|
summary="s",
|
|
scaffold_doc=_governed_doc(),
|
|
release_sha="f86b2d1eecde039f",
|
|
deployed_sha="d84060cc027f1a2b",
|
|
)
|
|
assert note.release_sha != note.deployed_sha
|
|
md = rn.render_markdown(note)
|
|
assert "final deployed" in md # distinction exposed honestly
|
|
|
|
|
|
def test_24i_missing_evidence_no_invented_prose():
|
|
"""§16/§24I: an empty governed pool produces a note with empty sections
|
|
(omitted in Markdown) — never invented prose."""
|
|
note = rn.build_note(version="9.9.9", summary="empty release", scaffold_doc=ss.new_doc())
|
|
md = rn.render_markdown(note)
|
|
assert "## Improved" not in md
|
|
assert "## Added" not in md
|
|
assert "empty release" in md # only the supplied summary exists
|
|
|
|
|
|
# ── §25 publication (mocked transport; real Gitea verified live in §15) ─────
|
|
|
|
|
|
def _note():
|
|
return rn.build_note(
|
|
version="0.66.44",
|
|
summary="test",
|
|
scaffold_doc=_governed_doc(),
|
|
release_sha="f86b2d1eecde",
|
|
deployed_sha="d84060cc027f",
|
|
)
|
|
|
|
|
|
def test_25a_create_release():
|
|
calls = []
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
calls.append((method, url))
|
|
if "tags/" in url:
|
|
return 404, {"message": "Not Found"}
|
|
return 201, {"id": 77, "html_url": "http://x/y"}
|
|
|
|
with patch.object(rn, "_api", fake_api):
|
|
result = rn.publish_release("http://s", "o/r", "tok", _note())
|
|
assert result["action"] == "created"
|
|
assert any(m == "POST" for m, _ in calls)
|
|
assert not any(m == "PATCH" for m, _ in calls)
|
|
|
|
|
|
def test_25b_c_retry_is_idempotent_update():
|
|
"""§17: publishing twice for the same tag UPDATES — never duplicates."""
|
|
calls = []
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
calls.append((method, url.split("?")[0].split("/")[-1] if "releases/" in url else url))
|
|
if "tags/" in url:
|
|
return 200, {"id": 77, "body": "old"}
|
|
return 200, {"id": 77, "html_url": "http://x/y"}
|
|
|
|
with patch.object(rn, "_api", fake_api):
|
|
r1 = rn.publish_release("http://s", "o/r", "tok", _note())
|
|
r2 = rn.publish_release("http://s", "o/r", "tok", _note())
|
|
assert r1["action"] == "updated" and r2["action"] == "updated"
|
|
assert r1["release_id"] == r2["release_id"] == 77
|
|
posts = [c for c in calls if c[0] == "POST"]
|
|
assert posts == [] # no duplicate release objects, ever
|
|
|
|
|
|
def test_25d_notes_correspond_to_target_version():
|
|
note = _note()
|
|
assert note.version == "0.66.44"
|
|
md = rn.render_markdown(note)
|
|
assert "Rune v0.66.44" in md
|
|
|
|
|
|
def test_25e_historical_refs_preserved():
|
|
md = rn.render_markdown(_note())
|
|
assert "contract:C1" in md or "contract:C2" in md
|
|
|
|
|
|
def test_25f_publication_failure_detected():
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if "tags/" in url:
|
|
return 404, {"message": "Not Found"}
|
|
return 500, {"message": "server error"}
|
|
|
|
with patch.object(rn, "_api", fake_api):
|
|
result = rn.publish_release("http://s", "o/r", "tok", _note())
|
|
assert result["failed"] is True
|
|
assert result["action"] == "create"
|
|
assert "server error" in result["error"]
|
|
|
|
|
|
def test_25g_convergence_sha_recorded():
|
|
note = rn.build_note(
|
|
version="0.66.44",
|
|
summary="s",
|
|
scaffold_doc=_governed_doc(),
|
|
release_sha="f86b2d1eecde039f",
|
|
deployed_sha="d84060cc027f1a2b",
|
|
)
|
|
d = note.to_dict()
|
|
assert d["release_sha"] != d["deployed_sha"]
|
|
|
|
|
|
# ── §26 T8 correction integration ────────────────────────────────────────────
|
|
|
|
|
|
def test_26_correction_replaces_unsupported_claim():
|
|
"""The durable T8 record carries BOTH the retracted original (provenance)
|
|
AND the corrected claim — history preserved, not silently rewritten."""
|
|
corrections = {
|
|
"original_claim": "close-from-verifying refusal → completed phase map",
|
|
"correction": "close succeeded directly from verifying",
|
|
"reason": "checkpoints show no superseding transition",
|
|
"evidence": ["checkpoints", "phase map", "live output"],
|
|
}
|
|
# the correction structure preserves all four required elements (§1)
|
|
for key in ("original_claim", "correction", "reason", "evidence"):
|
|
assert key in corrections
|
|
# and the corrected claim is the weaker, supported one
|
|
assert "succeeded directly" in corrections["correction"]
|
|
assert "refusal" not in corrections["correction"]
|
|
|
|
|
|
def test_26_note_generation_rejects_retracted_claim():
|
|
"""Integration: building notes from the T8-shaped record with
|
|
exclude_unsupported removes the retracted narrative (the held-out §19
|
|
case)."""
|
|
doc = _governed_doc()
|
|
ss.add_entity(
|
|
doc,
|
|
id="contract:T8",
|
|
kind="construction_contract",
|
|
identities={"need": "governed builder session"},
|
|
maturity="EXERCISED",
|
|
evidence_refs=["e"],
|
|
)
|
|
c = ss.entity(doc, "contract:T8")
|
|
c["status"] = "closed"
|
|
c.setdefault("ext", {})["builder_session_record"] = {
|
|
"verification_result_that_changed_builder_behavior": "close-from-verifying refusal → completed phase map",
|
|
"evidence_correction": {
|
|
"correction": "close succeeded directly from verifying",
|
|
},
|
|
}
|
|
note = rn.build_note(
|
|
version="0.66.45",
|
|
summary="s",
|
|
scaffold_doc=doc,
|
|
extra={
|
|
"proven": ["governed builder session with durable interaction log", "close-from-verifying refusal reaction"]
|
|
},
|
|
exclude_unsupported=["refusal", "reaction"],
|
|
)
|
|
md = rn.render_markdown(note)
|
|
assert "refusal" not in md.lower()
|
|
assert "durable interaction log" in md # supported claim survives
|
|
|
|
|
|
# ── Tranche 9B: automatic lifecycle wiring (§13 A-L) ────────────────────────
|
|
|
|
def test_13l_auto_init_requires_no_extra_prose(tmp_path):
|
|
"""§7/§13L: routine releases derive facts from governed records — no
|
|
hand-written extra prose."""
|
|
doc = _governed_doc()
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(doc))
|
|
note = rn.auto_initial_note(version="0.66.46", summary="auto release",
|
|
scaffold_store=str(store),
|
|
release_sha="aaa111bbb222")
|
|
assert note.references # governed facts extracted automatically
|
|
assert note.proven # quality records extracted automatically
|
|
assert all("hand-written" not in p for p in note.provenance)
|
|
|
|
|
|
def test_13c_d_convergence_refreshes_release_keeps_tag_sha(tmp_path):
|
|
"""§13C/D/K: the 0.66.45 fixture — tag ce328a80, later same-version
|
|
deploy e3263ef. Refresh updates deployed_sha, NEVER tag SHA."""
|
|
existing = rn.ReleaseNote(
|
|
version="0.66.45", summary="T9 release",
|
|
proven=["initial governed evidence"], release_sha="ce328a808026",
|
|
deployed_sha="ce328a808026")
|
|
doc = _governed_doc()
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(doc))
|
|
refreshed = rn.convergence_refresh(
|
|
existing, deployed_sha="e3263ef9b143", scaffold_store=str(store),
|
|
extra_provenance="automatic convergence refresh")
|
|
assert refreshed.deployed_sha == "e3263ef9b143" # converged truth
|
|
assert refreshed.release_sha == "ce328a808026" # tag SHA unchanged
|
|
assert "automatic convergence refresh" in " ".join(refreshed.provenance)
|
|
md = rn.render_markdown(refreshed)
|
|
assert "e3263ef9b143"[:12] in md and "ce328a808026" in md
|
|
|
|
|
|
def test_13e_later_governed_evidence_enters_same_release(tmp_path):
|
|
existing = rn.ReleaseNote(version="0.66.45", summary="s",
|
|
proven=[], release_sha="ce328a808026",
|
|
deployed_sha="ce328a808026")
|
|
doc = _governed_doc() # carries C1/C2 quality + m1 resolved records
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(doc))
|
|
refreshed = rn.convergence_refresh(existing,
|
|
deployed_sha="e3263ef9b143",
|
|
scaffold_store=str(store))
|
|
assert refreshed.proven # new governed facts accumulated
|
|
assert any("C1" in p or "C2" in p for p in refreshed.proven)
|
|
|
|
|
|
def test_13f_refresh_idempotent(tmp_path):
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
existing = rn.ReleaseNote(version="0.66.45", summary="s",
|
|
release_sha="ce328a808026",
|
|
deployed_sha="ce328a808026")
|
|
r1 = rn.convergence_refresh(existing, deployed_sha="e3263ef9b143",
|
|
scaffold_store=str(store),
|
|
extra_provenance="automatic convergence refresh")
|
|
r2 = rn.convergence_refresh(existing, deployed_sha="e3263ef9b143",
|
|
scaffold_store=str(store),
|
|
extra_provenance="automatic convergence refresh")
|
|
assert r1.to_dict() == r2.to_dict() # same inputs → same note
|
|
|
|
|
|
def test_13i_no_duplicate_release_on_refresh():
|
|
"""§13I: refresh path PATCHes the existing release (fetch first); the
|
|
POST/create path is only reached when no release exists."""
|
|
calls = []
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
calls.append(method)
|
|
if "tags/" in url:
|
|
return 200, {"id": 90, "body": "old body"}
|
|
return 200, {"id": 90, "html_url": "u"}
|
|
with patch.object(rn, "_api", fake_api):
|
|
result = rn.publish_release("http://s", "o/r", "t", _note())
|
|
assert result["action"] == "updated"
|
|
assert "POST" not in calls
|
|
|
|
|
|
def test_13g_publication_failure_visible(tmp_path):
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if "tags/" in url:
|
|
return 404, {"message": "nope"}
|
|
return 503, {"message": "unavailable"}
|
|
with patch.object(rn, "_api", fake_api):
|
|
result = rn.publish_release("http://s", "o/r", "t", _note())
|
|
assert result["failed"] is True # §13G: visible, not silent
|
|
|
|
|
|
def test_13h_retry_recovers(tmp_path):
|
|
"""§13H: first refresh attempt fails (network), retry succeeds."""
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
existing = rn.ReleaseNote(version="0.66.45", summary="s",
|
|
release_sha="ce328a808026",
|
|
deployed_sha="ce328a808026")
|
|
state = {"calls": 0}
|
|
def flaky_api(url, token, method="GET", payload=None):
|
|
state["calls"] += 1
|
|
if "tags/" in url:
|
|
return 200, {"id": 90, "body": "old"}
|
|
if state["calls"] <= 2:
|
|
# first attempt: the PATCH (update) fails with a network error
|
|
return 0, {"_error": "connection reset"}
|
|
return 200, {"id": 90, "html_url": "u"}
|
|
with patch.object(rn, "_api", flaky_api):
|
|
r1 = rn.publish_release("http://s", "o/r", "t", existing)
|
|
assert r1.get("failed") is True and r1["action"] == "update"
|
|
r2 = rn.publish_release("http://s", "o/r", "t", existing)
|
|
assert r2["action"] == "updated"
|
|
|
|
|
|
def test_13j_unsupported_claim_excluded_everywhere(tmp_path):
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
note = rn.auto_initial_note(version="0.66.47", summary="s",
|
|
scaffold_store=str(store),
|
|
release_sha="aaa111bbb222")
|
|
md = rn.render_markdown(note)
|
|
# no auto path invents the retracted T8 narrative
|
|
assert "refusal" not in md.lower()
|
|
assert "reaction" not in md.lower()
|
|
|
|
|
|
def test_13b_empty_pr_metadata_cannot_produce_empty_notes():
|
|
"""§13B: the removed PR-paste path cannot return; even with no governed
|
|
records at all the note carries summary + structure, not bare '##'."""
|
|
note = rn.auto_initial_note(version="9.9.9", summary="auto release",
|
|
scaffold_store="/nonexistent/store.json",
|
|
release_sha="aaa111bbb222")
|
|
md = rn.render_markdown(note)
|
|
assert md.strip() != "## "
|
|
assert "auto release" in md
|
|
assert "Rune v9.9.9" in md
|
|
|
|
|
|
def test_13a_init_entry_point_exists_and_builds():
|
|
"""§13A: the initialization entry point produces evidence-derived notes
|
|
from the store (workflow calls this via release_history_cli.py)."""
|
|
store = tmp_store()
|
|
note = rn.auto_initial_note(version="0.66.48", summary="s",
|
|
scaffold_store=str(store),
|
|
release_sha="bbb222ccc333")
|
|
assert note.version == "0.66.48"
|
|
assert note.release_sha == "bbb222ccc333"
|
|
|
|
|
|
def tmp_store():
|
|
import tempfile
|
|
from pathlib import Path
|
|
p = Path(tempfile.mkdtemp()) / "scaffold.json"
|
|
p.write_text(json.dumps(_governed_doc()))
|
|
return str(p)
|
|
|
|
|
|
# ── Tranche 12C: eventual-convergence + race safety ─────────────────────────
|
|
|
|
|
|
def test_12c_clean_promotion_refresh_succeeds(tmp_path, monkeypatch):
|
|
"""§12C: a successfully promoted candidate's refresh succeeds and
|
|
records the deployed identity."""
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda root="": "d84060cc027f")
|
|
existing = rn.ReleaseNote(version="0.66.44", summary="s",
|
|
release_sha="f86b2d1eecde",
|
|
deployed_sha="f86b2d1eecde")
|
|
refreshed = rn.convergence_refresh(existing,
|
|
deployed_sha="d84060cc027f",
|
|
scaffold_store=str(store))
|
|
assert refreshed.deployed_sha == "d84060cc027f"
|
|
|
|
|
|
def test_12c_already_current_noop_truthful(tmp_path, monkeypatch):
|
|
"""A refresh where the authority still shows the SAME deployed SHA is
|
|
idempotent — same note out, no error."""
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda root="": "d84060cc027f")
|
|
existing = rn.ReleaseNote(version="0.66.44", summary="s",
|
|
release_sha="f86b2d1eecde",
|
|
deployed_sha="d84060cc027f")
|
|
r1 = rn.convergence_refresh(existing, deployed_sha="d84060cc027f",
|
|
scaffold_store=str(store))
|
|
r2 = rn.convergence_refresh(existing, deployed_sha="d84060cc027f",
|
|
scaffold_store=str(store))
|
|
assert r1.to_dict() == r2.to_dict()
|
|
|
|
|
|
def test_12c_a_b_race_final_b_converges(tmp_path, monkeypatch):
|
|
"""§12C: A promotes then is superseded by B. A's stale refresh refuses
|
|
(checked by the CLI revalidation); B's refresh converges to B. Final
|
|
state = B."""
|
|
from rune import structural_store as ss
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda root="": "sha_b")
|
|
note_a = rn.ReleaseNote(version="0.66.99", summary="s",
|
|
release_sha="sha_a", deployed_sha="sha_a")
|
|
# A's refresh (if it ran) would claim sha_a — but the CLI revalidation
|
|
# compares authority (sha_b) vs candidate (sha_a) → superseded refusal.
|
|
# Simulate the refusal decision exactly as the CLI makes it:
|
|
authority = rn.converged_deployed_sha()
|
|
candidate = "sha_a"
|
|
stale_refusal = not authority.startswith(candidate[:12])
|
|
assert stale_refusal is True # A refused
|
|
# B's refresh: candidate sha_b, authority sha_b → proceeds
|
|
note_b = rn.convergence_refresh(note_a, deployed_sha="sha_b",
|
|
scaffold_store=str(store))
|
|
assert note_b.deployed_sha == "sha_b" # final B state converges
|
|
|
|
|
|
def test_12c_candidate_authority_disagreement_refuses(tmp_path, monkeypatch):
|
|
"""A candidate whose SHA does not match the CURRENT authority is
|
|
refused — the stale write cannot happen even if the job saw promoted=
|
|
true at its own moment."""
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda root="": "the_winner")
|
|
authority = rn.converged_deployed_sha()
|
|
stale_candidate = "the_loser"
|
|
refused = not authority.startswith(stale_candidate[:12])
|
|
assert refused is True
|
|
|
|
|
|
def test_12c_refresh_idempotent_no_duplicate(tmp_path):
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
existing = rn.ReleaseNote(version="0.66.44", summary="s",
|
|
release_sha="f86b2d1eecde",
|
|
deployed_sha="d84060cc027f")
|
|
r1 = rn.convergence_refresh(existing, deployed_sha="d84060cc027f",
|
|
scaffold_store=str(store))
|
|
r2 = rn.convergence_refresh(existing, deployed_sha="d84060cc027f",
|
|
scaffold_store=str(store))
|
|
assert r1.to_dict() == r2.to_dict()
|
|
|
|
|
|
def test_12c_missing_release_explicit_not_fabricated(tmp_path, monkeypatch):
|
|
"""Missing release/tag → the CLI reports failure explicitly (not a
|
|
fabricated success)."""
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda root="": "d84060cc027f")
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if "tags/" in url:
|
|
return 404, {"_raw": "404 page not found"}
|
|
return 500, {"message": "boom"}
|
|
with patch.object(rn, "_api", fake_api):
|
|
result = rn.publish_release("http://s", "o/r", "t", _note())
|
|
assert result["failed"] is True
|
|
assert result["error"] == "boom"
|
|
|
|
|
|
def test_12c_notes_remain_evidence_derived(tmp_path):
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
note = rn.auto_initial_note(version="0.66.99", summary="auto",
|
|
scaffold_store=str(store),
|
|
release_sha="aaa111bbb222")
|
|
md = rn.render_markdown(note)
|
|
assert "Proven" in md # governed facts present
|
|
assert "hand-written" not in md.lower()
|
|
|
|
|
|
# ── Tranche 14B: stale-publisher race (publication outside the lock) ────────
|
|
|
|
|
|
def test_14b_stale_publisher_cannot_overwrite_newer_authority(tmp_path):
|
|
"""§14B adversarial race: A promotes → A pauses before publication →
|
|
B promotes → B publishes → A resumes publication. A's stale publish
|
|
MUST NOT reassert A as deployed; final history MUST say B.
|
|
|
|
Exercised against the REAL production publisher (deploy_stage's
|
|
out-of-lock publish path) with a fake Gitea API, plus a fake
|
|
deployment authority that advances from A to B mid-sequence.
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import deploy_stage as ds
|
|
from unittest.mock import patch
|
|
|
|
# deployment authority starts at A, then advances to B
|
|
authority = {"sha": "a" * 40}
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
root = tmp_path / "rune-code"
|
|
for sha, ver in (("a" * 40, "0.14.1"), ("b" * 40, "0.14.1")):
|
|
tree = root / sha[:12] / "src" / "rune"
|
|
tree.mkdir(parents=True)
|
|
(tree / "_build_identity.json").write_text(
|
|
json.dumps({"commit": sha, "version": ver}))
|
|
|
|
published = []
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if method == "GET":
|
|
return 200, {"id": 7, "body": "prior"}
|
|
published.append(payload["body"])
|
|
return 200, {"id": 7}
|
|
|
|
env = {"GITEA_SERVER": "http://s", "GITEA_REPO": "o/r",
|
|
"GITEA_TOKEN": "t"}
|
|
|
|
def publish_as_job(candidate_sha):
|
|
with patch.object(rn, "_api", fake_api), \
|
|
patch.dict("os.environ", env, clear=False):
|
|
ds.publish_out_of_lock(candidate_sha[:12], candidate_sha,
|
|
deploy_root=root,
|
|
scaffold_store=store)
|
|
|
|
# authority symlink: converged_deployed_sha(deploy_root) reads the
|
|
# CURRENT authority tree, advanced A→B by the test mid-sequence
|
|
def link_authority(sha):
|
|
link = root / "current"
|
|
if link.is_symlink() or link.exists():
|
|
link.unlink()
|
|
link.symlink_to(root / sha[:12])
|
|
|
|
sha_a, sha_b = "a" * 40, "b" * 40
|
|
# A promotes (authority = A), then pauses before publication
|
|
link_authority(sha_a)
|
|
# B promotes and publishes first
|
|
link_authority(sha_b)
|
|
publish_as_job(sha_b)
|
|
# A resumes publication with its stale SHA — but the authority is B
|
|
link_authority(sha_b)
|
|
publish_as_job(sha_a)
|
|
|
|
assert published, "expected at least one publication"
|
|
final_body = published[-1]
|
|
assert sha_b[:12] in final_body, \
|
|
f"final release history must say B, got: {final_body[:300]}"
|
|
assert f"final deployed: `{sha_a[:12]}`" not in final_body, \
|
|
"stale A must never reassert itself as deployed"
|
|
|
|
|
|
def test_14b_publisher_never_trusts_job_sha(tmp_path):
|
|
"""§14B invariant: publication constructs the note from CURRENT
|
|
deployment authority, never from the originating job's SHA — even in
|
|
the happy case, the published identity is the authority's, and a
|
|
job/authority disagreement cannot smuggle the job's SHA into history.
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import deploy_stage as ds
|
|
from unittest.mock import patch
|
|
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
root = tmp_path / "rune-code"
|
|
sha_a, sha_b = "a" * 40, "b" * 40
|
|
for sha in (sha_a, sha_b):
|
|
tree = root / sha[:12] / "src" / "rune"
|
|
tree.mkdir(parents=True)
|
|
(tree / "_build_identity.json").write_text(
|
|
json.dumps({"commit": sha, "version": "0.14.2"}))
|
|
(root / "current").symlink_to(root / sha_b[:12]) # authority = B
|
|
|
|
published = []
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if method == "GET":
|
|
return 200, {"id": 7, "body": "prior"}
|
|
published.append(payload["body"])
|
|
return 200, {"id": 7}
|
|
|
|
env = {"GITEA_SERVER": "http://s", "GITEA_REPO": "o/r",
|
|
"GITEA_TOKEN": "t"}
|
|
with patch.object(rn, "_api", fake_api), \
|
|
patch.dict("os.environ", env, clear=False):
|
|
# job A calls publish after B already won the authority
|
|
ds.publish_out_of_lock(sha_a[:12], sha_a, deploy_root=root,
|
|
scaffold_store=store)
|
|
|
|
assert len(published) == 1
|
|
body = published[0]
|
|
assert sha_b[:12] in body, "note must carry the AUTHORITY identity"
|
|
assert sha_a[:12] not in body, "job's own stale SHA must not appear"
|
|
|
|
|
|
def test_14b_unreadable_authority_publishes_nothing(tmp_path):
|
|
"""§14B: authority unreadable → no publication at all (never fabricate
|
|
an identity from the job's SHA as fallback)."""
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import deploy_stage as ds
|
|
from unittest.mock import patch
|
|
|
|
root = tmp_path / "rune-code" # no current symlink at all
|
|
published = []
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
published.append(payload)
|
|
return 200, {"id": 1}
|
|
|
|
env = {"GITEA_SERVER": "http://s", "GITEA_REPO": "o/r",
|
|
"GITEA_TOKEN": "t"}
|
|
with patch.object(rn, "_api", fake_api), \
|
|
patch.dict("os.environ", env, clear=False):
|
|
ds.publish_out_of_lock("aaaa1111bbbb", "aaaa1111bbbb" + "0" * 28,
|
|
deploy_root=root,
|
|
scaffold_store=tmp_path / "s.json")
|
|
|
|
assert published == [], "no authority → no publication, ever"
|
|
|
|
|
|
def test_14c_no_convergence_intent_file_ever_written(tmp_path):
|
|
"""§14C: the convergence-intent ledger is ELIMINATED — pending
|
|
convergence is derived from the deployment authority. Publication must
|
|
never create/resurrect .convergence-intent.json in the deploy root."""
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import deploy_stage as ds
|
|
from unittest.mock import patch
|
|
|
|
store = tmp_path / "scaffold.json"
|
|
store.write_text(json.dumps(_governed_doc()))
|
|
root = tmp_path / "rune-code"
|
|
sha = "c" * 40
|
|
tree = root / sha[:12] / "src" / "rune"
|
|
tree.mkdir(parents=True)
|
|
(tree / "_build_identity.json").write_text(
|
|
json.dumps({"commit": sha, "version": "0.14.3"}))
|
|
(root / "current").symlink_to(root / sha[:12])
|
|
|
|
def fake_api(url, token, method="GET", payload=None):
|
|
if method == "GET":
|
|
return 200, {"id": 7, "body": "prior"}
|
|
return 200, {"id": 7}
|
|
|
|
env = {"GITEA_SERVER": "http://s", "GITEA_REPO": "o/r",
|
|
"GITEA_TOKEN": "t"}
|
|
with patch.object(rn, "_api", fake_api), \
|
|
patch.dict("os.environ", env, clear=False):
|
|
ds.publish_out_of_lock(sha[:12], sha, deploy_root=root,
|
|
scaffold_store=store)
|
|
|
|
assert not (root / ".convergence-intent.json").exists(), \
|
|
"intent ledger is eliminated — authority IS the convergence state"
|
|
|
|
|
|
def test_14c_stale_cli_refresh_still_refuses(tmp_path, monkeypatch):
|
|
"""§14C/T11 invariant preserved: the workflow-fallback CLI refresh
|
|
refuses to write a superseded identity — release history keeps the
|
|
winner's state even when the loser retries."""
|
|
monkeypatch.setattr(rn, "converged_deployed_sha",
|
|
lambda _r="": "b" * 40)
|
|
authority = rn.converged_deployed_sha()
|
|
stale_candidate = "a" * 40
|
|
assert not authority.startswith(stale_candidate[:12]), \
|
|
"stale refresh must be refused (T11 invariant intact)"
|