backend/tests/test_media_dir_temp_writers.py
Rama master · réplica de lectura del repositorio original.
"""Regression: files written into the media library through a same-directory temp.
Four writers created their work file with a bare ``tempfile.mkstemp(dir=...)``
next to the episode and ``os.replace``d it into place:
* ``ass_probe.extract_subtitle_stream`` (every embedded-track extraction)
* the subtitle-repair pass after an embedded extraction
* ``POST /tools/diff/apply``
* the ``remux_track`` subtitle-health fixer (rewrites the video itself)
``mkstemp`` creates its file 0600, so the result landed ``-rw-------``. On a
library with POSIX ACLs that is worse than it looks: chmod 600 also sets the
ACL mask to ``---``, which cancels every named ACL entry. Prod had 1 417
sidecars the media server (Emby, a different uid) could not open.
The same writers used the default ``tmp`` prefix, so a process killed mid-write
(container stop, host shutdown) left ``tmpXXXXXXXX.srt`` in the library for
good — prod had seven, all zero bytes, all dated to an outage.
"""
from __future__ import annotations
import os
import stat
import sys
import time
from unittest.mock import MagicMock, patch
import pytest
posix_only = pytest.mark.skipif(
sys.platform == "win32", reason="POSIX file-mode semantics do not apply on Windows"
)
SRT = b"1\n00:00:01,000 --> 00:00:02,000\nhello\n"
def _mode(path) -> int:
return stat.S_IMODE(os.stat(path).st_mode)
def _age(path, seconds: float) -> None:
past = time.time() - seconds
os.utime(path, (past, past))
def _fake_ffmpeg(captured: dict):
def fake_run(cmd, **kwargs):
captured["tmp_out"] = cmd[-1]
with open(cmd[-1], "wb") as fh:
fh.write(SRT)
return MagicMock(returncode=0, stderr="")
return fake_run
class TestExtractSubtitleStream:
@posix_only
def test_extracted_sidecar_is_world_readable(self, tmp_path, app_ctx):
from ass_probe import extract_subtitle_stream
mkv = tmp_path / "show.mkv"
mkv.write_bytes(b"\x1a\x45\xdf\xa3")
out = tmp_path / "show.de.srt"
old = os.umask(0o022)
try:
with patch("ass_probe.subprocess.run", side_effect=_fake_ffmpeg({})):
extract_subtitle_stream(str(mkv), {"sub_index": 0, "format": "srt"}, str(out))
finally:
os.umask(old)
assert _mode(out) == 0o644, f"expected 0644, got {oct(_mode(out))}"
def test_temp_file_carries_the_sublarr_prefix(self, tmp_path, app_ctx):
from ass_probe import extract_subtitle_stream
mkv = tmp_path / "show.mkv"
mkv.write_bytes(b"\x1a\x45\xdf\xa3")
captured: dict = {}
with patch("ass_probe.subprocess.run", side_effect=_fake_ffmpeg(captured)):
extract_subtitle_stream(
str(mkv), {"sub_index": 0, "format": "srt"}, str(tmp_path / "show.de.srt")
)
assert os.path.basename(captured["tmp_out"]).startswith(".sublarr-")
def test_stale_orphaned_temps_are_swept_before_extracting(self, tmp_path, app_ctx):
from ass_probe import extract_subtitle_stream
mkv = tmp_path / "show.mkv"
mkv.write_bytes(b"\x1a\x45\xdf\xa3")
stale_ours = tmp_path / ".sublarr-k2j4h5g6.srt"
stale_ours.write_bytes(b"partial")
_age(stale_ours, 7 * 3600)
stale_legacy = tmp_path / "tmp6fyuxlvz.srt"
stale_legacy.write_bytes(b"")
_age(stale_legacy, 7 * 3600)
fresh_ours = tmp_path / ".sublarr-inflight.srt" # another worker, right now
fresh_ours.write_bytes(b"partial")
legacy_with_content = tmp_path / "tmpabcdefgh.srt" # not provably ours
legacy_with_content.write_bytes(SRT)
_age(legacy_with_content, 7 * 3600)
user_file = tmp_path / "tmp.srt"
user_file.write_bytes(b"")
_age(user_file, 7 * 3600)
with patch("ass_probe.subprocess.run", side_effect=_fake_ffmpeg({})):
extract_subtitle_stream(
str(mkv), {"sub_index": 0, "format": "srt"}, str(tmp_path / "show.de.srt")
)
assert not stale_ours.exists()
assert not stale_legacy.exists()
assert fresh_ours.exists()
assert legacy_with_content.exists()
assert user_file.exists()
class TestSweepOrphanedTemps:
def test_never_touches_directories_or_video_temps_of_a_running_remux(self, tmp_path):
from utils.atomic_write import sweep_orphaned_temps
trash_dir = tmp_path / ".sublarr"
trash_dir.mkdir()
_age(trash_dir, 7 * 3600)
running_remux = tmp_path / ".sublarr-remux-abc.mkv"
running_remux.write_bytes(b"x")
assert sweep_orphaned_temps(str(tmp_path)) == 0
assert trash_dir.is_dir()
assert running_remux.exists()
def test_missing_directory_is_not_an_error(self, tmp_path):
from utils.atomic_write import sweep_orphaned_temps
assert sweep_orphaned_temps(str(tmp_path / "gone")) == 0
@posix_only
def test_repair_after_extract_keeps_sidecar_readable(tmp_path):
from services.embedded_extractor import _repair_sidecar_file
sub = tmp_path / "show.de.srt"
sub.write_bytes(b"\xef\xbb\xbf" + SRT.replace(b"\n", b"\r\n"))
os.chmod(sub, 0o644)
old = os.umask(0o022)
try:
changed = _repair_sidecar_file(str(sub))
finally:
os.umask(old)
assert changed is True, "fixture must actually need a repair"
assert _mode(sub) == 0o644, f"expected 0644, got {oct(_mode(sub))}"
@posix_only
def test_diff_apply_keeps_sidecar_readable(client, tmp_path):
from tests.test_diff_endpoint import MOD_ASS, ORIG_ASS
sub = tmp_path / "test.de.ass"
sub.write_text(ORIG_ASS, encoding="utf-8")
os.chmod(sub, 0o644)
old = os.umask(0o022)
try:
resp = client.post(
"/api/v1/tools/diff/apply",
json={
"file_path": str(sub),
"original": ORIG_ASS,
"modified": MOD_ASS,
"rejected_indices": [],
"last_modified": os.path.getmtime(sub),
},
)
finally:
os.umask(old)
assert resp.status_code == 200, resp.get_json()
assert _mode(sub) == 0o644, f"expected 0644, got {oct(_mode(sub))}"
@posix_only
def test_remux_track_keeps_the_videos_mode(tmp_path, app_ctx):
from services.subtitle_health.fixers import remux_track
video = tmp_path / "x.mkv"
video.write_bytes(b"original")
os.chmod(video, 0o664)
fake_json = {
"tracks": [
{"id": 0, "type": "video", "codec": "AVC/H.264"},
{"id": 1, "type": "subtitles", "codec": "SubRip/SRT", "properties": {}},
]
}
def fake_mkvmerge(src, out, *args, **kwargs):
with open(out, "wb") as fh:
fh.write(b"remuxed")
return True
prefix = "services.subtitle_health.fixers.remux_track"
with (
patch(f"{prefix}._mkvmerge_identify", return_value=fake_json),
patch(f"{prefix}.extract_track_raw", return_value=SRT),
patch(f"{prefix}._run_mkvmerge_replace", side_effect=fake_mkvmerge),
patch(f"{prefix}._make_backup", return_value="/trash/x.mkv"),
patch(f"{prefix}._validate_remux", return_value=None),
):
res = remux_track.apply(str(video), sub_index=0, codec="subrip", lang="ger")
assert res["changed"] is True
assert video.read_bytes() == b"remuxed"
assert _mode(video) == 0o664, f"expected 0664, got {oct(_mode(video))}"
class TestSweepIsCheap:
"""Cold review (Codex, 2026-09-17): the sweep ran on every atomic write and
stat-ed every entry of the directory first."""
def test_the_sweep_is_throttled_per_directory(self, tmp_path, monkeypatch):
import utils.atomic_write as aw
clock = {"now": 10_000.0}
monkeypatch.setattr(aw, "_now", lambda: clock["now"])
monkeypatch.setattr(aw, "_swept_at", {})
first = tmp_path / ".sublarr-one.srt"
first.write_bytes(b"x")
_age(first, 7 * 3600)
assert aw.sweep_orphaned_temps(str(tmp_path)) == 1
second = tmp_path / ".sublarr-two.srt"
second.write_bytes(b"x")
_age(second, 7 * 3600)
assert aw.sweep_orphaned_temps(str(tmp_path)) == 0, "a second sweep right after is skipped"
assert second.exists()
clock["now"] += aw._SWEEP_INTERVAL_S + 1
assert aw.sweep_orphaned_temps(str(tmp_path)) == 1
assert not second.exists()
def test_only_candidate_names_are_inspected(self, tmp_path, monkeypatch):
import utils.atomic_write as aw
monkeypatch.setattr(aw, "_swept_at", {})
for i in range(5):
(tmp_path / f"Episode {i}.mkv").write_bytes(b"v")
stale = tmp_path / ".sublarr-x.srt"
stale.write_bytes(b"x")
_age(stale, 7 * 3600)
stat_calls = []
real_stat = os.stat
def counting_stat(path, *args, **kwargs):
stat_calls.append(str(path))
return real_stat(path, *args, **kwargs)
monkeypatch.setattr(aw.os, "stat", counting_stat)
assert aw.sweep_orphaned_temps(str(tmp_path)) == 1
assert not [c for c in stat_calls if c.endswith(".mkv")], stat_calls