rein-aharness/tests/test_owner_renew.py

"""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"] == []


def test_renewal_to_observed_usage_with_run_limit(prepared, tmp_path):
    path, config, _, _ = prepared
    envelope = {**ENVELOPE, "settlement": "observed-usage", "max_runs": 3,
                "total_eur": "10", "daily_eur": "10"}
    paths = _paths(prepared, tmp_path, envelope)
    apply(plan(*paths))
    written = json.loads(paths[1].read_text())
    assert written["settlement"] == "observed-usage" and written["max_runs"] == 3
    policy = SpendPolicy.load(paths[1])
    assert json.loads(paths[0].read_text())["spend_policy_sha256"] == policy.sha256
    prepare(path, config)


def test_default_renewal_writes_no_optional_terms(prepared, tmp_path):
    paths = _paths(prepared, tmp_path)
    apply(plan(*paths))
    written = json.loads(paths[1].read_text())
    assert "settlement" not in written and "max_runs" not in written