"""Tests der Schlüssel-Automatik — ohne MakeMKV, ohne Laufwerk, ohne Netz. Die Testdaten sind gemessen (26.07.2026, Commander-PC, MakeMKV 1.18.4): `makemkvcon -r --cache=1 info disc:9999` auf einer Maschine OHNE optisches Laufwerk gibt sechzehn leere `DRV:`-Zeilen. Die BELEGTE Form ließ sich dort nicht messen — deshalb prüfen die Tests vor allem, dass die Automatik im Zweifel NICHTS tut. """ import schluessel # Wörtlich aus dem echten Lauf (gekürzt auf drei der sechzehn Zeilen). DRV_LEER = """MSG:1005,0,1,"MakeMKV v1.18.4 win(x64-release) gestartet","%1 gestartet","MakeMKV v1.18.4 win(x64-release)" MSG:5042,0,0,"Das Programm konnte keine verwendbaren optischen Laufwerke finden.","Das Programm konnte keine verwendbaren optischen Laufwerke finden." DRV:0,256,999,0,"","","" DRV:1,256,999,0,"","","" DRV:2,256,999,0,"","","" MSG:5010,0,0,"Das Öffnen der Disk schlug fehl ","Das Öffnen der Disk schlug fehl " TCOUNT:0 """ # So sieht eine belegte Zeile AUS SICHT DES FORMATS aus — nicht gemessen, # deshalb steht sie nur hier im Test und nicht als Annahme im Code. DRV_MIT_DISC = """DRV:0,2,999,12,"BD-RE HL-DT-ST BU40N","AKIRA","D:" DRV:1,256,999,0,"","","" """ def test_leere_laufwerke_ergeben_nichts(): """Der gemessene Fall: kein optisches Laufwerk → die Automatik tut nichts.""" assert schluessel.parse_laufwerke(DRV_LEER) == [] def test_kein_disc_name_gilt_als_nichts_da(): """Konservativ mit Absicht: Ein falsch-negativ ist harmlos (der Wächter greift trotzdem), ein falsch-positiv wäre ein makemkvcon-Lauf ins Leere.""" assert schluessel.parse_laufwerke('DRV:0,2,999,12,"BD-RE BU40N","",""') == [] def test_belegtes_laufwerk_wird_gefunden(): gefunden = schluessel.parse_laufwerke(DRV_MIT_DISC) assert len(gefunden) == 1 assert gefunden[0]["index"] == 0 assert gefunden[0]["disc"] == "AKIRA" assert gefunden[0]["geraet"] == "D:" def test_muell_und_leer_stuerzen_nicht_ab(): assert schluessel.parse_laufwerke("") == [] assert schluessel.parse_laufwerke(None) == [] assert schluessel.parse_laufwerke("irgendwas ganz anderes") == [] # --- Wann hochgeladen wird -------------------------------------------------- def test_erste_runde_laedt_immer_hoch(): """Rippy soll den Bestand kennen, auch wenn MakeMKV gerade nichts Neues gelernt hat.""" assert schluessel.hat_sich_geaendert(None, (6420480, 1753000000)) is True def test_unveraenderte_datei_wird_nicht_wieder_geschickt(): stand = (6420480, 1753000000) assert schluessel.hat_sich_geaendert(stand, stand) is False def test_geaenderte_datei_wird_geschickt(): assert schluessel.hat_sich_geaendert( (6420480, 1753000000), (6500000, 1753009999)) is True def test_fehlende_datei_loest_keinen_upload_aus(): """(0, 0) heißt „gibt es nicht" — dann gibt es auch nichts zu schicken.""" assert schluessel.hat_sich_geaendert(None, (0, 0)) is False assert schluessel.hat_sich_geaendert((6420480, 1753000000), (0, 0)) is False # --- Der Upload ------------------------------------------------------------- def test_upload_schickt_den_rohen_koerper(tmp_path): """Kein JSON, kein Multipart: genau so nimmt POST /system/keystore die Datei an (der API fehlt python-multipart).""" datei = tmp_path / "_private_data.tar" datei.write_bytes(b"TAR-INHALT") gesehen = {} class Antwort: def __enter__(self): return self def __exit__(self, *a): return False def read(self): return b"" def oeffner(anfrage, timeout=None): gesehen["url"] = anfrage.full_url gesehen["daten"] = anfrage.data gesehen["typ"] = anfrage.headers.get("Content-type") return Antwort() assert schluessel.hochladen("1.2.3.4", str(datei), oeffner=oeffner) == "" assert gesehen["url"] == "http://1.2.3.4/api/system/keystore" assert gesehen["daten"] == b"TAR-INHALT" assert gesehen["typ"] == "application/octet-stream" def test_upload_ohne_adresse_oder_datei_meldet_klartext(): assert "Adresse" in schluessel.hochladen("", "/x.tar") assert "Schlüsseldatei" in schluessel.hochladen("1.2.3.4", "") def test_upload_fehler_wird_gemeldet_nicht_geworfen(tmp_path): """Der Worker muss weiterarbeiten, auch wenn Rippy gerade weg ist.""" datei = tmp_path / "_private_data.tar" datei.write_bytes(b"x") def kaputt(anfrage, timeout=None): raise OSError("Netz weg") fehler = schluessel.hochladen("1.2.3.4", str(datei), oeffner=kaputt) assert "Hochladen fehlgeschlagen" in fehler # --- Eine ganze Runde ------------------------------------------------------- def test_runde_ohne_makemkv_tut_nichts(): """Auf einem PC ohne MakeMKV darf die Automatik nicht herumfummeln.""" stand, was = schluessel.runde("1.2.3.4", None, programm="") assert (stand, was) == (None, "kein-makemkv") def test_runde_stoesst_erst_an_und_prueft_dann(monkeypatch, tmp_path): """Reihenfolge mit Absicht: Ein gerade geholter Schluessel soll in DERSELBEN Runde mitgenommen werden, nicht erst eine Minute spaeter.""" datei = tmp_path / "_private_data.tar" datei.write_bytes(b"neu") monkeypatch.setattr(schluessel, "schluesseldatei", lambda: str(datei)) ablauf = [] def laufen(cmd, **kwargs): # disc:9999 = Laufwerke auflisten, disc:0 = Disc oeffnen ablauf.append(cmd[-1]) class E: stdout = DRV_MIT_DISC if cmd[-1] == "disc:9999" else "" return E() class Antwort: def __enter__(self): return self def __exit__(self, *a): return False def read(self): return b"" meldungen = [] stand, was = schluessel.runde( "1.2.3.4", None, melden=lambda lvl, t: meldungen.append((lvl, t)), programm="makemkvcon.exe", laufen=laufen, oeffner=lambda a, timeout=None: Antwort()) assert ablauf == ["disc:9999", "disc:0"] assert was == "hochgeladen" assert stand == schluessel.datei_stand(str(datei)) assert any("Disc erkannt" in t for _, t in meldungen) assert any("wirkt ab dem nächsten Rip" in t for _, t in meldungen) def test_runde_ohne_aenderung_schickt_nichts(monkeypatch, tmp_path): datei = tmp_path / "_private_data.tar" datei.write_bytes(b"alt") monkeypatch.setattr(schluessel, "schluesseldatei", lambda: str(datei)) vorher = schluessel.datei_stand(str(datei)) def laufen(cmd, **kwargs): class E: stdout = DRV_LEER return E() def darf_nicht(anfrage, timeout=None): raise AssertionError("kein Upload erwartet") stand, was = schluessel.runde( "1.2.3.4", vorher, programm="makemkvcon.exe", laufen=laufen, oeffner=darf_nicht) assert (stand, was) == (vorher, "unveraendert")