Hitonabi
146 lines
5.6 KiB
Python
146 lines
5.6 KiB
Python
"""Tests für die Erreichbarkeits-Prüfung vor der Kompression.
|
|
|
|
Diese Prüfung hatte KEINEN einzigen Test — und genau deshalb ist sie am
|
|
26.07.2026 im ersten vollständigen Durchlauf umgefallen: Der Rip lief sauber
|
|
durch (40 GB Akira), die Kompression brach 0,2 s später ab mit „Dieser Worker
|
|
erreicht das Ziel nicht", obwohl die Freigabe erreichbar war. Es fehlten nur
|
|
zwei noch nie angelegte Ordner.
|
|
"""
|
|
|
|
import tasks
|
|
|
|
|
|
# --- Der eigentliche Fehler: nur EINE fehlende Ebene war erlaubt -------------
|
|
|
|
def _ohne_schluss_trenner(p):
|
|
"""`\\\\NAS\\rippy\\` und `\\\\NAS\\rippy` sind derselbe Ordner.
|
|
|
|
Wichtig, weil `ntpath.dirname` für die UNC-Wurzel MIT abschließendem
|
|
Backslash zurückkommt. Das echte `os.path.isdir` stört das nicht — eine
|
|
Attrappe, die stur vergleicht, hätte hier einen Fehler erfunden, den es
|
|
nicht gibt.
|
|
"""
|
|
return p.rstrip("\\/") or p
|
|
|
|
|
|
def _fake_isdir(vorhandene):
|
|
"""isdir-Ersatz, der nur die genannten Pfade kennt."""
|
|
bekannt = {_ohne_schluss_trenner(p) for p in vorhandene}
|
|
return lambda p: _ohne_schluss_trenner(p) in bekannt
|
|
|
|
|
|
def test_findet_den_naechsten_vorhandenen_ordner():
|
|
da = {"\\\\NAS\\rippy"}
|
|
assert tasks.erster_vorhandener_ordner(
|
|
"\\\\NAS\\rippy\\movies\\Akira (1988)",
|
|
isdir=_fake_isdir(da), dirname=lambda p: p.rsplit("\\", 1)[0] if "\\" in p.strip("\\") else "",
|
|
) == "\\\\NAS\\rippy"
|
|
|
|
|
|
def test_zwei_fehlende_ebenen_sind_in_ordnung():
|
|
"""DER Vorfall: `movies` UND der Filmordner fehlten beide.
|
|
|
|
`os.makedirs` legt die ganze Kette an — es genügt also, dass irgendein
|
|
Vorfahre existiert. Die alte Prüfung sah nur nach `dirname` und erklärte
|
|
damit jeden ERSTEN Film in einer neuen Ablage für unerreichbar.
|
|
"""
|
|
da = {"/share"}
|
|
assert tasks.erster_vorhandener_ordner(
|
|
"/share/movies/Akira (1988)", isdir=_fake_isdir(da)) == "/share"
|
|
|
|
|
|
def test_gar_nichts_erreichbar_gibt_leer():
|
|
"""Ist die Freigabe wirklich weg, muss die Prüfung anschlagen."""
|
|
assert tasks.erster_vorhandener_ordner(
|
|
"/share/movies/Film", isdir=_fake_isdir(set())) == ""
|
|
|
|
|
|
def test_sucht_nicht_endlos_auf_einer_toten_freigabe():
|
|
"""Jeder Schritt kostet auf einer toten Freigabe einen Timeout."""
|
|
gefragt = []
|
|
|
|
def zaehlend(p):
|
|
gefragt.append(p)
|
|
return False
|
|
|
|
tiefer_pfad = "/" + "/".join(f"e{i}" for i in range(50))
|
|
assert tasks.erster_vorhandener_ordner(tiefer_pfad, isdir=zaehlend) == ""
|
|
assert len(gefragt) <= tasks.MAX_ELTERN_STUFEN
|
|
|
|
|
|
def test_leerer_pfad_stuerzt_nicht_ab():
|
|
assert tasks.erster_vorhandener_ordner("", isdir=_fake_isdir(set())) == ""
|
|
|
|
|
|
# --- Die Meldung: eine falsche Ursache ist teurer als gar keine -------------
|
|
|
|
def _pruefen(monkeypatch, vorhandene, mapping, im_container=False):
|
|
import ntpath
|
|
kennt = _fake_isdir(vorhandene)
|
|
monkeypatch.setattr(tasks.os.path, "isdir",
|
|
lambda p: kennt(p) or (im_container and p == "/app"))
|
|
monkeypatch.setattr(tasks.os.path, "dirname", ntpath.dirname)
|
|
monkeypatch.setenv("RIPPY_PATH_MAP", mapping)
|
|
return tasks._erreichbarkeit_pruefen(
|
|
"/app/media/rippy/job1", "\\\\NAS\\rippy\\job1",
|
|
"/app/media/rippy/movies/Akira (1988)", "\\\\NAS\\rippy\\movies\\Akira (1988)",
|
|
)
|
|
|
|
|
|
def test_alles_da_meldet_nichts(monkeypatch):
|
|
assert _pruefen(monkeypatch, {"\\\\NAS\\rippy\\job1", "\\\\NAS\\rippy\\movies\\Akira (1988)"},
|
|
"/app/media/rippy=\\\\NAS\\rippy") == ""
|
|
|
|
|
|
def test_ziel_fehlt_ganz_ist_kein_fehler(monkeypatch):
|
|
"""Der Vorfall vom 26.07.2026 — jetzt geht er durch."""
|
|
assert _pruefen(monkeypatch, {"\\\\NAS\\rippy\\job1", "\\\\NAS\\rippy"},
|
|
"/app/media/rippy=\\\\NAS\\rippy") == ""
|
|
|
|
|
|
def test_fehlende_quelle_bleibt_ein_fehler(monkeypatch):
|
|
meldung = _pruefen(monkeypatch, {"\\\\NAS\\rippy"}, "/app/media/rippy=\\\\NAS\\rippy")
|
|
assert "Quelle" in meldung
|
|
assert "Rohdaten sind NICHT verloren" in meldung
|
|
|
|
|
|
def test_uebersetzter_aber_unerreichbarer_pfad_beschuldigt_nicht_die_karte(monkeypatch):
|
|
"""⚠️ Genau die Falschaussage aus dem Vorfall.
|
|
|
|
Die Meldung lautete „RIPPY_PATH_MAP … deckt diesen Pfad aber nicht ab",
|
|
während sie ihn im selben Satz korrekt übersetzt hatte. Wer dem folgte,
|
|
suchte in der Karte statt in der Freigabe.
|
|
"""
|
|
meldung = _pruefen(monkeypatch, set(), "/app/media/rippy=\\\\NAS\\rippy")
|
|
assert "deckt diesen Pfad aber nicht ab" not in meldung
|
|
assert "Übersetzt wurde er korrekt" in meldung
|
|
assert "Freigabe verbunden" in meldung
|
|
|
|
|
|
def test_nicht_uebersetzter_pfad_beschuldigt_sehr_wohl_die_karte(monkeypatch):
|
|
"""Hier ist die Karte wirklich schuld — der Pfad blieb unverändert."""
|
|
monkeypatch.setattr(tasks.os.path, "isdir", lambda p: False)
|
|
monkeypatch.setenv("RIPPY_PATH_MAP", "/app/temp=Y:\\temp")
|
|
meldung = tasks._erreichbarkeit_pruefen(
|
|
"/app/media/rippy/job1", "/app/media/rippy/job1",
|
|
"/app/media/movies/X", "/app/media/movies/X",
|
|
)
|
|
assert "deckt diesen Pfad aber nicht ab" in meldung
|
|
assert "von keinem Eintrag übersetzt" in meldung
|
|
|
|
|
|
def test_ohne_mapping_nennt_die_wahre_ursache(monkeypatch):
|
|
meldung = _pruefen(monkeypatch, set(), "")
|
|
assert "nicht gesetzt" in meldung
|
|
assert "RIPPY_PATH_MAP" in meldung
|
|
|
|
|
|
def test_artikel_stimmt_in_beiden_meldungen(monkeypatch):
|
|
"""„erreicht die Ziel" stand wörtlich im Log des Commanders."""
|
|
quelle = _pruefen(monkeypatch, set(), "")
|
|
assert "erreicht die Quelle" in quelle
|
|
# Quelle da, Ziel unerreichbar (kein einziger Vorfahre) → Ziel-Meldung
|
|
ziel = _pruefen(monkeypatch, {"\\\\NAS\\rippy\\job1"},
|
|
"/app/media/rippy=\\\\NAS\\rippy")
|
|
assert "erreicht das Ziel" in ziel
|