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"