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(
        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 = []
    monkeypatch.setattr(mounts.os, "makedirs", lambda *a, **k: None)
    # 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, "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