"""Tests for transcribe-worker: the queue drain on the transcription host.

ffmpeg, whisper-cli and the diarizer are replaced by small fake executables (the
process boundary). The queue handling, the merge and the loop guard run for real.
"""

import importlib.machinery
import importlib.util
import json
import os
import stat
import sys
from pathlib import Path

import pytest

SRC = Path(__file__).resolve().parent.parent / "src"
sys.path.insert(0, str(SRC))


def _load_worker():
    loader = importlib.machinery.SourceFileLoader("transcribe_worker", str(SRC / "transcribe-worker"))
    spec = importlib.util.spec_from_loader("transcribe_worker", loader)
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules["transcribe_worker"] = module  # dataclasses looks the module up by name
    loader.exec_module(module)
    return module


worker = _load_worker()

FAKE_FFMPEG = """#!/usr/bin/env bash
# last argument is the output; the one after -i is the input
while [[ $# -gt 1 ]]; do [[ "$1" == "-i" ]] && in="$2"; shift; done
cp "$in" "$1"
exit "${FAKE_FFMPEG_EXIT:-0}"
"""

FAKE_WHISPER = """#!/usr/bin/env bash
printf '%s\\n' "$*" >> "$FAKE_LOG"
while [[ $# -gt 0 ]]; do [[ "$1" == "-of" ]] && prefix="$2"; shift; done
[[ "${FAKE_WHISPER_EXIT:-0}" == "0" ]] && cp "$FAKE_WHISPER_JSON" "$prefix.json"
exit "${FAKE_WHISPER_EXIT:-0}"
"""

FAKE_DIARIZE = """#!/usr/bin/env bash
printf 'diarize %s\\n' "$*" >> "$FAKE_LOG"
[[ "${FAKE_DIARIZE_EXIT:-0}" == "0" ]] && cp "$FAKE_TURNS_JSON" "$2"
echo "fake diarizer says hello" >&2
exit "${FAKE_DIARIZE_EXIT:-0}"
"""


def whisper_json(*words, step_ms=400):
    return {"transcription": [
        {"offsets": {"from": i * step_ms, "to": i * step_ms + 300}, "text": f" {w}"}
        for i, w in enumerate(words)
    ]}


class Rig:
    def __init__(self, tmp_path, monkeypatch):
        self.home = tmp_path / "state"
        self.bin = tmp_path / "bin"
        self.bin.mkdir()
        for name, body in (("ffmpeg", FAKE_FFMPEG), ("whisper-cli", FAKE_WHISPER), ("fake-diarize", FAKE_DIARIZE)):
            path = self.bin / name
            path.write_text(body)
            path.chmod(path.stat().st_mode | stat.S_IXUSR)
        self.log = tmp_path / "calls.log"
        self.whisper_json = tmp_path / "whisper.json"
        self.turns_json = tmp_path / "turns.json"
        self.set_whisper(whisper_json("Good", "morning."))
        self.turns_json.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}]))
        monkeypatch.setenv("PATH", f"{self.bin}:{os.environ['PATH']}")
        monkeypatch.setenv("FAKE_LOG", str(self.log))
        monkeypatch.setenv("FAKE_WHISPER_JSON", str(self.whisper_json))
        monkeypatch.setenv("FAKE_TURNS_JSON", str(self.turns_json))
        self.monkeypatch = monkeypatch
        self.config = worker.Config(
            home=self.home,
            whisper_model=tmp_path / "model.bin",
            diarize_cmd=[str(self.bin / "fake-diarize")],
            threads=2,
        )

    def set_whisper(self, data):
        self.whisper_json.write_text(json.dumps(data))

    def submit(self, job_id, job=None, audio=b"audio", where="incoming"):
        folder = self.home / where / job_id
        folder.mkdir(parents=True)
        (folder / "audio.m4a").write_bytes(audio)
        if job is not False:
            (folder / "job.json").write_text(json.dumps(job if job is not None else {"language": "en"}))
        return folder

    def calls(self):
        return self.log.read_text().splitlines() if self.log.exists() else []


