backend/tests/test_cleanup_executors.py
Rama master · réplica de lectura del repositorio original.
"""Tests for cleanup_executors — language_filter, format_upgrade, orphan_files, orphan_db."""
import os
from unittest.mock import MagicMock, patch
import pytest
def test_language_filter_deletes_non_kept_languages(tmp_path):
"""Files not in keep_languages should be deleted; NFO files are never touched."""
from services.cleanup_executors import execute_language_filter
(tmp_path / "show.de.ass").write_text("german sub")
(tmp_path / "show.en.srt").write_text("english sub")
(tmp_path / "show.fr.ass").write_text("french sub")
(tmp_path / "show.de.nfo").write_text("nfo file") # never deleted
# permanent_delete=True bypasses the trash-dir resolver, which on CI
# may fall outside tmp_path. The new audit tests in
# test_extract_cleanup_audit_2026_05_09.py cover the trash-by-default
# path with a patched media_path; this test covers the hard-delete path.
config = {"keep_languages": ["de"], "permanent_delete": True}
result = execute_language_filter(str(tmp_path), config, dry_run=False)
assert result["deleted"] == 2 # fr.ass + en.srt
assert (tmp_path / "show.de.ass").exists()
assert (tmp_path / "show.de.nfo").exists()
assert not (tmp_path / "show.fr.ass").exists()
assert not (tmp_path / "show.en.srt").exists()
def test_language_filter_keeps_language_variants(tmp_path):
"""keep_languages=['de'] must also keep .deu.ass and .ger.ass variants."""
from services.cleanup_executors import execute_language_filter
(tmp_path / "show.de.ass").write_text("german iso-1")
(tmp_path / "show.deu.ass").write_text("german iso-2")
(tmp_path / "show.ger.ass").write_text("german iso-legacy")
(tmp_path / "show.german.ass").write_text("german full")
(tmp_path / "show.fr.ass").write_text("french sub")
config = {"keep_languages": ["de"], "permanent_delete": True}
result = execute_language_filter(str(tmp_path), config, dry_run=False)
assert result["deleted"] == 1 # only fr.ass
assert (tmp_path / "show.de.ass").exists()
assert (tmp_path / "show.deu.ass").exists()
assert (tmp_path / "show.ger.ass").exists()
assert (tmp_path / "show.german.ass").exists()
assert not (tmp_path / "show.fr.ass").exists()
def test_language_filter_dry_run_deletes_nothing(tmp_path):
"""dry_run=True must not delete any files."""
from services.cleanup_executors import execute_language_filter
(tmp_path / "show.fr.ass").write_text("french sub")
config = {"keep_languages": ["de"]}
result = execute_language_filter(str(tmp_path), config, dry_run=True)
assert result["would_delete"] >= 1
assert (tmp_path / "show.fr.ass").exists()
def _foreign_probe(*_args, **_kwargs):
"""Fake ffprobe result: ger + eng kept, jpn + ita are foreign, und present."""
return {
"streams": [
{"codec_type": "video", "index": 0},
{"codec_type": "subtitle", "index": 1, "tags": {"language": "ger"}},
{"codec_type": "subtitle", "index": 2, "tags": {"language": "eng"}},
{"codec_type": "subtitle", "index": 3, "tags": {"language": "jpn"}},
{"codec_type": "subtitle", "index": 4, "tags": {"language": "ita"}},
{"codec_type": "subtitle", "index": 5, "tags": {"language": "und"}},
]
}
_ASS_HEADER = (
"[Script Info]\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR,"
" MarginV, Effect, Text\n"
)
_ASS_WITH_LINE = _ASS_HEADER + "Dialogue: 0,0:00:01.00,0:00:02.00,Default,,0,0,0,,Hallo\n"
_SRT_WITH_LINE = "1\n00:00:01,000 --> 00:00:02,000\nHallo\n"
def test_format_upgrade_removes_srt_when_ass_exists(tmp_path):
"""SRT should be deleted when ASS exists for same base+language."""
from services.cleanup_executors import execute_format_upgrade
# Real lines, not placeholders: since the sandbox VM found that an empty
# .ass trashed a working .srt, a file only counts as an upgrade when it
# holds at least one subtitle line.
(tmp_path / "show.de.ass").write_text(_ASS_WITH_LINE, encoding="utf-8")
(tmp_path / "show.de.srt").write_text(_SRT_WITH_LINE, encoding="utf-8")
(tmp_path / "show.en.srt").write_text(_SRT_WITH_LINE, encoding="utf-8") # no ASS counterpart
config = {"keep_format": "ass", "permanent_delete": True}
result = execute_format_upgrade(str(tmp_path), config, dry_run=False)
assert result["deleted"] == 1 # only de.srt
assert (tmp_path / "show.de.ass").exists()
assert not (tmp_path / "show.de.srt").exists()
assert (tmp_path / "show.en.srt").exists() # kept, no ASS counterpart
def test_orphan_files_deletes_subs_without_video(tmp_path):
"""Subtitle without a video file in same dir should be detected as orphan."""
from services.cleanup_executors import execute_orphan_files
subdir = tmp_path / "nosub"
subdir.mkdir()
(subdir / "orphan.de.ass").write_text("sub without video")
paired_dir = tmp_path / "paired"
paired_dir.mkdir()
(paired_dir / "movie.mkv").write_text("video")
(paired_dir / "movie.de.ass").write_text("paired sub")
result = execute_orphan_files(str(tmp_path), {"permanent_delete": True}, dry_run=False)
assert result["deleted"] == 1
assert not (subdir / "orphan.de.ass").exists()
assert (paired_dir / "movie.de.ass").exists()
def test_orphan_db_removes_stale_db_entries(tmp_path):
"""DB rows pointing to non-existent files should be batch-removed.
Audit C2-2: previously the executor called ``delete_by_path`` once
per missing row (N+1). It now collects every missing path and
issues a single ``delete_by_paths`` call.
"""
from services.cleanup_executors import execute_orphan_db
existing = str(tmp_path / "exists.de.ass")
missing = str(tmp_path / "missing.de.ass")
open(existing, "w").close() # create the existing file
mock_repo = MagicMock()
mock_repo.get_all_subtitle_paths.return_value = [existing, missing]
mock_repo.delete_by_paths.return_value = 1
# Pin media_path to tmp_path so the new mount-reachability probe
# passes during the unit test.
fake_settings = MagicMock()
fake_settings.media_path = str(tmp_path)
with (
patch("services.cleanup_executors.SubtitleRepository", return_value=mock_repo),
patch("config.get_settings", return_value=fake_settings),
):
result = execute_orphan_db({}, dry_run=False)
assert result["deleted"] == 1
mock_repo.delete_by_paths.assert_called_once_with([missing])
def test_subtitle_walk_skips_sublarr_trash(tmp_path):
"""Sidecar walks must NOT descend into the .sublarr trash subtree.
Already-trashed foreign sidecars live under .sublarr/trash/<date>/ with
no matching video. Before the exclusion, language_filter re-walked them
every night and tried to re-trash them (prod: 5922 Permission-denied
WARNINGs, and orphan_files counted 27k of them as orphans). The trash
subtree must be pruned exactly like _video_files already does.
"""
from services.cleanup_executors import execute_language_filter
# A live foreign sidecar at the top level — must be removed (control).
(tmp_path / "show.fr.ass").write_text("french sub")
# Trashed foreign sidecars under .sublarr — must be left untouched.
trash = tmp_path / ".sublarr" / "trash" / "2026-06-13"
trash.mkdir(parents=True)
(trash / "Old.ara.ass").write_text("trashed arabic")
(trash / "Old.spa.srt").write_text("trashed spanish")
config = {"keep_languages": ["de"], "permanent_delete": True}
result = execute_language_filter(str(tmp_path), config, dry_run=False)
assert result["deleted"] == 1 # only the live top-level fr.ass
assert not (tmp_path / "show.fr.ass").exists()
assert (trash / "Old.ara.ass").exists()
assert (trash / "Old.spa.srt").exists()
def test_subtitle_walk_skips_sidecar_trash_batches(tmp_path):
"""Sidecar walks must NOT descend into ``.sublarr_trash`` either.
Batch trash (UI delete, cleanup audits) lives in ``<media>/.sublarr_trash``,
a sibling of ``.sublarr`` that the walker used to enter. A nightly rule then
treated trashed files as live ones and could move them again, breaking the
batch manifest that restoring depends on.
"""
from services.cleanup_executors import execute_language_filter
(tmp_path / "show.fr.ass").write_text("french sub")
batch = tmp_path / ".sublarr_trash" / "0123abcd"
batch.mkdir(parents=True)
(batch / "Old.ara.ass").write_text("trashed arabic")
config = {"keep_languages": ["de"], "permanent_delete": True}
result = execute_language_filter(str(tmp_path), config, dry_run=False)
assert result["deleted"] == 1
assert (batch / "Old.ara.ass").exists()
# ── VM test of 1.14.4-rc.6: an empty preferred file is no upgrade ────────────
def _format_upgrade_case(tmp_path, ass_body):
(tmp_path / "Show - S01E01.mkv").write_bytes(b"v")
(tmp_path / "Show - S01E01.de.srt").write_text(_SRT_WITH_LINE, encoding="utf-8")
(tmp_path / "Show - S01E01.de.ass").write_text(ass_body, encoding="utf-8")
def test_an_empty_ass_does_not_replace_a_working_srt(tmp_path, app_ctx):
"""Prod flow found on the sandbox VM: extraction writes a good .de.srt, the
"keep ass" rule trashes it because an empty .de.ass sits next to it, and the
episode is left with a subtitle file that holds no line at all."""
from services.cleanup_executors import execute_format_upgrade
_format_upgrade_case(tmp_path, _ASS_HEADER)
result = execute_format_upgrade(str(tmp_path), {"keep_format": "ass"}, dry_run=False)
assert result["deleted"] == 0, result
assert (tmp_path / "Show - S01E01.de.srt").exists()
def test_a_zero_byte_ass_does_not_replace_a_working_srt(tmp_path, app_ctx):
from services.cleanup_executors import execute_format_upgrade
_format_upgrade_case(tmp_path, "")
result = execute_format_upgrade(str(tmp_path), {"keep_format": "ass"}, dry_run=False)
assert result["deleted"] == 0, result
assert (tmp_path / "Show - S01E01.de.srt").exists()
def test_a_real_ass_still_replaces_the_srt(tmp_path, app_ctx):
"""Control: the rule must keep working for files that actually hold lines."""
from services.cleanup_executors import execute_format_upgrade
_format_upgrade_case(tmp_path, _ASS_WITH_LINE)
result = execute_format_upgrade(str(tmp_path), {"keep_format": "ass"}, dry_run=False)
assert result["deleted"] == 1, result
assert not (tmp_path / "Show - S01E01.de.srt").exists()
def test_an_empty_srt_does_not_replace_a_working_ass(tmp_path, app_ctx):
"""The same rule the other way round (keep_format=srt)."""
from services.cleanup_executors import execute_format_upgrade
(tmp_path / "Show - S01E02.mkv").write_bytes(b"v")
(tmp_path / "Show - S01E02.de.ass").write_text(_ASS_WITH_LINE, encoding="utf-8")
(tmp_path / "Show - S01E02.de.srt").write_text("", encoding="utf-8")
result = execute_format_upgrade(str(tmp_path), {"keep_format": "srt"}, dry_run=False)
assert result["deleted"] == 0, result
assert (tmp_path / "Show - S01E02.de.ass").exists()
def test_the_dry_run_agrees_with_the_real_run(tmp_path, app_ctx):
from services.cleanup_executors import execute_format_upgrade
_format_upgrade_case(tmp_path, _ASS_HEADER)
preview = execute_format_upgrade(str(tmp_path), {"keep_format": "ass"}, dry_run=True)
assert preview["would_delete"] == 0, preview
def test_an_unreadable_ass_does_not_replace_a_working_srt(tmp_path, app_ctx):
from services.cleanup_executors import execute_format_upgrade
"""Sandbox VM, 1.14.4-rc.7: with the .ass unreadable (mode 000) the content
check could not read a single line — and counted that as proof of content,
so the working .srt was trashed anyway."""
import sys
if sys.platform == "win32":
pytest.skip("POSIX file modes do not apply on Windows")
_format_upgrade_case(tmp_path, _ASS_WITH_LINE)
os.chmod(tmp_path / "Show - S01E01.de.ass", 0o000)
try:
result = execute_format_upgrade(str(tmp_path), {"keep_format": "ass"}, dry_run=False)
finally:
os.chmod(tmp_path / "Show - S01E01.de.ass", 0o644)
assert result["deleted"] == 0, result
assert (tmp_path / "Show - S01E01.de.srt").exists()
def test_an_unreadable_preferred_file_is_not_proof_of_content(tmp_path):
"""Unit view of the same rule, so it is covered on every platform."""
from unittest.mock import patch as _patch
from services.cleanup_executors import _has_subtitle_lines
path = tmp_path / "x.de.ass"
path.write_text(_ASS_WITH_LINE, encoding="utf-8")
# Patch the module's own name, not builtins: a process-wide open() patch
# also hits background threads and has made unrelated tests fail.
with _patch(
"subtitle_content.open",
create=True,
side_effect=PermissionError("no read permission"),
):
assert _has_subtitle_lines(str(path)) is False