rippy/docker/api/test_eta.py

"""Tests der Restzeit-Schätzung — reine Funktionen, keine Infrastruktur.

Die Zahlen in den Szenarien sind die echten Messwerte vom 25.07.2026: Der
4K-Encode von Akira kam in 29 Minuten von 0 auf 1,44 % (an der Leseposition im
Quellstrom gemessen, `/proc/<pid>/fdinfo/`) — hochgerechnet 28-55 Stunden. Genau
dieser Fall muss eine ehrliche Antwort geben, statt „gleich fertig" zu behaupten.
"""

import eta


def test_ohne_genug_punkte_keine_aussage():
    assert eta.restzeit_sekunden(None, 100.0) == -1
    assert eta.restzeit_sekunden({"status": "transcoding", "punkte": []}, 100.0) == -1
    einer = {"status": "transcoding", "punkte": [[0.0, 5]]}
    assert eta.restzeit_sekunden(einer, 100.0) == -1


def test_zu_kurze_spanne_keine_aussage():
    """Zwei Punkte 10 s auseinander sagen nichts über einen Encode, der Stunden
    läuft — MINDEST_SPANNE_SEKUNDEN verhindert die Hochrechnung."""
    reihe = {"status": "transcoding", "punkte": [[0.0, 1], [10.0, 2]]}
    assert eta.restzeit_sekunden(reihe, 10.0) == -1


def test_kein_fortschritt_keine_aussage():
    """Eine Stunde ohne einen einzigen Prozentpunkt: die Rate ist unbekannt,
    nicht null. Ohne diese Sperre käme eine Division durch 0."""
    reihe = {"status": "transcoding", "punkte": [[0.0, 3]]}
    reihe = eta.beobachtung_hinzufuegen(reihe, "transcoding", 3, 3600.0)
    assert eta.restzeit_sekunden(reihe, 3600.0) == -1


def test_einfache_hochrechnung():
    """10 % in 10 Minuten → 90 % brauchen 90 Minuten."""
    reihe = {"status": "ripping", "punkte": [[0.0, 0], [600.0, 10]]}
    assert eta.restzeit_sekunden(reihe, 600.0) == 90 * 60


def test_der_echte_4k_fall_landet_in_der_groessenordnung_tage():
    """Gemessen: 1,44 % in 29 Minuten. Die Schätzung muss in der Größenordnung
    TAGE landen — das ist die Angabe, deren Fehlen den 50-Stunden-Lauf am
    25.07.2026 unsichtbar machte. Die genaue Stundenzahl ist Nebensache; wer
    „1 Tag 23 h" liest, bricht ab, wer nichts liest, wartet."""
    reihe = {"status": "transcoding", "punkte": [[0.0, 0], [29 * 60.0, 1]]}
    sekunden = eta.restzeit_sekunden(reihe, 29 * 60.0)
    stunden = sekunden / 3600
    assert 45 < stunden < 50           # 99 % × 29 min ≈ 47,85 h
    assert eta.formatiere_restzeit(sekunden) == "noch ca. 1 Tag 23 h"
    # Etwas langsamer, und es heißt nur noch „mehr als 2 Tage" — bei der
    # Größenordnung wäre jede Stundenangabe erfundene Genauigkeit.
    langsamer = {"status": "transcoding", "punkte": [[0.0, 0], [45 * 60.0, 1]]}
    assert eta.formatiere_restzeit(
        eta.restzeit_sekunden(langsamer, 45 * 60.0)) == "mehr als 2 Tage"


def test_stillstand_verlaengert_die_schaetzung():
    """Kernpunkt: Bleibt der Fortschritt stehen, MUSS die Restzeit steigen —
    sonst zeigt ein hängender Job stundenlang „noch 5 Minuten"."""
    reihe = {"status": "transcoding", "punkte": [[0.0, 0], [600.0, 50]]}
    frisch = eta.restzeit_sekunden(reihe, 600.0)
    # ... eine Stunde später steht der Fortschritt immer noch bei 50 %
    reihe = eta.beobachtung_hinzufuegen(reihe, "transcoding", 50, 4200.0)
    spaeter = eta.restzeit_sekunden(reihe, 4200.0)
    assert spaeter > frisch * 5


