rippy/docker/api/test_mounts_helpers.py

"""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(
        # Seit V2-4 steht der Ablauf in ablauf.py; tasks.py ist nur noch
        # die Celery-Huelle. Dieser Waechter muss dorthin schauen, wo der
        # Code WIRKLICH steht — sonst prueft er eine leere Datei und ist
        # gruen, ohne etwas zu beweisen.
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "worker", "ablauf.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 = []
    monkeypatch.setattr(mounts.os, "makedirs", lambda *a, **k: None)
    # pfad_lage gestubbt: Sie ruft selbst `timeout ls` auf, und hier geht es um
    # die Reihenfolge von Loesen und Mounten.
    monkeypatch.setattr(mounts, "pfad_lage", lambda ziel: "da")
    # Vorher tot (der Vor-Check), nach dem Mount dauerhaft erreichbar. Die
    # Doppelprobe wird hier gestubbt, damit der Test nicht 3 s echt wartet.
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False)
    monkeypatch.setattr(mounts, "wirklich_erreichbar", lambda name: True)
    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, "pfad_lage", lambda ziel: "da")
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False)
    monkeypatch.setattr(mounts, "wirklich_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


def test_wirklich_erreichbar_prueft_zweimal_mit_abstand(monkeypatch):
    """Befund 26.07.2026: Direkt nach einem frischen `mount` antwortete die
    Freigabe - und Sekunden spaeter lief jeder Zugriff in die Zeitgrenze
    (Wettlauf mit dem lazy umount, dessen Abbau hinter den neuen Mount fiel).
    Eine EINZIGE Probe kann das nicht sehen."""
    import mounts

    antworten = iter([True, False])   # erst ja, dann nein
    gewartet = []
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: next(antworten))
    assert mounts.wirklich_erreichbar("rippy", warten=gewartet.append) is False
    assert gewartet == [3]


def test_wirklich_erreichbar_bei_gesunder_freigabe(monkeypatch):
    import mounts

    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: True)
    assert mounts.wirklich_erreichbar("rippy", warten=lambda s: None) is True


def test_wirklich_erreichbar_spart_das_warten_wenn_schon_die_erste_probe_faellt(monkeypatch):
    import mounts

    gewartet = []
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False)
    assert mounts.wirklich_erreichbar("rippy", warten=gewartet.append) is False
    assert gewartet == []   # nicht drei Sekunden fuer nichts


def test_stale_loesen_wartet_den_lazy_abbau_ab(monkeypatch):
    """Der Grund fuer die 150 Sekunden (Befund 26.07.2026): `umount -l` ist lazy,
    der Abbau passiert spaeter. Wer direkt danach mountet, riskiert, dass der
    Abbau HINTER dem neuen Mount landet - der erste Reparaturversuch scheiterte
    dadurch regelmaessig, und der zweite kostete 30 s Timeout."""
    import time
    import types

    import mounts

    gewartet = []
    monkeypatch.setattr(time, "sleep", gewartet.append)
    aufrufe = []

    def fake_run(cmd, **kwargs):
        aufrufe.append(cmd)
        rc = 0 if len(aufrufe) <= 2 else 1
        return types.SimpleNamespace(returncode=rc, stdout=b"", stderr=b"")

    monkeypatch.setattr(mounts.subprocess, "run", fake_run)
    assert mounts._stale_mounts_loesen("/app/media/x") == 2
    assert gewartet == [1.5]


def test_stale_loesen_wartet_nicht_wenn_nichts_zu_loesen_war(monkeypatch):
    """War kein Mount da, gibt es auch keinen Abbau abzuwarten - dann darf die
    Reparatur nicht kuenstlich gebremst werden."""
    import time
    import types

    import mounts

    gewartet = []
    monkeypatch.setattr(time, "sleep", gewartet.append)
    monkeypatch.setattr(
        mounts.subprocess, "run",
        lambda cmd, **k: types.SimpleNamespace(returncode=1, stdout=b"", stderr=b""))
    assert mounts._stale_mounts_loesen("/app/media/x") == 0
    assert gewartet == []


def test_pfad_lage_unterscheidet_da_weg_unklar(monkeypatch):
    """DER Fund vom 26.07.2026: Eine Wiederanbindung brauchte 3 min 15 s, und die
    Zeit ging in die ERSTE Zeile von mounten() - `os.makedirs(ziel,
    exist_ok=True)`. `exist_ok` prueft mit os.path.isdir, und ein `stat` auf einen
    toten CIFS-Mount blockiert im Kernel bis zum SMB-Timeout. Deshalb wird die
    Lage jetzt mit einem abbrechbaren Kind-Prozess erfragt."""
    import types

    import mounts

    def antwort(rc):
        return lambda cmd, **k: types.SimpleNamespace(returncode=rc, stdout=b"", stderr=b"")

    monkeypatch.setattr(mounts.subprocess, "run", antwort(0))
    assert mounts.pfad_lage("/app/media/rippy") == "da"
    monkeypatch.setattr(mounts.subprocess, "run", antwort(2))
    assert mounts.pfad_lage("/app/media/neu") == "weg"
    # 124 = `timeout` hat abgeschossen: existiert, antwortet aber nicht
    monkeypatch.setattr(mounts.subprocess, "run", antwort(124))
    assert mounts.pfad_lage("/app/media/totes-nas") == "unklar"


def test_pfad_lage_nutzt_eine_zeitgrenze(monkeypatch):
    import types

    import mounts

    gesehen = {}
    monkeypatch.setattr(
        mounts.subprocess, "run",
        lambda cmd, **k: (gesehen.update(cmd=cmd, kw=k),
                          types.SimpleNamespace(returncode=0))[1])
    mounts.pfad_lage("/x")
    assert gesehen["cmd"][0] == "timeout"
    assert gesehen["kw"]["timeout"] is not None


def test_mounten_legt_den_ordner_nur_an_wenn_er_fehlt(monkeypatch):
    """Bei "unklar" (toter Mount) darf makedirs NICHT laufen - genau dort hing es
    drei Minuten. Der Ordner ist dann ohnehin da."""
    import types

    import mounts

    angelegt = []
    monkeypatch.setattr(mounts, "pfad_lage", lambda ziel: "unklar")
    monkeypatch.setattr(mounts.os, "makedirs",
                        lambda *a, **k: angelegt.append(a[0] if a else ""))
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False)
    monkeypatch.setattr(mounts, "wirklich_erreichbar", lambda name: True)
    monkeypatch.setattr(mounts, "_stale_mounts_loesen", lambda ziel: 0)
    monkeypatch.setattr(mounts, "schreibtest", lambda p: True)
    monkeypatch.setattr(
        mounts.subprocess, "run",
        lambda cmd, **k: types.SimpleNamespace(returncode=0, stdout="", stderr=""))

    assert mounts.mounten("rippy", "cifs", "//nas/rippy") is True
    assert angelegt == []


def test_mounten_legt_den_ordner_bei_erstinstallation_an(monkeypatch):
    import types

    import mounts

    angelegt = []
    monkeypatch.setattr(mounts, "pfad_lage", lambda ziel: "weg")
    monkeypatch.setattr(mounts.os, "makedirs",
                        lambda *a, **k: angelegt.append(a[0] if a else ""))
    monkeypatch.setattr(mounts, "ist_erreichbar", lambda name: False)
    monkeypatch.setattr(mounts, "wirklich_erreichbar", lambda name: True)
    monkeypatch.setattr(mounts, "_stale_mounts_loesen", lambda ziel: 0)
    monkeypatch.setattr(mounts, "schreibtest", lambda p: True)
    monkeypatch.setattr(
        mounts.subprocess, "run",
        lambda cmd, **k: types.SimpleNamespace(returncode=0, stdout="", stderr=""))

    mounts.mounten("rippy", "cifs", "//nas/rippy")
    assert angelegt == ["/app/media/rippy"]


def test_reparieren_fragt_ismount_nicht_mehr(monkeypatch):
    """os.path.ismount ist ein `stat` und blockiert auf einem toten CIFS-Mount.
    Gebraucht wird es nicht: `umount -l` auf einen leeren Pfad kostet nichts."""
    import mounts

    monkeypatch.setattr(mounts.os.path, "ismount",
                        lambda p: (_ for _ in ()).throw(AssertionError("nicht fragen!")))
    monkeypatch.setattr(mounts, "_lazy_umount", lambda ziel: None)
    monkeypatch.setattr(mounts, "mounten", lambda *a, **k: True)
    assert mounts.reparieren("rippy", "cifs", "//nas/rippy") is True