rippy/docker/api/test_api_smoke.py

"""Import-Smoke-Test: bricht, wenn main.py kaputte Imports oder Verdrahtung hat.

Warum: Kein anderer Test importiert main.py — ein Tippfehler dort fiele sonst
erst beim Container-Start auf (und die Ampel bliebe fälschlich grün).
## Warum das hier nicht mehr uebersprungen wird (28.08.2026)

Bis V2-4 stand hier ein `pytest.skip` fuer Windows: `main.py` importierte
`rippy.drives.linux` direkt, und das braucht `fcntl`. Seit die Treiberwahl
ueber den Port `rippy.ports.Drives` laeuft, laedt `main.py` auf BEIDEN
Plattformen — nachgemessen: 57 Routen, sauberer Import.

Das Ueberspringen war damit nicht mehr Vorsicht, sondern eine Luecke: Diese
siebzehn Tests liefen nur auf der Ampel. Und ein Test, der nur auf einer
Plattform greift, ist eine halbe Zusage — genau daran ist die Ampel am
28.08.2026 fuenf Laeufe lang unbemerkt rot gewesen.
"""

import pytest  # noqa: F401  (einzelne Tests brauchen ihn fuer skipif)


def test_main_importierbar_und_routen_verdrahtet():
    from main import app

    routen = {route.path for route in app.routes}
    for pfad in (
        "/health", "/jobs", "/devices", "/logs", "/settings",
        # KEYDB.cfg + AACS-Dumps: der Platz für die selbst mitgebrachte
        # Schlüsseldatei (Befund 25.07.2026 — MakeMKV holt UHD-Schlüssel nicht
        # mehr online nach). Ohne diese Routen ist die Seite im UI tot.
        "/system/keydb", "/system/aacs-dumps", "/system/aacs-dumps/{dateiname}",
        # Der Hauptweg für 4K-UHD (25.07.2026): makemkvcon holt Disc-Schluessel
        # unter Linux nie selbst, sie kommen von Hand über diesen Endpunkt.
        "/system/keystore",
        # Ohne die weiss das UI nicht, worauf es laeuft — und zeigt dann auf
        # einem Windows-PC "Pruefen: docker compose ps" (Befund 28.08.2026).
        "/betrieb",
    ):
        assert pfad in routen, f"Route {pfad} fehlt"


def test_dateiname_validierung_blockt_pfad_tricks():
    """Download-Endpoint: nur nackte Dateinamen — kein .., kein Slash, kein Dotfile."""
    from main import _sicherer_dateiname

    assert _sicherer_dateiname("film.mkv") is True
    assert _sicherer_dateiname("../../etc/passwd") is False
    assert _sicherer_dateiname("a/b.mkv") is False
    assert _sicherer_dateiname("a\\b.mkv") is False
    assert _sicherer_dateiname(".versteckt") is False
    assert _sicherer_dateiname("") is False


def test_worker_task_name_passt_zum_celery_client():
    """API schickt an 'worker.tasks.rip_disc' — der Name ist Vertrag mit dem Worker."""
    import inspect

    import celery_client

    quelle = inspect.getsource(celery_client.start_rip)
    assert '"worker.tasks.rip_disc"' in quelle


def test_remount_blockiert_den_api_start_nicht():
    """Regression (Vorfall 24.07.): ein haengender Netz-Mount (CIFS-Schreibtest kann
    im Kernel haengen, wait_for_response) darf den API-Start NICHT blockieren. remount
    muss als Hintergrund-Task laufen (create_task), nicht direkt awaited werden."""
    import inspect

    import main

    quelle = inspect.getsource(main.startup_event)
    assert "create_task(asyncio.to_thread(remount))" in quelle, \
        "remount muss als Hintergrund-Task laufen (nicht blockierend)"
    assert "await asyncio.to_thread(remount)" not in quelle, \
        "remount darf nicht mehr direkt awaited werden (blockiert sonst den Start)"


