"""The guards in front of a recursive delete. `sweep_orphans` rmtree's directories under `_HLS_ROOT`, and that root is admin-configurable (PLEX_HLS_DIR) — it can legitimately be a shared scratch path. `_SESSION_DIR_RE` is the ONLY thing that keeps the sweep off a neighbour's data, so the names it accepts are pinned here. Since the sweep stopped being awaited at startup it also runs CONCURRENTLY with playback, so the second half of this file pins what keeps it off a LIVE session's segments.""" import os import pytest from app.plex import stream from app.plex.stream import _SESSION_DIR_RE @pytest.mark.parametrize( "name", [ "12345_0", # legacy shape (pre multi-audio): {rating_key}_{start} "12345_842", "12345_0_None_+0.00", # current shape, single audio track (audio_ord None) "12345_0_2_+0.00", # current shape, explicit audio_ord "12345_842_multi_-1.25", # multi-audio, negative offset "12345_842_multi_+12.50", ], ) def test_our_session_directories_are_swept(name): assert _SESSION_DIR_RE.match(name) @pytest.mark.parametrize( "name", [ "backup_2024", # the class an "optional tail" pattern wrongly matched "release_10", "important_data_2024", "node_modules", "logs", "plex-hls", "12345", # no start offset — not a session dir "12345_0_multi", # truncated: missing the offset "12345_0_multi_1.25", # offset without its sign "snapshot_2024_backup_+1.00", # session-ish tail, but the key isn't a rating_key "backup_2024_multi_+0.00", ".config", ], ) def test_foreign_directories_are_left_alone(name): assert not _SESSION_DIR_RE.match(name) class _FakeSession: def __init__(self, directory): self.dir = directory @pytest.fixture def hls_root(tmp_path, monkeypatch): """Point the sweep at a temp root with no sessions registered.""" monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path) monkeypatch.setattr(stream, "_sessions", {}) return tmp_path def _session_dir(root, name, *, age_s=0.0): """A session directory, optionally backdated by `age_s` to look like a previous run's.""" d = root / name d.mkdir() (d / "seg_0.ts").write_bytes(b"x") if age_s: t = d.stat().st_mtime - age_s os.utime(d, (t, t)) return d def _cutoff(root): """A "process start" read from the FILESYSTEM, not from a clock. The guard compares a directory's mtime against a timestamp, and the two do not have to come from the same clock: on a bind-mounted dev volume the file times are stamped by the host while `time.time()` runs in the container VM, and the drift between them is real (it made an earlier version of these tests flaky). Anchoring the cutoff to a file this test just created removes the clock from the assertion; that the DEFAULT cutoff is `_PROCESS_START` is pinned separately below.""" probe = root / ".cutoff-probe" probe.write_bytes(b"") at = probe.stat().st_mtime probe.unlink() return at class TestSweepOrphans: def test_a_previous_runs_leftovers_are_deleted(self, hls_root): d = _session_dir(hls_root, "12345_0_None_+0.00", age_s=3600) assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 1 assert not d.exists() def test_foreign_directories_survive(self, hls_root): keep = hls_root / "backup_2024" keep.mkdir() (keep / "important.tar").write_bytes(b"x") assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0 assert (keep / "important.tar").exists() def test_a_live_sessions_directory_is_never_deleted(self, hls_root): # Registered session, but with an OLD directory — only the live check can save it. d = _session_dir(hls_root, "12345_0_multi_+0.00", age_s=3600) stream._sessions["k"] = _FakeSession(d) assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0 assert d.exists() def test_a_directory_this_run_created_is_never_deleted(self, hls_root): # THE RACE: the sweep is no longer awaited before the app serves, and `start_session` # REUSES the path of a same-key session — so between our listing and our rmtree, a stale # name can become a live directory. A fresh mtime is what tells them apart, and it must # hold even while `_sessions` is still empty (the session registers a beat later). cutoff = _cutoff(hls_root) fresh = _session_dir(hls_root, "12345_0_None_+0.00") # created "after boot" old = _session_dir(hls_root, "999_0_None_+0.00", age_s=3600) assert stream.sweep_orphans(older_than=cutoff) == 1 # only the leftover assert fresh.exists() assert not old.exists() def test_the_default_cutoff_is_this_processs_start(self, hls_root, monkeypatch): # The parameter exists for the tests; production calls it with no argument, so pin what # that means — otherwise the guard above could be right and unused. d = _session_dir(hls_root, "12345_0_None_+0.00") monkeypatch.setattr(stream, "_MTIME_GRACE_S", 0.0) monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 1) assert stream.sweep_orphans() == 1 # "older than boot" → swept assert not d.exists() def test_a_directory_stamped_just_before_boot_survives(self, hls_root, monkeypatch): # MEASURED in the container, not theory: a file's mtime comes from the kernel's COARSE # clock and `time.time()` from the fine one, so a directory created 2ms AFTER # `_PROCESS_START` was stamped 2ms BEFORE it. Without the grace the sweep deletes a session # that started in the first instants after boot — which is exactly when the sweep runs. d = _session_dir(hls_root, "12345_0_None_+0.00") monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 0.05) # the skew assert stream.sweep_orphans() == 0 assert d.exists() def test_a_missing_root_is_not_an_error(self, tmp_path, monkeypatch): monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path / "nope") monkeypatch.setattr(stream, "_sessions", {}) assert stream.sweep_orphans() == 0 def test_a_stray_file_with_a_session_name_is_left_alone(self, hls_root): # Only directories are ours; a same-named FILE belongs to someone else. f = hls_root / "12345_0" f.write_bytes(b"x") os.utime(f, (f.stat().st_mtime - 3600,) * 2) # old enough that only is_dir() saves it assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0 assert f.exists()