134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
"""Pure-function tests for outputs retention (v1.1 task add-outputs-retention)."""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
# We import the runner module to test helpers directly.
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location(
|
|
"lr", str(Path(__file__).resolve().parent.parent / "scripts" / "loop-runner.py")
|
|
)
|
|
lr = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(lr)
|
|
|
|
|
|
def _make_tick_file(out_dir: Path, tick_num: int, role: str, ext: str = ".json"):
|
|
"""Create a single tick output file for testing."""
|
|
name = f"tick{tick_num}-{role}{ext}"
|
|
(out_dir / name).write_text("{}")
|
|
return name
|
|
|
|
|
|
def _make_tick_group(out_dir: Path, tick_num: int):
|
|
"""Create all 6 files for a tick group."""
|
|
for role in ("implement", "verify", "orchestrate"):
|
|
_make_tick_file(out_dir, tick_num, role, ".json")
|
|
_make_tick_file(out_dir, tick_num, role + "-prompt", ".md")
|
|
|
|
|
|
class TestGetRetention:
|
|
def test_default_main(self):
|
|
assert lr._get_retention({}) == 20
|
|
assert lr._get_retention({"outputs": {}}) == 20
|
|
assert lr._get_retention({"outputs": {"retention": None}}) == 20
|
|
|
|
def test_explicit_value(self):
|
|
assert lr._get_retention({"outputs": {"retention": 5}}) == 5
|
|
assert lr._get_retention({"outputs": {"retention": 0}}) == 0
|
|
assert lr._get_retention({"outputs": {"retention": 100}}) == 100
|
|
|
|
def test_negative_is_zero(self):
|
|
assert lr._get_retention({"outputs": {"retention": -1}}) == 0
|
|
assert lr._get_retention({"outputs": {"retention": -100}}) == 0
|
|
|
|
def test_non_int_falls_back(self):
|
|
assert lr._get_retention({"outputs": {"retention": "garbage"}}) == 20
|
|
assert lr._get_retention({"outputs": {"retention": []}}) == 20
|
|
|
|
def test_none_cfg_falls_back(self):
|
|
assert lr._get_retention(None) == 20
|
|
|
|
|
|
class TestGcOutputs:
|
|
def test_gc_keeps_recent_deletes_old(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
for i in range(1, 31):
|
|
_make_tick_group(out, i)
|
|
assert len(os.listdir(str(out))) == 30 * 6
|
|
lr._gc_outputs(tmp_path, 20)
|
|
remaining = os.listdir(str(out))
|
|
assert len(remaining) == 20 * 6
|
|
for i in range(11, 31):
|
|
assert any(f"tick{i}-" in n for n in remaining), f"tick{i} should be kept"
|
|
for i in range(1, 11):
|
|
assert not any(f"tick{i}-" in n for n in remaining), f"tick{i} should be deleted"
|
|
|
|
def test_retention_zero_skips_gc(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
for i in range(1, 31):
|
|
_make_tick_group(out, i)
|
|
lr._gc_outputs(tmp_path, 0)
|
|
assert len(os.listdir(str(out))) == 30 * 6
|
|
|
|
def test_retention_greater_than_file_count(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
for i in range(1, 6):
|
|
_make_tick_group(out, i)
|
|
lr._gc_outputs(tmp_path, 20)
|
|
assert len(os.listdir(str(out))) == 5 * 6
|
|
|
|
def test_missing_outputs_dir(self, tmp_path):
|
|
lr._gc_outputs(tmp_path, 20) # should not crash
|
|
|
|
def test_non_tick_files_preserved(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
for i in range(1, 31):
|
|
_make_tick_group(out, i)
|
|
(out / "README.txt").write_text("keep me")
|
|
(out / "loop-info.md").write_text("also keep")
|
|
lr._gc_outputs(tmp_path, 20)
|
|
remaining = os.listdir(str(out))
|
|
assert "README.txt" in remaining
|
|
assert "loop-info.md" in remaining
|
|
|
|
def test_unrelated_tick_prefix_preserved(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
_make_tick_group(out, 1)
|
|
(out / "tick-foo.md").write_text("no numeric index")
|
|
lr._gc_outputs(tmp_path, 0) # no gc; just verify regex doesn't break on tick-foo
|
|
assert "tick-foo.md" in os.listdir(str(out))
|
|
|
|
|
|
def test_gc_single_tick_group(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
_make_tick_group(out, 1)
|
|
_make_tick_group(out, 2)
|
|
lr._gc_outputs(tmp_path, 1)
|
|
remaining = os.listdir(str(out))
|
|
assert len(remaining) == 6
|
|
assert any("tick2-" in n for n in remaining)
|
|
assert not any("tick1-" in n for n in remaining)
|
|
|
|
def test_gc_error_swallowed(self, tmp_path):
|
|
out = tmp_path / "outputs"
|
|
out.mkdir()
|
|
_make_tick_group(out, 1)
|
|
_make_tick_group(out, 2)
|
|
# Make the directory read-only so unlink fails.
|
|
prev_mode = os.stat(str(out)).st_mode
|
|
os.chmod(str(out), 0o555)
|
|
try:
|
|
lr._gc_outputs(tmp_path, 1)
|
|
finally:
|
|
os.chmod(str(out), prev_mode)
|
|
|