rune/tests/test_rollback_generation.py

"""test_rollback_generation.py — T21.4 §21.4F: governed rollback API tests.

Refusals must occur BEFORE any mutation of current/last-good. The live
B→A→B proof runs against the real deploy root; these tests prove the
contract on synthetic deploy roots (same code path, no production risk).
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

_SCRIPTS = str(Path(__file__).resolve().parent.parent / "scripts")
if _SCRIPTS not in sys.path:
    sys.path.insert(0, _SCRIPTS)

import rollback_generation as rg


def make_generation(root: Path, name: str, commit: str) -> Path:
    gen = root / name
    pkg = gen / "src" / "rune"
    pkg.mkdir(parents=True)
    (pkg / "__init__.py").write_text('__version__ = "test"\n')
    (gen / "src" / "rune" / "_build_identity.json").write_text(
        json.dumps({"commit": commit}))
    return gen


@pytest.fixture
def deploy_root(tmp_path, monkeypatch):
    root = tmp_path / "rune-code"
    root.mkdir()
    gen_a = make_generation(root, "gen-a", "a" * 40)
    gen_b = make_generation(root, "gen-b", "b" * 40)
    (root / "current").symlink_to(gen_a)
    (root / "last-good").symlink_to(gen_b)
    history = tmp_path / "history.jsonl"
    monkeypatch.setattr(rg.deploy_promotion, "ancestry_of",
                        lambda t, d, r: "ancestor")
    return root, history, gen_a, gen_b


def readlink(p: Path) -> str:
    return str(p.readlink())


# ── refusals: BEFORE mutation ──────────────────────────────────────────────

def test_refuse_no_reason(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    outcome, detail = rg.rollback_generation("gen-b", "a" * 40, "  ",
                                             deploy_root=root,
                                             history_path=history)
    assert outcome == rg.REFUSED_NO_REASON
    assert readlink(root / "current") == str(gen_a)


def test_refuse_expected_mismatch(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    outcome, detail = rg.rollback_generation("gen-b", "f" * 40, "stale caller",
                                             deploy_root=root,
                                             history_path=history)
    assert outcome == rg.REFUSED_EXPECTED_MISMATCH
    assert readlink(root / "current") == str(gen_a)
    assert readlink(root / "last-good") == str(gen_b)


def test_refuse_target_missing(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    outcome, _ = rg.rollback_generation("gen-ghost", "a" * 40, "ghost target",
                                        deploy_root=root, history_path=history)
    assert outcome == rg.REFUSED_TARGET_MISSING
    assert readlink(root / "current") == str(gen_a)


def test_refuse_target_unreadable(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    bad = root / "gen-bad"
    bad.mkdir()
    outcome, _ = rg.rollback_generation("gen-bad", "a" * 40, "no identity",
                                        deploy_root=root, history_path=history)
    assert outcome == rg.REFUSED_TARGET_UNREADABLE
    assert readlink(root / "current") == str(gen_a)


def test_refuse_not_eligible_divergent(deploy_root, monkeypatch):
    root, history, gen_a, gen_b = deploy_root
    monkeypatch.setattr(rg.deploy_promotion, "ancestry_of",
                        lambda t, d, r: "divergent")
    outcome, detail = rg.rollback_generation("gen-b", "a" * 40, "divergent",
                                             deploy_root=root,
                                             history_path=history)
    assert outcome == rg.REFUSED_NOT_ELIGIBLE
    assert "divergent" in detail["error"]
    assert readlink(root / "current") == str(gen_a)


# ── the operation itself ───────────────────────────────────────────────────

def test_rollback_success_switch(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    outcome, detail = rg.rollback_generation("gen-b", "a" * 40, "regression",
                                             deploy_root=root,
                                             history_path=history)
    assert outcome == rg.ROLLED_BACK, detail
    assert readlink(root / "current") == str(gen_b)
    assert readlink(root / "last-good") == str(gen_a)
    events = [json.loads(l) for l in
              history.read_text(encoding="utf-8").splitlines()]
    assert events[0]["outcome"] == rg.ROLLED_BACK
    assert events[0]["reason"] == "regression"


def test_noop_when_already_current(deploy_root):
    root, history, gen_a, gen_b = deploy_root
    outcome, _ = rg.rollback_generation("gen-a", "a" * 40, "same",
                                        deploy_root=root, history_path=history)
    assert outcome == rg.NOOP


def test_health_failure_reverts(deploy_root, monkeypatch):
    root, history, gen_a, gen_b = deploy_root
    monkeypatch.setattr(rg, "_health", lambda r, g, timeout=60:
                        (False, "import smoke failed: boom"))
    outcome, detail = rg.rollback_generation("gen-b", "a" * 40, "bad health",
                                             deploy_root=root,
                                             history_path=history)
    assert outcome == rg.ROLLED_BACK_REVERTED
    assert readlink(root / "current") == str(gen_a)
    events = [json.loads(l) for l in
              history.read_text(encoding="utf-8").splitlines()]
    assert events[0]["outcome"] == rg.ROLLED_BACK_REVERTED


def test_import_smoke_real(tmp_path):
    """Health check runs a REAL import against a generation tree."""
    root = tmp_path / "rune-code"
    root.mkdir()
    make_generation(root, "gen-x", "x" * 40)
    ok, detail = rg._health(root, "gen-x")
    assert ok, detail


def test_lock_shared_with_promotion(deploy_root):
    """Rollback serialises on the SAME lock file promotion uses."""
    root, history, gen_a, gen_b = deploy_root
    lock = root / ".promotion.lock"
    import fcntl
    with open(lock, "w") as fd:
        fcntl.flock(fd, fcntl.LOCK_EX)
        import threading
        results = {}

        def try_rollback():
            results["r"] = rg.rollback_generation(
                "gen-b", "a" * 40, "contended", deploy_root=root,
                history_path=history)

        t = threading.Thread(target=try_rollback)
        t.start()
        import time
        time.sleep(0.3)
        assert "r" not in results  # blocked on the shared lock
        fcntl.flock(fd, fcntl.LOCK_UN)
    t.join(timeout=10)
    assert results["r"][0] == rg.ROLLED_BACK