arm-scan-gateway/api/tests/unit/test_local_hotfolder.py

"""Горячая папка в обычном каталоге (spec image-recognition-gateway, D-1, R-4/R-5, slice-0-skeleton).

Быстрые тесты на `tmp_path`: атомарная запись входящего файла через служебный каталог `Tmp`, опрос набора ответов ИС
с готовностью по стабилизации (settle, часы подменены), проверка здоровья, валидация имён.
"""

import os
import uuid
from pathlib import Path
from typing import Any

import pytest

from src.gateway.hotfolder import AnswerFile, HotFolder, LocalHotFolder, SettleTracker, inbox_name, validate_name


def _job_id() -> str:
    return str(uuid.uuid4())


class FakeClock:
    """Монотонные часы теста: время двигается только вручную."""

    def __init__(self) -> None:
        self.now = 100.0

    def __call__(self) -> float:
        return self.now


def _write_answer(hf: LocalHotFolder, job_id: str, data: bytes, name: str | None = None) -> Path:
    path = hf.answer_dir(job_id) / (name or f"{job_id}.json")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    return path


def _settled(tmp_path: Path, settle: float = 2) -> tuple[LocalHotFolder, FakeClock]:
    clock = FakeClock()
    return LocalHotFolder(tmp_path, settle_seconds=settle, clock=clock), clock


