"""The download supervisor/watchdog: `_run_supervised` decides when to hard-kill a child, and
`_stop_child` escalates SIGTERM→SIGKILL. No real subprocess or DB here — a fake proc stands in and
the DB-liveness lookup + side effects are monkeypatched, so the decision logic is exercised on any
machine. These are regression tests for the class of bug that motivated the whole change: a wedged
download that a thread could neither kill nor cancel.
"""

import signal
import subprocess

import pytest

from app import worker


class _FakeProc:
    """Stands in for the child Popen. `waits` scripts each wait() call: "timeout" = still alive,
    ("exit", rc) = the child finished with that code."""

    def __init__(self, waits, pid=4242):
        self._waits = list(waits)
        self.pid = pid
        self.returncode = None
        self.signals = []

    def wait(self, timeout=None):
        step = self._waits.pop(0) if self._waits else "timeout"
        if step == "timeout":
            raise subprocess.TimeoutExpired("child", timeout or 0)
        self.returncode = step[1]
        return step[1]

    def poll(self):
        return self.returncode

    def send_signal(self, sig):
        self.signals.append(sig)


@pytest.fixture
def rec(monkeypatch):
    """Record supervisor side effects and stub out the real spawn/kill/liveness/staging so a test
    drives them. `calls` collects (name, kwargs) so assertions read the decisions, not the plumbing."""
    calls = []

    def record(name):
        def _f(*args, **kwargs):
            calls.append((name, args, kwargs))

        return _f

    monkeypatch.setattr(worker, "_reset_asset_pending", record("reset_asset"))
    monkeypatch.setattr(worker, "_requeue_if_running", record("requeue"))
    monkeypatch.setattr(worker, "_fail_running", record("fail"))
    monkeypatch.setattr(worker, "_cleanup_staging", record("cleanup"))
    monkeypatch.setattr(worker, "_stop_child", record("stop_child"))
    monkeypatch.setattr(worker._stop, "is_set", lambda: False)
    return calls


def _names(calls):
    return [c[0] for c in calls]


def _spawn(monkeypatch, proc):
    monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: proc)


def _liveness(monkeypatch, value):
    monkeypatch.setattr(worker, "_job_liveness", lambda job_id: value)


def test_clean_child_exit_leaves_the_outcome_alone(monkeypatch, rec):
    # Child finished on its own (exit 0) — it wrote the terminal state. The supervisor only reconciles
    # the asset (a no-op when the child finalized it) and must not kill or re-fail anything.
    _spawn(monkeypatch, _FakeProc([("exit", 0)]))
    worker._run_supervised(1, 99)
    assert _names(rec) == ["reset_asset"]


def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec):
    _spawn(monkeypatch, _FakeProc(["timeout"]))
    _liveness(monkeypatch, ("running", 9999.0))  # far past the stall timeout
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "reset_asset" in names
    fail = next(c for c in rec if c[0] == "fail")
    assert fail[2]["error_code"] == "network"
    assert "stalled" in fail[2]["error"].lower()
    assert fail[2]["cause"] == "stalled"  # the line the killed child never gets to write


def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec):
    # A route flipped the job to 'paused'. Kill + free the asset, but do NOT overwrite the
    # status with an error and do NOT requeue.
    _spawn(monkeypatch, _FakeProc(["timeout"]))
    _liveness(monkeypatch, ("paused", 1.0))
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "reset_asset" in names
    assert "fail" not in names and "requeue" not in names


def test_route_cancel_kills_but_keeps_the_route_status(monkeypatch, rec):
    # B-sub-1 regression: the cancel ROUTE writes the literal 'canceled' (models.py), but the supervisor
    # used to compare against 'cancelled' — so this branch was DEAD and a canceled job's child ran on to
    # completion, overwriting 'canceled' with 'done' and double-releasing its asset. The supervisor MUST
    # treat the route's actual literal as a stop: kill + free the asset, no error write, no requeue.
    _spawn(monkeypatch, _FakeProc(["timeout"]))
    _liveness(monkeypatch, ("canceled", 1.0))
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "reset_asset" in names
    assert "fail" not in names and "requeue" not in names


def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec):
    monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
    _spawn(monkeypatch, _FakeProc(["timeout"]))
    _liveness(monkeypatch, ("running", 1.0))  # fresh progress, but over the absolute cap
    worker._run_supervised(1, 99)
    fail = next(c for c in rec if c[0] == "fail")
    assert "time limit" in fail[2]["error"].lower()
    assert fail[2]["cause"] == "timeout"


def test_child_crash_without_finishing_is_failed(monkeypatch, rec):
    # Non-zero exit and the job is still 'running' → the child died (crash/OOM) before writing state.
    _spawn(monkeypatch, _FakeProc([("exit", 1)]))
    _liveness(monkeypatch, ("running", 1.0))
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "reset_asset" in names
    fail = next(c for c in rec if c[0] == "fail")
    assert "unexpectedly" in fail[2]["error"].lower()
    # Its line is `_fail_running`'s, so it is logged only if the job really failed here.
    assert (fail[2]["cause"], fail[2]["exit_code"]) == ("child_exit", 1)


def test_child_nonzero_exit_that_already_set_state_is_left_alone(monkeypatch, rec):
    # Child exited non-zero but the job is no longer 'running' (it recorded its own error) → no double.
    _spawn(monkeypatch, _FakeProc([("exit", 1)]))
    _liveness(monkeypatch, ("error", None))
    worker._run_supervised(1, 99)
    assert "fail" not in _names(rec)


