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).
Läuft nur unter Linux (detection.py nutzt fcntl/ioctl), also genau dort,
wo auch die Ampel läuft.
"""

import sys

import pytest

if sys.platform == "win32":  # pragma: no cover
    pytest.skip("detection.py braucht fcntl (Linux)", allow_module_level=True)


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",
    ):
        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)