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