Files
rippy/docker/worker/test_medien.py
T
HitonabiandClaude Opus 5 f4a8d77598
Ampel / ampel (push) Failing after 48s
feat(windows): Rippy arbeitet eigenstaendig — Rippen, Werkzeuge, Fähigkeiten
WAS: Der Ablauf ist aus tasks.py heraus (ablauf.py, ohne Celery), die
LocalQueue wird bedient, ein Laeufer arbeitet Auftraege im selben Prozess
ab, und Rippy meldet sich mit gemessenen Faehigkeiten selbst als Arbeiter.

WARUM: "Rippy fuer Windows soll standalone funktionieren" (Commander). Bis
hierher konnte die Windows-App alles ANZEIGEN und nichts TUN — ein Rip waere
eingereiht worden und fuer immer liegengeblieben, weil niemand ihn holt.

DIE TRENNUNG: rip_disc hing an GENAU DREI Celery-Stellen in 234 Zeilen —
self.update_state, _transcode_queue, transcode_files.apply_async. Alle drei
sind Fragen der ZUSTELLUNG, nicht des Ablaufs. Sie sind jetzt Rueckrufe:
tasks.py reicht die Celery-Fassung herein, standalone.py die lokale. OHNE
Rueckruf komprimiert derselbe Prozess weiter — genau das, was ein
Ein-Prozess-Rippy braucht. Der Ablauf selbst ist Zeile fuer Zeile derselbe;
der Docker-Betrieb merkt vom Umbau nichts (Task-Namen, Argumente, Queues
unveraendert).

EINE ZUSTELL-STELLE statt drei: celery_client.abschicken() bedient alle
Auftragsarten. Vorher rief jede Stelle send_task selbst auf — der
Standalone-Betrieb haette an drei Stellen umgebogen werden muessen, beim
naechsten Auftragstyp an einer vierten.

WEITERER BLOCKER GEFUNDEN: ablauf.py holte detect_disc_type fest aus dem
LINUX-Treiber, in einem try/except. Unter Windows waere es damit IMMER None
gewesen und Rippen "hart verriegelt" — Rippy haette alles angezeigt und
nichts gerippt, ohne dass irgendwo ein Fehler stuende. Jetzt fragt es den
Treiber-Port.

WERKZEUGE: ripping.py und caps.py suchten nur im PATH. Auf dem Commander-PC
gemessen, vorher/nachher:

  vorher   check_makemkv_installed()  -> False (obwohl installiert)
           erkenne_encoder()          -> nur CPU
  nachher  MakeMKV   1.18.4  C:\Program Files (x86)\MakeMKV\makemkvcon64.exe
           HandBrake 1.11.2  ueber die API geholt, in 2,7 s
           Encoder   cpu-x264, cpu-x265, cpu-av1, VCE, VCE-AV1
           107 Presets, Ryzen 7 9700X, 16 Kerne, avx512f

VCE ist die Hardwarebeschleunigung der Radeon — die hat Rippy auf diesem
Rechner vorher nie gesehen, weil es HandBrake gar nicht fand.

NEUE ROUTEN: GET /system/werkzeuge (was liegt wo, in welcher Fassung, gibt
es Neueres) und POST /system/werkzeuge/{name}/holen. HandBrake kommt
vollautomatisch von GitHub. MakeMKV wird NICHT mitgeliefert — Rippy laedt
die offizielle Datei und startet sie (Black-Box-Trennung, KONZEPT.md § 6).
makemkv.com antwortete beim Bauen mit HTTP 525; das wird im Klartext
gemeldet, und eine selbst geholte Datei bleibt moeglich.

HERZSCHLAG: /capabilities las die workers-Tabelle, die bisher nur der
Celery-Herzschlag fuellte. Im Standalone-Betrieb stand dort "0 Worker" und
die Encoder-Auswahl im UI blieb LEER — auf einem Rechner, der alles kann.
Jetzt meldet sich der Prozess selbst, mit dem, was caps.py MISST.

GEMESSEN, aus der fertigen EXE (29,0 MB):
  bereit nach 1 s, keine Fehler im Log
  Werkzeuge: beide gefunden, mit Version und Pfad
  Worker: 1 (TobisNicerPC), 5 Encoder, 107 Presets

