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_same_millisecond_collision_keeps_first_shape(tmp_path):
    """Same-millisecond records all land; only repeats get a `-N` suffix."""
    dumper = TriageDumper(True, directory=str(tmp_path))
    for n in (1, 2, 3):
        dumper.begin(TS, laya={"request": {}, "response": {"n": n}, "error": None})
        dumper.finalize(TS)

    dt = datetime.fromtimestamp(TS)
    base = dt.strftime("%H%M%S") + "-" + dt.strftime("%f")[:3]
    files = {path.name: path for path in tmp_path.rglob("*.json")}
    assert set(files) == {f"{base}.json", f"{base}-1.json", f"{base}-2.json"}
    assert re.fullmatch(r"\d{6}-\d{3}\.json", f"{base}.json")
    first = json.loads(files[f"{base}.json"].read_text(encoding="utf-8"))
    second = json.loads(files[f"{base}-1.json"].read_text(encoding="utf-8"))
    third = json.loads(files[f"{base}-2.json"].read_text(encoding="utf-8"))
    assert first["laya"]["response"] == {"n": 1}
    assert second["laya"]["response"] == {"n": 2}
    assert third["laya"]["response"] == {"n": 3}


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_llm_duration_ms_survives_response_merge(tmp_path):
    """The response-phase record carries duration_ms alongside the payload."""
    dumper = TriageDumper(True, directory=str(tmp_path))
    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
    dumper.record(
        TS, llm={"request": {"url": "u"}, "response": None, "error": None}
    )
    dumper.record(
        TS,
        llm={
            "request": None,
            "response": {"c": 1},
            "error": None,
            "duration_ms": 812.4,
        },
    )
    dumper.finalize(TS)

    data = json.loads(_only_file(tmp_path).read_text(encoding="utf-8"))
    assert data["llm"]["duration_ms"] == 812.4
    assert data["llm"]["request"] == {"url": "u"}


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("*")) == []