STT-Assistant/tests/test_triage_dump.py

"""Tests for the opt-in triage I/O dump (TriageDumper)."""

import json
import re
from datetime import datetime

from stt_daemon.assistant.triage_dump import TriageDumper

TS = 1759073916.19


def _only_file(directory):
    files = list(directory.rglob("*.json"))
    assert len(files) == 1
    return files[0]


def test_enabled_writes_one_shaped_file(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(
        TS,
        laya={"request": {"state": "s"}, "response": {"ok": True}, "error": None},
        llm={"request": {"url": "u", "body": {}}, "response": {"c": 1}, "error": None},
    )
    dumper.finalize(TS)

    path = _only_file(tmp_path)
    assert re.fullmatch(r"\d{6}-\d{3}\.json", path.name)
    expected_day = datetime.fromtimestamp(TS).strftime("%Y-%m-%d")
    assert path.parent.name == expected_day

    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["ts"] == TS
    assert isinstance(data["iso"], str) and data["iso"]
    assert data["laya"] == {
        "request": {"state": "s"},
        "response": {"ok": True},
        "error": None,
    }
    assert data["llm"] == {
        "request": {"url": "u", "body": {}},
        "response": {"c": 1},
        "error": None,
    }


def test_disabled_writes_nothing(tmp_path):
    dumper = TriageDumper(False, directory=str(tmp_path))
    assert dumper.enabled is False
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.record(TS, llm={"request": {}, "response": {}, "error": None})
    dumper.finalize(TS)
    dumper.clear()

    assert list(tmp_path.rglob("*")) == []


def test_record_merges_without_dropping_earlier_parts(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(
        TS, laya={"request": {"state": "s"}, "response": None, "error": None}
    )
    dumper.record(TS, llm={"request": {"url": "u"}, "response": None, "error": None})
    dumper.record(TS, laya={"response": {"ok": True}, "error": None})
    dumper.finalize(TS)

    data = json.loads(_only_file(tmp_path).read_text(encoding="utf-8"))
    assert data["laya"]["request"] == {"state": "s"}
    assert data["laya"]["response"] == {"ok": True}
    assert data["llm"]["request"] == {"url": "u"}


def test_llm_response_phase_keeps_request(tmp_path):
    """The response-phase record must not wipe the captured request."""
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.record(
        TS,
        llm={"request": {"url": "u", "body": {"m": 1}}, "response": None, "error": None},
    )
    dumper.record(TS, llm={"request": None, "response": {"c": 1}, "error": None})
    dumper.finalize(TS)

    data = json.loads(_only_file(tmp_path).read_text(encoding="utf-8"))
    assert data["llm"]["request"] == {"url": "u", "body": {"m": 1}}
    assert data["llm"]["response"] == {"c": 1}
    assert data["llm"]["error"] is None


def test_finalize_unknown_ts_is_noop(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.finalize(1234567890.0)

    assert list(tmp_path.rglob("*")) == []


def test_clear_drops_pending(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.clear()
    dumper.finalize(TS)

    assert list(tmp_path.rglob("*")) == []


def test_write_failure_is_swallowed(tmp_path):
    blocker = tmp_path / "blocker"
    blocker.write_text("not a directory")
    dumper = TriageDumper(True, directory=str(blocker / "sub"))

    # Must not raise even though no directory can be created here.
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.finalize(TS)

    assert blocker.is_file()


def test_discard_drops_pending_without_writing(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.discard(TS)
    dumper.finalize(TS)

    assert list(tmp_path.rglob("*")) == []


def test_discard_unknown_and_disabled_is_noop(tmp_path):
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.discard(1234567890.0)

    assert list(tmp_path.rglob("*")) == []

    disabled = TriageDumper(False, directory=str(tmp_path))
    disabled.discard(TS)
    disabled.finalize(TS)

    assert list(tmp_path.rglob("*")) == []