90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""Retained terminal classification and filesystem-failure edge coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from testing.tests.test_hermes_cli_support import (
|
|
_pending_terminal_record,
|
|
lanes,
|
|
)
|
|
|
|
|
|
class _Snapshot:
|
|
document = None
|
|
invalid_reason = "malformed"
|
|
|
|
def __init__(self):
|
|
self.closed = 0
|
|
|
|
def close(self):
|
|
self.closed += 1
|
|
|
|
|
|
def test_retained_classifier_handles_missing_snapshot(tmp_path: Path, monkeypatch):
|
|
path = tmp_path / "task.run-1.terminal.committed.json"
|
|
monkeypatch.setattr(lanes, "_open_terminal_recovery_snapshot", lambda _path: None)
|
|
assert lanes._quarantine_invalid_retained_terminal(path) == (True, False)
|
|
|
|
|
|
def test_retained_classifier_accepts_valid_committed_evidence(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
|
|
state_file = lanes.state_path("board", "task")
|
|
path = lanes._terminal_path(state_file, 2, "committed")
|
|
record = _pending_terminal_record("board", "task", 2, "accepted")
|
|
record["kanban_state"] = "committed"
|
|
lanes.atomic_json(path, record)
|
|
assert lanes._quarantine_invalid_retained_terminal(path) == (False, False)
|
|
|
|
|
|
def test_retained_classifier_bounds_post_quarantine_stat_error(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
path = tmp_path / "task.run-3.terminal.committed.json"
|
|
snapshot = _Snapshot()
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_open_terminal_recovery_snapshot",
|
|
lambda _path: snapshot,
|
|
)
|
|
monkeypatch.setattr(lanes, "_quarantine_terminal", lambda *_args, **_kwargs: path)
|
|
real_stat = Path.stat
|
|
|
|
def fail_selected(candidate, *args, **kwargs):
|
|
if candidate == path:
|
|
raise PermissionError("denied")
|
|
return real_stat(candidate, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Path, "stat", fail_selected)
|
|
assert lanes._quarantine_invalid_retained_terminal(path) == (True, False)
|
|
assert snapshot.closed == 1
|
|
|
|
|
|
def test_gc_skips_symlink_board_and_disappearing_candidate(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
root = tmp_path / "lanes"
|
|
root.mkdir()
|
|
target = tmp_path / "target"
|
|
target.mkdir()
|
|
(root / "linked-board").symlink_to(target, target_is_directory=True)
|
|
board = root / "board"
|
|
board.mkdir()
|
|
candidate = board / "task.candidate-1.json"
|
|
candidate.write_text("{}", encoding="utf-8")
|
|
monkeypatch.setattr(lanes, "STATE_ROOT", root)
|
|
real_stat = Path.stat
|
|
|
|
def disappear(selected, *args, **kwargs):
|
|
if selected == candidate:
|
|
raise FileNotFoundError
|
|
return real_stat(selected, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Path, "stat", disappear)
|
|
assert lanes.gc_lane_artifacts() == 0
|