import copy
import subprocess
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
import yaml

from rtut import pack as P, run as R1, run2 as R

ROOT = Path(__file__).resolve().parent.parent
SRC = yaml.safe_load((ROOT / "runbooks/showcase-lighthouse/pack.v0.2.yaml").read_text())
PHRASE = "I understand this cannot be undone"


def mkpack(tmp_path, mutate=None):
    d = copy.deepcopy(SRC)
    if mutate:
        mutate(d)
    f = tmp_path / "runbooks" / "lh" / "pack.yaml"
    f.parent.mkdir(parents=True, exist_ok=True)
    f.write_text(yaml.safe_dump(d, sort_keys=False))
    pk = P.load(f)
    assert pk.ok, pk.errors
    return pk


_SCRATCH: list[str] = []


@pytest.fixture(autouse=True)
def _cleanup_scratch():
    yield
    import shutil
    while _SCRATCH:
        shutil.rmtree(_SCRATCH.pop(), ignore_errors=True)


def new(tmp_path, mutate=None, **kw):
    pk = mkpack(tmp_path, mutate)
    logdir = f"/tmp/rtut-{uuid.uuid4().hex[:10]}"
    _SCRATCH.append(logdir)
    vals = {"logdir": logdir, **kw.pop("values", {})}
    return R.create(pk, "tester", vals, **kw)


def do_step(r, sid, alt=None, phrase=None, manual=True, run_cmd=True, evidence=None):
    """What an operator does: start, run the command, check, mark done."""
    R.start(r, sid, alt=alt, phrase=phrase)
    cmd = R.command_for(r, sid)
    if cmd and run_cmd:
        subprocess.run(cmd, shell=True, check=True, capture_output=True)
    R.verify(r, sid, confirm=True, manual_ok=manual)
    R.complete(r, sid, evidence)


def pass_gate(r, gid):
    for i, ck in enumerate(R.flow(r).gates[gid]["checks"]):
        R.gate_check(r, gid, i, confirm=True, manual_ok=True)


# ---- datapoints -----------------------------------------------------------------------------------

def test_create_sets_datapoint_states(tmp_path):
    r = new(tmp_path)
    dp = r.data["datapoints"]
    assert dp["keeper_name"]["state"] == "set" and dp["keeper_name"]["value"] == "Ada"
    assert dp["lamp_lit"]["state"] == "nil"
    assert r.data["params"]["weather"] == "fog"


def test_invalid_datapoint_is_err_and_not_stored(tmp_path):
    r = new(tmp_path)
    assert R.set_datapoint(r, "weather", "hail") == "err"
    assert "value" not in r.data["datapoints"]["weather"] and "weather" not in r.data["params"]
    assert R.set_datapoint(r, "weather", "storm") == "set"
    assert R.set_datapoint(r, "logdir", "/etc") == "err"
    assert R.set_datapoint(r, "motto", 'x"; rm -rf /') == "err"
    with pytest.raises(R.RunError):
        R.create(mkpack(tmp_path), "t", {"weather": "hail"})


# ---- readiness, recommendation, parallelism, alternatives ---------------------------------------------

def test_recommended_next_is_first_ready_in_canonical_order_and_others_are_parallel(tmp_path):
    r = new(tmp_path)
    na = R.next_actions(r)
    assert na["recommended"] == "read-the-sign"
    assert na["ready"] == ["read-the-sign", "light-lamp", "read-the-sky"]           # parallel, canonical order
    blocked = {b["id"]: b["unmet"] for b in na["blocked"]}
    assert "lamp-check" in blocked and "warn" in blocked and "close-log" in blocked
    R.start(r, "light-lamp")                                                        # any ready item may be taken
    assert R.next_actions(r)["recommended"] == "light-lamp"                         # a running item comes first


def test_alternatives_recommended_by_datapoint_and_explicit_choice(tmp_path):
    for weather, alt in (("fog", "sound-horn"), ("clear", "wave-flags"), ("storm", "shelter")):
        r = new(tmp_path, values={"weather": weather})
        assert R.recommended_alt(r, "warn") == alt
    r = new(tmp_path)
    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
        do_step(r, s)
    pass_gate(r, "lamp-check")
    with pytest.raises(R.RunError, match="choose one of"):
        R.start(r, "warn", alt="dance")
    R.start(r, "warn", alt="wave-flags")                                             # operator may override the recommendation
    assert R.command_for(r, "warn").startswith("echo") and "signals with flags" in R.command_for(r, "warn")
    assert r.data["steps"]["warn"]["alternative"] == "wave-flags"


