Ampel / ampel (push) Failing after 48s
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>
176 lines
7.1 KiB
Python
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 <im> Traum & 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
|