"""Tests for ratio-transcribe, the client.

ssh and scp are replaced by fakes that act on a temp directory standing in for the
remote home, so the client's real logic (job ids, upload-then-rename, polling,
collecting, the local fallback) runs against a filesystem it can't tell from the host.
"""

import json
import os
import subprocess
from pathlib import Path

import pytest

SCRIPT = Path(__file__).resolve().parent.parent / "src" / "ratio-transcribe"
STATE = ".local/state/meeting-transcribe"

FAKE_SSH = r"""#!/usr/bin/env bash
# ssh [-o k=v]... host command...   -> run the command with HOME at the fake remote
[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255
no_stdin=""
while [[ "$1" == -* ]]; do [[ "$1" == "-n" ]] && no_stdin=1; [[ "$1" == "-o" ]] && shift; shift; done
shift  # host
printf '%s\n' "$*" >> "$FAKE_SSH_LOG"
if [[ -n "$no_stdin" ]]; then
  ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" < /dev/null )
  rc=$?
else
  ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" )
  rc=$?
  cat > /dev/null  # like the real ssh, drain whatever stdin the command left behind
fi
if [[ "$*" == *"incoming/"* && "$*" == mv* && -n "${FAKE_ON_SUBMIT:-}" ]]; then
  ( cd "$FAKE_REMOTE_HOME" && bash -c "$FAKE_ON_SUBMIT" )
fi
exit $rc
"""

FAKE_SCP = r"""#!/usr/bin/env bash
[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255
while [[ "$1" == -* ]]; do [[ "$1" == "-o" ]] && shift; shift; done
printf '%s -> %s\n' "$1" "$2" >> "$FAKE_SCP_LOG"
cp "$1" "$FAKE_REMOTE_HOME/${2#*:}"
"""

# What the worker would do, compressed: finish or fail whatever sits in incoming/.
WORKER_OK = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d");
  printf '00:00:00 Speaker A: Hello from the host.\\n' > done/$id.txt; rm -rf "$d"; done"""
WORKER_FAIL = f"""cd {STATE}; mkdir -p failed; for d in incoming/*/; do id=$(basename "$d");
  printf 'stage: whisper\\nexit 3\\n' > failed/$id.log; rm -rf "$d"; done"""
WORKER_EMPTY = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d");
  : > done/$id.txt; rm -rf "$d"; done"""


class Rig:
    def __init__(self, tmp_path):
        self.tmp = tmp_path
        self.remote = tmp_path / "remote-home"
        self.remote.mkdir()
        self.local_home = tmp_path / "local-home"
        self.local_home.mkdir()
        self.bin = tmp_path / "bin"
        self.bin.mkdir()
        for name, body in (("ssh", FAKE_SSH), ("scp", FAKE_SCP)):
            path = self.bin / name
            path.write_text(body)
            path.chmod(0o755)
        self.ssh_log = tmp_path / "ssh.log"
        self.scp_log = tmp_path / "scp.log"
        self.audio = tmp_path / "meeting.m4a"
        self.audio.write_bytes(b"pretend audio")

    def run(self, args=None, on_submit=WORKER_OK, **env_extra):
        env = {
            "PATH": f"{self.bin}:{os.environ['PATH']}",
            "HOME": str(self.local_home),
            "FAKE_REMOTE_HOME": str(self.remote),
            "FAKE_SSH_LOG": str(self.ssh_log),
            "FAKE_SCP_LOG": str(self.scp_log),
            "TRANSCRIBE_HOST": "testhost",
            "TRANSCRIBE_POLL": "0",
            "TRANSCRIBE_TIMEOUT": "5",
        }
        if on_submit:
            env["FAKE_ON_SUBMIT"] = on_submit
        env.update({k: str(v) for k, v in env_extra.items()})
        if args is None:
            args = [str(self.audio)]
        return subprocess.run([str(SCRIPT), *args], env=env, capture_output=True, text=True, timeout=60)

    def state(self, *parts):
        return self.remote.joinpath(STATE, *parts)

    def uploads(self):
        return self.scp_log.read_text().splitlines() if self.scp_log.exists() else []


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


class TestClientNormal:
    def test_ratio_transcribe_new_recording_is_uploaded_and_transcript_printed(self, rig):
        """Normal: upload, wait, print. stdout is the transcript and nothing else."""
        result = rig.run()
        assert result.returncode == 0, result.stderr
        assert result.stdout == "00:00:00 Speaker A: Hello from the host.\n"
        assert len(rig.uploads()) == 1

    def test_ratio_transcribe_job_file_carries_language_speakers_and_name(self, rig):
        """Normal: the options reach the host inside job.json."""
        rig.run(args=[str(rig.audio), "es"], on_submit=None, SPEAKERS=3, TRANSCRIBE_TIMEOUT=0)
        jobs = list(rig.state("incoming").glob("*/job.json"))
        assert len(jobs) == 1
        assert json.loads(jobs[0].read_text()) == {
            "language": "es", "speakers": 3, "original_name": "meeting.m4a",
        }
        assert (jobs[0].parent / "audio.m4a").read_bytes() == b"pretend audio"

    def test_ratio_transcribe_finished_job_is_collected_without_uploading_again(self, rig):
        """Normal: rerunning after a dropped connection costs one ssh round trip."""
        first = rig.run()
        rig.scp_log.unlink()
        second = rig.run(on_submit=None)
        assert second.returncode == 0
        assert second.stdout == first.stdout
        assert rig.uploads() == []

    def test_ratio_transcribe_job_id_depends_on_audio_and_options(self, rig):
        """Normal: same file with a different speaker count is a different job."""
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0, SPEAKERS=3)
        assert len(list(rig.state("incoming").iterdir())) == 2

    def test_ratio_transcribe_earlier_failure_is_retried(self, rig):
        """Normal: a stale failure log does not block a fresh attempt."""
        assert rig.run(on_submit=WORKER_FAIL).returncode == 1
        result = rig.run(on_submit=WORKER_OK)
        assert result.returncode == 0
        assert "Hello from the host" in result.stdout
        assert list(rig.state("failed").glob("*.log")) == []


