93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
"""L-2: tests for prune_run_logs (run-log rotation).
|
|
|
|
Verifies that old / surplus *.log files are removed while fresh logs, non-.log
|
|
files, the active log, and subdirectories are left intact. Function is
|
|
best-effort and must never raise.
|
|
"""
|
|
import os
|
|
import time
|
|
|
|
from src.agents.launcher import prune_run_logs
|
|
|
|
|
|
def _touch(path, age_days=0):
|
|
with open(path, "w") as f:
|
|
f.write("x")
|
|
mtime = time.time() - age_days * 86400
|
|
os.utime(path, (mtime, mtime))
|
|
return path
|
|
|
|
|
|
def test_old_logs_removed_fresh_kept(tmp_path):
|
|
runs = tmp_path
|
|
fresh = _touch(str(runs / "1.log"), age_days=1)
|
|
old = _touch(str(runs / "2.log"), age_days=40)
|
|
|
|
removed = prune_run_logs(str(runs), keep_days=30, keep_max=500)
|
|
|
|
assert removed == 1
|
|
assert os.path.exists(fresh)
|
|
assert not os.path.exists(old)
|
|
|
|
|
|
def test_non_log_files_untouched(tmp_path):
|
|
runs = tmp_path
|
|
old_log = _touch(str(runs / "stale.log"), age_days=99)
|
|
keep_txt = _touch(str(runs / "notes.txt"), age_days=99)
|
|
keep_db = _touch(str(runs / "orchestrator.db"), age_days=99)
|
|
|
|
prune_run_logs(str(runs), keep_days=30, keep_max=500)
|
|
|
|
assert not os.path.exists(old_log)
|
|
assert os.path.exists(keep_txt)
|
|
assert os.path.exists(keep_db)
|
|
|
|
|
|
def test_keep_max_retains_newest(tmp_path):
|
|
runs = tmp_path
|
|
# 5 logs, all recent (within keep_days), increasing age 0..4 days.
|
|
paths = []
|
|
for i in range(5):
|
|
paths.append(_touch(str(runs / f"{i}.log"), age_days=i))
|
|
|
|
removed = prune_run_logs(str(runs), keep_days=365, keep_max=2)
|
|
|
|
# Only the 2 newest (age 0, 1) survive.
|
|
assert removed == 3
|
|
assert os.path.exists(paths[0])
|
|
assert os.path.exists(paths[1])
|
|
for p in paths[2:]:
|
|
assert not os.path.exists(p)
|
|
|
|
|
|
def test_active_log_never_removed(tmp_path):
|
|
runs = tmp_path
|
|
active = _touch(str(runs / "active.log"), age_days=99)
|
|
other = _touch(str(runs / "other.log"), age_days=99)
|
|
|
|
removed = prune_run_logs(
|
|
str(runs), keep_days=30, keep_max=500, active_paths=[active]
|
|
)
|
|
|
|
assert removed == 1
|
|
assert os.path.exists(active)
|
|
assert not os.path.exists(other)
|
|
|
|
|
|
def test_subdirs_untouched(tmp_path):
|
|
runs = tmp_path
|
|
sub = runs / "sub.log"
|
|
sub.mkdir() # a directory that happens to end in .log
|
|
old_log = _touch(str(runs / "old.log"), age_days=99)
|
|
|
|
prune_run_logs(str(runs), keep_days=30, keep_max=500)
|
|
|
|
assert sub.is_dir()
|
|
assert not os.path.exists(old_log)
|
|
|
|
|
|
def test_missing_dir_is_noop(tmp_path):
|
|
missing = tmp_path / "does-not-exist"
|
|
# Must not raise.
|
|
assert prune_run_logs(str(missing)) == 0
|