backend/tests/test_foreign_tracks_sweep_scope.py
Rama master · réplica de lectura del repositorio original.
"""The sweep's scope — include/exclude paths and the size limit — must hold
for the worklist it already has, not only for the next walk.
Prod 2026-09-28: the foreign_tracks rule was narrowed to ``include_paths:
["_Anime"]`` and ``foreign_track_sweep_max_file_gb`` set to 20. Both filter
only inside the walk. The config change reset the verdicts but, because the
previous walk had completed, jumped straight to PROBE over the old
whole-library worklist — so the next two nights remuxed 2160p films in
``_Filme`` (Sinners, Frozen, Bumblebee, Godzilla …) and the trash grew from
783 to 1421 GB. The state still named generation 1 from 2026-09-26; the new
scope would only have applied once that worklist drained, years away.
"""
import os
from types import SimpleNamespace
import pytest
from db.models import foreign_tracks as ft
from services.foreign_tracks import sweep as sw
from services.foreign_tracks.state import PHASE_STRIP, SweepState, config_hash, save_state
GB = 1024**3
CFG = {"keep_languages": ["de", "en"], "keep_und": True}
ANIME = dict(CFG, include_paths=["_Anime"])
SPANISH = {"streams": [{"index": 1, "codec_type": "subtitle", "tags": {"language": "spa"}}]}
class FakeClock:
def __init__(self):
self.t = 1_000_000.0
def __call__(self):
return self.t
@pytest.fixture()
def app(tmp_path):
"""Create a Flask app with an isolated SQLite DB for testing."""
from app import create_app
from config import reload_settings
os.environ["SUBLARR_DB_PATH"] = str(tmp_path / "test.db")
os.environ["SUBLARR_API_KEY"] = ""
os.environ["SUBLARR_LOG_LEVEL"] = "ERROR"
reload_settings()
application = create_app(testing=True)
application.config["TESTING"] = True
with application.app_context():
yield application
os.environ.pop("SUBLARR_DB_PATH", None)
os.environ.pop("SUBLARR_API_KEY", None)
os.environ.pop("SUBLARR_LOG_LEVEL", None)
@pytest.fixture
def repo(app):
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
return ForeignTrackScanRepository()
def _settings(monkeypatch, max_file_gb=0):
monkeypatch.setattr(
"config.get_settings",
lambda: SimpleNamespace(
cleanup_foreign_tracks_keep_languages=["de", "en"],
cleanup_foreign_tracks_keep_und=True,
foreign_track_sweep_max_file_gb=max_file_gb,
),
)
def _record_strips(monkeypatch) -> list[str]:
stripped: list[str] = []
def fake_strip(path, keep, keep_und, **kw):
stripped.append(path)
return ("/trash/x.bak", 5)
monkeypatch.setattr(sw, "_strip_file", fake_strip)
return stripped
def _record_walks(monkeypatch, files) -> list[list[str]]:
"""Each walk's include_paths, in order; every walk yields ``files``."""
walks: list[list[str]] = []
def fake_walk(root, include_paths, exclude_paths, **kw):
walks.append(list(include_paths))
return iter(files)
monkeypatch.setattr(sw, "iter_video_files", fake_walk)
return walks
def _prod_state_of_2026_09_28(root: str, config: dict) -> None:
"""A completed walk from before the scope existed, already in STRIP, and
with the narrowed config already hashed — exactly what prod held."""
save_state(
SweepState(
generation=1,
phase=PHASE_STRIP,
enumeration_complete=True,
config_hash=config_hash(config, root),
)
)
class TestTheOldWorklistObeysTheScope:
def test_an_affected_film_outside_include_paths_is_never_stripped(
self, app, repo, tmp_path, monkeypatch
):
root = str(tmp_path)
film = os.path.join(root, "_Filme", "Sinners (2025)", "Sinners.mkv")
episode = os.path.join(root, "_Anime", "Bleach", "S01E01.mkv")
for path in (film, episode):
repo.upsert_seen(path, 10, 1.0, generation=1)
repo.mark_probed(path, ["spa"])
_settings(monkeypatch)
_prod_state_of_2026_09_28(root, ANIME)
stripped = _record_strips(monkeypatch)
_record_walks(monkeypatch, [])
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
assert stripped == [episode]
assert repo._get(film) is None, "an out-of-scope row must leave the worklist"
def test_an_affected_file_over_the_size_limit_is_never_stripped(
self, app, repo, tmp_path, monkeypatch
):
root = str(tmp_path)
film = os.path.join(root, "_Anime", "Film", "huge.mkv")
episode = os.path.join(root, "_Anime", "Bleach", "S01E01.mkv")
repo.upsert_seen(film, 70 * GB, 1.0, generation=1)
repo.upsert_seen(episode, 1 * GB, 1.0, generation=1)
for path in (film, episode):
repo.mark_probed(path, ["spa"])
_settings(monkeypatch, max_file_gb=20)
_prod_state_of_2026_09_28(root, ANIME)
stripped = _record_strips(monkeypatch)
_record_walks(monkeypatch, [])
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
assert stripped == [episode]
assert repo._get(film) is None
def test_a_pending_file_outside_the_scope_is_not_even_probed(
self, app, repo, tmp_path, monkeypatch
):
root = str(tmp_path)
film = os.path.join(root, "_Filme", "Frozen (2013)", "Frozen.mkv")
repo.upsert_seen(film, 10, 1.0, generation=1)
_settings(monkeypatch)
save_state(
SweepState(
generation=1,
phase="probe",
enumeration_complete=True,
config_hash=config_hash(ANIME, root),
)
)
probed: list[str] = []
monkeypatch.setattr(sw, "_probe_file", lambda path: probed.append(path) or SPANISH)
_record_strips(monkeypatch)
_record_walks(monkeypatch, [])
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
assert probed == []
assert repo._get(film) is None
class TestAScopeChangeWalksAgain:
def test_narrowing_include_paths_after_a_complete_walk_walks_again(
self, app, repo, tmp_path, monkeypatch
):
root = str(tmp_path)
_settings(monkeypatch)
monkeypatch.setattr(sw, "_probe_file", lambda path: {"streams": []})
walks = _record_walks(monkeypatch, [(os.path.join(root, "_Anime", "a.mkv"), 10, 1.0)])
sw.run_slice(root, CFG, budget_s=60, now_fn=FakeClock(), repo=repo)
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
assert walks == [[], ["_Anime"]]
def test_widening_include_paths_walks_again_so_new_files_are_found(
self, app, repo, tmp_path, monkeypatch
):
"""The mirror case: without a walk, a widened scope would only reach
the new folders at the next rescan, `foreign_track_sweep_rescan_days`
later."""
root = str(tmp_path)
_settings(monkeypatch)
monkeypatch.setattr(sw, "_probe_file", lambda path: {"streams": []})
walks = _record_walks(monkeypatch, [])
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
sw.run_slice(root, CFG, budget_s=60, now_fn=FakeClock(), repo=repo)
assert walks == [["_Anime"], []]
def test_changing_the_size_limit_walks_again_without_resetting_verdicts(
self, app, repo, tmp_path, monkeypatch
):
"""The limit is not a verdict input — a clean file stays clean, it
only needs the walk that applies the limit."""
root = str(tmp_path)
episode = os.path.join(root, "_Anime", "a.mkv")
probed: list[str] = []
monkeypatch.setattr(sw, "_probe_file", lambda path: probed.append(path) or {"streams": []})
walks = _record_walks(monkeypatch, [(episode, 10, 1.0)])
_settings(monkeypatch, max_file_gb=0)
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
_settings(monkeypatch, max_file_gb=20)
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
assert len(walks) == 2
assert probed == [episode], "the size limit must not re-probe the library"
assert repo.counts_by_state()[ft.STATE_CLEAN] == 1
def test_a_keep_list_change_alone_still_skips_the_walk(self, app, repo, tmp_path, monkeypatch):
"""Unchanged behaviour: the files on disk did not change, only the
verdicts — re-probing the worklist is enough."""
root = str(tmp_path)
_settings(monkeypatch)
monkeypatch.setattr(sw, "_probe_file", lambda path: {"streams": []})
walks = _record_walks(monkeypatch, [(os.path.join(root, "_Anime", "a.mkv"), 10, 1.0)])
sw.run_slice(root, ANIME, budget_s=60, now_fn=FakeClock(), repo=repo)
sw.run_slice(
root, dict(ANIME, keep_languages=["de"]), budget_s=60, now_fn=FakeClock(), repo=repo
)
assert walks == [["_Anime"]]