"""Tests für die SMB-Fehlerübersetzung (Speicherziele → Freigaben auflisten).""" from mounts import ( pfad_map_vorschlag, pfad_map_zeile, uebersetze_smb_fehler, unc_aus_quelle, validiere_name, ) def test_access_denied_ohne_credentials_erklaert_gastproblem(): meldung = uebersetze_smb_fehler( "session setup failed: NT_STATUS_ACCESS_DENIED", mit_credentials=False ) assert "Gast" in meldung assert "Benutzername + Passwort" in meldung def test_access_denied_mit_credentials_verweist_auf_konto(): meldung = uebersetze_smb_fehler( "session setup failed: NT_STATUS_ACCESS_DENIED", mit_credentials=True ) assert "stimmen nicht" in meldung def test_logon_failure_wird_uebersetzt(): meldung = uebersetze_smb_fehler("NT_STATUS_LOGON_FAILURE", mit_credentials=True) assert "Passwort falsch" in meldung def test_unerreichbar_wird_uebersetzt(): meldung = uebersetze_smb_fehler( "do_connect: Connection to 10.0.0.9 failed (Error NT_STATUS_IO_TIMEOUT)", mit_credentials=False, ) assert "nicht erreichbar" in meldung def test_unbekannter_fehler_bleibt_erhalten_und_gekappt(): meldung = uebersetze_smb_fehler("X" * 500, mit_credentials=False) assert meldung == "X" * 200 def test_validiere_name_bleibt_streng(): assert validiere_name("nas-filme") assert not validiere_name("NAS Filme") def test_stale_mounts_loesen_loest_bis_nichts_mehr_geht(monkeypatch): """Löst gestapelte Schichten per lazy umount, bis umount nichts mehr findet (returncode != 0), und meldet die Zahl der gelösten Schichten.""" import types import mounts aufrufe = [] def fake_run(cmd, **kwargs): aufrufe.append(cmd) rc = 0 if len(aufrufe) <= 3 else 1 # 3 Schichten lösen, dann leer return types.SimpleNamespace(returncode=rc, stdout=b"", stderr=b"") monkeypatch.setattr(mounts.subprocess, "run", fake_run) assert mounts._stale_mounts_loesen("/app/media/x") == 3 assert all(cmd[:2] == ["umount", "-l"] for cmd in aufrufe) def test_mounten_loest_stale_vor_dem_mount(): """Regression (Vorfall 24.07.): mounten() muss Alt-Mounts LÖSEN, bevor es neu mountet — sonst stapelt es auf eine Mount-Leiche (12 Schichten, ls-Timeout).""" import inspect import mounts quelle = inspect.getsource(mounts.mounten) assert "_stale_mounts_loesen(ziel)" in quelle # --- RIPPY_PATH_MAP: der Anschluss für externe Worker (Befund 26.07.2026) ---- def test_unc_aus_quelle_uebersetzt_cifs(): assert unc_aus_quelle("cifs", "//192.168.178.62/rippy") == "\\\\192.168.178.62\\rippy" assert unc_aus_quelle("cifs", "//nas/medien/filme") == "\\\\nas\\medien\\filme" def test_unc_aus_quelle_raet_bei_nfs_nicht(): """NFS gibt "" — Windows-Schreibweise ist nicht ableitbar (AGENTS Regel D).""" assert unc_aus_quelle("nfs", "192.168.178.62:/volume1/rippy") == "" assert unc_aus_quelle("cifs", "") == "" assert unc_aus_quelle("cifs", "kein-unc-pfad") == "" def test_pfad_map_zeile_baut_was_pfad_lokal_liest(monkeypatch): """Der erzeugte Wert muss vom Worker gelesen werden können — genau dieses Format erwartet worker/tasks.pfad_lokal(): Paare, getrennt durch ';'.""" import mounts monkeypatch.setattr(mounts, "ist_gemountet", lambda name: True) vorschlaege = pfad_map_vorschlag([ {"name": "rippy", "typ": "cifs", "quelle": "//192.168.178.62/rippy"}, ]) assert pfad_map_zeile(vorschlaege) == "/app/media/rippy=\\\\192.168.178.62\\rippy" def test_pfad_map_zeile_laesst_nfs_weg(monkeypatch): """Ein halbes Mapping wäre schlimmer als keines: pfad_lokal() hört beim ersten passenden Präfix auf, ein NFS-Eintrag ohne Ziel würde also einen Pfad 'übersetzen', den der Worker danach trotzdem nicht sieht.""" import mounts monkeypatch.setattr(mounts, "ist_gemountet", lambda name: True) vorschlaege = pfad_map_vorschlag([ {"name": "nfs-ziel", "typ": "nfs", "quelle": "10.0.0.9:/export"}, {"name": "rippy", "typ": "cifs", "quelle": "//nas/rippy"}, ]) assert pfad_map_zeile(vorschlaege) == "/app/media/rippy=\\\\nas\\rippy" def test_pfad_map_zeile_ohne_mounts_ist_leer(): assert pfad_map_zeile([]) == "" assert pfad_map_zeile(None) == "" def test_erzeugtes_mapping_uebersetzt_den_echten_fehlerfall(monkeypatch): """Gegenprobe mit dem Pfad, an dem es am 26.07.2026 live scheiterte. Der Rohschnitt lag auf `/app/media/rippy/95afdc89-…/title_t00.mkv`; der Windows-Worker sah dort nichts. Mit dem hier erzeugten Mapping muss genau dieser Pfad auf die Freigabe zeigen. `pfad_lokal` ist eine reine Funktion im Worker — hier nachgebaut aufzurufen wäre wertlos, deshalb wird sie über den Pfad importiert. """ import importlib.util import os import mounts monkeypatch.setattr(mounts, "ist_gemountet", lambda name: True) mapping = pfad_map_zeile(pfad_map_vorschlag([ {"name": "rippy", "typ": "cifs", "quelle": "//192.168.178.62/rippy"}, ])) # Der Worker liegt neben der API im Repo; kein geteiltes Paket zwischen # den Containern, deshalb per Pfad laden statt importieren. worker_tasks = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "worker", "tasks.py" ) spec = importlib.util.spec_from_file_location("_worker_tasks_pfad", worker_tasks) quelltext = open(worker_tasks, encoding="utf-8").read() assert "def pfad_lokal" in quelltext and spec is not None # pfad_lokal ist bewusst rein und ohne Modul-Zustand — die Funktion aus dem # Quelltext zu holen, ohne tasks.py komplett zu laden (das braucht celery, # db, requests …), geht am ehrlichsten über exec des Funktionsblocks. anfang = quelltext.index("def pfad_lokal") ende = quelltext.index("\nRAW_DIR", anfang) umgebung = {"os": os} exec(compile(quelltext[anfang:ende], worker_tasks, "exec"), umgebung) # noqa: S102 pfad_lokal = umgebung["pfad_lokal"] assert pfad_lokal("/app/media/rippy/95afdc89/title_t00.mkv", mapping) == ( "\\\\192.168.178.62\\rippy\\95afdc89\\title_t00.mkv" ) # --- Der Mount kam nach einem Rebuild nicht zurueck (Befund 26.07.2026) ------ def test_mounten_geht_bei_totem_mount_den_reparatur_weg(monkeypatch): """Regression. Vorher galt `os.path.ismount` als Beweis, dass alles steht - und nach `docker compose up -d --build` war die CIFS-Freigabe TOT (4 von 4 Zugriffen 10 s Timeout), der Mountpunkt aber weiter vorhanden. Damit brach das Wiederherstellen genau dort ab, und `schreibtest()` (kein Timeout!) blockierte den Start-Thread im Kernel. Antwortet die Freigabe nicht, muss geloest und frisch gemountet werden - genau wie reparieren() es tut, nur automatisch. Der Test faengt das an der WIRKUNG: ismount darf dann gar nicht mehr gefragt werden.""" import types import mounts ablauf = [] antworten = iter([False, True]) # vorher tot, nach dem Mount erreichbar monkeypatch.setattr(mounts.os, "makedirs", lambda *a, **k: None) monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: next(antworten)) monkeypatch.setattr(mounts, "_stale_mounts_loesen", lambda ziel: ablauf.append("loesen")) monkeypatch.setattr(mounts, "schreibtest", lambda p: True) def fake_run(cmd, **kwargs): ablauf.append(cmd[0]) return types.SimpleNamespace(returncode=0, stdout="", stderr="") monkeypatch.setattr(mounts.subprocess, "run", fake_run) # ismount darf hier gar nicht mehr gefragt werden monkeypatch.setattr(mounts.os.path, "ismount", lambda p: (_ for _ in ()).throw(AssertionError("zu frueh gefragt"))) assert mounts.mounten("rippy", "cifs", "//nas/rippy") is True assert ablauf == ["loesen", "mount"] def test_mounten_prueft_das_ergebnis_und_versucht_es_zweimal(monkeypatch): """Befund 26.07.2026, dreimal reproduziert: `mount` meldete Erfolg, und die Freigabe antwortete danach TROTZDEM nicht (eine einzige, korrekt aussehende Schicht in /proc/mounts). Derselbe Ablauf ein zweites Mal stellte sie sofort her. Also wird das Ergebnis geprueft statt geglaubt.""" import types import mounts ablauf = [] # nie erreichbar: vorher, nach Versuch 1, nach Versuch 2 monkeypatch.setattr(mounts.os, "makedirs", lambda *a, **k: None) monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False) monkeypatch.setattr(mounts, "_stale_mounts_loesen", lambda ziel: None) monkeypatch.setattr(mounts, "_lazy_umount", lambda ziel: ablauf.append("lazy")) monkeypatch.setattr(mounts, "schreibtest", lambda p: True) monkeypatch.setattr( mounts.subprocess, "run", lambda cmd, **k: (ablauf.append(cmd[0]), types.SimpleNamespace(returncode=0, stdout="", stderr=""))[1]) import pytest with pytest.raises(RuntimeError) as fehler: mounts.mounten("rippy", "cifs", "//nas/rippy") # Zweimal gemountet, dazwischen einmal geloest assert ablauf == ["mount", "lazy", "mount"] # Und die Meldung sagt die Wahrheit statt "eingehaengt" assert "antwortet aber nicht" in str(fehler.value) def test_mounten_laesst_gesunden_mount_in_ruhe(monkeypatch): """Antwortet die Freigabe, bleibt sie unangetastet - kein Loesen, kein zweites Mounten (das wuerde stapeln).""" import mounts monkeypatch.setattr(mounts.os, "makedirs", lambda *a, **k: None) monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: True) monkeypatch.setattr(mounts.os.path, "ismount", lambda p: True) monkeypatch.setattr(mounts, "schreibtest", lambda p: True) monkeypatch.setattr(mounts, "_stale_mounts_loesen", lambda ziel: (_ for _ in ()).throw(AssertionError("nicht loesen!"))) assert mounts.mounten("rippy", "cifs", "//nas/rippy") is True