"""Orphan-file sweep: remove on-disk cruft no ready asset owns (old accented dirs, stale posters), while never touching system trees, owned files, or freshly-written downloads.""" import os import time from pathlib import Path from app.downloads.gc import _sweep_orphan_files class _Scalars: def __init__(self, rows): self._rows = rows def scalars(self): return list(self._rows) class _StubDB: """Minimal stand-in: `_sweep_orphan_files` only issues one select for ready rel_paths.""" def __init__(self, rel_paths): self._rel_paths = rel_paths def execute(self, _stmt): return _Scalars(self._rel_paths) def _aged(path: Path, hours: float) -> None: old = time.time() - hours * 3600 os.utime(path, (old, old)) def test_sweep_removes_orphans_but_keeps_owned_system_and_fresh(tmp_path: Path): root = tmp_path # An owned asset (in the keep set) + its sidecars. kept_dir = root / "Chan" / "Season_2026" kept_dir.mkdir(parents=True) kept_media = kept_dir / "Chan_-_2026-01-01_-_Title_vid_a1.mp4" for p in (kept_media, kept_media.with_suffix(".nfo"), kept_media.with_suffix(".jpg")): p.write_bytes(b"x") _aged(p, 5) # old, but owned → must survive # An orphan poster in a stale (e.g. old accented) channel dir — old, unowned → swept. orphan_dir = root / "OldChan" / "Season_2026" orphan_dir.mkdir(parents=True) orphan = orphan_dir / "OldChan_-_old_title_vid.jpg" orphan.write_bytes(b"x") _aged(orphan, 5) # A system tree (storyboards) — never touched, even when old. sysfile = root / ".storyboards" / "asset-1.jpg" sysfile.parent.mkdir() sysfile.write_bytes(b"x") _aged(sysfile, 5) # A brand-new unowned file (an in-flight download's just-placed output) — too young to sweep. fresh_dir = root / "FreshChan" / "Season_2026" fresh_dir.mkdir(parents=True) fresh = fresh_dir / "FreshChan_-_now_vid_a2.mp4" fresh.write_bytes(b"x") # mtime = now db = _StubDB(["Chan/Season_2026/Chan_-_2026-01-01_-_Title_vid_a1.mp4"]) removed = _sweep_orphan_files(db, str(root)) assert removed == 1 # only the aged orphan assert kept_media.exists() and kept_media.with_suffix(".jpg").exists() assert not orphan.exists() and not orphan_dir.exists() # orphan + its emptied dirs pruned assert not (root / "OldChan").exists() assert sysfile.exists() # system tree untouched assert fresh.exists() # too new to sweep