"""Retention: refuse before the volume is full, and delete only what is derived.
The library is not a cache. 745 GB of it sits on the volume, the overwhelming
majority never fetched by this addon, and an episode it did fetch is exactly as
permanent as one that arrived another way. So nothing here is allowed to delete
an episode -- what protects the disk is an ingest DECLINING to start, which is
the honest order of preference: a copy we have beats a copy we were about to
make.
What may be reclaimed is only what can be proved derived: a partial remux
nothing is working on, and a legacy cache copy the library has superseded.
Run: .venv/bin/python3 -m pytest test_retention.py -q
"""
import json
import os
import pytest
import addon
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "cache"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
os.path.join(str(tmp_path), "queue.json"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "verified.json"))
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
os.makedirs(str(tmp_path / "cache"), exist_ok=True)
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01"},
"tv/1/season/1/episode/1": {"name": "Pilot", "runtime": 50},
}[path])
yield
def _queue():
with open(addon.INGEST_QUEUE_FILE, "w") as f:
json.dump({"series:tmdb:1:1:1:720": {
"state": "done", "args": ["series", "tmdb:1", 1, 1, 720],
"ts": 100.0}}, f)
def _dest():
meta = addon._library_meta("series", "tmdb:1", 1, 1)
return addon._library_path("series", meta["title"], meta["year"],
meta["tmdb_id"], 1, 1, meta["episode_title"])
def _write(path, size=1024, age=0.0):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"x" * size)
if age:
old = os.stat(path).st_mtime - age
os.utime(path, (old, old))
# --- the floor --------------------------------------------------------------
def test_a_full_volume_stops_an_ingest_rather_than_deleting_anything(monkeypatch):
"""The whole policy in one assertion: the new copy yields, the old ones
stay."""
monkeypatch.setattr(addon, "_disk", lambda: (1000, 999, 1))
monkeypatch.setattr(addon, "ingest_source", lambda: object())
resolved = []
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: resolved.append(1))
state = addon._run_ingest("series", "tmdb:1", 1, 1, 720)
assert state == "failed: below the free-space floor"
assert resolved == [] # and it cost no browser resolve
def test_headroom_lets_an_ingest_proceed(monkeypatch):
monkeypatch.setattr(addon, "_disk",
lambda: (0, 0, addon.LIBRARY_MIN_FREE_BYTES * 2))
assert addon._enough_space() is True
def test_an_unreadable_disk_does_not_stop_anything(monkeypatch):
"""A brake that cannot read the road must not stop the car -- the same rule
the completeness gate follows for a missing runtime."""
monkeypatch.setattr(addon, "_disk", lambda: None)
assert addon._enough_space() is True
# --- what may be reclaimed --------------------------------------------------
def test_an_abandoned_partial_is_reclaimable():
_queue()
_write(addon._part_path(_dest()), age=addon.MP4_REMUX_TIMEOUT + 60)
assert [i["reason"] for i in addon._reclaimable()] == ["abandoned partial remux"]
def test_a_partial_that_could_still_be_running_is_left_alone():
"""A remux may legitimately run for MP4_REMUX_TIMEOUT."""
_queue()
_write(addon._part_path(_dest()), age=60)
assert addon._reclaimable() == []
def test_a_partial_belonging_to_a_running_job_is_left_alone(monkeypatch):
_queue()
_write(addon._part_path(_dest()), age=addon.MP4_REMUX_TIMEOUT + 60)
monkeypatch.setattr(addon, "_mp4_jobs", {"series:tmdb:1:1:1:720": "running"})
assert addon._reclaimable() == []
def test_a_legacy_copy_is_reclaimable_once_the_library_supersedes_it(monkeypatch):
_queue()
_write(addon._mp4_path("series", "tmdb:1", 1, 1, 720))
_write(_dest())
monkeypatch.setattr(addon, "_media_duration", lambda p: 3000.0)
assert [i["reason"] for i in addon._reclaimable()] == \
["superseded by the library copy"]
def test_a_legacy_copy_is_kept_when_the_library_copy_is_short(monkeypatch):
"""Both halves of that check matter. Without the completeness gate, a
truncated library copy would be grounds for deleting the good legacy one."""
_queue()
_write(addon._mp4_path("series", "tmdb:1", 1, 1, 720))
_write(_dest())
monkeypatch.setattr(addon, "_media_duration", lambda p: 600.0)
assert addon._reclaimable() == []
def test_a_legacy_copy_with_no_library_copy_is_never_touched():
"""The real one: mp4-cache holds X-Men '97 S02E07, never ingested to the
library, so it is the ONLY copy of that episode."""
_queue()
_write(addon._mp4_path("series", "tmdb:1", 1, 1, 720))
assert addon._reclaimable() == []
def test_a_file_we_cannot_tie_to_a_job_is_never_touched():
"""Nothing is in the queue, so nothing is ours to reason about -- including
the unrelated 2019 Box Left.png.part sitting in a scans folder."""
_write(os.path.join(addon.LIBRARY_ROOT, "Shows", "Scans", "Box Left.png.part"),
age=addon.MP4_REMUX_TIMEOUT + 60)
_write(addon._part_path(_dest()), age=addon.MP4_REMUX_TIMEOUT + 60)
assert addon._reclaimable() == []
def test_a_finished_library_episode_is_never_reclaimable(monkeypatch):
"""The line this must not cross."""
_queue()
_write(_dest())
monkeypatch.setattr(addon, "_media_duration", lambda p: 3000.0)
assert addon._reclaimable() == []
# --- reclaiming -------------------------------------------------------------
def test_reclaim_removes_only_the_derived_file():
_queue()
part = addon._part_path(_dest())
_write(part, size=2048, age=addon.MP4_REMUX_TIMEOUT + 60)
_write(_dest())
result = addon._reclaim()
assert result["removed"] == [part] and result["freed"] == 2048
assert not os.path.exists(part) and os.path.exists(_dest())
def test_reclaim_does_not_take_a_path_from_the_caller():
"""_reclaim() takes no arguments: the set is re-derived, so no request can
ever name something for deletion."""
import inspect
assert list(inspect.signature(addon._reclaim).parameters) == []
def test_the_route_is_not_reachable_by_a_get():
assert addon.app.test_client().get("/library/reclaim").status_code == 405
def test_the_page_shows_the_budget(monkeypatch):
monkeypatch.setattr(addon, "_disk", lambda: (2 * 1024 ** 4, 0, 1024 ** 4))
body = addon._disk_html()
assert "free of" in body and "floor" in body
def test_the_page_says_so_when_ingests_are_declining(monkeypatch):
monkeypatch.setattr(addon, "_disk", lambda: (1024 ** 4, 1024 ** 4 - 1, 1))
assert "declining" in addon._disk_html()