Die ganze Kette ist als Test festgehalten (test_kette.py): zustellen ->
einreihen -> Laeufer -> ablauf -> Job endet in einem EHRLICHEN Zustand.
Ohne Laufwerk geprueft, und das ist der wichtigere Fall: Ein Rip auf ein
totes Geraet muss zuegig scheitern, nicht auf "pending" haengenbleiben.

GEMESSEN: ruff sauber, 512 Tests gruen + 15 uebersprungen (vorher 489).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-28 11:27:54 +02:00

176 lines
7.1 KiB
Python

"""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 ablauf 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"
# Die beiden früheren Tests zu _original_aufheben sind nach
# test_original_aufheben.py gewandert und dort erweitert. Sie prüfen jetzt das
# richtige Kriterium: Vorher taeuschten sie per gefaelschtem os.stat
# „verschiedene Dateisysteme" vor — genau die Annahme, an der der Schutz in der
# Praxis scheiterte (st_dev war identisch, os.rename trotzdem EXDEV).
def test_arbeitsverzeichnis_wahl_des_rips_schlaegt_die_einstellung():
"""Pro Rip wählbar (Commander 25.07.2026), Einstellung bleibt Standard.
Reihenfolge: Wahl dieses Rips -> Setting -> Container-Default. Der
Setting-Wert ist genau der, der bei Vollautomatik-Rips greift, weil dort
niemand gefragt wird.
"""
import ablauf as tasks
einst = {"workDir": "/app/media/movies"}
assert tasks._arbeitsverzeichnis(einst, "/app/media/rippy") == "/app/media/rippy"
assert tasks._arbeitsverzeichnis(einst) == "/app/media/movies"
assert tasks._arbeitsverzeichnis({}) == tasks.RAW_DIR
# Ausbruchsversuche und Pfade außerhalb /app/media fallen durch
assert tasks._arbeitsverzeichnis({}, "/etc") == tasks.RAW_DIR
assert tasks._arbeitsverzeichnis({}, "/app/media/../etc") == tasks.RAW_DIR
# Leere Wahl fällt sauber auf die Einstellung zurück
assert tasks._arbeitsverzeichnis(einst, " ") == "/app/media/movies"
def test_unter_wurzel_faellt_nicht_auf_praefix_namen_herein():
"""Befund 25.07.2026: Elf Stellen prüften mit nacktem startswith().
„/app/media-boese/x" beginnt mit „/app/media", liegt aber außerhalb."""
import ablauf as tasks
assert tasks.unter_wurzel("/app/media", "/app/media") is True
assert tasks.unter_wurzel("/app/media/movies", "/app/media") is True
assert tasks.unter_wurzel("/app/media/movies/Film (2020)", "/app/media") is True
# Der Fall, den startswith() durchgelassen haette
assert tasks.unter_wurzel("/app/media-boese/x", "/app/media") is False
assert tasks.unter_wurzel("/app/mediaX", "/app/media") is False
assert tasks.unter_wurzel("/etc/passwd", "/app/media") is False
assert tasks.unter_wurzel("", "/app/media") is False
assert tasks.unter_wurzel("/app/media", "") is False
# Wurzel mit Schrägstrich am Ende darf nichts ändern
assert tasks.unter_wurzel("/app/media/movies", "/app/media/") is True
def test_zielbasis_lehnt_praefix_ausbruch_ab():
import ablauf as tasks
assert tasks._zielbasis("/app/media/movies", "bluray") == "/app/media/movies"
# Ausbruch per Praefix-Namen fällt auf den Standard zurück
assert tasks._zielbasis("/app/media-boese", "bluray") != "/app/media-boese"
assert tasks._zielbasis("/etc", "bluray") != "/etc"
def test_arbeitsverzeichnis_lehnt_praefix_ausbruch_ab():
import ablauf as tasks
assert tasks._arbeitsverzeichnis({}, "/app/media-boese") == tasks.RAW_DIR
assert tasks._arbeitsverzeichnis({"workDir": "/app/mediaX"}) == tasks.RAW_DIR