def test_unter_wurzel_faellt_nicht_auf_praefix_namen_herein():
    """Befund 25.07.2026: In main.py prüften neun Stellen mit nacktem
    startswith(MEDIA_ROOT) — darunter /browse und /browse/mkdir, wo der Pfad
    vom Nutzer kommt. „/app/media-boese/x" beginnt mit „/app/media", liegt
    aber außerhalb. Zwilling von tasks.unter_wurzel im Worker."""
    from main import unter_wurzel

    assert unter_wurzel("/app/media", "/app/media") is True
    assert unter_wurzel("/app/media/movies", "/app/media") is True
    assert unter_wurzel("/app/media-boese/x", "/app/media") is False
    assert unter_wurzel("/app/mediaX", "/app/media") is False
    assert unter_wurzel("/etc/passwd", "/app/media") is False
    assert unter_wurzel("", "/app/media") is False
    assert unter_wurzel("/app/media", "") is False
    assert unter_wurzel("/app/media/movies", "/app/media/") is True


def test_ping_vorrat_verhindert_die_wartesekunde(monkeypatch):
    """Gemessen 25.07.2026: /capabilities brauchte 1,010 s - jedes Mal. Der
    Celery-Ping sammelt Antworten bis zum Timeout und kann nicht früher
    aufhoeren. Fuenf UI-Stellen holen /capabilities, also zahlte jede Seite
    eine Sekunde, während alle anderen Endpunkte unter 25 ms lagen.

    Der Vorrat muss deshalb abgelesen und NICHT neu gepingt werden, solange er
    frisch ist - und bei altem Vorrat lieber einmal langsam als falsch.
    """
    import time as _t

    import main

    pings = []
    monkeypatch.setattr(main, "_ping_jetzt", lambda: pings.append(1) or ["celery@neu"])

    # Frischer Vorrat -> ablesen, kein Ping
    monkeypatch.setitem(main._PING, "knoten", ["celery@alt"])
    monkeypatch.setitem(main._PING, "stand", _t.monotonic())
    assert main._ping_knoten() == ["celery@alt"]
    assert pings == [], "bei frischem Vorrat darf NICHT gepingt werden"

    # Zu alter Vorrat -> einmal synchron pingen
    monkeypatch.setitem(main._PING, "stand", _t.monotonic() - main.PING_ALTER_MAX_SEKUNDEN - 1)
    assert main._ping_knoten() == ["celery@neu"]
    assert len(pings) == 1

    # Kalter Start (nie gepingt) -> ebenfalls pingen, nicht "alles offline" melden
    monkeypatch.setitem(main._PING, "stand", -1e9)
    main._ping_knoten()
    assert len(pings) == 2


def test_ping_takt_ist_kuerzer_als_die_haltbarkeit():
    """Sonst läuft der Vorrat zwischen zwei Hintergrund-Laeufen ab und der
    Endpunkt pingt doch wieder synchron."""
    import main

    assert main.PING_INTERVALL_SEKUNDEN < main.PING_ALTER_MAX_SEKUNDEN


def test_tote_routen_sind_und_bleiben_weg():
    """Entfernt am 25.07.2026, jede ein Überrest eines ersetzten Entwurfs und
    ohne einen einzigen Aufrufer (mechanisch gegengeprueft: alle api.*-Aufrufe
    des UI gegen alle Routen).

    Der Test hält sie draussen. /stream/jobs ist der Grund für diese
    Absicherung: er war schon einmal ein Placebo, wurde dann "repariert" statt
    entfernt - und war danach eine Endlosschleife je Verbindung ohne jeden
    Verbraucher. Wer echtes Push will, braucht BEIDE Seiten (Server UND ein
    EventSource im UI).
    """
    from main import app

    routen = {route.path for route in app.routes}
    for pfad in ("/prescan", "/jellyfin/format", "/stream/jobs",
                 "/worker-setup/windows-gui"):
        assert pfad not in routen, (
            f"{pfad} ist wieder da — entweder mit Verbraucher (dann diesen Test "
            "anpassen) oder versehentlich (dann wieder raus)"
        )


def test_worker_setup_routen_die_gebraucht_werden_sind_da():
    """Die Installer holen sich Code und .exe hierueber — /worker-setup/paket
    ruft install.ps1 UND install-gui.ps1 auf, /windows-exe der UI-Knopf."""
    from main import app

    routen = {route.path for route in app.routes}
    for pfad in ("/worker-setup/paket", "/worker-setup/windows",
                 "/worker-setup/windows-exe"):
        assert pfad in routen, f"Route {pfad} fehlt — Worker-Installation kaputt"

