"""Unit tests for the WhisperSTT source plumbing.

These do NOT load a model or spawn the worker (that needs the whisper312 venv and
is exercised by the manual TTS->STT round trip). They pin the SpeechToText
contract: how `utterances()` turns an audio source into Utterances, and how it
behaves with no source wired yet.
"""

import asyncio
import json
from typing import AsyncIterator

from wsai.backends.whisper import WhisperSTT
from wsai.interfaces import SpeechToText, Utterance


def _collect(stt: WhisperSTT) -> list[Utterance]:
    async def run():
        return [u async for u in stt.utterances()]

    return asyncio.run(asyncio.wait_for(run(), timeout=5))


async def _paths(items) -> AsyncIterator[str]:
    for it in items:
        yield it


def test_is_speech_to_text():
    assert isinstance(WhisperSTT(), SpeechToText)


def test_no_audio_source_yields_nothing():
    # Discord voice receiver not wired yet -> the loop idles and returns.
    assert _collect(WhisperSTT(audio_source=None)) == []


def test_utterances_transcribes_each_chunk(monkeypatch):
    stt = WhisperSTT(audio_source=_paths(["a.wav", "b.wav"]))

    async def fake_transcribe(wav_path, *, language=None):
        return f"text::{wav_path}"

    monkeypatch.setattr(stt, "transcribe", fake_transcribe)

    utts = _collect(stt)
    assert [u.text for u in utts] == ["text::a.wav", "text::b.wav"]
    assert all(u.source == "voice" for u in utts)


def test_empty_transcript_is_skipped(monkeypatch):
    # Silence / VAD-filtered audio transcribes to "" and must not become a turn.
    stt = WhisperSTT(audio_source=_paths(["silent.wav", "real.wav"]))

    async def fake_transcribe(wav_path, *, language=None):
        return "" if wav_path == "silent.wav" else "안녕"

    monkeypatch.setattr(stt, "transcribe", fake_transcribe)

    utts = _collect(stt)
    assert [u.text for u in utts] == ["안녕"]


def test_request_during_warmup_does_not_overlap_stdout(monkeypatch):
    """Regression: a transcribe() arriving while warmup() is still awaiting the
    worker's ready line must NOT read the same stdout StreamReader concurrently.

    Before the fix, _ensure()'s fast path returned as soon as the subprocess was
    spawned (proc set, returncode None) even though the ready handshake was still
    in flight, so the request's stdout.readline() overlapped warmup's and asyncio
    raised "readuntil() called while another coroutine is already waiting for
    incoming data" — the exact crash seen in the Discord voice server."""

    async def run():
        stt = WhisperSTT()
        stdout = asyncio.StreamReader()
        stderr = asyncio.StreamReader()
        stderr.feed_eof()  # nothing on stderr; let the drain task finish cleanly

        class FakeStdin:
            def write(self, _b):
                pass

            async def drain(self):
                pass

        class FakeProc:
            returncode = None

            def __init__(self):
                self.stdin = FakeStdin()
                self.stdout = stdout
                self.stderr = stderr

            def terminate(self):
                self.returncode = 0

            async def wait(self):
                return 0

        spawns = []

        async def fake_create(*_a, **_k):
            spawns.append(1)
            return FakeProc()

        monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create)

        # warmup enters _ensure and blocks awaiting the ready line on stdout.
        warm = asyncio.create_task(stt.warmup())
        await asyncio.sleep(0.05)

        # A concurrent request lands mid-warmup. It must wait for readiness, not
        # crash and not read stdout yet.
        tr = asyncio.create_task(stt.transcribe("x.wav"))
        await asyncio.sleep(0.05)
        assert not tr.done()  # blocked on the start lock, no overlapping read

        # Complete the handshake -> warmup finishes and releases the request.
        stdout.feed_data(
            (json.dumps({"ready": True, "ms": 1, "device": "cpu"}) + "\n").encode()
        )
        await asyncio.wait_for(warm, timeout=1)
        await asyncio.sleep(0.02)
        stdout.feed_data(
            (json.dumps({"ok": True, "text": "안녕", "ms": 2}) + "\n").encode()
        )
        assert await asyncio.wait_for(tr, timeout=1) == "안녕"
        assert sum(spawns) == 1  # one worker, not one-per-concurrent-caller

        await stt.aclose()

    asyncio.run(asyncio.wait_for(run(), timeout=5))