"""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}"] # --- Drei Antworten: "konnte nicht nachsehen" ist nicht "ist weg" ------------ def test_pruefen_unterscheidet_drei_faelle(): assert rohdaten.pruefen("/x", _Lauf(rc=0)) == "da" assert rohdaten.pruefen("/x", _Lauf(rc=2)) == "weg" # 124 = `timeout` hat das Kind abgeschossen (coreutils) -> nicht angesehen assert rohdaten.pruefen("/x", _Lauf(rc=124)) == "unklar" assert rohdaten.pruefen("/x", _Lauf(wirf=OSError("kein timeout"))) == "unklar" assert rohdaten.pruefen("", _Lauf(rc=0)) == "weg" def test_verzeichnis_da_bleibt_streng(): """Wer Ja/Nein braucht, bekommt im Zweifel Nein.""" assert rohdaten.verzeichnis_da("/x", _Lauf(rc=0)) is True assert rohdaten.verzeichnis_da("/x", _Lauf(rc=124)) is False def test_suche_mit_status_meldet_ungepruefte_orte(): """Der echte Fall: Die Freigabe antwortet nicht, der lokale Ort ist leer. Ein leeres Ergebnis darf dann NICHT als "nichts da" gelten.""" def pruefe(pfad): return "unklar" if pfad.startswith("/app/media/rippy/") else "weg" e = rohdaten.suche_mit_status( JOB, "", listdir=lambda p: ["rippy"], pruefer=pruefe) assert e == {"pfade": [], "unklar": True} def test_suche_mit_status_ohne_zweifel(): def pruefe(pfad): return "da" if pfad == f"/app/media/rippy/{JOB}" else "weg" e = rohdaten.suche_mit_status( JOB, "", listdir=lambda p: ["movies", "rippy"], pruefer=pruefe) assert e == {"pfade": [f"/app/media/rippy/{JOB}"], "unklar": False} def test_suche_mit_status_findet_trotz_unklarem_anderen_ort(): """Ein Treffer bleibt ein Treffer, auch wenn ein anderer Ort schweigt.""" def pruefe(pfad): if pfad == f"/app/media/rippy/{JOB}": return "da" return "unklar" if "totes-nas" in pfad else "weg" e = rohdaten.suche_mit_status( JOB, "", listdir=lambda p: ["rippy", "totes-nas"], pruefer=pruefe) assert e["pfade"] == [f"/app/media/rippy/{JOB}"] assert e["unklar"] is True