154 lines
4.7 KiB
Python
154 lines
4.7 KiB
Python
"""Retained terminal classification and filesystem-failure edge coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
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
|
|
|
|
|
|
def test_gc_reports_quarantine_that_left_the_entry_in_place(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""An invalid retained journal that survives quarantine is not counted."""
|
|
root = tmp_path / "lanes"
|
|
board = root / "board"
|
|
board.mkdir(parents=True)
|
|
retained = board / "task.run-1.terminal.committed.json"
|
|
retained.write_text("{}", encoding="utf-8")
|
|
monkeypatch.setattr(lanes, "STATE_ROOT", root)
|
|
monkeypatch.setattr(lanes, "_quarantine_terminal", lambda *_a, **_k: retained)
|
|
|
|
assert lanes.gc_lane_artifacts() == 0
|
|
assert retained.exists()
|
|
|
|
|
|
def test_gc_bounds_unpinnable_staging_and_failed_nonregular_unlinks(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""Unsnapshotable staging entries fall through to identity-safe removal."""
|
|
root = tmp_path / "lanes"
|
|
board = root / "board"
|
|
board.mkdir(parents=True)
|
|
dangling = board / ".retire.dangling"
|
|
dangling.symlink_to(board / "missing-target")
|
|
monkeypatch.setattr(lanes, "STATE_ROOT", root)
|
|
attempts = []
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_unlink_artifact_if_same",
|
|
lambda path, _observed: attempts.append(path.name) and False,
|
|
)
|
|
|
|
assert lanes.gc_lane_artifacts() == 0
|
|
assert attempts == [".retire.dangling"]
|
|
|
|
|
|
def test_gc_keeps_entries_whose_capped_unlink_loses_the_race(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
"""A lost inode race during count capping removes nothing else."""
|
|
root = tmp_path / "lanes"
|
|
board = root / "board"
|
|
board.mkdir(parents=True)
|
|
newer = board / "task.candidate-2.json"
|
|
newer.write_text("{}", encoding="utf-8")
|
|
older = board / "task.candidate-1.json"
|
|
older.write_text("{}", encoding="utf-8")
|
|
os.utime(older, (1, 1))
|
|
monkeypatch.setattr(lanes, "STATE_ROOT", root)
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_unlink_artifact_if_same",
|
|
lambda _path, _observed: False,
|
|
)
|
|
|
|
assert lanes.gc_lane_artifacts(now=100.0, max_age_seconds=10**9, max_count=1) == 0
|
|
assert older.exists() and newer.exists()
|