rippy/docker/worker/test_medien.py

"""Tests für die Media-Server-Aufbereitung (Benennung, NFO, Serien-Matching)."""

import os

from medien import baue_nfo, matche_episoden, serien_ordner, sicherer_name, zielordner


def test_sicherer_name_entfernt_verbotene_zeichen():
    assert sicherer_name('Alien: Die Wiedergeburt') == "Alien Die Wiedergeburt"
    assert sicherer_name('Was/ist\\das?<>|*"') == "Wasistdas"


def test_sicherer_name_mit_jahr():
    assert sicherer_name("Evangelion 2.22", 2009) == "Evangelion 2.22 (2009)"


def test_sicherer_name_kollabiert_leerraum_und_punkte():
    assert sicherer_name("  Viel   Raum ...  ") == "Viel Raum"


def test_sicherer_name_leer_bleibt_leer():
    assert sicherer_name("") == ""
    assert sicherer_name("???") == ""


def test_zielordner_nutzt_titel_und_jahr(tmp_path):
    pfad = zielordner(str(tmp_path), "Inception", 2010, "abc-123")
    assert pfad == os.path.join(str(tmp_path), "Inception (2010)")


def test_zielordner_faellt_auf_job_id_zurueck(tmp_path):
    pfad = zielordner(str(tmp_path), "", None, "abc-123")
    assert pfad == os.path.join(str(tmp_path), "abc-123")


def test_zielordner_weicht_bei_kollision_aus(tmp_path):
    os.makedirs(tmp_path / "Inception (2010)")
    pfad = zielordner(str(tmp_path), "Inception", 2010, "abcdef12-3456")
    assert pfad == os.path.join(str(tmp_path), "Inception (2010) [abcdef12]")


def test_baue_nfo_film_mit_plot_und_genres():
    nfo = baue_nfo(
        {"overview": "Ein Traum <im> Traum & so", "genres": ["Sci-Fi", "Action"]},
        "Inception", 2010,
    )
    assert "<movie>" in nfo and "</movie>" in nfo
    assert "<title>Inception</title>" in nfo
    assert "<year>2010</year>" in nfo
    # XML-Escaping: <, >, & dürfen den Parser nicht sprengen
    assert "Ein Traum &lt;im&gt; Traum &amp; so" in nfo
    assert "<genre>Sci-Fi</genre>" in nfo


def test_baue_nfo_serie_bekommt_tvshow_wurzel():
    nfo = baue_nfo({"type": "tv"}, "Neon Genesis Evangelion", 1995)
    assert "<tvshow>" in nfo and "</tvshow>" in nfo


def test_sicherer_name_haengt_jahr_nicht_doppelt_an():
    """OMDb-Titel tragen das Jahr teils schon — kein 'X (2009) (2009)' mehr."""
    assert sicherer_name("Summer Wars (2009)", 2009) == "Summer Wars (2009)"
    assert sicherer_name("Summer Wars", 2009) == "Summer Wars (2009)"


def test_serien_ordner_jellyfin_schema():
    assert serien_ordner("/app/media/series", "Neon Genesis: Evangelion", 2) == os.path.join(
        "/app/media/series", "Neon Genesis Evangelion", "Season 02"
    )


def test_matche_episoden_komplette_staffel_auch_bei_uniformen_laufzeiten():
    # Anime-Standardfall: ganze Staffel auf der Disc, alle Episoden ~24 min
    dateien = [1440, 1452, 1431]
    episoden = [(1, 1440), (2, 1440), (3, 1440)]
    assert matche_episoden(dateien, episoden) == [1, 2, 3]


def test_matche_episoden_teil_disc_mit_distinkten_laufzeiten():
    # Disc 2 einer Staffel: Laufzeiten unterscheiden sich klar → zuordenbar
    dateien = [1440, 2650]
    episoden = [(1, 700), (2, 1440), (3, 2650)]
    assert matche_episoden(dateien, episoden) == [2, 3]


def test_matche_episoden_haelt_reihenfolge():
    # Datei 2 ist ein Doppellänger → muss auf die SPÄTERE lange Episode fallen
    dateien = [1440, 2880]
    episoden = [(1, 1440), (2, 2880)]
    assert matche_episoden(dateien, episoden) == [1, 2]