def test_statuswechsel_verwirft_die_reihe():
    """Rip (eine Stunde) und Kompression (Tage) haben nichts miteinander zu tun.
    Ohne diesen Schnitt entstünde beim Phasenwechsel eine Phantasiezahl."""
    reihe = {"status": "ripping", "punkte": [[0.0, 0], [600.0, 50]]}
    neu = eta.beobachtung_hinzufuegen(reihe, "transcoding", 2, 610.0)
    assert neu["status"] == "transcoding"
    assert neu["punkte"] == [[610.0, 2]]
    assert eta.restzeit_sekunden(neu, 610.0) == -1


def test_nur_das_fenster_zaehlt():
    """Alte Punkte fliegen raus — HandBrake wird bei komplexen Szenen langsamer,
    und dann lügt der Anfang der Messreihe."""
    reihe = None
    for i in range(20):
        reihe = eta.beobachtung_hinzufuegen(reihe, "transcoding", i, float(i * 60))
    assert len(reihe["punkte"]) == eta.FENSTER
    assert reihe["punkte"][0][1] == 10  # die ersten zehn sind weg


def test_gleicher_fortschritt_haengt_keinen_punkt_an():
    reihe = eta.beobachtung_hinzufuegen(None, "transcoding", 7, 0.0)
    reihe = eta.beobachtung_hinzufuegen(reihe, "transcoding", 7, 30.0)
    reihe = eta.beobachtung_hinzufuegen(reihe, "transcoding", 7, 60.0)
    assert reihe["punkte"] == [[0.0, 7]]


def test_fertig_ist_null():
    reihe = {"status": "transcoding", "punkte": [[0.0, 50], [600.0, 100]]}
    assert eta.restzeit_sekunden(reihe, 600.0) == 0
    assert eta.formatiere_restzeit(0) == "unter einer Minute"


# --- Textform ---------------------------------------------------------------


def test_textform_deckt_alle_groessenordnungen():
    assert eta.formatiere_restzeit(-1) == ""
    assert eta.formatiere_restzeit(None) == ""
    assert eta.formatiere_restzeit(30) == "unter einer Minute"
    assert eta.formatiere_restzeit(90) == "noch ca. 1 min"
    assert eta.formatiere_restzeit(45 * 60) == "noch ca. 45 min"
    assert eta.formatiere_restzeit(3 * 3600 + 7 * 60) == "noch ca. 3 h 07 min"
    assert eta.formatiere_restzeit(30 * 3600) == "noch ca. 1 Tag 6 h"
    assert eta.formatiere_restzeit(60 * 3600) == "mehr als 2 Tage"


# --- Der Weg über den Cache -------------------------------------------------


def test_aktualisiere_und_schaetze_ueber_zwei_aufrufe():
    """So läuft es live: /jobs wird alle vier Sekunden abgefragt, jeder Aufruf
    schreibt die Reihe fort."""
    speicher = {}

    def hole(k):
        return speicher.get(k)

    def lege(k, wert, expire=None):
        speicher[k] = wert

    erst = eta.aktualisiere_und_schaetze(
        "job1", "transcoding", 10, 0.0, hole, lege)
    assert erst == {"sekunden": -1, "text": ""}  # ein Punkt sagt nichts

    dann = eta.aktualisiere_und_schaetze(
        "job1", "transcoding", 20, 600.0, hole, lege)
    assert dann["sekunden"] == 80 * 60
    assert dann["text"] == "noch ca. 1 h 20 min"


def test_kaputter_cache_bringt_nichts_zum_absturz():
    """Redis weg → keine ETA, aber die Job-Liste muss weiter funktionieren."""
    def kaputt_holen(k):
        raise RuntimeError("Redis weg")

    def kaputt_legen(k, wert, expire=None):
        raise RuntimeError("Redis weg")

    ergebnis = eta.aktualisiere_und_schaetze(
        "job1", "transcoding", 10, 0.0, kaputt_holen, kaputt_legen)
    assert ergebnis == {"sekunden": -1, "text": ""}


