Add one-command owner envelope renewal (REINAH-WP-0008-T03)
Some checks failed
Governed runtime contract / contract (push) Failing after 12s
Some checks failed
Governed runtime contract / contract (push) Failing after 12s
owner renew validates a settled ledger, builds the new spend policy from an envelope file of changed fields only, retires the old ledger by rename, initializes a fresh one at the same path and rewrites the policy and owner.json pins. Dry run by default; prints the Secrets Engine file pins. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 269692@bnt-lap001 Assistant-Session: 8077e68a-0985-4b41-898c-2dc602b0d6be
This commit is contained in:
parent
caaea1e90e
commit
b48c715e9a
6 changed files with 361 additions and 4 deletions
118
tests/test_owner_renew.py
Normal file
118
tests/test_owner_renew.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
"""One-command owner renewal (REINAH-WP-0008-T03)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from rein_aharness.cli import main
|
||||
from rein_aharness.owner_bootstrap import prepare
|
||||
from rein_aharness.owner_renew import RenewalRefused, apply, plan
|
||||
from rein_aharness.spend_admission import SpendLedger, SpendPolicy
|
||||
from test_owner_bootstrap import prepared as prepared
|
||||
from test_spend_admission import NOW, ledger as ledger, run
|
||||
|
||||
ENVELOPE = {
|
||||
"envelope_id": "renewed-2026-10-13",
|
||||
"authority_ref": "fixture:owner-renewal",
|
||||
"total_eur": "30",
|
||||
}
|
||||
|
||||
|
||||
def _paths(prepared, tmp_path, envelope=ENVELOPE):
|
||||
path, config, _, runtime = prepared
|
||||
envelope_path = tmp_path / "envelope.json"
|
||||
envelope_path.write_text(json.dumps(envelope))
|
||||
return (path, Path(config.spend_policy_path), Path(config.spend_ledger_path),
|
||||
runtime, envelope_path)
|
||||
|
||||
|
||||
def _digests(*paths):
|
||||
return {p: hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}
|
||||
|
||||
|
||||
def test_dry_run_changes_nothing(prepared, tmp_path, capsys):
|
||||
paths = _paths(prepared, tmp_path)
|
||||
before = _digests(*paths[:3])
|
||||
argv = ["owner", "renew", "--owner-config", str(paths[0]), "--policy", str(paths[1]),
|
||||
"--ledger", str(paths[2]), "--runtime-path", str(paths[3]),
|
||||
"--envelope", str(paths[4])]
|
||||
assert main(argv) == 0
|
||||
out = json.loads(capsys.readouterr().out)
|
||||
assert out["dry_run"] is True and "catalog_file_pins" not in out
|
||||
assert out["envelope"] == {"from": "test-only", "to": "renewed-2026-10-13"}
|
||||
assert out["limits"]["total_eur"] == "30"
|
||||
assert _digests(*paths[:3]) == before
|
||||
|
||||
|
||||
def test_apply_renews_and_owner_check_passes(prepared, tmp_path):
|
||||
path, config, _, runtime = prepared
|
||||
paths = _paths(prepared, tmp_path)
|
||||
old_ledger = SpendLedger(paths[2], SpendPolicy.load(paths[1]))
|
||||
result = apply(plan(*paths))
|
||||
|
||||
new_policy = SpendPolicy.load(paths[1])
|
||||
owner = json.loads(paths[0].read_text())
|
||||
assert new_policy.envelope_id == "renewed-2026-10-13" and new_policy.total_eur == "30"
|
||||
assert owner["spend_policy_sha256"] == new_policy.sha256
|
||||
assert owner["authority_ref"] == "fixture:owner-renewal"
|
||||
assert owner["runtime"]["path"] == str(runtime)
|
||||
# The real owner check accepts the renewed files at the unchanged paths.
|
||||
prepare(path, config)
|
||||
# Fresh ledger for the new envelope; the old one is retired, not reset.
|
||||
assert SpendLedger(paths[2], new_policy).status()["reservations"] == []
|
||||
retired = paths[2].with_name(f"{paths[2].name}.retired-test-only")
|
||||
assert retired.exists()
|
||||
assert paths[0].with_name("owner.json.test-only.bak").exists()
|
||||
assert result["catalog_file_pins"] == {
|
||||
str(p): d for p, d in _digests(paths[0], paths[1]).items()}
|
||||
assert oct(paths[1].stat().st_mode & 0o777) == "0o600"
|
||||
|
||||
|
||||
def test_held_reservation_refuses(prepared, tmp_path):
|
||||
paths = _paths(prepared, tmp_path)
|
||||
db = sqlite3.connect(paths[2])
|
||||
db.execute("INSERT INTO reservations (run_id, definition_id, idempotency_key, attempt, state, "
|
||||
"liability, start_day) VALUES ('r', 'd', 'k', 1, 'held', 1, '2026-10-13')")
|
||||
db.commit(); db.close()
|
||||
with pytest.raises(RenewalRefused, match="held reservation"):
|
||||
plan(*paths)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("envelope,match", [
|
||||
({"total_eur": "30"}, "new envelope_id"),
|
||||
({**ENVELOPE, "envelope_id": "test-only"}, "new envelope_id"),
|
||||
({**ENVELOPE, "worker_id": "other"}, "envelope may only set"),
|
||||
({**ENVELOPE, "total_eur": "1"}, "new spend policy invalid"),
|
||||
])
|
||||
def test_bad_envelope_refuses(prepared, tmp_path, envelope, match):
|
||||
with pytest.raises(RenewalRefused, match=match):
|
||||
plan(*_paths(prepared, tmp_path, envelope))
|
||||
|
||||
|
||||
def test_second_apply_from_same_envelope_refuses(prepared, tmp_path):
|
||||
paths = _paths(prepared, tmp_path)
|
||||
renewal = plan(*paths)
|
||||
apply(renewal)
|
||||
with pytest.raises(RenewalRefused):
|
||||
apply(renewal)
|
||||
|
||||
|
||||
def test_charged_reservation_renews_and_survives_in_retired_ledger(prepared, tmp_path):
|
||||
# The railiance01 state after the 2026-10-13 reconciliation.
|
||||
paths = _paths(prepared, tmp_path)
|
||||
db = sqlite3.connect(paths[2])
|
||||
db.execute("INSERT INTO reservations (run_id, definition_id, idempotency_key, attempt, state, "
|
||||
"liability, start_day, end_day, observed_usd, receipt) VALUES ('6efa9436', 'd', 'k', "
|
||||
"1, 'charged', 10000000, '2026-09-27', '2026-10-13', '0.05', 'owner-estimate')")
|
||||
db.commit(); db.close()
|
||||
apply(plan(*paths))
|
||||
retired = sqlite3.connect(paths[2].with_name(f"{paths[2].name}.retired-test-only"))
|
||||
assert retired.execute("SELECT state, liability FROM reservations").fetchall() == [
|
||||
("charged", 10000000)]
|
||||
retired.close()
|
||||
assert SpendLedger(paths[2], SpendPolicy.load(paths[1])).status()["reservations"] == []
|
||||
Loading…
Add table
Add a link
Reference in a new issue