class TestClientBoundary:
    def test_ratio_transcribe_job_still_queued_is_not_uploaded_twice(self, rig):
        """Boundary: a second run while the first job waits just joins the wait."""
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        assert len(rig.uploads()) == 1

    def test_ratio_transcribe_upload_lands_in_incoming_only_by_rename(self, rig):
        """Boundary: the audio is copied into uploading/, never straight into incoming/."""
        rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        assert "/uploading/" in rig.uploads()[0]
        assert list(rig.state("uploading").iterdir()) == []

    def test_ratio_transcribe_awkward_filename_survives(self, rig):
        """Boundary: spaces and quotes in the recording's name."""
        odd = rig.tmp / "someone's \"weekly\" sync.m4a"
        odd.write_bytes(b"x")
        result = rig.run(args=[str(odd)], on_submit=None, TRANSCRIBE_TIMEOUT=0)
        job = json.loads(next(rig.state("incoming").glob("*/job.json")).read_text())
        assert job["original_name"] == "someone's \"weekly\" sync.m4a", result.stderr

    def test_ratio_transcribe_colon_in_a_relative_filename_still_uploads(self, rig):
        """Boundary: scp reads "standup 14:30.m4a" as host:path; the client hands it an absolute path."""
        (rig.tmp / "standup 14:30.m4a").write_bytes(b"x")
        env = {
            "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home),
            "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log),
            "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK,
            "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5",
        }
        result = subprocess.run(
            [str(SCRIPT), "standup 14:30.m4a"], cwd=rig.tmp, env=env, capture_output=True, text=True, timeout=60,
        )
        assert result.returncode == 0, result.stderr
        assert rig.uploads()[0].startswith("/")

    def test_ratio_transcribe_host_unreachable_falls_back_to_local_worker(self, rig):
        """Boundary: offline, the same queue and worker run on this machine."""
        worker = rig.tmp / "fake-worker"
        worker.write_text(
            "#!/usr/bin/env bash\n"
            f"cd \"$HOME/{STATE}\" && mkdir -p done && for d in incoming/*/; do id=$(basename \"$d\");\n"
            "printf '00:00:00 Speaker A: Local fallback.\\n' > done/$id.txt; rm -rf \"$d\"; done\n"
        )
        worker.chmod(0o755)
        result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(worker))
        assert result.returncode == 0, result.stderr
        assert result.stdout == "00:00:00 Speaker A: Local fallback.\n"
        assert "local" in result.stderr.lower()