def test_datapoint_ok_via_confirmed_by_and_gate_flow(tmp_path):
    r = new(tmp_path)
    assert R.unmet(r, "lamp-check")                                                  # stage steps not done yet
    do_step(r, "light-lamp")
    dp = r.data["datapoints"]["lamp_lit"]
    assert dp["state"] == "ok" and dp["how"]                                          # produced, then confirmed by the step's own check
    for s in ("read-the-sign", "read-the-sky"):
        do_step(r, s)
    assert R.unmet(r, "lamp-check") == []
    R.gate_check(r, "lamp-check", 0, manual_ok=True)
    assert r.data["gates"]["lamp-check"]["state"] == "open"                          # not all checks evaluated yet
    R.gate_check(r, "lamp-check", 1)                                                  # datapoints_ok is automatic
    assert r.data["gates"]["lamp-check"]["state"] == "passed"
    assert "warn" in R.next_actions(r)["ready"]


def test_failed_step_marks_produced_datapoint_err_and_retry_clears_it(tmp_path):
    r = new(tmp_path)
    R.start(r, "light-lamp"); R.fail(r, "light-lamp", "no oil")
    assert r.data["datapoints"]["lamp_lit"]["state"] == "err"
    R.retry(r, "light-lamp")
    assert r.data["datapoints"]["lamp_lit"]["state"] == "nil"


# ---- overrides, waivers, deviations ------------------------------------------------------------------

def test_override_needs_reason_and_acknowledged_risk_and_is_recorded(tmp_path):
    r = new(tmp_path)
    with pytest.raises(R.RunError, match="Risk:"):
        R.start(r, "warn", alt="sound-horn")                                          # gate not passed: names unmet + risk
    with pytest.raises(R.RunError, match="acknowledge"):
        R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text")
    R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text", acknowledge_risk=True)
    dev = r.data["deviations"][0]
    assert dev["kind"] == "override" and dev["item"] == "warn" and dev["mode"] == "manual"
    assert any("lamp-check" in u for u in dev["unmet"]) and dev["risk_acknowledged"] and dev["reason"]
    assert "not complete" in dev["unmet"][0]


def test_waiver_only_where_declared_with_reason_and_recorded(tmp_path):
    r = new(tmp_path)
    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
        do_step(r, s)
    R.gate_check(r, "lamp-check", 0, manual_ok=False)
    R.gate_check(r, "lamp-check", 1)
    assert r.data["gates"]["lamp-check"]["state"] == "failed"
    with pytest.raises(R.RunError, match="reason"):
        R.waive(r, "lamp-check", "  ")
    R.waive(r, "lamp-check", "harbour master accepts the risk")
    assert r.data["gates"]["lamp-check"]["state"] == "waived"
    assert r.data["deviations"][-1]["kind"] == "waiver" and "check 0 not passed" in r.data["deviations"][-1]["unmet"]
    assert "warn" in R.next_actions(r)["ready"]                                       # a waived gate unblocks dependents
    with pytest.raises(R.RunError, match="not waivable"):
        R.waive(r, "night-on-record", "try")


def test_deviations_block_exercise_but_not_finalize(tmp_path):
    r = new(tmp_path)
    R.start(r, "warn", alt="sound-horn", override_reason="probe", acknowledge_risk=True)
    r.data["steps"]["warn"]["state"] = "failed"
    assert r.data["deviations"]
    with pytest.raises(R1.RunError, match="deviations"):
        for k in r.data["steps"]:
            r.data["steps"][k]["state"] = "done"
        R1.exercise_update(r)


# ---- confirmation styles ------------------------------------------------------------------------------

def test_irreversible_step_needs_the_typed_phrase(tmp_path):
    r = new(tmp_path)
    assert R.confirmation_style(r, "read-the-sign") == "confirm" and R.confirmation_style(r, "close-log") == "phrase"
    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
        do_step(r, s)
    pass_gate(r, "lamp-check")
    do_step(r, "warn")
    do_step(r, "write-logbook")
    with pytest.raises(R.RunError, match="confirmation phrase"):
        R.start(r, "close-log")
    with pytest.raises(R.RunError, match="confirmation phrase"):
        R.start(r, "close-log", phrase="yes")
    R.start(r, "close-log", phrase=PHRASE)
    assert r.data["steps"]["close-log"]["authorized_by"] == "tester"