class TestInboxName:
    def test_joins_job_id_and_name(self) -> None:
        job_id = _job_id()
        assert inbox_name(job_id, "aep_5.PDF") == f"{job_id}_aep_5.PDF"

    @pytest.mark.parametrize(
        ("original", "expected"),
        [
            ("../../etc/passwd", "passwd"),
            ("dir/sub/report.pdf", "report.pdf"),
            ("C:\\scans\\report.pdf", "report.pdf"),
            ("/abs/report.pdf", "report.pdf"),
            ("..\\..\\report.pdf", "report.pdf"),
        ],
    )
    def test_sanitizes_to_basename(self, original: str, expected: str) -> None:
        assert inbox_name("job", original) == f"job_{expected}"

    @pytest.mark.parametrize("original", ["", "/", "..", ".", "a/..", "dir/", "\\", "...", ". .", " ", "dir/. "])
    def test_rejects_unusable_names(self, original: str) -> None:
        with pytest.raises(ValueError, match="имя файла"):
            inbox_name("job", original)

    @pytest.mark.parametrize(
        ("original", "expected"),
        [
            ("a:b?.pdf", "a_b_.pdf"),
            ("Отчёт: 2024.pdf", "Отчёт_ 2024.pdf"),
            ('Баланс "итог".pdf', "Баланс _итог_.pdf"),
            ("форма?.pdf", "форма_.pdf"),
            ("a|b<c>*.pdf", "a_b_c__.pdf"),
            ("C:report.pdf", "C_report.pdf"),
            ("a\x01b.pdf", "a_b.pdf"),
            ("bad\x00name.pdf", "bad_name.pdf"),
            ("tab\tname.pdf", "tab_name.pdf"),
        ],
    )
    def test_replaces_chars_illegal_on_windows(self, original: str, expected: str) -> None:
        """Имя от пользователя законно на Linux/S3/macOS: заменяем, а не отвергаем (контракт: только minLength 1)."""
        assert inbox_name("job", original) == f"job_{expected}"

    @pytest.mark.parametrize(
        ("original", "expected"),
        [("report.pdf.", "report.pdf"), ("report.pdf . ", "report.pdf"), ("report .pdf", "report .pdf")],
    )
    def test_strips_trailing_dots_and_spaces(self, original: str, expected: str) -> None:
        assert inbox_name("job", original) == f"job_{expected}"

    def test_keeps_spaces_and_cyrillic(self) -> None:
        job_id = _job_id()
        assert inbox_name(job_id, "Отчёт за 2024 год (копия).PDF") == f"{job_id}_Отчёт за 2024 год (копия).PDF"

    @pytest.mark.parametrize(
        "original",
        [
            "a:b?.pdf",
            'Баланс "итог".pdf',
            "Отчёт за 2024 год (копия).PDF",
            ".hidden.pdf",
            "..pdf",
            "x\x00\x1f<>|*?.pdf. ",
            "C:\\dir\\..\\скан.png",
            " leading.pdf",
        ],
    )
    def test_result_always_passes_validate_name_and_put(self, tmp_path: Path, original: str) -> None:
        hf = LocalHotFolder(tmp_path)
        name = inbox_name(_job_id(), original)
        validate_name(name)
        hf.put(name, b"x")
        assert (hf.input_dir / name).read_bytes() == b"x"

    def test_result_is_accepted_by_put(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        name = inbox_name(_job_id(), "../../scan.pdf")
        hf.put(name, b"x")
        assert (hf.input_dir / name).read_bytes() == b"x"


class TestInboxNameLength:
    """T7.3: длинное имя укорачивается так, чтобы `.<job_id>_<имя>.part` занимал не больше 240 байт UTF-8 (ext4 —
    255 байт, SMB/NTFS — 255 единиц UTF-16), расширение (до 16 символов после последней точки) сохраняется."""

    @staticmethod
    def _part_bytes(name: str) -> int:
        return len(f".{name}.part".encode())

    def test_long_cyrillic_name_is_trimmed_and_put(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        job_id = _job_id()
        name = inbox_name(job_id, "я" * 200 + ".pdf")
        assert self._part_bytes(name) <= 240
        assert name.startswith(f"{job_id}_яя")
        assert name.endswith("я.pdf")
        hf.put(name, b"x")
        assert (hf.input_dir / name).read_bytes() == b"x"

    def test_short_name_is_unchanged(self) -> None:
        job_id = _job_id()
        original = "Отчёт за 2024 год (копия).PDF"
        assert inbox_name(job_id, original) == f"{job_id}_{original}"

    def test_name_at_limit_is_unchanged(self) -> None:
        job_id = _job_id()
        original = "a" * (240 - len(f".{job_id}_.part") - 4) + ".pdf"
        name = inbox_name(job_id, original)
        assert name == f"{job_id}_{original}"
        assert self._part_bytes(name) == 240

    def test_long_name_without_extension(self, tmp_path: Path) -> None:
        name = inbox_name(_job_id(), "б" * 300)
        assert self._part_bytes(name) <= 240
        assert name.endswith("б")
        LocalHotFolder(tmp_path).put(name, b"x")

    def test_long_suffix_is_not_treated_as_extension(self) -> None:
        name = inbox_name(_job_id(), "ж" * 150 + "." + "x" * 17)
        assert self._part_bytes(name) <= 240
        assert not name.endswith("x" * 17)

    def test_four_byte_chars_are_not_split(self) -> None:
        name = inbox_name(_job_id(), "\U0001f600" * 100 + ".png")
        assert self._part_bytes(name) <= 240
        assert name.endswith("\U0001f600.png")


class TestPaths:
    def test_default_layout(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        job_id = _job_id()
        assert hf.root == tmp_path
        assert hf.input_dir == tmp_path / "Input"
        assert hf.output_dir == tmp_path / "Output"
        assert hf.tmp_dir == tmp_path / "Tmp"
        assert hf.answer_dir(job_id) == tmp_path / "Output" / job_id
        assert hf.answer_path(job_id) == tmp_path / "Output" / job_id / f"{job_id}.json"

    def test_custom_dir_names_and_str_root(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(str(tmp_path), input_dir="Входящие", output_dir="Исходящие", tmp_dir="Служебный")
        assert hf.root == tmp_path
        assert hf.input_dir == tmp_path / "Входящие"
        assert hf.output_dir == tmp_path / "Исходящие"
        assert hf.tmp_dir == tmp_path / "Служебный"
        assert hf.answer_path("abc") == tmp_path / "Исходящие" / "abc" / "abc.json"

    @pytest.mark.parametrize("job_id", ["", ".", "..", "../x", "a/b", "a\\b", "a:b", "a?b", "a*b", "a\x00b", "a\nb"])
    def test_answer_path_rejects_unsafe_job_id(self, tmp_path: Path, job_id: str) -> None:
        hf = LocalHotFolder(tmp_path)
        with pytest.raises(ValueError, match="job_id"):
            hf.answer_path(job_id)
        with pytest.raises(ValueError, match="job_id"):
            hf.answer_dir(job_id)
        with pytest.raises(ValueError, match="job_id"):
            hf.poll(job_id)

    def test_negative_settle_is_rejected(self, tmp_path: Path) -> None:
        with pytest.raises(ValueError, match="settle"):
            LocalHotFolder(tmp_path, settle_seconds=-1)

    def test_implements_protocol(self, tmp_path: Path) -> None:
        hf: HotFolder = LocalHotFolder(tmp_path)
        assert hf.healthy() is True


class TestPut:
    def test_writes_bytes_under_inbox_name(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        name = inbox_name(_job_id(), "aep_5.pdf")
        data = bytes(range(256)) * 10
        hf.put(name, data)
        assert (tmp_path / "Input" / name).read_bytes() == data

    def test_leaves_no_part_files(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        name = inbox_name(_job_id(), "aep_5.pdf")
        hf.put(name, b"data")
        assert [p.name for p in hf.input_dir.iterdir()] == [name]
        assert list(hf.tmp_dir.iterdir()) == []

    def test_part_is_written_in_tmp_not_in_input(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """OQ-79: во время записи во входящем каталоге нет временного файла — он в `Tmp/`, затем rename в `Input`."""
        hf = LocalHotFolder(tmp_path)
        name = inbox_name(_job_id(), "aep_5.pdf")
        seen: dict[str, Any] = {}
        real_replace = os.replace

        def spy(src: Any, dst: Any) -> None:
            seen["src"], seen["dst"] = Path(src), Path(dst)
            seen["input"] = sorted(p.name for p in hf.input_dir.iterdir())
            seen["tmp"] = sorted(p.name for p in hf.tmp_dir.iterdir())
            seen["part"] = Path(src).read_bytes()
            real_replace(src, dst)

        monkeypatch.setattr("src.gateway.hotfolder.os.replace", spy)
        hf.put(name, b"data")
        assert seen == {
            "src": hf.tmp_dir / f".{name}.part",
            "dst": hf.input_dir / name,
            "input": [],
            "tmp": [f".{name}.part"],
            "part": b"data",
        }
        assert [p.name for p in hf.input_dir.iterdir()] == [name]
        assert list(hf.tmp_dir.iterdir()) == []

    def test_creates_input_and_tmp_dirs_only(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        assert not hf.input_dir.exists()
        assert not hf.tmp_dir.exists()
        hf.put("a.pdf", b"1")
        assert hf.input_dir.is_dir()
        assert hf.tmp_dir.is_dir()
        assert not hf.output_dir.exists()

    def test_existing_tmp_dir_is_fine(self, tmp_path: Path) -> None:
        (tmp_path / "Tmp").mkdir()
        LocalHotFolder(tmp_path).put("a.pdf", b"1")
        assert (tmp_path / "Input" / "a.pdf").read_bytes() == b"1"

    def test_custom_tmp_dir_name(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path, tmp_dir="Служебный")
        hf.put("a.pdf", b"1")
        assert (tmp_path / "Служебный").is_dir()
        assert not (tmp_path / "Tmp").exists()

    def test_existing_input_dir_is_fine(self, tmp_path: Path) -> None:
        (tmp_path / "Input").mkdir()
        LocalHotFolder(tmp_path).put("a.pdf", b"1")
        assert (tmp_path / "Input" / "a.pdf").read_bytes() == b"1"

    def test_missing_root_raises_and_creates_nothing(self, tmp_path: Path) -> None:
        root = tmp_path / "absent"
        with pytest.raises(FileNotFoundError):
            LocalHotFolder(root).put("a.pdf", b"1")
        assert not root.exists()
        assert list(tmp_path.iterdir()) == []

    def test_overwrites_existing_file(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        hf.put("a.pdf", b"old")
        hf.put("a.pdf", b"new")
        assert (hf.input_dir / "a.pdf").read_bytes() == b"new"
        assert [p.name for p in hf.input_dir.iterdir()] == ["a.pdf"]

    def test_empty_data_is_written(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        hf.put("empty.pdf", b"")
        assert (hf.input_dir / "empty.pdf").read_bytes() == b""

    @pytest.mark.parametrize(
        "name",
        ["", "../escape.pdf", "sub/a.pdf", "sub\\a.pdf", "/abs.pdf", ".hidden.pdf", ".a.pdf.part", "..", ".", "a\x00b"],
    )
    def test_rejects_invalid_names(self, tmp_path: Path, name: str) -> None:
        hf = LocalHotFolder(tmp_path)
        with pytest.raises(ValueError, match="имя файла"):
            hf.put(name, b"x")
        assert not hf.input_dir.exists()
        assert not (tmp_path.parent / "escape.pdf").exists()

    @pytest.mark.parametrize("name", ["a:b.pdf", "a<b.pdf", "a>b.pdf", 'a"b.pdf', "a|b.pdf", "a?b.pdf", "a*b.pdf"])
    def test_rejects_names_illegal_on_windows(self, tmp_path: Path, name: str) -> None:
        hf = LocalHotFolder(tmp_path)
        with pytest.raises(ValueError, match="имя файла"):
            hf.put(name, b"x")
        assert not hf.input_dir.exists()

    @pytest.mark.parametrize("name", ["a\x01b.pdf", "a\tb.pdf", "a\nb.pdf", "a\x1fb.pdf"])
    def test_rejects_control_chars(self, tmp_path: Path, name: str) -> None:
        with pytest.raises(ValueError, match="имя файла"):
            LocalHotFolder(tmp_path).put(name, b"x")

    def test_accepts_spaces_cyrillic_and_inbox_names(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path)
        names = [inbox_name(_job_id(), "Отчёт за 2024 год (копия).PDF"), inbox_name(_job_id(), "aep_8.pdf"), "a b.pdf"]
        for name in names:
            hf.put(name, b"x")
        assert sorted(p.name for p in hf.input_dir.iterdir()) == sorted(names)

    def test_cleans_part_file_on_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        hf = LocalHotFolder(tmp_path)

        def boom(src: object, dst: object) -> None:
            raise OSError("диск недоступен")

        monkeypatch.setattr("src.gateway.hotfolder.os.replace", boom)
        with pytest.raises(OSError, match="диск недоступен"):
            hf.put("a.pdf", b"data")
        assert list(hf.input_dir.iterdir()) == []
        assert list(hf.tmp_dir.iterdir()) == []

    def test_cleanup_failure_does_not_mask_original_error(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """T7.5: ошибка удаления `.part` подавляется, наружу выходит исходное исключение записи."""
        hf = LocalHotFolder(tmp_path)

        def boom(src: object, dst: object) -> None:
            raise OSError("диск недоступен")

        def unlink_fails(self: Path, missing_ok: bool = False) -> None:
            raise PermissionError("удаление .part запрещено")

        monkeypatch.setattr("src.gateway.hotfolder.os.replace", boom)
        monkeypatch.setattr(Path, "unlink", unlink_fails)
        with pytest.raises(OSError, match="диск недоступен"):
            hf.put("a.pdf", b"data")


class TestPoll:
    """`poll` — все `*.json` каталога `<output>/<job_id>/` по имени, когда набор настоялся `settle_seconds`."""

    def test_none_until_answer_dir_has_json(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        assert hf.poll(job_id) is None
        (tmp_path / "Output").mkdir()
        assert hf.poll(job_id) is None
        (tmp_path / "Output" / job_id).mkdir()
        assert hf.poll(job_id) is None

    def test_missing_root_is_none(self, tmp_path: Path) -> None:
        assert LocalHotFolder(tmp_path / "absent", settle_seconds=0).poll(_job_id()) is None

    def test_answer_dir_is_a_file_is_none(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        (tmp_path / "Output").mkdir()
        (tmp_path / "Output" / job_id).write_bytes(b"{}")
        assert hf.poll(job_id) is None

    def test_settle_zero_returns_at_once(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        payload = b'{"file_uid": "x", "data_ocr": {}}'
        _write_answer(hf, job_id, payload)
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", payload)]

    def test_one_json_after_settle(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None  # набор впервые увиден
        clock.now += 1.9
        assert hf.poll(job_id) is None
        clock.now += 0.1
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")]

    def test_two_jsons_ascending_by_name(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b'{"n": 2}', f"{job_id}_ofr.json")
        _write_answer(hf, job_id, b'{"n": 1}', f"{job_id}.json")
        assert hf.poll(job_id) is None
        clock.now += 2
        assert hf.poll(job_id) == [
            AnswerFile(f"{job_id}.json", b'{"n": 1}'),
            AnswerFile(f"{job_id}_ofr.json", b'{"n": 2}'),
        ]

    def test_non_json_files_and_subdirs_are_ignored(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"text", "readme.txt")
        _write_answer(hf, job_id, b"{}", f".{job_id}.json.part")
        (hf.answer_dir(job_id) / "nested.json").mkdir()
        assert hf.poll(job_id) is None
        clock.now += 10
        assert hf.poll(job_id) is None  # json-файлов нет — ответа нет, сколько ни жди
        _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None
        clock.now += 2
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")]

    def test_json_extension_is_case_insensitive(self, tmp_path: Path) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}", "B.JSON")
        _write_answer(hf, job_id, b"[]", "a.Json")
        assert hf.poll(job_id) == [AnswerFile("B.JSON", b"{}"), AnswerFile("a.Json", b"[]")]

    def test_growing_file_is_not_ready_until_stable(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        path = _write_answer(hf, job_id, b'{"part": ')
        assert hf.poll(job_id) is None
        clock.now += 1.5
        path.write_bytes(b'{"part": 1, "more": ')  # размер вырос — таймер заново
        assert hf.poll(job_id) is None
        clock.now += 1.5
        path.write_bytes(b'{"part": 1, "more": 2}')
        assert hf.poll(job_id) is None
        clock.now += 1.9
        assert hf.poll(job_id) is None
        clock.now += 0.1
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b'{"part": 1, "more": 2}')]

    def test_new_file_resets_timer(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None
        clock.now += 1.5
        _write_answer(hf, job_id, b"[]", f"{job_id}_ofr.json")
        assert hf.poll(job_id) is None
        clock.now += 1.5
        assert hf.poll(job_id) is None  # с момента появления второго файла прошло 1,5 с
        clock.now += 0.5
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}"), AnswerFile(f"{job_id}_ofr.json", b"[]")]

    def test_removed_file_resets_timer(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        extra = _write_answer(hf, job_id, b"[]", f"{job_id}_ofr.json")
        assert hf.poll(job_id) is None
        clock.now += 1.5
        extra.unlink()
        assert hf.poll(job_id) is None
        clock.now += 1.5
        assert hf.poll(job_id) is None
        clock.now += 0.5
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")]

    def test_removal_of_all_json_is_none_and_restarts(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        path = _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None
        clock.now += 1
        path.unlink()
        assert hf.poll(job_id) is None
        clock.now += 5
        path.write_bytes(b"{}")  # тот же набор снова: настаивается заново, а не «с прошлого раза»
        assert hf.poll(job_id) is None
        clock.now += 2
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"{}")]

    def test_state_is_dropped_after_answer(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None
        clock.now += 2
        assert hf.poll(job_id) is not None
        assert hf.pending_jobs() == 0
        assert hf.poll(job_id) is None  # состояние сброшено: повторный опрос настаивает набор заново

    def test_other_job_is_unaffected(self, tmp_path: Path) -> None:
        hf, clock = _settled(tmp_path)
        job_id, other = _job_id(), _job_id()
        _write_answer(hf, other, b"{}")
        assert hf.poll(other) is None
        clock.now += 1
        _write_answer(hf, job_id, b"[]")
        assert hf.poll(job_id) is None
        clock.now += 1
        assert hf.poll(job_id) is None
        assert hf.poll(other) == [AnswerFile(f"{other}.json", b"{}")]
        clock.now += 1
        assert hf.poll(job_id) == [AnswerFile(f"{job_id}.json", b"[]")]

    def test_file_changed_between_listing_and_read_is_none(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        monkeypatch.setattr(Path, "read_bytes", lambda self: b'{"longer": true}')
        assert hf.poll(job_id) is None

    def test_file_removed_between_listing_and_read_is_none(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        path = _write_answer(hf, job_id, b"{}")

        def vanished(self: Path) -> bytes:
            path.unlink()
            raise FileNotFoundError(str(self))

        monkeypatch.setattr(Path, "read_bytes", vanished)
        assert hf.poll(job_id) is None

    def test_new_file_appeared_while_reading_is_none_and_set_resettles(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """T2.1: после чтения каталог листается заново; набор изменился (появился файл) — `None`, набор настаивается
        заново и отдаётся целиком."""
        hf, clock = _settled(tmp_path)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        assert hf.poll(job_id) is None
        clock.now += 2
        real_read = Path.read_bytes

        def read_then_add(self: Path) -> bytes:
            data = real_read(self)
            _write_answer(hf, job_id, b'{"n": 2}', f"{job_id}_ofr.json")
            return data

        monkeypatch.setattr(Path, "read_bytes", read_then_add)
        assert hf.poll(job_id) is None
        monkeypatch.setattr(Path, "read_bytes", real_read)
        assert hf.poll(job_id) is None  # состояние сброшено: новый набор настаивается с нуля
        clock.now += 2
        assert hf.poll(job_id) == [
            AnswerFile(f"{job_id}.json", b"{}"),
            AnswerFile(f"{job_id}_ofr.json", b'{"n": 2}'),
        ]

    def test_file_removed_after_read_is_none(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """T2.1: файл прочитан, но пропал до повторного листинга — набор изменился, `None`."""
        hf = LocalHotFolder(tmp_path, settle_seconds=0)
        job_id = _job_id()
        _write_answer(hf, job_id, b"{}")
        second = _write_answer(hf, job_id, b"[]", f"{job_id}_ofr.json")
        real_read = Path.read_bytes

        def read_then_remove(self: Path) -> bytes:
            data = real_read(self)
            if self == second:
                second.unlink()
            return data

        monkeypatch.setattr(Path, "read_bytes", read_then_remove)
        assert hf.poll(job_id) is None


class TestSettleTracker:
    def test_snapshot_must_be_stable_for_settle_seconds(self) -> None:
        clock = FakeClock()
        tracker = SettleTracker(2, clock)
        snap = (("a.json", 2),)
        assert tracker.ready("job", snap) is False
        clock.now += 2
        assert tracker.ready("job", snap) is True
        assert len(tracker) == 0

    def test_empty_snapshot_is_never_ready(self) -> None:
        assert SettleTracker(0).ready("job", ()) is False

    def test_zero_settle_keeps_no_state(self) -> None:
        tracker = SettleTracker(0)
        assert tracker.ready("job", (("a.json", 2),)) is True
        assert len(tracker) == 0

    def test_state_is_bounded(self) -> None:
        """Задания, ушедшие по таймауту с ненастоявшимся набором, не копятся: старые вытесняются."""
        clock = FakeClock()
        tracker = SettleTracker(2, clock, max_jobs=3)
        for job in ("j1", "j2", "j3", "j4"):
            assert tracker.ready(job, (("a.json", 1),)) is False
        assert len(tracker) == 3
        clock.now += 2
        assert tracker.ready("j1", (("a.json", 1),)) is False  # вытеснено — настаивается заново
        assert tracker.ready("j4", (("a.json", 1),)) is True

    def test_forget(self) -> None:
        clock = FakeClock()
        tracker = SettleTracker(2, clock)
        tracker.ready("job", (("a.json", 1),))
        tracker.forget("job")
        tracker.forget("unknown")
        assert len(tracker) == 0

    def test_negative_settle_is_rejected(self) -> None:
        with pytest.raises(ValueError, match="settle"):
            SettleTracker(-0.1)


class TestHealthy:
    def test_true_for_existing_dir(self, tmp_path: Path) -> None:
        assert LocalHotFolder(tmp_path).healthy() is True

    def test_false_for_missing_root(self, tmp_path: Path) -> None:
        assert LocalHotFolder(tmp_path / "absent").healthy() is False

    def test_false_for_file_root(self, tmp_path: Path) -> None:
        file_root = tmp_path / "file"
        file_root.write_bytes(b"x")
        assert LocalHotFolder(file_root).healthy() is False

    def test_never_raises(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        def boom(self: Path) -> bool:
            raise PermissionError("нет доступа")

        monkeypatch.setattr(Path, "is_dir", boom)
        assert LocalHotFolder(tmp_path).healthy() is False