class TestClientStdin:
    def test_ratio_transcribe_leaves_the_callers_stdin_alone(self, rig):
        """Boundary: in a `while read` loop the client must not eat the loop's input."""
        env = {
            "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home),
            "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log),
            "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK,
            "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5",
        }
        result = subprocess.run(
            ["bash", "-c", '"$0" "$1" > /dev/null 2>&1; cat', str(SCRIPT), str(rig.audio)],
            env=env, input="next line of the caller's loop\n", capture_output=True, text=True, timeout=60,
        )
        assert result.stdout == "next line of the caller's loop\n"


class TestClientError:
    def test_ratio_transcribe_no_arguments_prints_usage(self, rig):
        """Error: usage."""
        result = rig.run(args=[])
        assert result.returncode == 1 and "Usage: ratio-transcribe" in result.stderr and result.stdout == ""

    def test_ratio_transcribe_missing_file_fails_before_any_ssh(self, rig):
        """Error: no such recording."""
        result = rig.run(args=[str(rig.tmp / "nope.m4a")])
        assert result.returncode == 1 and "not found" in result.stderr
        assert not rig.ssh_log.exists()

    @pytest.mark.parametrize("env", [{"SPEAKERS": "0"}, {"SPEAKERS": "three"}, {"MIN_SPEAKERS": "-1"},
                                      {"MIN_SPEAKERS": "4", "MAX_SPEAKERS": "2"}, {"SPEAKERS": "3", "MAX_SPEAKERS": "5"}])
    def test_ratio_transcribe_bad_speaker_settings_are_rejected(self, rig, env):
        """Error: non-positive, non-numeric or contradictory counts."""
        result = rig.run(**env)
        assert result.returncode == 1 and "speaker" in result.stderr.lower()
        assert not rig.ssh_log.exists()

    def test_ratio_transcribe_bad_language_is_rejected(self, rig):
        """Error: the language ends up in a job file and a command line."""
        result = rig.run(args=[str(rig.audio), "en; rm -rf /"])
        assert result.returncode == 1 and "language" in result.stderr.lower()
        assert not rig.ssh_log.exists()

    def test_ratio_transcribe_failed_job_reports_the_log_and_prints_nothing(self, rig):
        """Error: the host's failure log comes back on stderr."""
        result = rig.run(on_submit=WORKER_FAIL)
        assert result.returncode == 1
        assert result.stdout == ""
        assert "stage: whisper" in result.stderr

    def test_ratio_transcribe_timeout_says_the_job_is_still_running(self, rig):
        """Error: giving up waiting is not the job failing."""
        result = rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0)
        assert result.returncode == 1
        assert result.stdout == ""
        assert "again" in result.stderr.lower()

    def test_ratio_transcribe_empty_transcript_is_a_failure(self, rig):
        """Error: a zero-byte result is never passed off as a transcript."""
        result = rig.run(on_submit=WORKER_EMPTY, TRANSCRIBE_TIMEOUT=1)
        assert result.returncode == 1
        assert result.stdout == ""

    def test_ratio_transcribe_offline_without_local_worker_names_what_is_missing(self, rig):
        """Error: no host and nothing installed locally."""
        result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(rig.tmp / "absent"))
        assert result.returncode == 1
        assert "absent" in result.stderr