STT-Assistant/tests/test_audio.py

"""Tests for audio capture backends using fakes (no hardware, no subprocesses)."""

import queue

import pytest

from stt_daemon.audio import ParecCapture, SoundDeviceCapture
from stt_daemon.config import AudioConfig


class FakeStdout:
    def __init__(self, chunks):
        self._chunks = list(chunks)

    def read(self, _n):
        return self._chunks.pop(0) if self._chunks else b""


class FakeProc:
    def __init__(self, chunks=()):
        self.stdout = FakeStdout(chunks)
        self.terminated = 0
        self.killed = 0
        self.waits = 0

    def terminate(self):
        self.terminated += 1

    def wait(self, timeout=None):
        self.waits += 1
        return 0

    def kill(self):
        self.killed += 1

    def poll(self):
        return None


class FakeStream:
    def __init__(self):
        self.stopped = 0
        self.closed = 0
        self.active = True

    def stop(self):
        self.stopped += 1
        self.active = False

    def close(self):
        self.closed += 1


def _parec(chunks=()):
    cap = ParecCapture.__new__(ParecCapture)
    cap.cfg = AudioConfig()
    cap._frame_bytes = 4
    cap._proc = FakeProc(chunks)
    return cap


def _sounddevice():
    cap = SoundDeviceCapture.__new__(SoundDeviceCapture)
    cap.cfg = AudioConfig()
    cap._queue = queue.Queue()
    cap._read_timeout_s = 0.05
    cap._stream = FakeStream()
    return cap


def test_parec_frames_yield_until_eof():
    cap = _parec(chunks=[b"abcd", b"efgh", b""])
    gen = cap.frames()
    assert next(gen) == b"abcd"
    assert next(gen) == b"efgh"
    with pytest.raises(RuntimeError, match="parec EOF"):
        next(gen)


def test_parec_short_read_raises():
    cap = _parec(chunks=[b"ab"])
    with pytest.raises(RuntimeError, match="short read"):
        next(cap.frames())


def test_parec_close_is_idempotent():
    cap = _parec()
    proc = cap._proc
    cap.close()
    cap.close()
    assert proc.terminated == 1
    assert proc.waits == 1
    assert cap.healthy() is False


def test_sounddevice_frames_yield_queued_frames():
    cap = _sounddevice()
    cap._queue.put(b"frame1")
    gen = cap.frames()
    assert next(gen) == b"frame1"


def test_sounddevice_stall_raises_instead_of_silent_exit():
    cap = _sounddevice()
    with pytest.raises(RuntimeError, match="stalled"):
        next(cap.frames())


def test_sounddevice_close_is_idempotent():
    cap = _sounddevice()
    stream = cap._stream
    cap.close()
    cap.close()
    assert stream.stopped == 1
    assert stream.closed == 1
    assert cap.healthy() is False