"""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}" # Die Wurzeln des CONTAINER-Betriebs, eingespritzt. Ohne sie hingen diese # Tests am laufenden Rechner: Unter Windows liefert `wurzeln()` echte # Windows-Pfade, und die Erwartungen hier gelten dort nicht (am 29.08.2026 # prompt sieben Tests rot geworden). CONTAINER = ("/app/temp/raw", "/app/media", False) 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, orte_wurzeln=CONTAINER) 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 #: Die Wurzeln eines NATIVEN Betriebs (Windows, freies Blättern) — #: Gegenstück zu CONTAINER weiter oben. NATIV = ("C:\\Rippy\\_arbeit", "C:\\Rippy", True) def test_laufwerks_wurzel_bleibt_absolut(): """Der Fall des Commanders (30.08.2026): Arbeitsordner F: — die Wurzel. Hier wurde der Schluss-Trenner abgestreift und mit dem Rest dann auch VERBUNDEN. Ein blosses "F:" ist unter Windows aber der AKTUELLE Ordner auf Laufwerk F, nicht dessen Wurzel — die Suche sah damit an einer ganz anderen Stelle nach. Ergebnis: 16,5 GB Rohschnitt unsichtbar, und der Wiederholen-Dialog bot nur "Neu rippen" an: Stunden am beschädigten Datenträger für etwas, das schon dalag. """ orte = rohdaten.kandidaten(JOB, "F:\\", [], NATIV) assert "F:" + chr(92) + JOB in orte assert "F:" + JOB not in orte # Ohne Schluss-Trenner muss dasselbe herauskommen assert rohdaten.kandidaten(JOB, "F:\\Roh\\", [], NATIV)[-1] == ( "F:" + chr(92) + "Roh" + chr(92) + JOB) def test_media_root_mit_schluss_trenner_zaehlt_auch(): """/app/media/ und /app/media sind derselbe Ort — der Vergleich darf nicht am Trenner scheitern.""" orte = rohdaten.kandidaten(JOB, "/app/media/", [], CONTAINER) 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, orte_wurzeln=CONTAINER) 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, orte_wurzeln=CONTAINER) 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), orte_wurzeln=CONTAINER) 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, orte_wurzeln=CONTAINER) 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, orte_wurzeln=CONTAINER) 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, orte_wurzeln=CONTAINER) assert e["pfade"] == [f"/app/media/rippy/{JOB}"] assert e["unklar"] is True def test_nativ_wird_kein_prozess_gestartet(monkeypatch, tmp_path): """Befund 30.08.2026, an der laufenden Instanz beobachtet: 14:54:40 timeout.exe timeout 4 ls -d C:...d7ee6c06-... 14:54:40 WindowsTerminal.exe Der Commander sah Fenster aufblitzen. `timeout` und `ls` sind Linux-Befehle; die Windows-eigene timeout.exe kennt weder `ls` noch `-d`, braucht aber eine Konsole. Dreifach falsch: ein Fenster je Durchlauf, ein Prozess fuer nichts, und Rueckgabewert ungleich 0 — also die Antwort weg fuer JEDES Verzeichnis. """ gestartet = [] monkeypatch.setattr(rohdaten, "nativ_nachsehen", lambda: True) monkeypatch.setattr(rohdaten.subprocess, "run", lambda *a, **k: gestartet.append(a)) assert rohdaten.pruefen(str(tmp_path)) == "da" assert rohdaten.pruefen(str(tmp_path / "gibt-es-nicht")) == "weg" assert gestartet == [], "es darf kein Prozess gestartet werden" def test_im_container_bleibt_der_kindprozess(monkeypatch): """Dort ist der Umweg richtig: os.path.isdir kann an einem toten CIFS-Mount im Kernel haengen (Begruendung in verzeichnis_da).""" monkeypatch.setattr(rohdaten, "nativ_nachsehen", lambda: False) aufrufe = [] class Antwort: returncode = 0 monkeypatch.setattr(rohdaten.subprocess, "run", lambda *a, **k: aufrufe.append(a[0]) or Antwort()) assert rohdaten.pruefen("/app/media/x") == "da" assert aufrufe[0][0] == "timeout" assert aufrufe[0][-1] == "/app/media/x"