test_output.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. """Tests for transcript output sinks."""
  2. import json
  3. from stt_daemon.output import TranscriptSink
  4. from stt_daemon.transcriber import TranscriptResult
  5. def _result(text="hello world"):
  6. return TranscriptResult(language="en", language_probability=0.98, text=text)
  7. def test_jsonl_includes_level_and_reason(tmp_path):
  8. sink = TranscriptSink(stdout=False, jsonl_dir=str(tmp_path))
  9. sink.emit(_result(), 2.5, rms=0.041234, reason="pause")
  10. files = list(tmp_path.glob("*.jsonl"))
  11. assert len(files) == 1
  12. record = json.loads(files[0].read_text().splitlines()[0])
  13. assert record["rms"] == round(0.041234, 5)
  14. assert record["end_reason"] == "pause"
  15. assert record["duration_s"] == 2.5
  16. assert record["text"] == "hello world"
  17. assert record["lang"] == "en"
  18. def test_stdout_includes_level_and_reason(capsys):
  19. sink = TranscriptSink(stdout=True, jsonl_dir=None)
  20. sink.emit(_result(), 1.0, rms=0.02, reason="end_of_turn")
  21. out = capsys.readouterr().out
  22. assert "rms=0.020" in out
  23. assert "end_of_turn" in out
  24. assert "hello world" in out
  25. def test_empty_text_is_skipped(tmp_path, capsys):
  26. sink = TranscriptSink(stdout=True, jsonl_dir=str(tmp_path))
  27. sink.emit(_result(text=""), 1.0, rms=0.02, reason="pause")
  28. assert capsys.readouterr().out == ""
  29. assert list(tmp_path.glob("*.jsonl")) == []