rippy/src/rippy/test_kette.py

"""Die ganze Kette im Standalone-Betrieb — von der Zustellung bis zum Job-Ende.

## Was hier bewiesen wird

    celery_client.abschicken(...)   die API stellt zu
      -> standalone.zustellung      uebersetzt in einen Auftrag
        -> lokal.einreihen          Auftragstabelle mit Lease
          -> Laeufer.einmal()       holt ihn und arbeitet ihn ab
            -> ablauf.rippen()      der echte Ablauf, ohne Celery
              -> db.update_job()    der Job endet in einem ehrlichen Zustand

Das ist der Weg, der unter Windows bis V2-4 gar nicht existierte: Dort haette
die API einen Rip eingereiht, und er waere fuer immer liegengeblieben, weil
niemand ihn holt.

## Warum ohne Laufwerk getestet wird

Es gibt hier keins (es haengt an der VM). Das ist kein Mangel, sondern der
wichtigere Fall: Ein Rip auf ein Geraet, das nicht antwortet, muss **ehrlich
scheitern** — mit einem Zustand und einer Begruendung, die im UI ankommen.
Was NICHT passieren darf: dass der Job auf „pending" stehenbleibt und
niemand erfaehrt, warum nichts geschieht.
"""

import time

import pytest

from rippy import standalone, store
from rippy.queue import laeufer as laeufer_modul
from rippy.queue import lokal


@pytest.fixture
def umgebung(tmp_path):
    vorher = store.zustand_sichern()
    store.verbinden(f"sqlite:///{(tmp_path / 'kette.db').as_posix()}")
    store.init_db()
    yield store
    store.engine_holen().dispose()
    store.zustand_wiederherstellen(vorher)


class FakeBus:
    def __init__(self):
        self.gesendet = []

    def senden(self, typ, daten=None, entitaet=None, entitaet_id=None):
        self.gesendet.append((typ, entitaet_id, daten or {}))


def test_die_ganze_kette_bis_zum_ehrlichen_fehlschlag(umgebung):
    """Ein Rip auf ein Geraet, das es nicht gibt.

    Erwartet wird KEIN Erfolg — erwartet wird, dass der Job in einem
    ENDZUSTAND landet und die Begruendung im Log steht. Ein Job, der auf
    „pending" haengenbleibt, waere der schlimmere Ausgang: Der Nutzer saehe
    einen Balken, der sich nie bewegt, und nirgends stuende warum.
    """
    umgebung.insert_job("job-kette", "/dev/gibt-es-nicht", disc_type=None)

    bus = FakeBus()
    zustellen = standalone.zustellung_bauen(bus)
    zustellen("worker.tasks.rip_disc", ["/dev/gibt-es-nicht", "job-kette", None])

    assert len(lokal.offene_auftraege()) == 1, "Der Auftrag wurde nicht eingereiht"

    lauf = laeufer_modul.Laeufer(standalone.ausfuehren, "test-knoten",
                                 kann={"art=rip"}, bus=bus, store=umgebung)
    assert lauf.einmal() is True, "Der Laeufer hat den Auftrag nicht geholt"

    job = umgebung.get_job("job-kette")
    assert job["status"] in ("failed", "completed"), (
        f"Der Job steht auf {job['status']!r} — er muss in einem Endzustand "
        "landen, sonst wartet der Nutzer auf etwas, das nie passiert.")
    assert job["status"] == "failed"
    assert job["error"], "Ein Fehlschlag ohne Begruendung ist im UI wertlos"

    meldungen = " ".join(z["message"] for z in umgebung.list_logs(limit=50))
    assert "job-kette" in meldungen, "Im Log steht nichts ueber diesen Job"

    assert lokal.offene_auftraege() == [], "Der Auftrag blieb in der Queue liegen"


def test_der_laeufer_bleibt_danach_arbeitsfaehig(umgebung):
    """Nach einem Fehlschlag muss der naechste Auftrag trotzdem laufen —
    sonst legt ein einziger kaputter Rip die ganze Installation lahm."""
    umgebung.insert_job("job-a", "/dev/gibt-es-nicht")
    umgebung.insert_job("job-b", "/dev/gibt-es-auch-nicht")
    zustellen = standalone.zustellung_bauen()
    zustellen("worker.tasks.rip_disc", ["/dev/gibt-es-nicht", "job-a", None])
    zustellen("worker.tasks.rip_disc", ["/dev/gibt-es-auch-nicht", "job-b", None])

    lauf = laeufer_modul.Laeufer(standalone.ausfuehren, "test-knoten",
                                 kann={"art=rip"}, store=umgebung)
    assert lauf.einmal() is True
    assert lauf.einmal() is True
    assert umgebung.get_job("job-a")["status"] == "failed"
    assert umgebung.get_job("job-b")["status"] == "failed"


def test_die_api_stellt_ueber_dieselbe_stelle_zu(umgebung, monkeypatch):
    """`celery_client.start_rip` ist der Weg, den POST /jobs geht. Er MUSS
    im Standalone-Betrieb in der lokalen Queue landen — sonst reiht die API
    an einen Broker ein, den es nicht gibt, und der Job verschwindet."""
    import sys

    from rippy.daemon import _api_pfad

    if _api_pfad() not in sys.path:
        sys.path.insert(0, _api_pfad())
    import celery_client

    umgebung.insert_job("job-api", "/dev/sr0")
    monkeypatch.setattr(celery_client, "_zusteller", standalone.zustellung_bauen())

    celery_client.start_rip("/dev/sr0", "job-api", "/app/media/x")

    offen = lokal.offene_auftraege()
    assert len(offen) == 1
    assert offen[0]["job_id"] == "job-api"
    assert offen[0]["art"] == "rip"


def test_ein_rip_haengt_nicht_ewig(umgebung):
    """Zeitgrenze als Zusage: Ein Auftrag auf ein totes Geraet muss ZUEGIG
    scheitern. Haengt er, merkt es im Betrieb niemand — der Job steht auf
    „laeuft", und der Laeufer nimmt keinen weiteren an."""
    umgebung.insert_job("job-zeit", "/dev/nichts")
    standalone.zustellung_bauen()("worker.tasks.rip_disc", ["/dev/nichts", "job-zeit", None])

    lauf = laeufer_modul.Laeufer(standalone.ausfuehren, "k", kann={"art=rip"},
                                 store=umgebung)
    start = time.monotonic()
    lauf.einmal()
    dauer = time.monotonic() - start
    assert dauer < 30, f"Der Fehlschlag brauchte {dauer:.1f}s — das ist zu lang"