def test_spawn_failure_frees_the_asset_and_requeues(monkeypatch, rec):
    def boom(job_id):
        raise OSError("no fork")

    monkeypatch.setattr(worker, "_spawn_download_child", boom)
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "reset_asset" in names and "requeue" in names


def test_shutdown_kills_and_requeues_for_next_boot(monkeypatch, rec):
    monkeypatch.setattr(worker._stop, "is_set", lambda: True)
    _spawn(monkeypatch, _FakeProc(["timeout"]))
    _liveness(monkeypatch, ("running", 1.0))
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "requeue" in names and "fail" not in names


def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
    # A download that outlives the stall timeout must NOT be killed as long as it keeps writing
    # progress (updated_at stays fresh via the model's onupdate). Model: alive across several polls,
    # liveness always fresh, then a clean exit — nothing killed, nothing failed.
    monkeypatch.setattr(worker.settings, "download_stall_timeout_seconds", 10)
    _spawn(monkeypatch, _FakeProc(["timeout", "timeout", "timeout", ("exit", 0)]))
    _liveness(monkeypatch, ("running", 2.0))  # always well under the 10s stall
    worker._run_supervised(1, 99)
    assert _names(rec) == ["reset_asset"]  # only the clean-exit reconciliation (a no-op), no kill


def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec):
    # The child committed its terminal state (done/error) but hasn't exited yet (e.g. slow staging
    # cleanup outlasting a poll). A non-'running' status must NOT be treated as a route-cancel and
    # kill the finisher (which could drop its completion notification) — let it exit on its own.
    _spawn(monkeypatch, _FakeProc(["timeout", ("exit", 0)]))  # alive one poll, then a clean exit
    _liveness(monkeypatch, ("done", 1.0))
    worker._run_supervised(1, 99)
    assert _names(rec) == ["reset_asset"]  # not killed; only the clean-exit reconciliation (no-op)


def test_a_wedged_finished_child_is_killed_by_the_wall_clock(monkeypatch, rec):
    # A child that reached a terminal state (done) but never exits (e.g. wedged in teardown on a stuck
    # mount) must still be killed by the absolute wall-clock cap to free the worker slot — WITHOUT
    # overwriting its DB status (it already succeeded).
    monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
    _spawn(monkeypatch, _FakeProc(["timeout"]))  # never exits within the test
    _liveness(monkeypatch, ("done", 1.0))  # finished, but the process lingers
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "fail" not in names  # killed to free the slot, status kept


def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):
    # If the liveness probe raises (e.g. a transient DB error), the child must NOT be left running
    # unsupervised — the feature's whole point. It is killed and the job is failed.
    _spawn(monkeypatch, _FakeProc(["timeout"]))

    def boom(job_id):
        raise RuntimeError("db down")

    monkeypatch.setattr(worker, "_job_liveness", boom)
    worker._run_supervised(1, 99)
    names = _names(rec)
    assert "stop_child" in names and "reset_asset" in names
    assert "fail" in _names(rec)


# --- _stop_child escalation -----------------------------------------------------------------


@pytest.fixture
def killpg_rec(monkeypatch):
    sent = []
    monkeypatch.setattr(worker.os, "getpgid", lambda pid: pid)
    monkeypatch.setattr(worker.os, "killpg", lambda pgid, sig: sent.append(sig))
    return sent


def test_stop_child_sigterm_is_enough(killpg_rec):
    proc = _FakeProc([("exit", 0)])  # dies on the SIGTERM wait
    worker._stop_child(proc, grace_s=0.01)
    assert killpg_rec == [signal.SIGTERM]


def test_stop_child_escalates_to_sigkill(killpg_rec):
    proc = _FakeProc(["timeout", ("exit", -9)])  # survives SIGTERM, dies on SIGKILL
    worker._stop_child(proc, grace_s=0.01)
    assert killpg_rec == [signal.SIGTERM, signal.SIGKILL]


def test_stop_child_noop_if_already_exited(killpg_rec):
    proc = _FakeProc([])
    proc.returncode = 0  # already gone
    worker._stop_child(proc, grace_s=0.01)
    assert killpg_rec == []


def test_a_crashed_jobs_crash_line_and_cleanup_carry_the_job_id(monkeypatch):
    """The job id is bound for the WHOLE job, the crash handler included: bound inside the `try`, it
    was already reset when the one line that says the job crashed was logged."""
    from app import logs

    seen: dict[str, object] = {}
    claims = iter([42])

    def claim():
        job = next(claims, None)
        if job is None:
            worker._stop.set()
        return job

    def crash(_job_id):
        raise RuntimeError("boom")

    monkeypatch.setattr(worker, "_claim_job", claim)
    monkeypatch.setattr(worker, "_process_job", crash)
    monkeypatch.setattr(
        worker.log, "error", lambda *a, **k: seen.setdefault("crash", logs._job_id.get())
    )
    monkeypatch.setattr(
        worker, "_fail_running", lambda *a, **k: seen.setdefault("cleanup", logs._job_id.get())
    )
    monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
    worker._stop.clear()
    try:
        worker._worker_loop(0)
    finally:
        worker._stop.clear()
    assert seen == {"crash": 42, "cleanup": 42}