STT-Assistant/tests/test_output.py

"""Tests for transcript output sinks."""

import json
import wave

import numpy as np

from stt_daemon.output import TranscriptSink
from stt_daemon.transcriber import TranscriptResult


def _result(text="hello world"):
    return TranscriptResult(language="en", language_probability=0.98, text=text)


def test_jsonl_includes_level_and_reason(tmp_path):
    sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path))
    sink.emit(_result(), 2.5, rms=0.041234, reason="pause")

    files = list(tmp_path.glob("*.jsonl"))
    assert len(files) == 1

    record = json.loads(files[0].read_text().splitlines()[0])
    assert record["rms"] == round(0.041234, 5)
    assert record["end_reason"] == "pause"
    assert record["duration_s"] == 2.5
    assert record["text"] == "hello world"
    assert record["lang"] == "en"


def test_stdout_includes_level_and_reason(capsys):
    sink = TranscriptSink(stdout=True, jsonl_dir=None)
    sink.emit(_result(), 1.0, rms=0.02, reason="end_of_turn")

    out = capsys.readouterr().out
    assert "rms=0.020" in out
    assert "end_of_turn" in out
    assert "hello world" in out


def test_empty_text_is_skipped(tmp_path, capsys):
    sink = TranscriptSink(stdout=True, jsonl_dir=str(tmp_path))
    sink.emit(_result(text=""), 1.0, rms=0.02, reason="pause")

    assert capsys.readouterr().out == ""
    assert list(tmp_path.glob("*.jsonl")) == []


def test_saves_wav_when_enabled(tmp_path):
    audio = np.zeros(16000, dtype=np.float32)
    audio[:8000] = 0.5
    logs = tmp_path / "logs"

    sink = TranscriptSink(
        stdout=False,
        jsonl_dir=str(logs),
        save_audio=True,
        audio_dir=str(tmp_path / "audio"),
        sample_rate=16000,
    )
    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=audio)

    wavs = list((tmp_path / "audio").rglob("*.wav"))
    assert len(wavs) == 1

    with wave.open(str(wavs[0]), "rb") as wf:
        assert wf.getnchannels() == 1
        assert wf.getsampwidth() == 2
        assert wf.getframerate() == 16000
        assert wf.getnframes() == 16000

    record = json.loads(next(logs.glob("*.jsonl")).read_text().splitlines()[0])
    assert record["audio"] == str(wavs[0])


def test_no_wav_when_disabled(tmp_path):
    sink = TranscriptSink(
        stdout=False,
        jsonl_dir=None,
        save_audio=False,
        audio_dir=str(tmp_path / "audio"),
        sample_rate=16000,
    )
    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32))

    assert not (tmp_path / "audio").exists()


def test_save_audio_without_samples_is_safe(tmp_path):
    sink = TranscriptSink(
        stdout=False,
        jsonl_dir=None,
        save_audio=True,
        audio_dir=str(tmp_path / "audio"),
        sample_rate=16000,
    )
    sink.emit(_result(), 1.0, rms=0.3, reason="pause", audio=None)

    assert not (tmp_path / "audio").exists()


def test_failed_audio_save_is_non_fatal(tmp_path, monkeypatch, capsys, caplog):
    import stt_daemon.output as output

    def boom(*args, **kwargs):
        raise OSError("disk full")

    monkeypatch.setattr(output, "write_wav", boom)

    logs = tmp_path / "logs"
    sink = TranscriptSink(
        stdout=True,
        jsonl_dir=str(logs),
        save_audio=True,
        audio_dir=str(tmp_path / "audio"),
        sample_rate=16000,
    )
    with caplog.at_level("WARNING"):
        sink.emit(
            _result(), 1.0, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32)
        )

    assert "hello world" in capsys.readouterr().out
    record = json.loads(next(logs.glob("*.jsonl")).read_text().splitlines()[0])
    assert record["text"] == "hello world"
    assert "audio" not in record
    assert "Failed to save audio" in caplog.text