"""Tests für die Auswertung des llama-swap-Journals — Zeilenformat wie auf der Box (23. und 25.09.2026).""" from datetime import datetime import pytest from services import modell_nutzung ZEILEN = [ ('2026-09-23T03:12:44+02:00 box llama-swap[1]: [INFO] Request 192.168.178.28 ' '"POST /v1/chat/completions HTTP/1.1" 200 10254 "Go-http-client/1.1" 1.2s'), ('2026-09-23T03:12:50+02:00 box llama-swap[1]: [INFO] Request 192.168.178.28 ' '"POST /v1/chat/completions HTTP/1.1" 200 9000 "Go-http-client/1.1" 1.1s'), ('2026-09-23T07:01:00+02:00 box llama-swap[1]: [INFO] Request 127.0.0.1 ' '"POST /v1/chat/completions HTTP/1.1" 200 800 "python-httpx/0.28.1" 3s'), ('2026-09-23T07:02:00+02:00 box llama-swap[1]: [INFO] Request 127.0.0.1 ' '"POST /v1/embeddings HTTP/1.1" 200 800 "python-httpx/0.28.1" 20ms'), "2026-09-17T19:59:00+02:00 box llama-swap[1]: [INFO] Health check passed on http://localhost:5805/health", ('2026-09-10T03:00:00+02:00 box llama-swap[1]: [INFO] Request 192.168.178.28 ' '"POST /v1/chat/completions HTTP/1.1" 200 1 "Go-http-client/1.1" 1s'), ] NAMEN = {"127.0.0.1": "Lucy und OpenChamber über MC2", "192.168.178.28": "NerdQuiz auf Arcane"} JETZT = datetime.fromisoformat("2026-09-23T20:45:00+02:00") def anfrage(zeit: str, ip: str = "192.168.178.28", status: int = 200, dauer: str = "1s", programm: str = "Go-http-client/1.1") -> str: return (f'{zeit} tobisniceaiarbeitstier llama-swap[237863]: [INFO] Request {ip} ' f'"POST /v1/chat/completions HTTP/1.1" {status} 805 "{programm}" {dauer}') def je_name(d: dict) -> dict: return {a["name"]: a for a in d["absender"]} def test_zaehlt_nur_chat_anfragen_der_letzten_sieben_tage_je_absender(): d = modell_nutzung.werte_aus(ZEILEN, JETZT, NAMEN) # Der Lauf vom 10.09. liegt vor dem Fenster (17.09. 00:00 bis jetzt), Embeddings zählen nicht. assert [(a["name"], a["anfragen"]) for a in d["absender"]] == [ ("NerdQuiz auf Arcane", 2), ("Lucy und OpenChamber über MC2", 1), ] assert d["seit"] == "2026-09-17T00:00:00+02:00" assert d["gesamt"]["anfragen"] == 3 def test_letzte_24h_je_stunde_und_zuletzt_geladen(): d = modell_nutzung.werte_aus(ZEILEN, JETZT, NAMEN) stunden = {s["stunde"]: s["je_absender"] for s in d["letzte_24h"]} assert stunden["03"] == {"NerdQuiz auf Arcane": 2} # der Lauf vom 10.09. liegt außerhalb assert stunden["07"] == {"Lucy und OpenChamber über MC2": 1} assert len(d["letzte_24h"]) == 24 assert d["zuletzt_geladen"]["Qwen3.8-27B"].startswith("2026-09-17T19:59") def test_unbekannte_ip_bleibt_als_ip_stehen_und_kommt_nach_den_bekannten(): zeile = anfrage("2026-09-23T10:00:00+02:00", ip="10.0.0.9", programm="curl/8") d = modell_nutzung.werte_aus([zeile], JETZT, NAMEN) assert [(a["name"], a["anfragen"], a["reihe"]) for a in d["absender"]] == [("10.0.0.9", 1, 2)] @pytest.mark.parametrize(("text", "sekunden"), [ ("388.092603ms", 0.388092603), ("3.416448406s", 3.416448406), ("31.899196571s", 31.899196571), ("1m2.5s", 62.5), ("2h0m0s", 7200.0), ("850µs", 0.00085), ("850μs", 0.00085), ("12ns", 0.000000012), ("0s", 0.0), ]) def test_dauer_liest_go_dauern(text, sekunden): assert modell_nutzung.dauer_s(text) == pytest.approx(sekunden) @pytest.mark.parametrize("text", ["", "abc", "805", "1.5x", "s", '"Go-http-client/1.1"', "1s2"]) def test_dauer_ohne_einheit_ist_keine_dauer(text): assert modell_nutzung.dauer_s(text) is None def test_antwortzeiten_fehler_und_letzte_anfrage_je_absender(): zeilen = [ anfrage("2026-09-23T09:00:00+02:00", dauer="400ms"), anfrage("2026-09-23T09:01:00+02:00", dauer="1.2s"), anfrage("2026-09-23T09:02:00+02:00", dauer="1m2s"), anfrage("2026-09-23T09:03:00+02:00", status=502, dauer="31.9s"), anfrage("2026-09-23T12:00:00+02:00", ip="127.0.0.1", status=499, dauer="5s", programm="python-httpx/0.28.1"), anfrage("2026-09-23T12:01:00+02:00", ip="127.0.0.1", dauer="600ms", programm="python-httpx/0.28.1"), # Ohne Dauer am Ende (anderes Format): zählt, geht aber in keine Zeit ein. '2026-09-23T12:02:00+02:00 box llama-swap[1]: [INFO] Request 127.0.0.1 "POST /v1/chat/completions HTTP/1.1"', ] d = modell_nutzung.werte_aus(zeilen, JETZT, NAMEN) nq, lucy = je_name(d)["NerdQuiz auf Arcane"], je_name(d)["Lucy und OpenChamber über MC2"] assert nq["anfragen"] == 4 assert nq["fehler"] == 1 and nq["fehler_codes"] == {"502": 1} # Typisch und 9 von 10 nur aus den erfolgreichen Antworten; die Rechenzeit zählt alle. assert nq["dauer_typisch_s"] == pytest.approx(1.2) assert nq["dauer_p90_s"] == pytest.approx(62.0) assert nq["dauer_summe_s"] == pytest.approx(0.4 + 1.2 + 62 + 31.9, abs=0.05) assert nq["zuletzt"] == "2026-09-23T09:03:00+02:00" assert lucy["anfragen"] == 3 assert lucy["fehler_codes"] == {"499": 1} assert lucy["dauer_typisch_s"] == pytest.approx(0.6) assert lucy["zuletzt"] == "2026-09-23T12:02:00+02:00" assert d["gesamt"]["fehler"] == 2 assert d["gesamt"]["fehler_codes"] == {"502": 1, "499": 1} assert d["gesamt"]["dauer_typisch_s"] == pytest.approx(0.9) # Median aus 0,4 · 0,6 · 1,2 · 62 assert d["gesamt"]["dauer_p90_s"] == pytest.approx(62.0) def test_ohne_anfragen_bleiben_die_zeiten_leer(): d = modell_nutzung.werte_aus([], JETZT, NAMEN) assert d["absender"] == [] assert d["gesamt"] == {"anfragen": 0, "fehler": 0, "fehler_codes": {}, "dauer_typisch_s": None, "dauer_p90_s": None, "dauer_summe_s": 0.0} assert list(d["je_tag"].values()) == [{}] * 7 assert all(s["je_absender"] == {} for s in d["stunden"]) def test_farbe_folgt_dem_absender_nicht_dem_rang(): """Überholt Lucy NerdQuiz, bleibt die Reihe (= Farbe) gleich — sie kommt aus der Namensliste.""" zeilen = [anfrage(f"2026-09-23T10:0{i}:00+02:00", ip="127.0.0.1") for i in range(5)] zeilen.append(anfrage("2026-09-23T11:00:00+02:00")) namen = {"192.168.178.28": "NerdQuiz auf Arcane", "127.0.0.1": "Lucy", "::1": "Lucy", "192.168.178.22": "PC"} d = modell_nutzung.werte_aus(zeilen, JETZT, namen) assert [(a["name"], a["reihe"]) for a in d["absender"]] == [("Lucy", 1), ("NerdQuiz auf Arcane", 0)] def test_verlauf_je_tag_deckt_sieben_kalendertage_bis_heute(): zeilen = [ anfrage("2026-09-16T23:59:59+02:00"), # vor dem Fenster anfrage("2026-09-17T00:00:00+02:00"), anfrage("2026-09-20T22:30:00+00:00"), # in UTC geschrieben: in Berlin schon der 21.09. anfrage("2026-09-23T20:00:00+02:00", ip="127.0.0.1"), ] d = modell_nutzung.werte_aus(zeilen, JETZT, NAMEN) assert list(d["je_tag"]) == [f"2026-09-{t}" for t in range(17, 24)] assert d["je_tag"]["2026-09-17"] == {"NerdQuiz auf Arcane": 1} assert d["je_tag"]["2026-09-20"] == {} assert d["je_tag"]["2026-09-21"] == {"NerdQuiz auf Arcane": 1} assert d["je_tag"]["2026-09-23"] == {"Lucy und OpenChamber über MC2": 1} def test_stunden_laufen_chronologisch_bis_zur_laufenden_stunde(): zeilen = [ anfrage("2026-09-22T20:59:00+02:00"), # vor der ersten Stunde (21:00 gestern) anfrage("2026-09-22T21:10:00+02:00"), anfrage("2026-09-23T00:30:00+02:00", ip="127.0.0.1"), anfrage("2026-09-23T20:44:00+02:00"), ] d = modell_nutzung.werte_aus(zeilen, JETZT, NAMEN) stunden = d["stunden"] assert len(stunden) == 24 assert stunden[0]["beginn"] == "2026-09-22T21:00:00+02:00" assert stunden[-1]["beginn"] == "2026-09-23T20:00:00+02:00" assert stunden[0]["je_absender"] == {"NerdQuiz auf Arcane": 1} assert stunden[3]["je_absender"] == {"Lucy und OpenChamber über MC2": 1} assert stunden[-1]["je_absender"] == {"NerdQuiz auf Arcane": 1} def test_stunden_ueber_die_zeitumstellung(): """Am 25.10.2026 gibt es 02:00 zweimal — die Stunden bleiben 24 echte Stunden.""" d = modell_nutzung.werte_aus([], datetime.fromisoformat("2026-10-25T12:30:00+01:00"), NAMEN) beginne = [s["beginn"] for s in d["stunden"]] assert beginne[0] == "2026-10-24T14:00:00+02:00" assert "2026-10-25T02:00:00+02:00" in beginne and "2026-10-25T02:00:00+01:00" in beginne assert beginne[-1] == "2026-10-25T12:00:00+01:00" def test_ladevorgaenge_je_modell_neueste_zuerst(): geladen = "box llama-swap[1]: [INFO] <{}> Health check passed on http://localhost:5804/health" zeilen = [ f"2026-09-20T06:09:02+02:00 {geladen.format('Qwen3.6-35B-A3B')}", f"2026-09-21T14:57:07+02:00 {geladen.format('Qwen3-VL-30B-A3B-Instruct')}", f"2026-09-23T11:25:09+02:00 {geladen.format('Qwen3.6-35B-A3B')}", f"2026-09-23T11:25:46+02:00 {geladen.format('Qwen3.6-35B-A3B')}", f"2026-09-10T11:25:46+02:00 {geladen.format('Altes-Modell')}", # vor dem Fenster ] d = modell_nutzung.werte_aus(zeilen, JETZT, NAMEN) assert d["ladevorgaenge"] == [ {"modell": "Qwen3.6-35B-A3B", "anzahl": 3, "zeiten": ["2026-09-23T11:25:46+02:00", "2026-09-23T11:25:09+02:00", "2026-09-20T06:09:02+02:00"]}, {"modell": "Qwen3-VL-30B-A3B-Instruct", "anzahl": 1, "zeiten": ["2026-09-21T14:57:07+02:00"]}, ] assert d["zuletzt_geladen"] == {"Qwen3.6-35B-A3B": "2026-09-23T11:25:46+02:00", "Qwen3-VL-30B-A3B-Instruct": "2026-09-21T14:57:07+02:00"}