"""Tests for stt_daemon.debug.DebugSink (per-file fakes, no conftest.py).""" import json import logging import os import threading from datetime import datetime from pathlib import Path import pytest from stt_daemon.debug import DebugSink def _read_lines(path): with open(path, encoding="utf-8") as f: return [json.loads(line) for line in f if line.strip()] def test_emit_writes_valid_jsonl_per_day(tmp_path): sink = DebugSink(str(tmp_path / "debug"), "daemon") sink.emit("heartbeat", queue_depth=0, device_ok=True) sink.emit("capture_reopen", attempt=1) day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert len(records) == 2 assert records[0]["proc"] == "daemon" assert records[0]["type"] == "heartbeat" assert records[0]["queue_depth"] == 0 assert records[0]["device_ok"] is True assert "ts" in records[0] assert records[1]["type"] == "capture_reopen" def test_forbidden_fields_never_persisted(tmp_path): sink = DebugSink(str(tmp_path / "debug"), "assistant") sink.emit("heartbeat", text="secret transcript", audio="clip.wav", prompt="sys", rms=0.1) day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert len(records) == 1 assert "text" not in records[0] assert "audio" not in records[0] assert "prompt" not in records[0] assert records[0]["rms"] == 0.1 def test_invalid_proc_rejected(tmp_path): with pytest.raises(ValueError): DebugSink(str(tmp_path), "worker") @pytest.mark.parametrize("proc", ["daemon", "assistant", "supervisor"]) def test_all_procs_accepted(tmp_path, proc): sink = DebugSink(str(tmp_path / proc), proc) sink.emit("heartbeat") day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / proc / f"{day}.jsonl") assert records[0]["proc"] == proc def test_disabled_path_creates_no_files(tmp_path): # Disabled path = caller holds None instead of a sink (zero overhead). debug = None target = tmp_path / "debug" if debug is not None: debug.emit("heartbeat") assert not target.exists() def test_heartbeat_gating_respected_by_caller_contract(tmp_path): # Gating lives with the caller: heartbeat lines are skipped unless enabled, # anomaly lines always flow. class FakeCaller: def __init__(self, sink, heartbeat): self.sink = sink self.heartbeat = heartbeat def tick(self, is_heartbeat, type, **fields): if is_heartbeat and not self.heartbeat: return self.sink.emit(type, **fields) sink = DebugSink(str(tmp_path / "debug"), "daemon") caller = FakeCaller(sink, heartbeat=False) caller.tick(True, "heartbeat", queue_depth=0) caller.tick(False, "capture_reopen", attempt=1) day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert [r["type"] for r in records] == ["capture_reopen"] caller.heartbeat = True caller.tick(True, "heartbeat", queue_depth=0) records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert [r["type"] for r in records] == ["capture_reopen", "heartbeat"] def test_mkdir_failure_disables_without_raising(tmp_path, caplog): blocker = tmp_path / "blocker" blocker.write_text("not a dir") with caplog.at_level(logging.WARNING): sink = DebugSink(str(blocker / "debug"), "daemon") assert sink._disabled assert "mkdir failed" in caplog.text sink.emit("heartbeat") # no-op, must not raise def test_open_failure_warns_not_raises_and_breaker_latches(tmp_path, caplog, monkeypatch): sink = DebugSink(str(tmp_path / "debug"), "daemon") def _boom(path, flags, mode=0o644): raise OSError("disk gone") monkeypatch.setattr(os, "open", _boom) with caplog.at_level(logging.WARNING): sink.emit("heartbeat", queue_depth=1) assert "write failure" in caplog.text # Breaker latched: even with os.open restored, emits are silent no-ops. monkeypatch.undo() caplog.clear() with caplog.at_level(logging.WARNING): sink.emit("heartbeat", queue_depth=2) assert caplog.text == "" day = datetime.now().strftime("%Y-%m-%d") assert not (tmp_path / "debug" / f"{day}.jsonl").exists() def test_write_failure_warns_once_then_latches(tmp_path, caplog, monkeypatch): sink = DebugSink(str(tmp_path / "debug"), "daemon") sink.emit("heartbeat", seq=0) calls = {"n": 0} real_write = os.write def _flaky(fd, data): calls["n"] += 1 if calls["n"] == 1: raise OSError("transient") return real_write(fd, data) monkeypatch.setattr(os, "write", _flaky) with caplog.at_level(logging.WARNING): sink.emit("heartbeat", seq=1) assert "write failure" in caplog.text monkeypatch.undo() caplog.clear() with caplog.at_level(logging.WARNING): sink.emit("heartbeat", seq=2) assert caplog.text == "" day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert [r.get("seq") for r in records] == [0] def test_concurrent_emits_produce_no_partial_lines(tmp_path): sink = DebugSink(str(tmp_path / "debug"), "daemon") threads = 8 per_thread = 100 def _work(tid): for i in range(per_thread): sink.emit("heartbeat", tid=tid, seq=i, pad="x" * 200) workers = [threading.Thread(target=_work, args=(t,)) for t in range(threads)] for w in workers: w.start() for w in workers: w.join() day = datetime.now().strftime("%Y-%m-%d") path = tmp_path / "debug" / f"{day}.jsonl" with open(path, encoding="utf-8") as f: raw = f.readlines() assert len(raw) == threads * per_thread seen = set() for line in raw: record = json.loads(line) # raises on any partial/interleaved line seen.add((record["tid"], record["seq"])) assert len(seen) == threads * per_thread def test_cap_skips_past_limit_with_single_warning(tmp_path, caplog): sink = DebugSink(str(tmp_path / "debug"), "daemon", max_bytes_per_day=200) with caplog.at_level(logging.WARNING): for i in range(20): sink.emit("heartbeat", seq=i, pad="y" * 50) warnings = [r for r in caplog.records if "cap reached" in r.message] assert len(warnings) == 1 day = datetime.now().strftime("%Y-%m-%d") size = Path(tmp_path / "debug" / f"{day}.jsonl").stat().st_size assert size <= 200 def test_cap_counter_resets_on_rollover(tmp_path): clock = {"now": datetime(2026, 9, 29, 23, 59, 59)} sink = DebugSink(str(tmp_path / "debug"), "daemon", max_bytes_per_day=200, now_fn=lambda: clock["now"]) for i in range(20): # exhaust day one's cap sink.emit("heartbeat", seq=i, pad="z" * 50) clock["now"] = datetime(2026, 9, 30, 0, 0, 1) sink.emit("heartbeat", seq="next-day") records = _read_lines(tmp_path / "debug" / "2026-09-30.jsonl") assert [r.get("seq") for r in records] == ["next-day"] def test_close_is_noop(tmp_path): sink = DebugSink(str(tmp_path / "debug"), "supervisor") sink.emit("spawn", pid=123) sink.close() sink.emit("exit", code=0) day = datetime.now().strftime("%Y-%m-%d") records = _read_lines(tmp_path / "debug" / f"{day}.jsonl") assert [r["type"] for r in records] == ["spawn", "exit"]