def test_matche_episoden_mehrdeutig_gibt_none():
    # Alle Episoden gleich lang und mehr Kandidaten als Dateien in Toleranz →
    # Datei 1 passt auf Ep 1 UND Ep 2 gleich gut → keine eindeutige Wahl
    dateien = [1440]
    episoden = [(1, 1440), (2, 1440)]
    assert matche_episoden(dateien, episoden) is None


def test_matche_episoden_ohne_treffer_gibt_none():
    dateien = [600]  # 10 min passt auf keine 24-min-Episode
    episoden = [(1, 1440), (2, 1440)]
    assert matche_episoden(dateien, episoden) is None


def test_pfad_lokal_uebersetzt_fuer_windows_worker():
    from tasks import pfad_lokal

    mapping = "/app/media=Z:\\media;/app/temp=Y:\\temp"
    assert pfad_lokal("/app/temp/raw/abc", mapping) == "Y:\\temp\\raw\\abc"
    assert pfad_lokal("/app/media/movies/Film (2020)", mapping) == "Z:\\media\\movies\\Film (2020)"
    assert pfad_lokal("/woanders/x", mapping) == "/woanders/x"
    assert pfad_lokal("/app/temp/raw/abc", "") == "/app/temp/raw/abc"


def test_original_aufheben_laesst_job_nie_scheitern(tmp_path, monkeypatch):
    """Der Schritt "Original behalten" darf einen fertigen Job NICHT killen.

    Schaden vom 25.07.2026 (Akira-UHD): shutil.move() ueber Dateisystemgrenzen
    kopierte 74 GB, lief in ENOSPC, und der Job galt als fehlgeschlagen —
    obwohl die komprimierte Datei laengst fertig war. Der Nutzer sah nur eine
    leere Queue.
    """
    import tasks

    meldungen = []
    monkeypatch.setattr(tasks.db, "add_log", lambda *a: meldungen.append(a))

    raw = tmp_path / "raw"
    raw.mkdir()
    (raw / "title_t00.mkv").write_bytes(b"x" * 1024)
    final = tmp_path / "final"
    final.mkdir()

    def platzt(*_a, **_k):
        raise OSError(28, "No space left on device")

    monkeypatch.setattr(tasks.shutil, "move", platzt)
    # Darf NICHT werfen
    tasks._original_aufheben("job-1", str(raw), str(final))
    assert meldungen and meldungen[-1][0] == "warning"
    assert "Roh-Datei bleibt" in meldungen[-1][2]


def test_original_aufheben_warnt_statt_platte_vollzuschreiben(tmp_path, monkeypatch):
    """Zu wenig Platz auf einem ANDEREN Dateisystem -> gar nicht erst anfangen."""
    import tasks

    meldungen = []
    monkeypatch.setattr(tasks.db, "add_log", lambda *a: meldungen.append(a))
    bewegt = []
    monkeypatch.setattr(tasks.shutil, "move", lambda *a: bewegt.append(a))

    raw = tmp_path / "raw"
    raw.mkdir()
    (raw / "gross.mkv").write_bytes(b"x" * 4096)
    final = tmp_path / "final"
    final.mkdir()

    # Verschiedene Dateisysteme vortaeuschen und den freien Platz kleinrechnen.
    # _ordner_groesse wird mitgefaelscht, weil ein selbstgebautes os.stat sonst
    # os.path.getsize mitreisst (das liest st_size an derselben Stelle).
    class Stat:
        def __init__(self, dev): self.st_dev = dev
    monkeypatch.setattr(tasks.os, "stat", lambda p: Stat(1 if "raw" in str(p) else 2))
    monkeypatch.setattr(tasks, "_ordner_groesse", lambda _p: 4096)
    monkeypatch.setattr(tasks, "_frei_bytes", lambda _p: 100)

    tasks._original_aufheben("job-2", str(raw), str(final))
    assert bewegt == [], "es darf NICHT kopiert werden, wenn der Platz nicht reicht"
    assert meldungen[-1][0] == "warning"
    assert "NICHT aufgehoben" in meldungen[-1][2]