def test_fertige_und_wartende_jobs_bekommen_keine_eta():
    speicher = {}
    for status, progress in (("completed", 100), ("failed", 42), ("pending", 0)):
        ergebnis = eta.aktualisiere_und_schaetze(
            "x", status, progress, 0.0,
            speicher.get, lambda k, v, expire=None: speicher.__setitem__(k, v))
        assert ergebnis == {"sekunden": -1, "text": ""}
    assert speicher == {}  # nichts geschrieben


# ── Ohne Redis muss die Schaetzung trotzdem zustande kommen ─────────────
#
# Commander 29.08.2026: „Über den kompletten vorgang steht dort ‚Restzeit wird
# gemessen' aber messung wird nicht abgeschlossen. Das heißt man hat kein ETA"
#
# Die Messreihe lag nur im Cache, und der Modul-Kopf sagte „Der Cache (Redis)
# ist schon da". Im Container stimmt das. Auf einem Windows-PC gab `cache_get`
# bei JEDEM Aufruf None zurueck — also jedes Mal eine frische Reihe mit EINEM
# Punkt, und `restzeit_sekunden` braucht zwei.


def _ohne_cache():
    """Ein Cache, der nichts behaelt — genau wie Redis, das es nicht gibt."""
    return (lambda k: None), (lambda k, v, expire=None: False)


def test_ohne_cache_kommt_trotzdem_eine_schaetzung():
    """DER Fall des Commanders."""
    import eta

    eta._REIHEN.clear()
    lesen, schreiben = _ohne_cache()
    # Zwei Messpunkte, 120 s auseinander, 10 % Fortschritt.
    eta.aktualisiere_und_schaetze("j1", "running", 10, 1000.0, lesen, schreiben)
    ergebnis = eta.aktualisiere_und_schaetze("j1", "running", 20, 1120.0, lesen, schreiben)

    assert ergebnis["sekunden"] > 0, "ohne Redis kam nie eine Zahl heraus"
    assert ergebnis["text"], "und damit stand dauerhaft „wird gemessen"


def test_der_cache_hat_weiterhin_vorrang():
    """Wo es Redis gibt, bleibt Redis die Quelle — es ueberlebt einen
    API-Neustart, das Woerterbuch im Prozess nicht."""
    import eta

    eta._REIHEN.clear()
    aus_dem_cache = {"status": "running", "punkte": [[500.0, 5], [560.0, 15]]}
    ergebnis = eta.aktualisiere_und_schaetze(
        "j2", "running", 25, 620.0,
        lambda k: aus_dem_cache, lambda k, v, expire=None: True)
    # Drei Punkte, 120 s fuer 20 % -> die Reihe aus dem Cache wurde benutzt.
    assert ergebnis["sekunden"] > 0


def test_ein_statuswechsel_beginnt_auch_ohne_cache_neu():
    """Rip und Kompression haben nichts miteinander zu tun."""
    import eta

    eta._REIHEN.clear()
    lesen, schreiben = _ohne_cache()
    eta.aktualisiere_und_schaetze("j3", "running", 50, 1000.0, lesen, schreiben)
    eta.aktualisiere_und_schaetze("j3", "running", 90, 1100.0, lesen, schreiben)
    neu = eta.aktualisiere_und_schaetze("j3", "transcoding", 5, 1110.0, lesen, schreiben)
    assert neu["sekunden"] == -1, "nach dem Wechsel gibt es noch keine Aussage"


def test_alte_reihen_wachsen_nicht_unbegrenzt():
    """Wer einen Vorrat anlegt, raeumt ihn auch weg (AGENTS.md)."""
    import eta

    eta._REIHEN.clear()
    lesen, schreiben = _ohne_cache()
    eta.aktualisiere_und_schaetze("alt", "running", 10, 0.0, lesen, schreiben)
    # Einen Tag spaeter ein anderer Job -> der alte fliegt raus.
    eta.aktualisiere_und_schaetze("neu", "running", 10,
                                  eta.REIHE_HALTBAR_SEKUNDEN + 10.0, lesen, schreiben)
    assert eta.schluessel("alt") not in eta._REIHEN
    assert eta.schluessel("neu") in eta._REIHEN