STT-Assistant/tests/test_brief_file.py

"""Tests for the atomic best-effort durable brief writer (BriefFile)."""

import os
import threading

from stt_daemon.assistant.brief_file import BriefFile


def _path(tmp_path):
    return tmp_path / "brief.md"


def _tmps(tmp_path):
    return [p.name for p in tmp_path.iterdir() if p.suffix == ".tmp"]


def test_write_enabled_property(tmp_path):
    assert BriefFile(True, str(tmp_path)).write_enabled is True
    assert BriefFile(False, str(tmp_path)).write_enabled is False


def test_disabled_write_and_clear_create_no_file(tmp_path):
    # misleading-success probe: disabled means the path must never appear.
    brief = BriefFile(False, str(tmp_path))
    brief.write("hello")
    assert not _path(tmp_path).exists()
    brief.clear()
    assert not _path(tmp_path).exists()
    assert not any(tmp_path.iterdir())


def test_enabled_write_writes_exact_text(tmp_path):
    brief = BriefFile(True, str(tmp_path))
    brief.write("hello")
    assert _path(tmp_path).read_text(encoding="utf-8") == "hello"
    assert _path(tmp_path).read_bytes() == b"hello"
    assert _tmps(tmp_path) == []


def test_second_write_replaces_atomically_no_tmp(tmp_path):
    # stale-state probe: only the newest text survives, no .tmp leaks.
    brief = BriefFile(True, str(tmp_path))
    brief.write("first")
    brief.write("second")
    assert _path(tmp_path).read_text(encoding="utf-8") == "second"
    assert _tmps(tmp_path) == []


def test_clear_empties_file(tmp_path):
    brief = BriefFile(True, str(tmp_path))
    brief.write("content")
    brief.clear()
    assert _path(tmp_path).read_text(encoding="utf-8") == ""
    assert _path(tmp_path).exists()


def test_write_failure_path_swallowed_no_raise(tmp_path):
    # failure probe: parent is a regular file, so makedirs/open fail -> swallow.
    blocker = tmp_path / "blocker"
    blocker.write_text("x", encoding="utf-8")
    brief = BriefFile(True, str(blocker / "nested"))
    brief.write("hello")  # must not raise
    assert not os.path.exists(str(blocker / "nested"))


def test_clear_failure_path_swallowed_no_raise(tmp_path):
    brief = BriefFile(True, str(tmp_path / "missing"))
    brief.clear()  # missing dir -> open fails -> swallowed, no raise


def test_default_directory_is_brief(tmp_path, monkeypatch):
    # constructor default: `brief/brief.md` relative to the cwd.
    monkeypatch.chdir(tmp_path)
    brief = BriefFile(True)
    brief.write("x")
    assert (tmp_path / "brief" / "brief.md").read_text(encoding="utf-8") == "x"


def test_set_directory_redirects_writes(tmp_path):
    brief = BriefFile(True)
    brief.set_directory(str(tmp_path))
    brief.write("x")
    assert (tmp_path / "brief.md").read_text(encoding="utf-8") == "x"


def test_disabled_after_set_directory_writes_nothing(tmp_path):
    brief = BriefFile(False)
    brief.set_directory(str(tmp_path))
    brief.write("x")
    brief.clear()
    assert not any(tmp_path.iterdir())


def test_concurrent_set_directory_and_write_stay_within_dirs(tmp_path):
    # race probe: a writer flipping between two dirs must never land elsewhere.
    dir_a = tmp_path / "a"
    dir_b = tmp_path / "b"
    brief = BriefFile(True)
    brief.set_directory(str(dir_a))
    stop = threading.Event()

    def swapper():
        while not stop.is_set():
            brief.set_directory(str(dir_a))
            brief.set_directory(str(dir_b))

    def writer():
        for _ in range(500):
            brief.write("x")
        stop.set()

    threads = [threading.Thread(target=swapper), threading.Thread(target=writer)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    allowed = {dir_a.resolve(), dir_b.resolve()}
    produced = [p for p in tmp_path.rglob("*") if p.is_file()]
    assert produced, "writer should have produced at least one file"
    for path in produced:
        assert path.resolve().parent in allowed, f"write escaped to {path}"
    assert not (tmp_path / "brief").exists()
    assert not (tmp_path / "brief.md").exists()