backend/tests/test_foreign_tracks_preview.py
Rama master · réplica de lectura del repositorio original.
"""Preview and manual-run behaviour for the foreign-track rule."""
import os
import pytest
from db.models import foreign_tracks as ft
@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
db_path = str(tmp_path / "test.db")
os.environ["SUBLARR_DB_PATH"] = db_path
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():
try:
yield application
finally:
# Same teardown the conftest fixtures do: without it every test in
# this file leaks its event-dispatcher threads.
from app_shutdown import shutdown_event_dispatchers
shutdown_event_dispatchers(application)
os.environ.pop("SUBLARR_DB_PATH", None)
os.environ.pop("SUBLARR_API_KEY", None)
os.environ.pop("SUBLARR_LOG_LEVEL", None)
def _make_rule(name="Foreign-Track Cleanup"):
"""`config_json` is stored as a JSON *string*, not a dict."""
import json
from db.repositories.cleanup import CleanupRepository
repo = CleanupRepository()
rule = repo.create_rule(
name=name,
rule_type="foreign_tracks",
enabled=True,
config_json=json.dumps({"keep_languages": ["de", "en"]}),
)
return rule["id"]
def test_preview_reports_counts_from_the_scan_table(app):
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
from services.cleanup_rule_runner import _preview_by_rule
rule_id = _make_rule()
scan = ForeignTrackScanRepository()
scan.upsert_seen("/media/a.mkv", 1, 1.0, generation=1)
scan.mark_probed("/media/a.mkv", ["spa", "ita"])
scan.upsert_seen("/media/b.mkv", 1, 1.0, generation=1)
scan.mark_probed("/media/b.mkv", [])
result = _preview_by_rule({"rule_id": rule_id}, "/media")
assert result["affected_files"] == 1
assert result["clean_files"] == 1
assert result["examples"][0]["path"] == "/media/a.mkv"
assert result["examples"][0]["langs"] == ["ita", "spa"]
def test_preview_examples_carry_the_per_track_verdicts(app):
"""Task 7 stores a verdict per track (mark_probed_verdicts); the rule
preview must pass that `verdicts` key through on each example, keeping
the existing keys (`path`, `tracks`, `langs`) untouched."""
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
from services.cleanup_rule_runner import _preview_by_rule
rule_id = _make_rule()
scan = ForeignTrackScanRepository()
scan.upsert_seen("/media/a.mkv", 1, 1.0, generation=1)
verdicts = [
{
"index": 2,
"sub_index": 0,
"language": "fr",
"fmt": "text",
"kind": "full",
"keep": False,
"reason": "stripped_language",
}
]
scan.mark_probed_verdicts("/media/a.mkv", verdicts)
result = _preview_by_rule({"rule_id": rule_id}, "/media")
example = result["examples"][0]
assert example["path"] == "/media/a.mkv"
assert example["verdicts"] == verdicts
def test_preview_says_so_when_no_scan_has_run(app):
from services.cleanup_rule_runner import _preview_by_rule
rule_id = _make_rule()
result = _preview_by_rule({"rule_id": rule_id}, "/media")
assert result["affected_files"] == 0
assert "no scan" in result["message"].lower()
def test_preview_no_longer_claims_it_is_unavailable(app):
from services.cleanup_rule_runner import _preview_by_rule
rule_id = _make_rule()
result = _preview_by_rule({"rule_id": rule_id}, "/media")
assert "not available" not in result["message"].lower()
def test_manual_run_executes_one_slice_and_reports_the_remainder(app, monkeypatch):
from services import cleanup_rule_runner as runner
rule_id = _make_rule()
monkeypatch.setattr(
runner,
"_run_foreign_track_slice",
lambda media_path, config: {
"phase": "strip",
"stripped_files": 3,
"tracks_removed": 9,
"bytes_freed": 100,
"pending": 5,
"affected": 7,
"paused_reason": None,
},
)
out = runner.execute_rule(rule_id)
assert out["status"] == "ok"
assert out["result"]["stripped_files"] == 3
assert out["result"]["affected"] == 7
def test_a_paused_slice_is_reported_as_aborted(app, monkeypatch):
from db.repositories.cleanup import CleanupRepository
from services import cleanup_rule_runner as runner
rule_id = _make_rule()
monkeypatch.setattr(
runner,
"_run_foreign_track_slice",
lambda media_path, config: {
"phase": "strip",
"stripped_files": 0,
"tracks_removed": 0,
"bytes_freed": 0,
"pending": 0,
"affected": 4,
"paused_reason": "disk floor reached",
},
)
out = runner.execute_rule(rule_id)
assert out["status"] == "aborted"
assert not CleanupRepository().get_history()["items"], (
"a slice that swept nothing must not leave a history row — that is what "
"makes a no-op distinguishable from a real sweep"
)
def test_a_pause_after_real_work_still_records_what_was_swept(app, monkeypatch):
"""A pause is not the same as a no-op.
The disk-floor check sits INSIDE the strip loop (`sweep.py`), so a slice
can rewrite forty files and only then stop. Treating every `paused_reason`
as "nothing happened" dropped both the history row and `last_run_at`, so
that work became invisible. The floor is not an edge case for the first
real sweep — it is the designed stopping condition (16.85 TB affected
against 14 TB free).
"""
from db.repositories.cleanup import CleanupRepository
from services import cleanup_rule_runner as runner
rule_id = _make_rule()
monkeypatch.setattr(
runner,
"_run_foreign_track_slice",
lambda media_path, config: {
"phase": "strip",
"stripped_files": 40,
"tracks_removed": 91,
"bytes_freed": 1234,
"pending": 0,
"affected": 300,
"paused_reason": "disk floor reached (min_free_gb=500)",
},
)
out = runner.execute_rule(rule_id)
assert out["status"] == "aborted", "the caller must still learn that the sweep paused"
items = CleanupRepository().get_history()["items"]
assert [i["files_deleted"] for i in items] == [40], (
"the forty rewritten files must appear in the cleanup history"
)
def test_the_rule_preview_endpoint_answers_from_the_scan_table(app, monkeypatch):
"""`preview_rule` is the one the UI calls — `POST /cleanup/rules/{id}/preview`.
It used to fall through to `execute_foreign_tracks(dry_run=True)`, which
walks and ffprobes the whole library inside the request: 754 s and 2,825 s
respectively on the production library. The scan table already holds every
verdict, so the answer is a set of counts.
"""
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
from services import cleanup_executors
from services import cleanup_rule_runner as runner
def _no_walking(*args, **kwargs):
raise AssertionError("the preview must not walk the library")
# The old executor is gone (1.15.0); guard the library walk itself.
monkeypatch.setattr(cleanup_executors, "_video_files", _no_walking)
rule_id = _make_rule()
scan = ForeignTrackScanRepository()
scan.upsert_seen("/media/a.mkv", 1, 1.0, generation=1)
scan.mark_probed("/media/a.mkv", ["spa", "ita"])
scan.upsert_seen("/media/b.mkv", 1, 1.0, generation=1)
scan.mark_probed("/media/b.mkv", [])
scan.upsert_seen("/media/c.mkv", 1, 1.0, generation=1)
scan.mark_probed("/media/c.mkv", ["fra"])
preview = runner.preview_rule(rule_id)["preview"]
# The keys PreviewPanel already renders — no frontend change needed.
assert preview["would_strip_files"] == 2
assert preview["would_keep"] == 1
assert preview["would_strip_tracks"] == 3
assert {e["path"] for e in preview["examples"]} == {"/media/a.mkv", "/media/c.mkv"}
def test_failed_rows_are_visible_in_the_preview(app):
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
from services.cleanup_rule_runner import _preview_by_rule
rule_id = _make_rule()
scan = ForeignTrackScanRepository()
scan.upsert_seen("/media/bad.mkv", 1, 1.0, generation=1)
scan.mark_failed("/media/bad.mkv", "ffprobe exploded", ft.ERROR_VERIFY)
result = _preview_by_rule({"rule_id": rule_id}, "/media")
assert result["failed_files"] == 1
def test_repeated_slices_converge_to_idle_over_real_files(app, tmp_path, monkeypatch):
"""Several ticks against a real directory tree must finish the work.
ffprobe and the remux are stubbed — the point is the state machine's
convergence, not ffmpeg.
"""
import os
from db.repositories.foreign_track_scan import ForeignTrackScanRepository
from services.foreign_tracks import sweep as sw
from services.foreign_tracks.state import PHASE_IDLE
for name in ("a.mkv", "b.mkv", "c.mkv"):
path = tmp_path / "Show" / name
os.makedirs(path.parent, exist_ok=True)
path.write_bytes(b"x" * 100)
os.utime(path, (1_000.0, 1_000.0))
monkeypatch.setattr(
sw,
"_probe_file",
lambda p: {
"streams": [{"index": 1, "codec_type": "subtitle", "tags": {"language": "spa"}}]
},
)
stripped = []
monkeypatch.setattr(
sw,
"_strip_file",
lambda p, keep, keep_und, **kw: (stripped.append(p), ("/trash.bak", 7))[1],
)
cfg = {"keep_languages": ["de", "en"], "keep_und": True, "min_free_gb": 0}
repo = ForeignTrackScanRepository()
for _ in range(6):
result = sw.run_slice(str(tmp_path), cfg, budget_s=600, repo=repo)
if result["phase"] == PHASE_IDLE:
break
assert result["phase"] == PHASE_IDLE
assert len(stripped) == 3
assert repo.counts_by_state().get("affected", 0) == 0