Ctrl K
ComunidadGuías…

backend/tests/test_foreign_tracks_preview_file.py

Rama master · réplica de lectura del repositorio original.

Archivo original ↗Preguntar a KiBot
"""Read-only per-track foreign-track cleanup preview for one video file.

`POST /api/v1/foreign-tracks/preview-file` must reach the SAME keep/strip
verdict a real strip would (via `services.foreign_track_cleanup.maybe_run_
foreign_track_cleanup`) and must never touch the file. Both entry points
share `services.foreign_track_cleanup.keep_tags_for` for the keep-set, so a
test here also pins that they actually agree for the same item.
"""

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest


def test_preview_lists_every_track_with_its_reason(client, tmp_path, monkeypatch):
    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    probe = {
        "streams": [
            {
                "index": 2,
                "codec_type": "subtitle",
                "codec_name": "subrip",
                "tags": {"language": "fre", "title": "French"},
            },
        ]
    }
    with patch("remux.get_media_streams", return_value=probe):
        resp = client.post("/api/v1/foreign-tracks/preview-file", json={"path": str(video)})
    assert resp.status_code == 200
    body = resp.get_json()
    assert body["path"] == str(video)
    assert "policy" in body
    assert body["verdicts"][0]["reason"] == "stripped_language"


def test_preview_refuses_paths_outside_the_media_root(client):
    resp = client.post("/api/v1/foreign-tracks/preview-file", json={"path": "/etc/passwd"})
    assert resp.status_code == 403


def test_preview_requires_a_path(client):
    resp = client.post("/api/v1/foreign-tracks/preview-file", json={})
    assert resp.status_code == 400


def test_preview_404s_on_a_missing_file(client, tmp_path, monkeypatch):
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    resp = client.post(
        "/api/v1/foreign-tracks/preview-file",
        json={"path": str(tmp_path / "nope.mkv")},
    )
    assert resp.status_code == 404


def test_preview_never_remuxes(client, tmp_path, monkeypatch):
    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    with (
        patch("remux.get_media_streams", return_value={"streams": []}),
        patch("remux.remove_subtitle_streams") as strip,
        patch("remux.remove_foreign_subtitle_streams") as strip_foreign,
    ):
        client.post("/api/v1/foreign-tracks/preview-file", json={"path": str(video)})
    assert not strip.called
    assert not strip_foreign.called


@pytest.mark.parametrize(
    "body",
    [
        {"series_id": "abc"},
        {"series_id": -1},
        {"series_id": 3.5},
        {"series_id": [1, 2]},
        {"series_id": True},  # JSON true -- bool is an int subclass, must not bind as id 1
        {"movie_id": "abc"},
        {"movie_id": -1},
        {"movie_id": 3.5},
        {"movie_id": [1, 2]},
        {"movie_id": False},  # JSON false -- must not silently resolve as "no id"
        {"series_id": 1, "movie_id": 1},  # mutually exclusive
        {"path": 123},  # non-string path
        {"path": ""},  # empty string path
    ],
)
def test_preview_rejects_invalid_series_or_movie_ids(client, tmp_path, monkeypatch, body):
    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    payload = {"path": str(video), **body}
    with patch("remux.get_media_streams", return_value={"streams": []}):
        resp = client.post("/api/v1/foreign-tracks/preview-file", json=payload)
    assert resp.status_code == 400, f"body={body!r} got {resp.status_code}: {resp.get_json()}"


def test_preview_accepts_valid_series_id_and_resolves_the_override(client, tmp_path, monkeypatch):
    from services.foreign_tracks.select import TrackPolicy

    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    override_policy = TrackPolicy(mode="one_per_language", keep_sdh=True)
    with (
        patch("remux.get_media_streams", return_value={"streams": []}),
        patch(
            "services.foreign_tracks.policy.resolve_policy", return_value=override_policy
        ) as resolve,
    ):
        resp = client.post(
            "/api/v1/foreign-tracks/preview-file",
            json={"path": str(video), "series_id": 42},
        )
    assert resp.status_code == 200
    resolve.assert_called_once_with(series_id=42, movie_id=None)
    assert resp.get_json()["policy"]["mode"] == "one_per_language"
    assert resp.get_json()["policy"]["keep_sdh"] is True


def test_preview_accepts_valid_movie_id_and_resolves_the_override(client, tmp_path, monkeypatch):
    from services.foreign_tracks.select import TrackPolicy

    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)
    override_policy = TrackPolicy(mode="one_per_language")
    with (
        patch("remux.get_media_streams", return_value={"streams": []}),
        patch(
            "services.foreign_tracks.policy.resolve_policy", return_value=override_policy
        ) as resolve,
    ):
        resp = client.post(
            "/api/v1/foreign-tracks/preview-file",
            json={"path": str(video), "movie_id": 7},
        )
    assert resp.status_code == 200
    resolve.assert_called_once_with(series_id=None, movie_id=7)


def test_preview_and_hook_reach_the_same_keep_set(client, tmp_path, monkeypatch):
    """The preview must reach the SAME keep-set the real post-download hook
    would for the same item — both go through `keep_tags_for`, never a
    re-implementation local to the route.
    """
    from services import foreign_track_cleanup as ftc

    fake_settings = SimpleNamespace(
        media_path=str(tmp_path),
        cleanup_foreign_tracks_default=True,
        cleanup_foreign_tracks_keep_und=True,
        cleanup_foreign_tracks_keep_languages=["de", "en"],
    )

    video = tmp_path / "e1.mkv"
    video.write_bytes(b"v")
    # Same item shape the preview constructs from its request body — no
    # missing_languages/target_language, since the preview never sees those.
    item = {"sonarr_series_id": None, "radarr_movie_id": None}

    calls: list[tuple[set, set]] = []
    real_keep_tags_for = ftc.keep_tags_for

    def spy(item_arg, target_languages=None):
        result = real_keep_tags_for(item_arg, target_languages)
        calls.append(result)
        return result

    monkeypatch.setattr(ftc, "keep_tags_for", spy)
    monkeypatch.setattr("routes.foreign_tracks_preview.is_safe_path", lambda *a, **k: True)

    strip = MagicMock(return_value=None)
    with (
        patch("config.get_settings", return_value=fake_settings),
        patch("remux.get_media_streams", return_value={"streams": []}),
        patch("remux.remove_foreign_subtitle_streams", strip),
    ):
        resp = client.post("/api/v1/foreign-tracks/preview-file", json={"path": str(video)})
        assert resp.status_code == 200

        ftc.maybe_run_foreign_track_cleanup(item, str(video))

    assert len(calls) == 2, "both the preview and the hook must call keep_tags_for"
    assert calls[0] == calls[1], "preview and hook must resolve the same keep-set"