def test_preauthorization_only_for_irreversible_steps(tmp_path):
    with pytest.raises(R.RunError, match="only irreversible"):
        new(tmp_path, mode="auto-confirm", preauthorize=["light-lamp"])


# ---- auto-confirm ------------------------------------------------------------------------------------------------

def drive(r, ticks=60, now=None, operator=None):
    log = []
    for _ in range(ticks):
        out = R.auto_confirm_tick(r, now)
        log.append(out)
        if out["action"] in ("finished", "stopped", "awaiting-operator", "waiting"):
            break
        if out["action"] == "started" and operator:
            operator(out["item"])                   # the operator runs the step's command while auto-confirm confirms
    return log


def test_auto_confirm_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
    r = new(tmp_path, mode="auto-confirm")

    def operator(sid):
        cmd = R.command_for(r, sid)
        if cmd:
            subprocess.run(cmd, shell=True, check=True, capture_output=True)

    log = drive(r, operator=operator)
    assert log[-1] == {"action": "awaiting-operator", "item": "close-log", "detail": "step is manual-only"}
    done = [x["item"] for x in log if x["action"] == "completed"]
    assert done[:3] == ["read-the-sign", "light-lamp", "read-the-sky"] and "lamp-check" in done and "write-logbook" in done
    assert r.data["steps"]["warn"]["alternative"] == "sound-horn"                    # recommended by the fog datapoint
    assert r.data["gates"]["lamp-check"]["state"] == "passed"
    assert r.data["steps"]["read-the-sign"]["verification"].get("auto_confirmed")
    assert r.data["deviations"] == []
    R.start(r, "close-log", phrase=PHRASE)                                            # the operator takes over the last step
    subprocess.run(R.command_for(r, "close-log"), shell=True, check=True, capture_output=True)
    R.verify(r, "close-log", confirm=True); R.complete(r, "close-log")
    assert R.auto_confirm_tick(r)["action"] == "completed"                               # gate night-on-record
    assert R.auto_confirm_tick(r)["action"] == "finished"
    R1.finalize(r, "succeeded")
    assert r.data["outcome"] == "succeeded" and not r.data["deviations"]


def test_auto_confirm_timeout_stop(tmp_path):
    def m(d):
        d["timeout"] = {"seconds": 30, "on_timeout": "stop"}
    r = new(tmp_path, m, mode="auto-confirm")
    log = drive(r)                                     # write-logbook's file check cannot pass: nobody ran its command
    assert log[-1]["action"] == "waiting" and log[-1]["item"] == "write-logbook"
    late = datetime.now(timezone.utc) + timedelta(seconds=31)
    out = R.auto_confirm_tick(r, late)
    assert out["action"] == "stopped" and "timed out after 30s" in out["detail"]
    assert R.auto_confirm_tick(r)["action"] == "stopped"                                # stays stopped
    assert r.data["deviations"] == []                                                # stopping is not a deviation


def test_auto_confirm_push_through_records_deviation_and_moves_on(tmp_path):
    def m(d):
        d["timeout"] = {"seconds": 30, "on_timeout": "push-through"}
    r = new(tmp_path, m, mode="auto-confirm")
    drive(r)
    late = datetime.now(timezone.utc) + timedelta(seconds=31)
    out = R.auto_confirm_tick(r, late)
    assert out["action"] == "push-through" and out["item"] == "write-logbook"
    assert r.data["steps"]["write-logbook"]["state"] == "pushed"                     # its own state: not done, not failed
    dev = r.data["deviations"][0]
    assert dev["kind"] == "push-through" and dev["mode"] == "auto-confirm" and dev["risk_acknowledged"] is False
    nxt = R.auto_confirm_tick(r, late)                                                    # continues to the manual-only step
    assert nxt["item"] == "close-log" and nxt["action"] == "awaiting-operator"
    with pytest.raises(R1.RunError, match="deviations"):
        for k in r.data["steps"]:
            r.data["steps"][k]["state"] = "done"
        R1.exercise_update(r)


