"""Tests for transcript output sinks.""" import json import re import wave import numpy as np import pytest 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): session = tmp_path / "2026-09-30-12-47-29" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 2.5, rms=0.041234, reason="pause", session_dir=session) record = json.loads((session / "transcripts.jsonl").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_jsonl_includes_transcribe_seconds(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 2.5, session_dir=session, rms=0.041234, transcribe_s=0.81234) record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["transcribe_s"] == round(0.81234, 3) def test_transcribe_seconds_omitted_when_unknown(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 2.5, session_dir=session, rms=0.041234, reason="pause") record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert "transcribe_s" not in record def test_jsonl_disabled_writes_nothing(tmp_path, capsys): """jsonl=False keeps transcripts off disk even when a session dir is given.""" sink = TranscriptSink(stdout=True, jsonl=False) sink.emit(_result(), 1.0, session_dir=tmp_path / "sess", rms=0.02, reason="pause") assert "hello world" in capsys.readouterr().out assert not (tmp_path / "sess" / "transcripts.jsonl").exists() def test_stdout_includes_level_and_reason(capsys): sink = TranscriptSink(stdout=True) 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): session = tmp_path / "sess" sink = TranscriptSink(stdout=True, jsonl=True) sink.emit(_result(text=""), 1.0, session_dir=session, rms=0.02, reason="pause") assert capsys.readouterr().out == "" assert not (session / "transcripts.jsonl").exists() def test_jsonl_without_session_dir_logs_warning(tmp_path, caplog): sink = TranscriptSink(stdout=False, jsonl=True) with caplog.at_level("WARNING"): sink.emit(_result(), 1.0, rms=0.02, reason="pause", session_dir=None) assert list(tmp_path.rglob("transcripts.jsonl")) == [] assert "no session_dir" in caplog.text def test_jsonl_records_session_id(tmp_path): session = tmp_path / "2026-09-30-12-47-29" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 1.0, session_dir=session, rms=0.02, reason="pause") record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["session_id"] == "2026-09-30-12-47-29" def test_saves_wav_when_enabled(tmp_path): audio = np.zeros(16000, dtype=np.float32) audio[:8000] = 0.5 session = tmp_path / "2026-09-30-12-47-29" sink = TranscriptSink(stdout=False, jsonl=True, save_audio=True, sample_rate=16000) sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=audio) wavs = list((session / "audio").glob("*.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((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["audio"] == str(wavs[0]) def test_session_wav_name_and_location(tmp_path): session = tmp_path / "2026-09-30-12-47-29" audio = np.zeros(160, dtype=np.float32) sink = TranscriptSink(stdout=False, save_audio=True, sample_rate=16000) sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=audio) wavs = list((session / "audio").glob("*.wav")) assert len(wavs) == 1 assert re.fullmatch(r"\d{6}-\d{3}-pause\.wav", wavs[0].name) assert wavs[0].parent == session / "audio" def test_single_timestamp_per_emit(tmp_path, monkeypatch): """All artifacts of one emit() share one timestamp, even across midnight.""" import stt_daemon.output as output from datetime import datetime as real_datetime calls = [] t1 = real_datetime(2026, 9, 20, 23, 59, 59) t2 = real_datetime(2026, 9, 21, 0, 0, 1) class FakeDatetime: @staticmethod def now(): calls.append(1) return t1 if len(calls) == 1 else t2 monkeypatch.setattr(output, "datetime", FakeDatetime) session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True, save_audio=True, sample_rate=16000) sink.emit( _result(), 1.0, session_dir=session, rms=0.1, reason="pause", audio=np.zeros(160, dtype=np.float32), ) assert len(calls) == 1 record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["ts"].startswith("2026-09-20") wav = next((session / "audio").glob("*.wav")) assert wav.name.startswith("235959-") def test_no_wav_when_disabled(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, save_audio=False, sample_rate=16000) sink.emit( _result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32), ) assert not (session / "audio").exists() def test_save_audio_without_samples_is_safe(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, save_audio=True, sample_rate=16000) sink.emit(_result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=None) assert not (session / "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) session = tmp_path / "sess" sink = TranscriptSink(stdout=True, jsonl=True, save_audio=True, sample_rate=16000) with caplog.at_level("WARNING"): sink.emit( _result(), 1.0, session_dir=session, rms=0.3, reason="pause", audio=np.zeros(1600, dtype=np.float32), ) assert "hello world" in capsys.readouterr().out record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["text"] == "hello world" assert "audio" not in record assert "Failed to save audio" in caplog.text def test_failed_serialization_leaves_jsonl_untouched(tmp_path): """A record that cannot be serialized must not leave a partial line behind.""" session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 2.5, session_dir=session, rms=0.04, reason="pause") path = session / "transcripts.jsonl" before = path.read_text() bad = TranscriptResult(language="en", language_probability=np.float32(0.98), text="boom") with pytest.raises(TypeError): sink.emit(bad, 2.5, session_dir=session, rms=0.04, reason="pause") assert path.read_text() == before def test_jsonl_includes_speaker_fields(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit( _result(), 3.2, session_dir=session, rms=0.031, reason="pause", seg_id=142, speaker_id="spk_0001", speaker_status="matched", speaker="Efren", speaker_score=0.712345, ) record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["seg_id"] == 142 assert record["speaker_id"] == "spk_0001" assert record["speaker_status"] == "matched" assert record["speaker"] == "Efren" assert record["speaker_score"] == round(0.712345, 4) def test_jsonl_speaker_fields_default_to_nulls(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 2.5, session_dir=session, rms=0.04, reason="pause") record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["seg_id"] is None assert record["speaker_id"] is None assert record["speaker_status"] == "none" assert record["speaker"] is None assert record["speaker_score"] is None def test_jsonl_speaker_score_rounded_to_4dp(tmp_path): session = tmp_path / "sess" sink = TranscriptSink(stdout=False, jsonl=True) sink.emit(_result(), 1.0, session_dir=session, rms=0.02, reason="pause", speaker_score=0.123456) record = json.loads((session / "transcripts.jsonl").read_text().splitlines()[0]) assert record["speaker_score"] == round(0.123456, 4)