| 1 | import hashlib |
| 2 | import importlib.util |
| 3 | import json |
| 4 | import os |
| 5 | import sys |
| 6 | import time |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 12 | if str(PROJECT_ROOT) not in sys.path: |
| 13 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 14 | |
| 15 | MOD_PATH = PROJECT_ROOT / "plugins" / "_time_travel" / "helpers" / "retention.py" |
| 16 | _spec = importlib.util.spec_from_file_location("time_travel_retention", MOD_PATH) |
| 17 | retention = importlib.util.module_from_spec(_spec) |
| 18 | assert _spec and _spec.loader |
| 19 | _spec.loader.exec_module(retention) |
| 20 | |
| 21 | DAY = 86400 |
| 22 | HOUR = 3600 |
| 23 | NOW = time.time() |
| 24 | |
| 25 | CFG_AGING = {"retention_max_age_days": 30} |
| 26 | |
| 27 | |
| 28 | def _hex_id(seed: str) -> str: |
| 29 | return hashlib.sha256(seed.encode()).hexdigest()[:32] |
| 30 | |
| 31 | |
| 32 | def _mk_repo(shadow_root, name, age_s, lock_age_s=None, invalid_age_s=None): |
| 33 | entry = os.path.join(shadow_root, name) |
| 34 | ref = os.path.join(entry, "repo.git", "refs", "heads") |
| 35 | os.makedirs(ref, exist_ok=True) |
| 36 | cur = os.path.join(ref, "current") |
| 37 | with open(cur, "w") as f: |
| 38 | f.write("deadbeef") |
| 39 | stamp = NOW - age_s |
| 40 | for path in (cur, os.path.join(entry, "repo.git"), entry): |
| 41 | os.utime(path, (stamp, stamp)) |
| 42 | if lock_age_s is not None: |
| 43 | lock = os.path.join(entry, "repo.git", "index.lock") |
| 44 | with open(lock, "w") as f: |
| 45 | f.write("") |
| 46 | os.utime(lock, (NOW - lock_age_s, NOW - lock_age_s)) |
| 47 | if invalid_age_s is not None: |
| 48 | backup = os.path.join(entry, "repo.git.invalid-20260101") |
| 49 | os.makedirs(backup, exist_ok=True) |
| 50 | with open(os.path.join(backup, "junk"), "w") as f: |
| 51 | f.write("x" * 100) |
| 52 | os.utime(backup, (NOW - invalid_age_s, NOW - invalid_age_s)) |
| 53 | return entry |
| 54 | |
| 55 | |
| 56 | def test_effective_config_defaults_and_clamps(): |
| 57 | cfg = retention.effective_config({}) |
| 58 | assert cfg["retention_enabled"] is True |
| 59 | assert cfg["retention_max_age_days"] == 0 |
| 60 | assert cfg["retention_sweep_interval_hours"] == 6 |
| 61 | clamped = retention.effective_config( |
| 62 | { |
| 63 | "retention_sweep_interval_hours": 0, |
| 64 | "retention_max_age_days": -5, |
| 65 | "retention_orphan_grace_hours": 0, |
| 66 | "retention_stale_lock_minutes": 1, |
| 67 | "retention_enabled": 1, |
| 68 | } |
| 69 | ) |
| 70 | assert clamped["retention_sweep_interval_hours"] == 1 |
| 71 | assert clamped["retention_max_age_days"] == 0 |
| 72 | assert clamped["retention_orphan_grace_hours"] == 1 |
| 73 | assert clamped["retention_stale_lock_minutes"] == 5 |
| 74 | assert clamped["retention_enabled"] is True |
| 75 | garbage = retention.effective_config({"retention_sweep_interval_hours": "nope"}) |
| 76 | assert garbage["retention_sweep_interval_hours"] == 6 |
| 77 | |
| 78 | |
| 79 | def test_sweep_matrix(tmp_path): |
| 80 | shadow = str(tmp_path / "workspaces") |
| 81 | state = str(tmp_path / "state") |
| 82 | os.makedirs(shadow) |
| 83 | |
| 84 | live_recent = _hex_id("alpha") |
| 85 | live_aged = _hex_id("beta") |
| 86 | live_locked = _hex_id("gamma") |
| 87 | live_fresh_lock = _hex_id("delta") |
| 88 | live_invalid = _hex_id("epsilon") |
| 89 | live_ids = {live_recent, live_aged, live_locked, live_fresh_lock, live_invalid} |
| 90 | |
| 91 | _mk_repo(shadow, live_recent, age_s=1 * HOUR) |
| 92 | _mk_repo(shadow, live_aged, age_s=40 * DAY) |
| 93 | _mk_repo(shadow, live_locked, age_s=1 * HOUR, lock_age_s=1 * HOUR) |
| 94 | _mk_repo(shadow, live_fresh_lock, age_s=1 * HOUR, lock_age_s=60) |
| 95 | _mk_repo(shadow, live_invalid, age_s=1 * HOUR, invalid_age_s=48 * HOUR) |
| 96 | _mk_repo(shadow, "0" * 32, age_s=48 * HOUR) # orphan past grace |
| 97 | _mk_repo(shadow, "1" * 32, age_s=1 * HOUR) # orphan inside grace |
| 98 | with open(os.path.join(shadow, "stray-file"), "w") as f: |
| 99 | f.write("ignore me") |
| 100 | |
| 101 | stats = retention.sweep( |
| 102 | cfg=CFG_AGING, shadow_root=shadow, live_ids=live_ids, now_ts=NOW, state_dir=state |
| 103 | ) |
| 104 | |
| 105 | assert os.path.isdir(os.path.join(shadow, live_recent)) |
| 106 | assert not os.path.exists(os.path.join(shadow, live_aged)) |
| 107 | assert stats["aged_removed"] == 1 |
| 108 | assert not os.path.exists(os.path.join(shadow, "0" * 32)) |
| 109 | assert os.path.isdir(os.path.join(shadow, "1" * 32)) |
| 110 | assert stats["orphans_removed"] == 1 |
| 111 | assert os.path.isdir(os.path.join(shadow, live_locked)) |
| 112 | assert not os.path.exists(os.path.join(shadow, live_locked, "repo.git", "index.lock")) |
| 113 | assert os.path.exists(os.path.join(shadow, live_fresh_lock, "repo.git", "index.lock")) |
| 114 | assert stats["stale_locks_removed"] == 1 |
| 115 | assert os.path.isdir(os.path.join(shadow, live_invalid)) |
| 116 | assert not os.path.exists( |
| 117 | os.path.join(shadow, live_invalid, "repo.git.invalid-20260101") |
| 118 | ) |
| 119 | assert stats["invalid_backups_removed"] == 1 |
| 120 | assert stats["bytes_reclaimed"] > 0 |
| 121 | assert os.path.isfile(os.path.join(shadow, "stray-file")) |
| 122 | |
| 123 | |
| 124 | def test_max_age_zero_keeps_history_forever(tmp_path): |
| 125 | shadow = str(tmp_path / "workspaces") |
| 126 | state = str(tmp_path / "state") |
| 127 | os.makedirs(shadow) |
| 128 | ancient = _hex_id("ancient") |
| 129 | _mk_repo(shadow, ancient, age_s=400 * DAY) |
| 130 | |
| 131 | stats = retention.sweep( |
| 132 | cfg={"retention_max_age_days": 0}, |
| 133 | shadow_root=shadow, |
| 134 | live_ids={ancient}, |
| 135 | now_ts=NOW, |
| 136 | state_dir=state, |
| 137 | ) |
| 138 | assert stats["aged_removed"] == 0 |
| 139 | assert os.path.isdir(os.path.join(shadow, ancient)) |
| 140 | |
| 141 | |
| 142 | def test_disabled_sweep_is_noop(tmp_path): |
| 143 | shadow = str(tmp_path / "workspaces") |
| 144 | os.makedirs(shadow) |
| 145 | _mk_repo(shadow, "0" * 32, age_s=48 * HOUR) |
| 146 | stats = retention.sweep( |
| 147 | cfg={"retention_enabled": False}, |
| 148 | shadow_root=shadow, |
| 149 | live_ids=set(), |
| 150 | now_ts=NOW, |
| 151 | state_dir=str(shadow), |
| 152 | ) |
| 153 | assert stats["orphans_removed"] == 0 |
| 154 | assert os.path.isdir(os.path.join(shadow, "0" * 32)) |
| 155 | |
| 156 | |
| 157 | def test_remove_tree_refuses_outside_root(tmp_path): |
| 158 | shadow = str(tmp_path / "workspaces") |
| 159 | outside = str(tmp_path / "outside") |
| 160 | os.makedirs(shadow) |
| 161 | os.makedirs(outside) |
| 162 | assert retention._remove_tree(outside, shadow) == 0 |
| 163 | assert os.path.isdir(outside) |
| 164 | |
| 165 | |
| 166 | def test_marker_and_history(tmp_path): |
| 167 | shadow = str(tmp_path / "workspaces") |
| 168 | state = str(tmp_path / "state") |
| 169 | os.makedirs(shadow) |
| 170 | _mk_repo(shadow, "0" * 32, age_s=48 * HOUR) |
| 171 | retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state) |
| 172 | _mk_repo(shadow, "2" * 32, age_s=48 * HOUR) |
| 173 | retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state) |
| 174 | |
| 175 | marker = json.load(open(os.path.join(state, retention.MARKER_FILE))) |
| 176 | assert marker["sweeps"] == 2 |
| 177 | assert marker["orphans_removed"] == 2 |
| 178 | assert marker["last_sweep_at"] |
| 179 | |
| 180 | history = retention.read_history(state_dir=state) |
| 181 | assert len(history) == 2 |
| 182 | assert history[0]["removed"]["orphans"] == ["0" * 32] |
| 183 | assert history[1]["removed"]["orphans"] == ["2" * 32] |
| 184 | assert history[0]["at"] |
| 185 | |
| 186 | |
| 187 | def test_history_tail_cap(tmp_path): |
| 188 | state = str(tmp_path / "state") |
| 189 | os.makedirs(state) |
| 190 | with open(os.path.join(state, retention.HISTORY_FILE), "w") as f: |
| 191 | for i in range(retention.HISTORY_MAX_LINES + 20): |
| 192 | f.write('{"at": "old-%d"}\n' % i) |
| 193 | retention._append_history(state, {"at": "newest"}) |
| 194 | history = retention.read_history(limit=retention.HISTORY_MAX_LINES + 100, state_dir=state) |
| 195 | assert len(history) == retention.HISTORY_MAX_LINES |
| 196 | assert history[-1]["at"] == "newest" |
| 197 | assert history[0]["at"] != "old-0" |
| 198 | |
| 199 | |
| 200 | def test_due_throttle(tmp_path): |
| 201 | state = str(tmp_path / "state") |
| 202 | shadow = str(tmp_path / "workspaces") |
| 203 | os.makedirs(shadow) |
| 204 | assert retention.due(cfg={}, state_dir=state) |
| 205 | assert not retention.due(cfg={"retention_enabled": False}, state_dir=state) |
| 206 | |
| 207 | retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state) |
| 208 | assert not retention.due(cfg={}, now_ts=time.time(), state_dir=state) |
| 209 | assert retention.due(cfg={}, now_ts=time.time() + 7 * HOUR, state_dir=state) |
| 210 | assert not retention.due( |
| 211 | cfg={"retention_sweep_interval_hours": 12}, |
| 212 | now_ts=time.time() + 7 * HOUR, |
| 213 | state_dir=state, |
| 214 | ) |
| 215 | |
| 216 | |
| 217 | def test_workspace_id_parity_with_time_travel(): |
| 218 | time_travel = pytest.importorskip( |
| 219 | "plugins._time_travel.helpers.time_travel", |
| 220 | reason="requires the full runtime environment", |
| 221 | ) |
| 222 | path = "/a0/usr/projects/example" |
| 223 | expected = hashlib.sha256( |
| 224 | time_travel.canonical_workspace_display_path(path).rstrip("/").encode("utf-8") |
| 225 | ).hexdigest()[:32] |
| 226 | assert time_travel.workspace_id_for(path) == expected |