# --- Mount-Wache: heilt, was nach jedem Rebuild kaputt ist -------------------
#
# Hier statt in test_mounts_helpers.py, weil diese Tests `main` brauchen und
# main.py haengt an fcntl (Linux). Dieses Modul ueberspringt sich unter Windows
# selbst — es laeuft also genau da, wo auch die Ampel laeuft.


def test_wache_ruehrt_nichts_an_solange_ein_job_laeuft(monkeypatch):
    """Neu verbinden heisst `umount -l`. Mitten in einem Rip oder Encode waere
    das ein Datenverlust - die Wache muss dann stillstehen."""
    import main

    monkeypatch.setattr(main.db, "list_mounts", lambda: [
        {"name": "rippy", "typ": "cifs", "quelle": "//nas/rippy"}])
    monkeypatch.setattr(main.db, "hat_arbeit", lambda: True)
    monkeypatch.setattr(main.mount_verwaltung, "ist_erreichbar",
                        lambda name: (_ for _ in ()).throw(AssertionError("nicht anfassen!")))

    main._mounts_nachsehen()  # darf einfach nichts tun


def test_wache_verbindet_eine_stumme_freigabe_neu(monkeypatch):
    import main

    repariert, gelogged = [], []
    monkeypatch.setattr(main.db, "list_mounts", lambda: [
        {"name": "rippy", "typ": "cifs", "quelle": "//nas/rippy"}])
    monkeypatch.setattr(main.db, "hat_arbeit", lambda: False)
    monkeypatch.setattr(main.db, "add_log",
                        lambda lvl, src, msg: gelogged.append((lvl, msg)))
    monkeypatch.setattr(main.mount_verwaltung, "ist_erreichbar", lambda name: False)
    monkeypatch.setattr(main.mount_verwaltung, "reparieren",
                        lambda *a, **k: repariert.append(a[0]))
    main._MOUNT_STAND.clear()

    main._mounts_nachsehen()

    assert repariert == ["rippy"]
    assert any("neu verbunden" in m for _, m in gelogged)


def test_wache_meckert_nicht_jede_minute(monkeypatch):
    """Ist das NAS ausgeschaltet, waere ein Log je Minute ein Wasserfall.
    Gemeldet wird nur der UEBERGANG."""
    import main

    gelogged = []
    monkeypatch.setattr(main.db, "list_mounts", lambda: [
        {"name": "rippy", "typ": "cifs", "quelle": "//nas/rippy"}])
    monkeypatch.setattr(main.db, "hat_arbeit", lambda: False)
    monkeypatch.setattr(main.db, "add_log",
                        lambda lvl, src, msg: gelogged.append(msg))
    monkeypatch.setattr(main.mount_verwaltung, "ist_erreichbar", lambda name: False)

    def reparieren_scheitert(*a, **k):
        raise RuntimeError("NAS aus")

    monkeypatch.setattr(main.mount_verwaltung, "reparieren", reparieren_scheitert)
    main._MOUNT_STAND.clear()

    for _ in range(5):
        main._mounts_nachsehen()

    # "antwortet nicht" genau EINMAL (der Uebergang), die Fehlschlaege sind
    # jeweils eigene Meldungen - aber kein wiederholtes "antwortet nicht".
    assert len([m for m in gelogged if "antwortet nicht" in m]) == 1


def test_wache_meldet_wenn_es_wieder_geht(monkeypatch):
    import main

    gelogged = []
    zustand = {"da": False}
    monkeypatch.setattr(main.db, "list_mounts", lambda: [
        {"name": "rippy", "typ": "cifs", "quelle": "//nas/rippy"}])
    monkeypatch.setattr(main.db, "hat_arbeit", lambda: False)
    monkeypatch.setattr(main.db, "add_log",
                        lambda lvl, src, msg: gelogged.append(msg))
    monkeypatch.setattr(main.mount_verwaltung, "ist_erreichbar",
                        lambda name: zustand["da"])
    monkeypatch.setattr(main.mount_verwaltung, "reparieren",
                        lambda *a, **k: None)
    main._MOUNT_STAND.clear()
    main._MOUNT_STAND["rippy"] = False

    zustand["da"] = True
    main._mounts_nachsehen()
    assert any("antwortet wieder" in m for m in gelogged)


