"""Tests der Rohdaten-Suche — mit dem echten Fall, der sie nötig gemacht hat. Job `95afdc89` (26.07.2026, an der laufenden Instanz gemessen): 79,6 GB Rohschnitt unter /app/media/rippy/, Einstellung `workDir` leer, `can_retry` = false. Der Knopf „Neu komprimieren" fehlte, obwohl die Datei intakt war. """ import rohdaten JOB = "95afdc89-2426-4d44-829e-ad6ce1411905" def test_der_echte_fall_wird_gefunden(): """workDir ist LEER (so stand es live) und der Rohschnitt liegt trotzdem auf der NAS — weil beim Start eine Wahl nur für diesen Rip getroffen wurde.""" orte = rohdaten.kandidaten(JOB, "", ["movies", "music", "rippy", "series"]) assert f"/app/media/rippy/{JOB}" in orte # Der Container-Standard bleibt der erste Kandidat (schnellster Treffer) assert orte[0] == f"/app/temp/raw/{JOB}" def test_suche_liefert_nur_was_existiert(): vorhanden = {f"/app/media/rippy/{JOB}"} gefunden = rohdaten.suche( JOB, "", listdir=lambda p: ["movies", "rippy"], isdir=lambda p: p in vorhanden, ) assert gefunden == [f"/app/media/rippy/{JOB}"] def test_eingestelltes_arbeitsverzeichnis_kommt_vor_den_zielen(): orte = rohdaten.kandidaten(JOB, "/app/media/rippy", ["movies", "rippy"]) assert orte[1] == f"/app/media/rippy/{JOB}" # ... und taucht nicht doppelt auf, obwohl „rippy" auch Ablageziel ist assert orte.count(f"/app/media/rippy/{JOB}") == 1 def test_arbeitsverzeichnis_ausserhalb_media_wird_ignoriert(): """Nur /app/media ist erlaubt (so entscheidet _arbeitsverzeichnis im Worker) — ein Pfad daneben darf hier nicht durchrutschen.""" orte = rohdaten.kandidaten(JOB, "/etc", []) assert orte == [f"/app/temp/raw/{JOB}"] # Der Klassiker: ein Pfad, der nur mit dem Präfix ANFÄNGT orte = rohdaten.kandidaten(JOB, "/app/media-boese", []) assert orte == [f"/app/temp/raw/{JOB}"] def test_media_root_selbst_ist_erlaubt(): orte = rohdaten.kandidaten(JOB, "/app/media", []) assert f"/app/media/{JOB}" in orte def test_ohne_job_id_nichts(): assert rohdaten.kandidaten("", "/app/media", ["x"]) == [] def test_kaputter_mount_reisst_die_suche_nicht_mit(): """Ein toter CIFS-Mount lässt isdir mit OSError fliegen. Das darf die Job-Liste nicht mitnehmen (Befund 24.07. bei /storage-targets).""" def isdir_kaputt(p): if "totes-nas" in p: raise OSError("Stale file handle") return p == f"/app/temp/raw/{JOB}" gefunden = rohdaten.suche( JOB, "", listdir=lambda p: ["totes-nas", "movies"], isdir=isdir_kaputt) assert gefunden == [f"/app/temp/raw/{JOB}"] def test_listdir_kaputt_faellt_auf_den_standard_zurueck(): def listdir_kaputt(p): raise OSError("kein /app/media") gefunden = rohdaten.suche( JOB, "", listdir=listdir_kaputt, isdir=lambda p: True) assert gefunden == [f"/app/temp/raw/{JOB}"] # --- Größe ------------------------------------------------------------------ def test_groesse_zaehlt_nur_dateien(): dateien = { f"/app/media/rippy/{JOB}/title_t00.mkv": 79604951639, f"/app/media/rippy/{JOB}/title_t01.mkv": 1000, } bytes_gesamt, anzahl = rohdaten.groesse( [f"/app/media/rippy/{JOB}"], listdir=lambda p: ["title_t00.mkv", "title_t01.mkv", "unterordner"], isfile=lambda p: p in dateien, getsize=lambda p: dateien[p], ) # Die echte Größe des Akira-Rohschnitts, plus eine zweite Datei assert bytes_gesamt == 79604952639 assert anzahl == 2 assert round(bytes_gesamt / 1024**3, 1) == 74.1 def test_groesse_ohne_pfade_ist_null(): assert rohdaten.groesse([], lambda p: [], lambda p: True, lambda p: 1) == (0, 0) assert rohdaten.groesse(None, lambda p: [], lambda p: True, lambda p: 1) == (0, 0) def test_groesse_ueberspringt_unlesbares(): def getsize_kaputt(p): raise OSError("weg") bytes_gesamt, anzahl = rohdaten.groesse( ["/x"], lambda p: ["a.mkv"], lambda p: True, getsize_kaputt) assert (bytes_gesamt, anzahl) == (0, 0) # --- Die harte Zeitgrenze: ein haengender Mount darf nichts toeten ----------- # # Am 26.07.2026 hing os.path.isdir an der CIFS-Freigabe im # KERNEL (Prozess-Zustand D). asyncio.to_thread kam nie zurueck, die # Hintergrund-Schleife erreichte ihr sleep nie und war dauerhaft tot. class _Lauf: """Merkt sich das Kommando und liefert einen gesetzten Rueckgabewert.""" def __init__(self, rc=0, wirf=None): self.rc, self.wirf, self.cmd = rc, wirf, None def __call__(self, cmd, **kwargs): self.cmd = cmd if self.wirf: raise self.wirf class E: returncode = self.rc return E() def test_verzeichnis_da_nutzt_timeout_und_ls(): """Ein Kind-PROZESS laesst sich abbrechen, ein im Kernel haengender Thread nicht. Deshalb genau dieses Kommando - wie mounts.ist_erreichbar.""" lauf = _Lauf(rc=0) assert rohdaten.verzeichnis_da("/app/media/rippy/x", lauf) is True assert lauf.cmd[0] == "timeout" assert lauf.cmd[1] == str(rohdaten.PRUEF_TIMEOUT_SEKUNDEN) assert lauf.cmd[2:] == ["ls", "-d", "/app/media/rippy/x"] def test_verzeichnis_da_nicht_vorhanden(): assert rohdaten.verzeichnis_da("/gibt/es/nicht", _Lauf(rc=2)) is False def test_verzeichnis_in_der_zeitgrenze_gilt_als_nicht_da(): """`timeout` beendet ls mit 124. Ein Ort, den man nicht in Sekunden ansehen kann, ist fuer einen Rip ohnehin unbrauchbar.""" assert rohdaten.verzeichnis_da("/app/media/totes-nas/x", _Lauf(rc=124)) is False def test_verzeichnis_da_ueberlebt_kaputte_umgebung(): import subprocess as sp assert rohdaten.verzeichnis_da("/x", _Lauf(wirf=OSError("kein timeout"))) is False assert rohdaten.verzeichnis_da( "/x", _Lauf(wirf=sp.TimeoutExpired("ls", 6))) is False assert rohdaten.verzeichnis_da("", _Lauf(rc=0)) is False def test_suche_mit_der_zeitgrenze_findet_den_echten_fall(): """Zusammenspiel: os.listdir fuer /app/media (lokal, sicher), verzeichnis_da fuer die Kandidaten (koennen im Netz liegen).""" def laufen(cmd, **kwargs): pfad = cmd[-1] class E: returncode = 0 if pfad == f"/app/media/rippy/{JOB}" else 1 return E() gefunden = rohdaten.suche( JOB, "", listdir=lambda p: ["bluray", "movies", "rippy"], isdir=lambda p: rohdaten.verzeichnis_da(p, laufen), ) assert gefunden == [f"/app/media/rippy/{JOB}"]