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.
    #
    # `*a, **k`, nicht `lambda: {}` (Befund 30.08.2026): Die echte
    # Funktion heisst `get_settings(key="ui", bei_fehler_leer=False)`.
    # Der zu enge Doppelgaenger warf `TypeError`, sobald ein Aufrufer
    # einen der Parameter benutzte — `system_info` fiel damit aus dem
    # Schnappschuss, und dieser Test zeigte auf den Code statt auf sich
    # selbst. Ein Doppelgaenger muss die Schnittstelle abbilden, die er
    # ersetzt, nicht nur den einen Aufruf, den es gerade gibt.
    monkeypatch.setattr(main.db, "get_settings", lambda *a, **k: {})

    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]


def test_eine_laufende_erkennung_ist_SICHTBAR(monkeypatch):
    """Commander 29.08.2026: „das die disc erkennung noch läuft muss sichtbar
    sein."

    Vorher liess `laufwerke_mit_disc` einen laufenden Eintrag KOMPLETT weg —
    damit keine halbfertige Disc-Karte mit „Wird erkannt…" als Titel und
    einem „Rippen starten"-Knopf erscheint. Richtig gedacht, falsch geloest:
    Fuer die Oberflaeche sah es aus wie ein leeres Laufwerk. Und das dauert:
    `makemkvcon info` laeuft an einer Blu-ray in seine 120-Sekunden-Grenze
    (gemessen: Disc nach 119 s erkannt).
    """
    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…"})

    geraet = main.laufwerke_mit_disc()[0]
    assert geraet["disc_wird_erkannt"] is True
    # UND weiterhin keinen Titel behaupten, den es noch nicht gibt.
    assert "disc" not in geraet