@pytest.fixture
def rig(tmp_path, monkeypatch):
    return Rig(tmp_path, monkeypatch)


class TestDrainNormal:
    def test_transcribe_worker_drain_one_job_writes_transcript_to_done(self, rig):
        """Normal: a job in incoming/ ends as done/<id>.txt and leaves nothing behind."""
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n"
        assert not (rig.home / "incoming" / "job1").exists()
        assert not (rig.home / "work" / "job1").exists()
        assert not (rig.home / "failed" / "job1.log").exists()

    def test_transcribe_worker_drain_writes_metadata_beside_the_transcript(self, rig):
        """Normal: done/<id>.json records what ran, for the client and for debugging."""
        rig.submit("job1", {"language": "en", "speakers": 3, "original_name": "standup.mkv"})
        worker.drain(rig.config)
        meta = json.loads((rig.home / "done" / "job1.json").read_text())
        assert meta["original_name"] == "standup.mkv"
        assert meta["speakers_found"] == 1
        assert meta["lines"] == 1
        assert meta["seconds"] >= 0

    def test_transcribe_worker_drain_passes_language_and_speaker_count_through(self, rig):
        """Normal: job options reach whisper and the diarizer."""
        rig.submit("job1", {"language": "es", "speakers": 3})
        worker.drain(rig.config)
        whisper_call = next(c for c in rig.calls() if not c.startswith("diarize"))
        diarize_call = next(c for c in rig.calls() if c.startswith("diarize"))
        assert "-l es" in whisper_call
        assert "-mc 0" in whisper_call and "-ojf" in whisper_call
        assert diarize_call.endswith("--speakers 3")

    def test_transcribe_worker_drain_min_and_max_speakers_are_forwarded(self, rig):
        """Normal: a range instead of an exact count."""
        rig.submit("job1", {"min_speakers": 2, "max_speakers": 5})
        worker.drain(rig.config)
        diarize_call = next(c for c in rig.calls() if c.startswith("diarize"))
        assert "--min-speakers 2" in diarize_call and "--max-speakers 5" in diarize_call

    def test_transcribe_worker_drain_processes_every_job_oldest_first(self, rig):
        """Normal: the queue drains completely, in arrival order."""
        first = rig.submit("b-first")
        rig.submit("a-second")
        os.utime(first, (1, 1))
        assert worker.drain(rig.config) == {"done": 2, "failed": 0}
        diarize_calls = [c for c in rig.calls() if c.startswith("diarize")]
        assert "b-first" in diarize_calls[0] and "a-second" in diarize_calls[1]


class TestDrainBoundary:
    def test_transcribe_worker_drain_empty_queue_is_a_no_op(self, rig):
        """Boundary: nothing to do, and the folders get created."""
        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
        assert (rig.home / "incoming").is_dir() and (rig.home / "done").is_dir()

    def test_transcribe_worker_drain_ignores_uploads_still_in_flight(self, rig):
        """Boundary: a dot-prefixed folder is left alone (second line of defence behind uploading/)."""
        rig.submit(".tmp-job9")
        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
        assert (rig.home / "incoming" / ".tmp-job9").exists()

    def test_transcribe_worker_drain_job_already_done_is_not_rerun(self, rig):
        """Boundary: resubmitting finished work costs nothing."""
        (rig.home / "done").mkdir(parents=True)
        (rig.home / "done" / "job1.txt").write_text("already here\n")
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 0}
        assert (rig.home / "done" / "job1.txt").read_text() == "already here\n"
        assert not (rig.home / "incoming" / "job1").exists()
        assert rig.calls() == []

    def test_transcribe_worker_drain_missing_job_file_uses_defaults(self, rig):
        """Boundary: audio with no job.json still transcribes, in English, count estimated."""
        rig.submit("job1", job=False)
        assert worker.drain(rig.config) == {"done": 1, "failed": 0}

    def test_transcribe_worker_drain_retries_a_job_left_in_work_by_a_crash(self, rig):
        """Boundary: a job stranded in work/ is picked up again."""
        rig.submit("job1", where="work")
        assert worker.drain(rig.config) == {"done": 1, "failed": 0}

    def test_transcribe_worker_drain_gives_up_on_a_job_that_keeps_crashing(self, rig):
        """Boundary: the second stranding is a failure, not a loop."""
        rig.submit("job1", {"language": "en", "attempts": 2}, where="work")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert "attempt" in (rig.home / "failed" / "job1.log").read_text()