# --- Live-Ereignisse (SSE), Etappe V2-3 --------------------------------------


def test_events_route_ist_verdrahtet():
    """Ohne diese Route fällt das UI stumm auf seinen letzten Stand zurück —
    und weil ein Abriss KEINE Aussage ist, sähe der Nutzer einfach nichts
    Neues, ohne Fehlermeldung. Deshalb hier festgenagelt."""
    from main import app

    assert "/events" in {route.path for route in app.routes}


def test_sse_rahmen_hat_das_format_das_der_browser_erwartet():
    """`id:` ist nicht Kosmetik — der Browser schickt genau diesen Wert beim
    Wiederverbinden als Last-Event-ID zurück. Fehlt er, gibt es keine
    lückenlose Wiederaufnahme, und jeder WLAN-Wechsel reisst ein Loch."""
    from main import _sse_rahmen

    rahmen = _sse_rahmen({"seq": 42, "typ": "job.progress", "daten": {"prozent": 7}})
    zeilen = rahmen.split("\n")
    assert zeilen[0] == "id: 42"
    assert zeilen[1] == "event: job.progress"
    assert zeilen[2].startswith("data: {")
    # Zwei Leerzeilen am Ende: eine schliesst das Ereignis, die zweite ist der
    # Trenner. Ohne den doppelten Umbruch haelt der Browser das Ereignis fuer
    # unvollstaendig und liefert es NIE aus.
    assert rahmen.endswith("\n\n")


def test_sse_rahmen_uebersteht_umlaute():
    """Job-Titel und Log-Zeilen sind deutsch. Mit ensure_ascii=True kaeme
    "Gr\u00f6\u00dfe" beim Nutzer an."""
    from main import _sse_rahmen

    rahmen = _sse_rahmen({"seq": 1, "typ": "log.line", "daten": {"text": "Größe"}})
    assert "Größe" in rahmen


def test_snapshot_meldet_unlesbare_laufwerke_als_none(monkeypatch):
    """„konnte nicht nachsehen" ist etwas anderes als „es gibt keine".

    Genau diese Vermischung hat in v1 die Job-Liste im Sekundentakt geleert
    (fuenfmal `catch(() => [])` im UI). Ein Snapshot mit devices=[] wuerde dem
    UI sagen „du hast kein Laufwerk"; None sagt „ich weiss es gerade nicht",
    und das UI behaelt seinen Stand.

    OHNE DATENBANK: Die Ampel hat keine Postgres — der erste Anlauf dieses
    Tests lief deshalb rot (Lauf 170). Der Store wird hier ersetzt, denn
    geprueft wird die Snapshot-LOGIK, nicht die Datenbank.
    """
    import asyncio

    import main

    monkeypatch.setattr(main.db, "list_jobs", lambda limit=50: [])
    monkeypatch.setattr(main.db, "list_workers", lambda: [])
    monkeypatch.setattr(main.db, "list_logs", lambda limit=50: [])

    def kaputt():
        raise OSError("Laufwerk haengt")

    monkeypatch.setattr(main.device_discovery, "list_optical_devices", kaputt)

    zustand = asyncio.run(main._snapshot())

    assert zustand["devices"] is None, "Ein unlesbares Laufwerk darf nicht als [] durchgehen"
    assert zustand["jobs"] == []


