"""Tests for backup retention pruning (mempalace.backups.prune_backups). These guard the fix for unbounded backup growth: ``mempalace migrate`` and ``mempalace repair max-seq-id`` each drop a fresh full-size, timestamped copy every run, and used to never delete the old ones — a palace was found with hundreds of GB of stale backups beside a few hundred MB of live data. """ import os import pytest from mempalace.backups import prune_backups def _make_backup_dir(parent, name, mtime): """Create a directory backup with a fixed mtime.""" path = parent / name path.mkdir() (path / "chroma.sqlite3").write_text("db") os.utime(path, (mtime, mtime)) return path def _make_backup_file(parent, name, mtime): """Create a file backup with a fixed mtime.""" path = parent / name path.write_text("db") os.utime(path, (mtime, mtime)) return path def test_prune_keeps_newest_and_removes_oldest(tmp_path): # 5 backups, mtimes 100..500; keep 2 newest (400, 500). paths = [_make_backup_file(tmp_path, f"b.{i}", mtime=i * 100) for i in range(1, 6)] removed = prune_backups(str(tmp_path / "b.*"), max_backups=2) surviving = {p.name for p in tmp_path.iterdir()} assert surviving == {"b.4", "b.5"} assert set(removed) == {str(paths[0]), str(paths[1]), str(paths[2])} def test_prune_removes_directory_backups(tmp_path): """migrate writes directory backups (full copytree) — must rmtree them.""" _make_backup_dir(tmp_path, "palace.pre-migrate.1", mtime=100) _make_backup_dir(tmp_path, "palace.pre-migrate.2", mtime=200) keep = _make_backup_dir(tmp_path, "palace.pre-migrate.3", mtime=300) removed = prune_backups(str(tmp_path / "palace.pre-migrate.*"), max_backups=1) assert keep.is_dir() assert len(removed) == 2 assert not (tmp_path / "palace.pre-migrate.1").exists() assert not (tmp_path / "palace.pre-migrate.2").exists() def test_prune_noop_when_under_limit(tmp_path): _make_backup_file(tmp_path, "b.1", mtime=100) _make_backup_file(tmp_path, "b.2", mtime=200) removed = prune_backups(str(tmp_path / "b.*"), max_backups=10) assert removed == [] assert len(list(tmp_path.iterdir())) == 2 def test_prune_noop_when_exactly_at_limit(tmp_path): _make_backup_file(tmp_path, "b.1", mtime=100) _make_backup_file(tmp_path, "b.2", mtime=200) removed = prune_backups(str(tmp_path / "b.*"), max_backups=2) assert removed == [] @pytest.mark.parametrize("disabled", [0, -1, None]) def test_prune_disabled_keeps_everything(tmp_path, disabled): for i in range(1, 6): _make_backup_file(tmp_path, f"b.{i}", mtime=i * 100) removed = prune_backups(str(tmp_path / "b.*"), max_backups=disabled) assert removed == [] assert len(list(tmp_path.iterdir())) == 5 def test_prune_no_matches(tmp_path): assert prune_backups(str(tmp_path / "nope.*"), max_backups=3) == [] def test_prune_only_touches_matching_pattern(tmp_path): """Live data and unrelated files must never be swept up by a backup glob.""" _make_backup_file(tmp_path, "chroma.sqlite3.max-seq-id-backup-1", mtime=100) _make_backup_file(tmp_path, "chroma.sqlite3.max-seq-id-backup-2", mtime=200) _make_backup_file(tmp_path, "chroma.sqlite3.max-seq-id-backup-3", mtime=300) # The live database and an unrelated file — must survive. live = _make_backup_file(tmp_path, "chroma.sqlite3", mtime=400) other = _make_backup_file(tmp_path, "tunnels.json", mtime=400) prune_backups( str(tmp_path / "chroma.sqlite3.max-seq-id-backup-*"), max_backups=1, ) assert live.exists() assert other.exists() assert (tmp_path / "chroma.sqlite3.max-seq-id-backup-3").exists() assert not (tmp_path / "chroma.sqlite3.max-seq-id-backup-1").exists() assert not (tmp_path / "chroma.sqlite3.max-seq-id-backup-2").exists() def test_prune_respects_glob_escape_for_metacharacter_paths(tmp_path): """Palace paths can contain glob metacharacters like ``[``. Without ``glob.escape`` the pattern would silently match nothing (the bracket is read as a character class), leaving backups unpruned. Callers escape the literal prefix; this confirms the helper prunes correctly once they do. """ import glob weird = tmp_path / "weird[name]" weird.mkdir() for i in range(1, 4): _make_backup_file(weird, f"chroma.sqlite3.max-seq-id-backup-{i}", mtime=i * 100) pattern = os.path.join(glob.escape(str(weird)), "chroma.sqlite3.max-seq-id-backup-*") removed = prune_backups(pattern, max_backups=1) assert len(removed) == 2 assert (weird / "chroma.sqlite3.max-seq-id-backup-3").exists() def test_prune_is_best_effort_on_delete_failure(tmp_path, monkeypatch): """A failed deletion is logged and skipped, never raised — pruning must not undo a migrate/repair that already succeeded.""" for i in range(1, 5): _make_backup_file(tmp_path, f"b.{i}", mtime=i * 100) real_remove = os.remove def flaky_remove(path): if path.endswith("b.1"): raise OSError("permission denied") return real_remove(path) monkeypatch.setattr(os, "remove", flaky_remove) logs = [] removed = prune_backups(str(tmp_path / "b.*"), max_backups=2, log=logs.append) # b.1 and b.2 were over the limit; b.1 failed, b.2 succeeded. assert str(tmp_path / "b.2") in removed assert str(tmp_path / "b.1") not in removed assert (tmp_path / "b.1").exists() assert any("could not remove" in line for line in logs)