tenant.openagents/omega

1from __future__ import annotations
2
3import os
4import tempfile
5import time
6import unittest
7from pathlib import Path
8
9from zed_eval import cleanup
10
11
12def age_path(path: Path, days: float) -> None:
13    past = time.time() - days * 86400.0
14    os.utime(path, (past, past))
15
16
17class PruneArtifactsTests(unittest.TestCase):
18    def setUp(self) -> None:
19        self._tmp = tempfile.TemporaryDirectory()
20        self.root = Path(self._tmp.name)
21        self.addCleanup(self._tmp.cleanup)
22
23    def _make_build(self, build_id: str, age_days: float) -> Path:
24        build_dir = self.root / "builds" / build_id
25        build_dir.mkdir(parents=True)
26        (build_dir / "eval-cli").write_text("binary")
27        ready = build_dir / "READY"
28        ready.write_text("done")
29        age_path(ready, age_days)
30        return build_dir
31
32    def test_eval_results_are_never_touched(self) -> None:
33        runs = self.root / "runs" / "anant" / "swe-atlas-rf" / "run-1"
34        runs.mkdir(parents=True)
35        result_file = runs / "summary.json"
36        result_file.write_text("{}")
37        age_path(result_file, 999)
38
39        cleanup.prune_artifacts(self.root, dry_run=False, build_retention_days=14)
40
41        self.assertTrue(result_file.exists())
42
43    def test_removes_old_builds_keeps_recent(self) -> None:
44        self._make_build("bld-old", age_days=30)
45        self._make_build("bld-new", age_days=1)
46
47        result = cleanup.prune_artifacts(
48            self.root, dry_run=False, build_retention_days=14
49        )
50
51        self.assertFalse((self.root / "builds" / "bld-old").exists())
52        self.assertTrue((self.root / "builds" / "bld-new").exists())
53        self.assertEqual(result["counts"]["builds"], 1)
54
55    def test_dry_run_deletes_nothing(self) -> None:
56        self._make_build("bld-old", age_days=30)
57        result = cleanup.prune_artifacts(
58            self.root, dry_run=True, build_retention_days=14
59        )
60        self.assertTrue((self.root / "builds" / "bld-old").exists())
61        self.assertEqual(result["counts"]["builds"], 1)
62
63    def test_stale_lock_removed(self) -> None:
64        locks = self.root / "build-locks"
65        locks.mkdir(parents=True)
66        stale = locks / "bld-x.json"
67        stale.write_text("{}")
68        age_path(stale, 1)  # 1 day > 6h ttl
69        fresh = locks / "bld-y.json"
70        fresh.write_text("{}")
71
72        cleanup.prune_artifacts(self.root, dry_run=False, lock_ttl_hours=6)
73
74        self.assertFalse(stale.exists())
75        self.assertTrue(fresh.exists())
76
77
78if __name__ == "__main__":
79    unittest.main()
80