class TestDrainError:
    def test_transcribe_worker_drain_whisper_failure_is_logged_with_its_stage(self, rig):
        """Error: whisper exits non-zero."""
        rig.monkeypatch.setenv("FAKE_WHISPER_EXIT", "3")
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        log = (rig.home / "failed" / "job1.log").read_text()
        assert "whisper" in log
        assert not (rig.home / "done" / "job1.txt").exists()
        assert not (rig.home / "work" / "job1").exists()

    def test_transcribe_worker_drain_diarizer_failure_keeps_its_stderr(self, rig):
        """Error: the diarizer fails; its own words land in the log."""
        rig.monkeypatch.setenv("FAKE_DIARIZE_EXIT", "1")
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        log = (rig.home / "failed" / "job1.log").read_text()
        assert "diarize" in log and "fake diarizer says hello" in log

    def test_transcribe_worker_drain_looping_transcription_fails(self, rig):
        """Error: whisper's repetition loop is a failed job, never a transcript."""
        rig.set_whisper(whisper_json(*("we do not know where we are going".split() * 5), step_ms=1000))  # 40 s
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert "looped" in (rig.home / "failed" / "job1.log").read_text()
        assert not (rig.home / "done" / "job1.txt").exists()

    def test_transcribe_worker_drain_short_loop_is_collapsed_and_recorded(self, rig):
        """Normal: a stutter of a few seconds is collapsed to one copy and noted in the metadata."""
        rig.set_whisper(whisper_json("Okay,", *("fair, it's not going to be".split() * 4), "easy."))
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Okay, fair, it's not going to be easy.\n"
        meta = json.loads((rig.home / "done" / "job1.json").read_text())
        assert meta["loops_collapsed"] == [["fair, it's not going to be", 4]]

    def test_transcribe_worker_drain_malformed_job_file_fails_that_job_only(self, rig):
        """Error: one bad job does not stop the queue."""
        bad = rig.submit("bad")
        (bad / "job.json").write_text("{not json")
        os.utime(bad, (1, 1))
        rig.submit("good")
        assert worker.drain(rig.config) == {"done": 1, "failed": 1}
        assert (rig.home / "done" / "good.txt").exists()
        assert "job.json" in (rig.home / "failed" / "bad.log").read_text()

    def test_transcribe_worker_drain_job_without_audio_fails(self, rig):
        """Error: a job folder holding no audio file."""
        folder = rig.submit("job1")
        (folder / "audio.m4a").unlink()
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert "audio" in (rig.home / "failed" / "job1.log").read_text()

    @pytest.mark.parametrize("bad", [{"speakers": 0}, {"speakers": "three"}, {"language": "en; rm -rf"}])
    def test_transcribe_worker_drain_rejects_bad_option_values(self, rig, bad):
        """Error: options are validated before they reach a command line."""
        rig.submit("job1", bad)
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert rig.calls() == []


class TestDrainSilenceHallucinations:
    def _silent_stretch(self, rig):
        spoken = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}]
        invented = [
            {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."}
            for i in range(12)
        ]
        rig.set_whisper({"transcription": spoken + invented})
        rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]))

    def test_transcribe_worker_drain_words_invented_in_silence_are_dropped_not_failed(self, rig):
        """Normal: a quiet meeting transcribes; the invented run never reaches the transcript."""
        self._silent_stretch(rig)
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 1, "failed": 0}
        assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n"

    def test_transcribe_worker_drain_metadata_counts_the_dropped_words(self, rig):
        """Normal: the count is on record, so a transcript that lost a lot is visible."""
        self._silent_stretch(rig)
        rig.submit("job1")
        worker.drain(rig.config)
        assert json.loads((rig.home / "done" / "job1.json").read_text())["dropped_outside_speech"] == 12

    def test_transcribe_worker_drain_recording_with_no_speech_at_all_fails_cleanly(self, rig):
        """Error: everything whisper produced sits in silence."""
        rig.set_whisper({"transcription": [{"offsets": {"from": 60_000, "to": 61_000}, "text": " Thank you."}]})
        rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]))
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert "no speech" in (rig.home / "failed" / "job1.log").read_text()


