backend/tests/test_foreign_tracks_verify_strip.py
Rama master · réplica de lectura del repositorio original.
"""Deleting the original after a verified rewrite, built into the sweep (owner 2026-09-26).
The option is a NEW key, ``delete_original_after_verify``: the old
``verify_then_delete_backup`` never had an effect in 1.15.0 RCs, and an
auto-update must not silently arm a box someone ticked back then (owner).
A backup may only be deleted when the rewritten file is exactly what the
policy asked for: it probes, keeps every video/audio stream of the original,
and carries precisely the subtitle tracks the policy keeps from the original
— not one more (nothing left to strip) and not one less (nothing lost).
Anything else keeps the backup: a false negative costs disk for the
retention period, a false positive costs the original.
"""
from unittest.mock import patch
from services.foreign_tracks.select import TrackPolicy
_B = TrackPolicy(
mode="one_per_language", keep_forced=True, keep_sdh=False, sidecar_policy="drop_if_real_sidecar"
)
_KEEP = {"de", "ger", "deu", "en", "eng"}
def _v(index):
return {"index": index, "codec_type": "video", "codec_name": "h264"}
def _a(index, lang="jpn"):
return {"index": index, "codec_type": "audio", "codec_name": "aac", "tags": {"language": lang}}
def _s(index, lang, codec="ass", title=None, frames=None, **disposition):
tags = {"language": lang}
if title:
tags["title"] = title
if frames is not None:
tags["NUMBER_OF_FRAMES"] = str(frames)
stream = {"index": index, "codec_type": "subtitle", "codec_name": codec, "tags": tags}
if disposition:
stream["disposition"] = {k: int(v) for k, v in disposition.items()}
return stream
ORIGINAL = [
_v(0),
_a(1),
_s(2, "eng", "ass", "English", 400),
_s(3, "eng", "hdmv_pgs_subtitle", "English", 800),
_s(4, "ger", "ass", "German", 390),
_s(5, "fre", "ass", "French", 380),
]
GOOD = [_v(0), _a(1), _s(2, "eng", "ass", "English", 400), _s(3, "ger", "ass", "German", 390)]
def _verify(original, result, tmp_path, real=frozenset(), policy=_B):
from services.foreign_tracks.verify import verify_strip
video = tmp_path / "e1.mkv"
video.write_bytes(b"new")
backup = tmp_path / "e1.mkv.bak"
backup.write_bytes(b"old")
def probe(path, use_cache=True):
assert use_cache is False, "verification must never trust a cached probe"
return {"streams": original if str(path) == str(backup) else result}
with patch("ass_probe.run_ffprobe", side_effect=probe):
return verify_strip(str(video), str(backup), policy, _KEEP, False, set(real))
def test_an_exact_result_verifies(tmp_path):
ok, why = _verify(ORIGINAL, GOOD, tmp_path)
assert ok, why
def test_a_lost_kept_track_fails(tmp_path):
ok, why = _verify(ORIGINAL, [_v(0), _a(1), _s(2, "eng", "ass", "English", 400)], tmp_path)
assert not ok
assert "de" in why
def test_a_track_the_policy_would_strip_fails(tmp_path):
left = GOOD + [_s(4, "fre", "ass", "French", 380)]
ok, why = _verify(ORIGINAL, left, tmp_path)
assert not ok
def test_a_lost_audio_track_fails(tmp_path):
ok, why = _verify(ORIGINAL + [_a(6, "ger")], GOOD, tmp_path)
assert not ok
assert "audio" in why
def test_no_video_fails(tmp_path):
ok, _ = _verify(ORIGINAL, [s for s in GOOD if s["codec_type"] != "video"], tmp_path)
assert not ok
def test_an_empty_file_fails(tmp_path):
from services.foreign_tracks.verify import verify_strip
video = tmp_path / "e1.mkv"
video.write_bytes(b"")
backup = tmp_path / "e1.mkv.bak"
backup.write_bytes(b"old")
ok, why = verify_strip(str(video), str(backup), _B, _KEEP, False, set())
assert not ok
def test_an_unprobeable_result_fails(tmp_path):
from services.foreign_tracks.verify import verify_strip
video = tmp_path / "e1.mkv"
video.write_bytes(b"new")
backup = tmp_path / "e1.mkv.bak"
backup.write_bytes(b"old")
with patch("ass_probe.run_ffprobe", side_effect=RuntimeError("ffprobe died")):
ok, why = verify_strip(str(video), str(backup), _B, _KEEP, False, set())
assert not ok
assert "probe" in why
def test_policy_b_with_a_real_sidecar_expects_the_main_track_gone(tmp_path):
"""Under B with a real German sidecar the German main track is stripped —
a result WITHOUT it is exactly right."""
result = [_v(0), _a(1), _s(2, "eng", "ass", "English", 400)]
ok, why = _verify(ORIGINAL, result, tmp_path, real={"de"})
assert ok, why
# --- the kept tracks must be the RIGHT ones, not just the right count (review C1) ---
def test_the_wrong_track_of_the_right_language_fails(tmp_path):
"""The policy keeps ENG ASS; a rewrite that kept ENG PGS instead has the
same per-language count and must still keep its backup."""
wrong = [
_v(0),
_a(1),
_s(2, "eng", "hdmv_pgs_subtitle", "English", 800),
_s(3, "ger", "ass", "German", 390),
]
ok, why = _verify(ORIGINAL, wrong, tmp_path)
assert not ok, why
def test_a_forced_track_kept_instead_of_the_full_one_fails(tmp_path):
"""keep_forced=False: the full German track is kept, the forced one goes.
A rewrite that kept only the forced track re-decides it as the lone
``kept_last_track`` — the count matches, the track does not."""
policy = TrackPolicy(mode="one_per_language", keep_forced=False, keep_sdh=False)
original = [
_v(0),
_a(1),
_s(2, "ger", "ass", "German", 390),
_s(3, "ger", "ass", "Signs & Songs", 40, forced=1),
]
wrong = [_v(0), _a(1), _s(2, "ger", "ass", "Signs & Songs", 40, forced=1)]
right = [_v(0), _a(1), _s(2, "ger", "ass", "German", 390)]
assert _verify(original, right, tmp_path, policy=policy)[0]
ok, why = _verify(original, wrong, tmp_path, policy=policy)
assert not ok, why
def test_values_the_remux_rewrites_do_not_fail_the_check(tmp_path):
"""mkvmerge renumbers streams and rewrites its statistics tags — neither
may make a correct rewrite look wrong."""
rewritten = [
_v(0),
_a(1),
_s(2, "eng", "ass", "English", 401),
_s(3, "ger", "ass", "German", None),
]
ok, why = _verify(ORIGINAL, rewritten, tmp_path)
assert ok, why
def test_a_changed_forced_flag_fails(tmp_path):
original = [_v(0), _a(1), _s(2, "ger", "ass", "German", 390, forced=0)]
result = [_v(0), _a(1), _s(2, "ger", "ass", "German", 390, forced=1)]
ok, _ = _verify(original, result, tmp_path)
assert not ok
def test_a_changed_default_flag_is_not_a_different_track(tmp_path):
"""ffmpeg (MP4 path) sets default on the first track when none is set — a
correct remux; failing on it would only keep backups for no reason."""
original = [_v(0), _a(1), _s(2, "ger", "ass", "German", 390, default=0)]
result = [_v(0), _a(1), _s(2, "ger", "ass", "German", 390, default=1)]
ok, why = _verify(original, result, tmp_path)
assert ok, why
# --- probing (review M2) ---------------------------------------------------------
def test_verification_probes_with_ffprobe_whatever_the_metadata_engine(tmp_path, monkeypatch):
"""With scan_metadata_engine=mediainfo, the engine-routed probe drops the
video stream, so every rewrite would fail with "no video". The check must
use ffprobe itself, uncached."""
from services.foreign_tracks.verify import verify_strip
video = tmp_path / "e1.mkv"
video.write_bytes(b"new")
backup = tmp_path / "e1.mkv.bak"
backup.write_bytes(b"old")
def ffprobe(path, use_cache=True):
assert use_cache is False
return {"streams": ORIGINAL if str(path) == str(backup) else GOOD}
def engine(path, use_cache=True):
streams = ORIGINAL if str(path) == str(backup) else GOOD
return {"streams": [s for s in streams if s["codec_type"] != "video"]}
with (
patch("ass_probe.run_ffprobe", side_effect=ffprobe),
patch("remux.get_media_streams", side_effect=engine),
patch("ass_probe.get_media_streams", side_effect=engine),
):
ok, why = verify_strip(str(video), str(backup), _B, _KEEP, False, set())
assert ok, why
def test_the_strip_decision_uses_a_fresh_probe(tmp_path):
"""The keep/strip decision must never come from a probe cached for a file
that has since been rewritten under the same path."""
import remux
video = tmp_path / "e1.mkv"
video.write_bytes(b"v")
calls = []
def probe(path, *args, **kwargs):
calls.append(kwargs.get("use_cache", args[0] if args else True))
return {"streams": [_v(0), _a(1), _s(2, "ger", "ass", "German", 390)]}
with (
patch("remux.get_media_streams", side_effect=probe),
patch("remux.check_hardlink_policy", return_value=True),
):
assert (
remux.remove_foreign_subtitle_streams(video_path=str(video), target_languages=_KEEP)
is None
)
assert calls == [False]
# --- the sweep uses it ---------------------------------------------------------
def _sweep_strip(tmp_path, monkeypatch, config, verified, batch=None):
import remux
from services.foreign_tracks import sweep as sw
video = tmp_path / "e1.mkv"
video.write_bytes(b"new")
backup = tmp_path / "e1.mkv.bak"
backup.write_bytes(b"old")
notified = []
class _Row:
path = str(video)
track_count = 2
rows = [_Row()]
class _Repo:
def claim_next_affected(self):
return rows.pop() if rows else None
def mark_stripped(self, path):
pass
def mark_failed(self, *a, **k):
raise AssertionError("must not fail")
def mark_probed_verdicts(self, *a, **k):
pass
monkeypatch.setattr(sw, "_media_root_reachable", lambda root: True)
monkeypatch.setattr(sw, "_free_bytes", lambda root: 10**15)
monkeypatch.setattr(sw, "_strip_file", lambda *a, **k: (str(backup), 1000))
monkeypatch.setattr(sw, "_real_sidecars_for_policy", lambda *a, **k: set())
monkeypatch.setattr(sw, "verify_strip", lambda *a, **k: (verified, "reason"), raising=False)
monkeypatch.setattr(
"services.media_server_notify.notify_media_servers",
lambda path, item_type="": notified.append(path),
)
monkeypatch.setattr(
"services.media_server_notify.notify_media_servers_batch",
batch or (lambda paths: notified.extend(paths)),
raising=False,
)
from services.foreign_tracks.state import SweepState
state = SweepState(phase="strip", enumeration_complete=True)
result = {"stripped_files": 0, "tracks_removed": 0, "bytes_freed": 0}
clock = iter(range(0, 10_000))
sw._strip_phase(
str(tmp_path),
config,
state,
_Repo(),
_KEEP,
False,
_B,
[],
10_000,
lambda: next(clock),
result,
(),
)
return backup.exists(), notified, result, remux
def test_the_sweep_deletes_a_verified_backup_when_the_option_is_on(tmp_path, monkeypatch):
exists, notified, result, _ = _sweep_strip(
tmp_path, monkeypatch, {"delete_original_after_verify": True}, verified=True
)
assert not exists
assert result.get("backups_deleted") == 1
def test_the_sweep_keeps_an_unverified_backup(tmp_path, monkeypatch):
exists, _, result, _ = _sweep_strip(
tmp_path, monkeypatch, {"delete_original_after_verify": True}, verified=False
)
assert exists
assert result.get("verify_failed") == 1
def test_the_sweep_keeps_the_backup_when_the_option_is_off(tmp_path, monkeypatch):
exists, _, _, _ = _sweep_strip(tmp_path, monkeypatch, {}, verified=True)
assert exists
def test_the_sweep_tells_the_media_servers_about_a_rewrite(tmp_path, monkeypatch):
_, notified, _, _ = _sweep_strip(tmp_path, monkeypatch, {}, verified=True)
assert notified == [str(tmp_path / "e1.mkv")]
def test_the_legacy_key_alone_never_deletes_a_backup(tmp_path, monkeypatch):
"""Ticked before the upgrade, when it did nothing: must stay inert until
the user confirms it through the new option."""
exists, _, result, _ = _sweep_strip(
tmp_path, monkeypatch, {"verify_then_delete_backup": True}, verified=True
)
assert exists
assert "backups_deleted" not in result
def test_the_sweep_refreshes_the_media_servers_once_per_slice(tmp_path, monkeypatch):
"""Night review I1: one synchronous refresh per rewritten file (with a
library-scan fallback on every miss) stalled the slice; the rewritten
paths are collected and handed over once, after the loop."""
from services.foreign_tracks import sweep as sw
from services.foreign_tracks.state import SweepState
paths = []
for name in ("e1.mkv", "e2.mkv"):
(tmp_path / name).write_bytes(b"new")
(tmp_path / (name + ".bak")).write_bytes(b"old")
paths.append(str(tmp_path / name))
class _Row:
def __init__(self, path):
self.path = path
self.track_count = 1
rows = [_Row(p) for p in reversed(paths)]
per_file, batches = [], []
class _Repo:
def claim_next_affected(self):
return rows.pop() if rows else None
def mark_stripped(self, path):
assert batches == [], "the refresh must come after the slice's rewrites"
def mark_failed(self, *a, **k):
raise AssertionError("must not fail")
def mark_probed_verdicts(self, *a, **k):
pass
monkeypatch.setattr(sw, "_media_root_reachable", lambda root: True)
monkeypatch.setattr(sw, "_free_bytes", lambda root: 10**15)
monkeypatch.setattr(sw, "_strip_file", lambda path, *a, **k: (path + ".bak", 10))
monkeypatch.setattr(sw, "_real_sidecars_for_policy", lambda *a, **k: set())
monkeypatch.setattr(
"services.media_server_notify.notify_media_servers",
lambda path, item_type="": per_file.append(path),
)
monkeypatch.setattr(
"services.media_server_notify.notify_media_servers_batch",
lambda p: batches.append(list(p)),
raising=False,
)
clock = iter(range(0, 10_000))
sw._strip_phase(
str(tmp_path),
{},
SweepState(phase="strip", enumeration_complete=True),
_Repo(),
_KEEP,
False,
_B,
[],
10_000,
lambda: next(clock),
{"stripped_files": 0, "tracks_removed": 0, "bytes_freed": 0},
(),
)
assert per_file == []
assert batches == [paths]
def test_a_failing_refresh_never_breaks_the_slice(tmp_path, monkeypatch):
def boom(paths):
raise RuntimeError("media server down")
exists, _, result, _ = _sweep_strip(
tmp_path, monkeypatch, {"delete_original_after_verify": True}, verified=True, batch=boom
)
assert result["stripped_files"] == 1
assert result["backups_deleted"] == 1
assert not exists