def test_snapshot_liefert_das_ganze_bild(monkeypatch):
    """Ein Client, der sich verbindet, bekommt das GANZE Bild — sonst muesste
    er den Rest raten und faellt auf Polling zurueck.

    ## Warum der Server-Zustand dazugehoert (Befund 28.08.2026)

    Er fehlte im Schnappschuss, und der Waechter schickt ihn nur alle 15
    Sekunden. Ein frisch geladenes Dashboard stand deshalb bis zu einer
    Viertelminute auf „Platz fuer Rippy: unbekannt" — und das sieht aus wie
    eine Auskunft, obwohl es keine ist. Genau das hat der Commander im
    Windows-Fenster gesehen.
    """
    import asyncio

    import main

    monkeypatch.setattr(main.db, "list_jobs", lambda limit=50: [])
    monkeypatch.setattr(main.db, "list_workers", lambda: [{"name": "pc"}])
    monkeypatch.setattr(main.db, "list_logs", lambda limit=50: [])
    monkeypatch.setattr(main.device_discovery, "list_optical_devices", lambda: [])
    # Ohne DB liefe system_info() in eine Ausnahme, und die Ampel hat keine
    # Datenbank (Lauf 170). Geprueft wird hier die FORM des Schnappschusses.
    monkeypatch.setattr(main.db, "get_settings", lambda: {})

    zustand = asyncio.run(main._snapshot())

    assert {"jobs", "workers", "logs", "devices"} <= set(zustand)
    assert "info" in zustand, (
        "Ohne den Server-Zustand steht das Dashboard nach jedem Neuladen "
        "bis zu 15 Sekunden auf 'unbekannt'."
    )
    assert zustand["devices"] == []          # wirklich leer, nicht „unbekannt"
    assert zustand["workers"] == [{"name": "pc"}]


def test_ein_kaputter_server_zustand_kippt_den_schnappschuss_nicht(monkeypatch):
    """Der Schnappschuss ist die Grundlage fuer ALLES im UI. Lieber ohne
    Platzangabe (das UI behaelt seinen Stand) als gar kein Schnappschuss."""
    import asyncio

    import main

    monkeypatch.setattr(main.db, "list_jobs", lambda limit=50: [])
    monkeypatch.setattr(main.db, "list_workers", lambda: [])
    monkeypatch.setattr(main.db, "list_logs", lambda limit=50: [])
    monkeypatch.setattr(main.device_discovery, "list_optical_devices", lambda: [])

    async def platzt():
        raise RuntimeError("Platte weg")

    monkeypatch.setattr(main, "system_stand_fuer_snapshot", platzt)
    zustand = asyncio.run(main._snapshot())
    assert {"jobs", "workers", "logs", "devices"} <= set(zustand)


def test_betrieb_meldet_faehigkeiten_statt_nur_einen_namen():
    """Der Befund vom 28.08.2026: Das Windows-Fenster zeigte "Worker
    erreichbar: 0 von 1", "Container-Platte" und "Pruefen: docker compose ps"
    — auf einem PC ohne Container und ohne zweiten Worker.

    Das UI muss FAEHIGKEITEN bekommen, keinen Modus-Namen: Aus einem Namen
    auf Verhalten zu schliessen bricht beim naechsten Betriebsfall (ein
    Docker-All-in-One hat Container-Pfade, aber keine externen Worker).
    """
    from fastapi.testclient import TestClient

    from main import app

    # OHNE `with`: Der Kontextmanager loest den Lebenszyklus aus, und der legt
    # Tabellen an — auf der Ampel gibt es keine Datenbank. Genau daran ist
    # Lauf 170 rot geworden. /betrieb braucht keine.
    antwort = TestClient(app).get("/betrieb")
    assert antwort.status_code == 200
    daten = antwort.json()
    assert daten["modus"] in ("standalone", "verteilt")
    assert daten["plattform"] in ("windows", "linux", "macos")
    assert set(daten["kann"]) == {"externe_worker", "freigaben_einhaengen",
                                  "container_pfade", "werkzeuge_verwalten",
                                  "frei_blaettern"}
    # Beide Orte muessen dabei sein. Fehlte einer, muesste die Oberflaeche
    # wieder raten — und genau daraus wurde „Container-Platte" auf einem PC,
    # der keinen Container hat (Befund 28.08.2026).
    assert daten["ablage_vorgabe"] and daten["arbeits_vorgabe"]
    assert all(isinstance(w, bool) for w in daten["kann"].values())
    # Ein Hinweis auf docker compose darf NUR im Container erscheinen.
    if not daten["im_container"]:
        assert daten["hilfe_befehl"] == ""