def test_nach_der_erkennung_ist_die_marke_wieder_weg(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"})

    geraet = main.laufwerke_mit_disc()[0]
    assert not geraet.get("disc_wird_erkannt")
    assert geraet["disc"]["title"] == "Evangelion 2.22"


def test_das_geraetemodell_gibt_die_marke_auch_heraus():
    """Ohne Feld im Modell schneidet FastAPI sie weg — und das UI saehe
    wieder nichts (derselbe Fehler wie 25.07. bei `meta`)."""
    import main

    assert "disc_wird_erkannt" in main.Device.model_fields


# ── Die Notdurft-Auskunft (Befund 29.08.2026) ──────────────────────────
#
# Waehrend der Disc-Erkennung haelt makemkvcon das Laufwerk. Gemessen:
# /devices braucht dann 14,0 s, die Zeitgrenze des Schnappschusses ist 5,0 s.
# Der lieferte deshalb `devices = None` („konnte nicht nachsehen") — und nach
# einem frischen Laden hatte das UI keinen alten Stand, den es haette behalten
# koennen. Der Bildschirm blieb leer, ausgerechnet in der Phase, die sichtbar
# sein soll.


def test_notdurft_meldet_den_letzten_stand_plus_die_erkennung(monkeypatch):
    import main

    monkeypatch.setattr(main, "LETZTE_LAUFWERKE",
                        [{"id": "G", "name": "Laufwerk G:", "path": r"\.\G:",
                          "type": "bluray", "status": "ready"}])
    monkeypatch.setitem(main.DISC_CACHE, r"\.\G:", {"_laeuft": True})

    geraete = main.laufwerke_notdurft()
    assert geraete[0]["disc_wird_erkannt"] is True
    assert geraete[0]["type"] == "bluray", "der letzte bekannte Typ bleibt"


def test_notdurft_ohne_jeden_stand_erfindet_nichts(monkeypatch):
    """Wer nie erfolgreich gelesen hat, soll schweigen — `None` heisst
    „behalte deinen Stand", und das ist die ehrliche Antwort."""
    import main

    monkeypatch.setattr(main, "LETZTE_LAUFWERKE", [])
    monkeypatch.setattr(main.device_discovery, "list_optical_devices", lambda: [])
    assert main.laufwerke_notdurft() is None


def test_notdurft_beim_kaltstart_nennt_wenigstens_das_laufwerk(monkeypatch):
    """Kaltstart mitten in der Erkennung: Die LISTE der Laufwerke ist billig,
    nur das Abfragen des Laufwerks ist teuer."""
    import main

    monkeypatch.setattr(main, "LETZTE_LAUFWERKE", [])
    monkeypatch.setattr(main.device_discovery, "list_optical_devices",
                        lambda: [r"\.\G:"])
    monkeypatch.setitem(main.DISC_CACHE, r"\.\G:", {"_laeuft": True})

    geraete = main.laufwerke_notdurft()
    assert geraete[0]["disc_wird_erkannt"] is True
    assert geraete[0]["type"] == "unknown", "kein Typ wird erfunden"


def test_ein_fertiges_ergebnis_ueberschreibt_die_alte_marke(monkeypatch):
    """Sonst bliebe „wird erkannt" stehen, nachdem die Disc laengst da ist."""
    import main

    monkeypatch.setattr(main, "LETZTE_LAUFWERKE",
                        [{"id": "G", "name": "Laufwerk G:", "path": r"\.\G:",
                          "type": "bluray", "status": "ready",
                          "disc_wird_erkannt": True}])
    monkeypatch.setitem(main.DISC_CACHE, r"\.\G:", {"title": "Evangelion 2.22"})

    geraet = main.laufwerke_notdurft()[0]
    assert "disc_wird_erkannt" not in geraet
    assert geraet["disc"]["title"] == "Evangelion 2.22"


# ── Die Disc-Wache: kein Dauerscan (Befund 29.08.2026) ─────────────────
#
# Commander: „liest er die disc NOCHMAL ein das macht aber keinen sinn, wenn
# er sie bereits erkannt hat. Außerdem öffnen sich nun immer irgendwelche
# fenster ganz kurz im hintergrund."
#
# Beides war dieselbe Schleife: `except OSError: continue` liess
# `bekannt[pfad]` ungesetzt, der naechste Durchlauf hielt das Laufwerk fuer
# neu und stiess einen weiteren Vor-Scan an. Der haelt das Laufwerk, die
# naechste Statusabfrage scheitert — und so weiter.


def _CDS():
    import main
    return main.CDS_DISC_OK, main.CDS_NO_DISC, main.CDS_TRAY_OPEN


def test_erster_blick_auf_eine_liegende_disc_erkennt_sie():
    import main
    ok, _, _ = _CDS()
    assert main.disc_entscheidung(False, None, ok) == "erststart"


def test_ein_leeres_laufwerk_beim_start_loest_nichts_aus():
    import main
    _, leer, _ = _CDS()
    assert main.disc_entscheidung(False, None, leer) == ""


def test_dieselbe_disc_wird_NICHT_nochmal_gelesen():
    """Der Kern des Befundes."""
    import main
    ok, _, _ = _CDS()
    assert main.disc_entscheidung(True, ok, ok) == ""


def test_ein_fehlgeschlagener_lesevorgang_startet_KEINEN_neuen_scan():
    """DER Fehler: Nach einem Fehlschlag blieb `vorher` leer. Wer das mit
    „noch nie gesehen" verwechselt, scannt endlos.

    Ablauf wie in echt: erst erfolgreich gelesen, dann scheitert der Lesevor-
    gang (makemkvcon haelt das Laufwerk), dann klappt er wieder.
    """
    import main
    ok, _, _ = _CDS()
    # Der Fehlschlag setzt `bekannt` nicht — `vorher` ist danach None.
    # `schon_gesehen` bleibt aber True, und genau daran haengt alles.
    assert main.disc_entscheidung(True, None, ok) == "eingelegt", \
        "eine Statusaenderung nach unbekannt->Disc ist ein Einlegen …"
    # … aber NICHT ein Erststart, der ohne jede Aenderung scannt:
    assert main.disc_entscheidung(True, ok, ok) == ""


def test_auswurf_wird_erkannt():
    import main
    ok, leer, offen = _CDS()
    assert main.disc_entscheidung(True, ok, leer) == "entfernt"
    assert main.disc_entscheidung(True, ok, offen) == "entfernt"


def test_ein_wackliger_zwischenstatus_wirft_die_disc_nicht_weg():
    """3 = „nicht bereit" heisst nicht „ausgeworfen". Sonst faellt der Vorrat
    weg, sobald das Laufwerk kurz beschaeftigt ist."""
    import main
    ok, _, _ = _CDS()
    assert main.disc_entscheidung(True, ok, 3) == ""


def test_die_wache_setzt_gesehen_auch_nach_einem_fehlschlag_nicht_zurueck():
    """Waechter gegen die Rueckkehr des `continue`-Fehlers."""
    import inspect

    import main

    quelle = inspect.getsource(main.disc_watcher)
    assert "gesehen" in quelle, "die Wache muss sich merken, was sie je las"
    assert "disc_entscheidung(" in quelle, "die Entscheidung gehoert in die pure Funktion"


def test_schnappschuss_und_jobs_liefern_DIESELBE_jobliste():
    """Waechter gegen die zweite Fassung (Befund 29.08.2026).

    Commander: „Über den kompletten vorgang steht dort ‚Restzeit wird
    gemessen' aber messung wird nicht abgeschlossen."

    Die Restzeit wurde nur in `/jobs` berechnet. Der SSE-Schnappschuss baute
    seine Jobs mit dem nackten `_job_row_to_model` — also ohne. Seit V2-3
    liest die Oberflaeche aber genau diesen Schnappschuss. Die Restzeit wurde
    also brav berechnet und niemandem gezeigt.
    """
    import inspect

    import main

    quelle = inspect.getsource(main._snapshot)
    assert "jobs_fuer_ui(" in quelle, "der Schnappschuss baut die Jobs selbst"
    assert "_job_row_to_model(z).model_dump() for z in db.list_jobs" not in quelle


def test_die_jobliste_traegt_die_restzeit_felder():
    """Ohne die Felder im Modell schneidet FastAPI sie weg."""
    import main

    for feld in ("eta_sekunden", "eta_text", "can_retry", "retry_art"):
        assert feld in main.Job.model_fields, feld


# ── Kein Vor-Scan waehrend eines Rips (Befund 29.08.2026) ───────────────
#
# Commander: „der ‚Disk wird gelesen' panel braucht eeeeewig. Der Rip ist
# bereits bei 30% und er liest immernoch."
#
# Waehrend eines Rips haelt makemkvcon das Laufwerk. Ein zweites
# `makemkvcon info` daneben wartet bis zu seiner Zeitgrenze — und die Marke
# `_laeuft` blieb solange stehen, also stand auch das Panel. Es gibt keinen
# Grund, dann zu scannen: Das Laufwerk ist belegt, und WELCHE Disc drin ist,
# wissen wir — der Job laeuft ja auf ihr.


def test_kein_vorscan_solange_ein_job_auf_dem_laufwerk_laeuft(monkeypatch):
    import asyncio

    import main

    gescannt = []

    class NieBenutzt:
        def scan(self, pfad):
            gescannt.append(pfad)
            raise AssertionError("waehrend eines Rips darf nicht gescannt werden")

    monkeypatch.setattr(main, "PreScan", NieBenutzt)
    monkeypatch.setattr(main.db, "has_active_job", lambda p: True)
    main.DISC_CACHE.pop(r"\.\G:", None)

    asyncio.run(main._auto_prescan(r"\.\G:"))
    assert gescannt == []
    assert r"\.\G:" not in main.DISC_CACHE, "auch keine Marke setzen"


def test_ohne_laufenden_job_wird_normal_gescannt(monkeypatch):
    import asyncio

    import main

    class Ergebnis:
        title, year, disc_type, confidence, fingerprint = "Akira", 1988, "Blu-ray", 0.9, "x"

        def to_dict(self):
            return {"title": "Akira"}

    class Scanner:
        def scan(self, pfad):
            return Ergebnis()

    monkeypatch.setattr(main, "PreScan", Scanner)
    monkeypatch.setattr(main.db, "has_active_job", lambda p: False)
    monkeypatch.setattr(main.db, "add_log", lambda *a, **k: None)
    monkeypatch.setattr(main, "_duplikat_suchen", lambda f: None)
    monkeypatch.setattr(main, "_auto_rip_wenn_aktiviert",
                        lambda p: asyncio.sleep(0))
    main.DISC_CACHE.pop(r"\.\G:", None)

    asyncio.run(main._auto_prescan(r"\.\G:"))
    assert main.DISC_CACHE[r"\.\G:"]["title"] == "Akira"
    main.DISC_CACHE.pop(r"\.\G:", None)


def test_der_job_start_raeumt_eine_haengende_erkennung_weg():
    """Eine Marke, die KURZ VOR dem Rip gesetzt wurde, bliebe sonst bis zum
    Ende stehen — der Scan davor haelt sie ja fest."""
    import inspect

    import main

    quelle = inspect.getsource(main.create_job) if hasattr(main, "create_job") else ""
    if not quelle:
        # Der Endpunkt heisst anders — dann ueber das Modul suchen.
        quelle = inspect.getsource(main)
    assert 'DISC_CACHE.pop(device_path, None)' in quelle, \
        "beim Job-Start muss eine haengende Erkennungs-Marke weg"


# ── Kein zweiter Rip waehrend der Kompression (Befund 29.08.2026) ───────
#
# Der Commander: „Der Rip an sich war bereits fertig, bei der komprimierung
# passiert das" — Fehlertext: „Das Öffnen der Disk schlug fehl".
#
# Ablauf dahinter: Rip fertig → Disc ausgeworfen → Status `transcoding`.
# `has_active_job` kennt nur pending/running und meldete „Laufwerk frei".
# Die Disc-Wache sah den Auswurf als Statuswechsel, die Vollautomatik startete
# einen ZWEITEN Rip — in eine Schublade, die gerade herausfuhr.


def test_automatik_startet_nicht_waehrend_der_kompression(monkeypatch):
    import asyncio

    import main

    gestartet = []
    monkeypatch.setattr(main.db, "get_settings", lambda: {"autoRipStart": True})
    monkeypatch.setattr(main.db, "job_offen", lambda p: True)   # Job komprimiert
    monkeypatch.setattr(main.db, "has_active_job", lambda p: False)  # Laufwerk frei
    monkeypatch.setattr(main, "start_rip", lambda *a, **k: gestartet.append(a))
    main.DISC_CACHE[r"\.\G:"] = {"title": "Akira"}

    asyncio.run(main._auto_rip_wenn_aktiviert(r"\.\G:"))
    assert gestartet == [], "waehrend der Kompression faengt die Automatik nichts Neues an"


def test_job_offen_zaehlt_die_kompression_mit():
    from rippy.store import JOB_OFFEN

    assert "transcoding" in JOB_OFFEN, "die Kompression gehoert zum Vorgang"
    assert "canceling" in JOB_OFFEN, "ein Abbruch ist auch noch nicht durch"
    assert "completed" not in JOB_OFFEN and "failed" not in JOB_OFFEN


def test_auswurf_bleibt_waehrend_der_kompression_erlaubt():
    """Waechter gegen einen zu breiten Riegel: `has_active_job` fragt „haelt
    jemand das Laufwerk?" — waehrend der Kompression tut das niemand, die Disc
    darf raus. Nur die AUTOMATIK haelt sich zurueck."""
    import inspect

    from rippy import store

    assert "transcoding" not in inspect.getsource(store.has_active_job)


def test_laufwerk_bleibt_waehrend_eines_rips_unberuehrt(monkeypatch):
    """Befund 30.08.2026, aus dem Protokoll des Commanders:

        12:49:52  bluray-Rip gestartet
        12:50:09  [watcher] Laufwerk G: beantwortet keine Medien-Abfragen
        12:50:12  MSG 2003 SCSI-Fehler ILLEGAL REQUEST:INVALID FIELD IN CDB
        12:50:12  makemkvcon endete mit Code 11

    Der Waechter fragt alle drei Sekunden ab — drei CreateFileW plus IOCTLs
    auf ein Geraet, das makemkvcon gerade liest. Sein Befund: das Laufwerk
    hoert einfach auf zu lesen. `_auto_prescan` haelt sich seit dem
    29.08.2026 an dieselbe Regel; die Laufwerksabfrage tat es nicht.
    """
    import main

    gefragt = []
    monkeypatch.setattr(main.device_discovery, "list_optical_devices",
                        lambda: ["/dev/sr0"])
    monkeypatch.setattr(main.device_discovery, "device_info",
                        lambda p: gefragt.append(p) or dict({"id": "sr0", "name": "Laufwerk", "type": "bluray", "path": "/dev/sr0", "status": "ready", "model": "X", "serial": "Y", "grund": ""}))
    monkeypatch.setattr(main, "LETZTE_LAUFWERKE", [dict({"id": "sr0", "name": "Laufwerk", "type": "bluray", "path": "/dev/sr0", "status": "ready", "model": "X", "serial": "Y", "grund": ""})])
    monkeypatch.setattr(main.db, "has_active_job", lambda p: True)

    stand = main.laufwerke_mit_disc()
    assert gefragt == [], "waehrend eines Rips darf niemand das Laufwerk anfassen"
    assert stand[0]["type"] == "bluray"      # letzter bekannter Stand gilt


def test_ohne_rip_wird_das_laufwerk_normal_abgefragt(monkeypatch):
    """Die Ausnahme darf nur fuer den laufenden Rip gelten."""
    import main

    gefragt = []
    monkeypatch.setattr(main.device_discovery, "list_optical_devices",
                        lambda: ["/dev/sr0"])
    monkeypatch.setattr(main.device_discovery, "device_info",
                        lambda p: gefragt.append(p) or dict({"id": "sr0", "name": "Laufwerk", "type": "bluray", "path": "/dev/sr0", "status": "ready", "model": "X", "serial": "Y", "grund": ""}))
    monkeypatch.setattr(main, "LETZTE_LAUFWERKE", [dict({"id": "sr0", "name": "Laufwerk", "type": "bluray", "path": "/dev/sr0", "status": "ready", "model": "X", "serial": "Y", "grund": ""})])
    monkeypatch.setattr(main.db, "has_active_job", lambda p: False)

    main.laufwerke_mit_disc()
    assert gefragt == ["/dev/sr0"]