"""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"] == "" # ── Die Grenze DB → Oberflaeche (Befund 29.08.2026) ───────────────────── # # Commander mit Bildschirmfoto: „schau mal hier, das Datum ist falsch." # # TYP „DISC" STARTZEIT „1.1.1970, 01:00:00" STATUS „running" # # Der Waechter schickte die ROHE Datenbankzeile ueber den Ereignisstrom. Dort # heissen die Spalten `disc_type` und `created_at`, im UI aber `type` und # `startTime` — die Uebersetzung macht `_job_row_to_model`, und sie bildet # auch `running` auf `processing` ab. # # Der Test, den ich beim ersten Anlauf geschrieben habe, hat das NICHT # gefunden: Ich hatte seine Beispielzeile selbst erfunden, mit meinen # falschen Feldnamen. Ein Test, der dieselbe Annahme macht wie der Code, # prueft nichts. Dieser hier geht von der ECHTEN Zeile aus. def _echte_db_zeile(): """Eine Zeile mit den Spaltennamen, die wirklich in der Datenbank stehen.""" from datetime import datetime return { "id": "j1", "disc_type": "bluray", # NICHT "type" "created_at": datetime(2026, 8, 29, 12, 7, 2), # NICHT "startTime" "finished_at": None, "status": "running", # wird zu "processing" "device": r"\.\G:", "progress": 12, "title": "Evangelion 2.22", "error": None, "meta": None, } def test_die_umwandlung_liefert_die_namen_die_das_ui_kennt(): import main ui = main._job_row_to_model(_echte_db_zeile()).model_dump() assert ui["type"] == "bluray" assert ui["startTime"].startswith("2026-08-29") assert ui["status"] == "processing", "der Worker-Status heisst im UI anders" def test_das_live_ereignis_traegt_dieselben_namen_wie_die_jobliste(): """DER Waechter gegen „1.1.1970". Ohne die Umwandlung sind alle Pflichtfelder leer — und leer sieht im Browser aus wie eine Auskunft.""" import main from rippy.bus.waechter import UI_PFLICHTFELDER, _job_kurz ui = main._job_row_to_model(_echte_db_zeile()).model_dump() kurz = _job_kurz(ui) for feld in UI_PFLICHTFELDER: assert kurz.get(feld) is not None, "%s waere im Browser leer" % feld assert kurz["type"] == "bluray" assert kurz["status"] == "processing" def test_der_waechter_ist_mit_der_umwandlung_verdrahtet(): """Sonst waere die Reparatur wieder nur eine Moeglichkeit, keine Tatsache.""" import inspect import main quelle = inspect.getsource(main.startup_event) assert "job_form=" in quelle, "Waechter bekommt die Umwandlung nicht" assert "_job_row_to_model" in quelle def test_die_laufwerksliste_kommt_aus_EINER_quelle(): """Waechter gegen die vierte Kopie (Befund 29.08.2026). Commander: „Es ist eine Disk im laufwerk, aber er erkennt sie dort nicht." Die Laufwerksliste wurde an DREI Stellen gebaut: im `/devices`-Endpunkt (mit angehaengter Disc), im Ereignis-Waechter und im SSE-Schnappschuss (beide ohne). Das Dashboard liest den Schnappschuss — und schloss aus der fehlenden Disc auf ein leeres Laufwerk, waehrend ihr Titel eine Zeile weiter oben stand. Eine Auskunft in drei Fassungen ist zwei Fassungen zu viel. """ import inspect import main quelle = inspect.getsource(main) # Die eine erlaubte Fundstelle ist die Funktion selbst. stellen = quelle.count("device_discovery.device_info(") assert stellen == 1, ( "device_info wird an %d Stellen aufgerufen — die Liste gehoert in " "laufwerke_mit_disc(), sonst fehlt irgendwo die Disc" % stellen) for name in ("_snapshot", "startup_event", "get_devices"): text = inspect.getsource(getattr(main, name)) assert "device_info(" not in text, \ "%s baut die Laufwerksliste selbst" % name def test_laufwerke_mit_disc_haengt_die_erkannte_disc_an(monkeypatch): import main monkeypatch.setattr(main.device_discovery, "list_optical_devices", lambda: [r"\.\G:"]) monkeypatch.setattr(main.device_discovery, "device_info", lambda p: {"id": "G", "name": "Laufwerk G:", "path": p, "type": "bluray", "status": "ready"}) monkeypatch.setitem(main.DISC_CACHE, r"\.\G:", {"title": "Evangelion 2.22"}) geraete = main.laufwerke_mit_disc() assert geraete[0]["disc"]["title"] == "Evangelion 2.22" def test_eine_noch_laufende_erkennung_wird_NICHT_als_disc_gemeldet(monkeypatch): """`_laeuft` heisst „wird gerade erkannt" — das ist noch keine Auskunft.""" import main monkeypatch.setattr(main.device_discovery, "list_optical_devices", lambda: [r"\.\G:"]) monkeypatch.setattr(main.device_discovery, "device_info", lambda p: {"id": "G", "name": "Laufwerk G:", "path": p, "type": "bluray", "status": "ready"}) monkeypatch.setitem(main.DISC_CACHE, r"\.\G:", {"_laeuft": True, "title": "Wird erkannt…"}) assert "disc" not in main.laufwerke_mit_disc()[0]