def test_auto_confirm_gate_push_through_and_stop(tmp_path):
    def m(d):
        d["stages"][0]["gates"][0]["checks"].append({"done_when": "always fails", "command": "false", "expect": "exit-0"})
        d["stages"][0]["gates"][0]["timeout"] = {"seconds": 10, "on_timeout": "push-through"}
    r = new(tmp_path, m, mode="auto-confirm")
    drive(r)
    assert r.data["gates"]["lamp-check"]["state"] == "failed"        # all checks evaluated, one failing
    out = R.auto_confirm_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
    assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "pushed"   # not "waived": nobody approved it
    assert r.data["deviations"][0]["item"] == "lamp-check" and r.data["deviations"][0]["kind"] == "push-through"


def test_auto_confirm_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
    def m(d):
        for st in d["stages"]:
            for s in st.get("steps", []):
                s.pop("auto_confirm", None)                                             # allow automation of close-log
    r = new(tmp_path, m, mode="auto-confirm")
    drive(r, operator=lambda sid: (subprocess.run(R.command_for(r, sid), shell=True, check=True, capture_output=True) if R.command_for(r, sid) else None))
    last = R.auto_confirm_tick(r)
    assert last["action"] == "awaiting-operator" and "not pre-authorized" in last["detail"]
    r2 = new(tmp_path, m, mode="auto-confirm", preauthorize=["close-log"])
    drive(r2, operator=lambda sid: (subprocess.run(R.command_for(r2, sid), shell=True, check=True, capture_output=True) if R.command_for(r2, sid) else None))
    assert r2.data["steps"]["close-log"]["state"] in ("running", "done")
    assert r2.data["steps"]["close-log"]["authorized_by"] == "tester"


def test_manual_mode_rejects_auto_confirm_and_auto_confirm_rejects_overrides(tmp_path):
    with pytest.raises(R.RunError, match="auto-confirm run"):
        R.auto_confirm_tick(new(tmp_path))
    r = new(tmp_path, mode="auto-confirm")
    with pytest.raises(R.RunError, match="manual mode"):
        R.start(r, "warn", alt="sound-horn", override_reason="x", acknowledge_risk=True)


def test_finished_v02_run_is_read_only_and_scores_count_it(tmp_path):
    r = new(tmp_path)
    R1.finalize(r, "abandoned")
    with pytest.raises(R.RunError, match="finished"):
        R.start(r, "read-the-sign")
    with pytest.raises(R.RunError, match="finished"):
        R.set_datapoint(r, "weather", "clear")
    assert R1.scorecard(r.pack)["abandoned"] == 1
    assert R1.load(r.pack, r.data["run_id"]).data["spec"] == "runbook-run/v0.2"      # reloads through the shared loader


def test_pushed_items_are_a_distinct_state_that_lets_the_run_continue_but_never_succeed(tmp_path):
    def m(d):
        d["timeout"] = {"seconds": 5, "on_timeout": "push-through"}
        light = next(s for st in d["stages"] for s in st.get("steps", []) if s["id"] == "light-lamp")
        light["verify"] = {"done_when": "never true here", "command": "false", "expect": "exit-0"}
        d["stages"][0]["gates"][0]["timeout"] = {"seconds": 5, "on_timeout": "push-through"}   # the gate's own timeout wins
    r = new(tmp_path, m, mode="auto-confirm")
    late = datetime.now(timezone.utc) + timedelta(seconds=6)
    seen = []
    for _ in range(40):
        out = R.auto_confirm_tick(r, late)
        seen.append(out["action"])
        if out["action"] in ("stopped", "awaiting-operator", "finished"):
            break
        late = late + timedelta(seconds=6)
    assert r.data["steps"]["light-lamp"]["state"] == "pushed"
    dp = r.data["datapoints"]["lamp_lit"]
    assert dp["state"] == "pushed" and "value" not in dp and "error" not in dp        # not set, not err
    assert "lamp_lit" not in r.data["params"]                                          # never usable as a value
    assert R.is_complete(r, "light-lamp")                                              # dependents may proceed
    assert r.data["gates"]["lamp-check"]["state"] == "pushed"                          # datapoints_ok needs 'ok', so the gate was pushed too
    kinds = [d["kind"] for d in r.data["deviations"]]
    assert kinds.count("push-through") >= 2 and "override" not in kinds
    with pytest.raises(R1.RunError, match="succeeds only when every step is done"):
        R1.finalize(r, "succeeded")
    R1.finalize(r, "failed", "pushed through to collect problems")                     # a clean-completion failure, finishable as failed
    assert r.data["outcome"] == "failed"