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 = []
    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