STT-Assistant/tests/test_daemon.py

"""Tests for the daemon's worker processing and transcription backlog warning."""

import logging

import numpy as np
import pytest

from stt_daemon import events
from stt_daemon import transcriber as transcriber_mod
from stt_daemon.config import Config
from stt_daemon.daemon import Daemon
from stt_daemon.segmenter import Utterance
from stt_daemon.transcriber import TranscriptResult


def _daemon(threshold=2, queue_size=10):
    cfg = Config()
    cfg.daemon.backlog_warn_at = threshold
    cfg.daemon.queue_size = queue_size
    return Daemon(cfg)


def test_queue_size_comes_from_config():
    daemon = _daemon(queue_size=7)
    assert daemon.segment_queue.maxsize == 7


def test_backlog_warning_is_edge_triggered(caplog):
    daemon = _daemon(threshold=2)

    with caplog.at_level(logging.WARNING, logger="stt_daemon"):
        daemon.segment_queue.put("a")
        daemon._check_backlog()
        assert "backlog" not in caplog.text.lower()

        daemon.segment_queue.put("b")
        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 1

        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 1

        daemon.segment_queue.get()
        daemon.segment_queue.get()
        daemon._check_backlog()
        assert daemon._backlog_warned is False

        daemon.segment_queue.put("c")
        daemon.segment_queue.put("d")
        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 2


class _FakeWhisperModel:
    """Constructor-only fake; keeps faster-whisper from loading a real model."""

    def __init__(self, *args, **kwargs):
        self.init_kwargs = kwargs


class _FakeCppModel:
    """Constructor-only fake; keeps pywhispercpp from loading a real model."""

    def __init__(self, *args, **kwargs):
        self.init_kwargs = kwargs


def test_build_transcriber_defaults_to_faster_whisper(monkeypatch):
    monkeypatch.setattr(transcriber_mod, "WhisperModel", _FakeWhisperModel)
    assert isinstance(_daemon()._build_transcriber(), transcriber_mod.Transcriber)


def test_build_transcriber_honors_pywhispercpp_backend(monkeypatch):
    cpp_model = pytest.importorskip("pywhispercpp.model")
    monkeypatch.setattr(cpp_model, "Model", _FakeCppModel)
    from stt_daemon.pywhispercpp_transcriber import PyWhisperCppTranscriber

    cfg = Config()
    cfg.whisper.backend = "pywhispercpp"
    assert isinstance(Daemon(cfg)._build_transcriber(), PyWhisperCppTranscriber)


# --- worker processing ------------------------------------------------------


class _FakeTranscriber:
    """Returns a canned result without loading a real model."""

    def __init__(self, result):
        self.result = result

    def transcribe(self, audio):
        return self.result

    def close(self):
        pass


class _RaisingSink:
    """Sink that fails on emit, simulating a non-JSON-serializable record."""

    def emit(self, *args, **kwargs):
        raise TypeError("Object of type float32 is not JSON serializable")

    def close(self):
        pass


def _utterance(seg_id=1):
    return Utterance(
        audio=np.zeros(1600, dtype=np.float32),
        duration_s=1.0,
        started_at=0.0,
        rms=0.05,
        reason="pause",
        seg_id=seg_id,
    )


def _transcript_result():
    return TranscriptResult(language="en", language_probability=0.9, text="hello world")


def _daemon_with_raising_sink():
    daemon = _daemon()
    daemon.transcriber = _FakeTranscriber(_transcript_result())
    daemon.sink = _RaisingSink()
    published = []
    daemon._publish = lambda type_, **fields: published.append((type_, fields))
    return daemon, published


def test_sink_failure_still_publishes_transcript(caplog):
    daemon, published = _daemon_with_raising_sink()

    with caplog.at_level(logging.ERROR, logger="stt_daemon"):
        daemon._process(_utterance())

    transcripts = [fields for type_, fields in published if type_ == events.TRANSCRIPT]
    assert len(transcripts) == 1
    assert transcripts[0]["text"] == "hello world"
    assert transcripts[0]["lang_prob"] == 0.9
    assert transcripts[0]["audio"] is None
    assert "Output error" in caplog.text


def test_sink_failure_reports_segment_done_as_emitted():
    daemon, published = _daemon_with_raising_sink()

    daemon._process(_utterance())

    done = [fields for type_, fields in published if type_ == events.SEGMENT_DONE]
    assert len(done) == 1
    assert done[0]["emitted"] is True