class TestRunStdin:
    def test_transcribe_worker_run_gives_tools_no_stdin(self, tmp_path):
        """Boundary: ffmpeg reads stdin when it can, which would eat a calling loop's input."""
        out = tmp_path / "stdin-target"
        read_end, write_end = os.pipe()
        saved = os.dup(0)
        os.dup2(read_end, 0)
        try:
            worker._run("probe", ["bash", "-c", f"readlink /proc/self/fd/0 > {out}"])
        finally:
            os.dup2(saved, 0)
            for fd in (saved, read_end, write_end):
                os.close(fd)
        assert out.read_text().strip() == "/dev/null"


class TestDrainResilience:
    def test_transcribe_worker_drain_null_attempts_counts_as_a_first_try(self, rig):
        """Error: a job.json with "attempts": null must not take the worker down; it is a first try."""
        odd = rig.submit("odd", {"language": "en", "attempts": None})
        os.utime(odd, (1, 1))
        rig.submit("good")
        assert worker.drain(rig.config) == {"done": 2, "failed": 0}
        assert (rig.home / "done" / "odd.txt").exists()
        assert (rig.home / "done" / "good.txt").exists()

    def test_transcribe_worker_drain_unexpected_error_in_one_job_is_logged_and_the_queue_goes_on(self, rig):
        """Error: an exception the pipeline never anticipated lands in failed/ with its type, not on the run."""
        first = rig.submit("first")
        os.utime(first, (1, 1))
        rig.submit("second")

        def explode(job: dict) -> list[str]:
            raise RuntimeError("boom")

        rig.monkeypatch.setattr(worker, "_diarize_options", explode)
        assert worker.drain(rig.config) == {"done": 0, "failed": 2}  # both reached, neither crashed the run
        for job_id in ("first", "second"):
            log = (rig.home / "failed" / f"{job_id}.log").read_text()
            assert "RuntimeError" in log and "boom" in log
            assert not (rig.home / "work" / job_id).exists()

    def test_transcribe_worker_drain_unwritable_done_dir_is_a_failed_job_not_a_crash(self, rig):
        """Error: an OSError while writing the transcript lands in failed/, and the run survives."""
        if os.geteuid() == 0:
            pytest.skip("root ignores directory permissions")
        (rig.home / "done").mkdir(parents=True)
        (rig.home / "done").chmod(0o500)
        try:
            rig.submit("job1")
            assert worker.drain(rig.config) == {"done": 0, "failed": 1}
            assert "job1" in "".join(p.name for p in (rig.home / "failed").iterdir())
        finally:
            (rig.home / "done").chmod(0o700)

    def test_transcribe_worker_drain_ffmpeg_failure_is_logged_with_its_stage(self, rig):
        """Error: ffmpeg exits non-zero."""
        rig.monkeypatch.setenv("FAKE_FFMPEG_EXIT", "2")
        rig.submit("job1")
        assert worker.drain(rig.config) == {"done": 0, "failed": 1}
        assert "ffmpeg" in (rig.home / "failed" / "job1.log").read_text()

    def test_transcribe_worker_drain_returns_at_once_when_another_worker_holds_the_lock(self, rig):
        """Boundary: a second drain does not touch the queue while the first holds the lock."""
        import fcntl
        rig.home.mkdir(parents=True, exist_ok=True)
        rig.submit("job1")
        with open(rig.home / "lock", "w", encoding="utf-8") as held:
            fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB)
            assert worker.drain(rig.config) == {"done": 0, "failed": 0}
        assert (rig.home / "incoming" / "job1").exists()
        assert rig.calls() == []