HitonabiandClaude Opus 5.5
91aa16eee1
Ampel / ampel (push) Failing after 21s
llama.cpp kann Draft-Beschleunigung und Bilder nicht zusammen (HTTP 500 "failed to process speculative batch", b11057 und b11157 geprueft; speculative.n_max=0 je Anfrage hilft nicht). Darum bekommen Hirn und Coder je einen Bild-Zwilling: gleiche Gewichte plus Projektor, ohne Draft (vision, coder-bild), in einer eigenen llama-swap-Gruppe, die den Coder nicht verdraengt. Probe 24.09.: beide 8/8 Bildmerkmale; Hirn-Zwilling 68 t/s, Coder-Zwilling 12,5 t/s. Bild-Weiche v3 im Gateway: Bild im aktuellen Schritt geht an den Zwilling der Rolle, aeltere Bilder werden einmal beschrieben (gemerkt) und als Text mitgeschickt, damit der Rest einer Agenten-Aufgabe wieder beim schnellen Modell laeuft. Qwen3-VL gibt "vision" ab, der Coder verliert den Projektor, der mit Draft nur HTTP 500 lieferte. Radar misst die Bildfaehigkeit des heutigen Modells ueber dessen Zwilling (sonst gewaenne jeder bildfaehige Kandidat mit "versteht Bilder"). Pruefstand: Coder darf vor dem Aendern lesen (Version 3). Modelle-Seite zeigt "Bilder: ja" ueber den Zwilling. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
657 lines
35 KiB
Python
657 lines
35 KiB
Python
"""Tests für das Modell-Radar: reine Logik (Zeitfenster, Wochengrenze, Filter, Speicher-Rechnung,
|
||
Zustandsübergänge) und die Urteile und Programmieraufgaben des Prüfstands — ohne Netz und ohne Box.
|
||
Zahlen wie auf der Box am 23.09.2026 (Ornith 21,7 GB + mmproj, Flash-Next UD-IQ3_XXS 82 GB)."""
|
||
|
||
import json
|
||
import os
|
||
import struct
|
||
import time
|
||
import zlib
|
||
from datetime import date, datetime
|
||
|
||
import pytest
|
||
from services import radar
|
||
|
||
BERLIN = radar.LOCAL_TZ
|
||
|
||
|
||
def _zeit(text: str) -> datetime:
|
||
return datetime.fromisoformat(text).replace(tzinfo=BERLIN)
|
||
|
||
|
||
def _kandidat(kid: str, rolle: str = "hirn", status: str = "neu", quelle: str = "watchlist", rang: int = 0,
|
||
**extra) -> dict:
|
||
return {"id": kid, "name": kid, "rolle": rolle, "repo": f"org/{kid}-GGUF", "quant": "Q4_K_M", "quelle": quelle,
|
||
"rang": rang, "datei": f"{kid}-Q4_K_M.gguf", "dateien": [[f"{kid}-Q4_K_M.gguf", 100]],
|
||
"mmproj": ["mmproj-F16.gguf", 10], "draft": None, "flags": [], "groesse_gb": 21.7, "passt": True,
|
||
"eng": False, "ctx": 131072, "status": status, "begruendung": "", **extra}
|
||
|
||
|
||
@pytest.fixture
|
||
def zustand(monkeypatch, tmp_path):
|
||
"""Radar-Zustand und Radar-Ordner in einem Temp-Ordner statt unter /srv/models."""
|
||
monkeypatch.setattr(radar, "STORE_PATH", tmp_path / "mc2-radar.json")
|
||
monkeypatch.setattr(radar, "RADAR_DIR", tmp_path / "radar")
|
||
(tmp_path / "radar").mkdir()
|
||
return tmp_path
|
||
|
||
|
||
# --- Zeitfenster und Wochengrenze ------------------------------------------------------------
|
||
|
||
def test_zeitfenster_nur_von_0030_bis_0230():
|
||
assert not radar.im_fenster(_zeit("2026-09-29T00:29:59"))
|
||
assert radar.im_fenster(_zeit("2026-09-29T00:30:00"))
|
||
assert radar.im_fenster(_zeit("2026-09-29T02:29:59"))
|
||
assert not radar.im_fenster(_zeit("2026-09-29T02:30:00"))
|
||
assert not radar.im_fenster(_zeit("2026-09-29T14:00:00"))
|
||
assert radar.fenster_ende(_zeit("2026-09-29T00:45:00")) == _zeit("2026-09-29T02:30:00")
|
||
|
||
|
||
def test_fensterende_am_umstellungstag_nie_nach_0300():
|
||
# 28.03.2027: 02:00 springt auf 03:00, 02:30 gibt es nicht. Lieber früher enden als in den NerdQuiz-Lauf.
|
||
ende = radar.fenster_ende(_zeit("2027-03-28T00:45:00"))
|
||
assert ende.timestamp() <= datetime(2027, 3, 28, 3, 0, tzinfo=BERLIN).timestamp()
|
||
|
||
|
||
def test_naechster_start_ist_die_naechste_nacht():
|
||
assert radar.naechster_start(_zeit("2026-09-29T00:10:00")) == _zeit("2026-09-29T00:30:00")
|
||
assert radar.naechster_start(_zeit("2026-09-29T01:00:00")) == _zeit("2026-09-30T00:30:00")
|
||
assert radar.naechster_start(_zeit("2026-09-29T14:00:00")) == _zeit("2026-09-30T00:30:00")
|
||
|
||
|
||
def test_wochengrenze_ein_neuer_kandidat_pro_woche():
|
||
stand = radar._leer()
|
||
stand["kandidaten"] = {"a": _kandidat("a", status="bestanden",
|
||
erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||
"b": _kandidat("b")}
|
||
assert radar.wochen_frei_ab(stand, date(2026, 10, 3)) == date(2026, 10, 6)
|
||
assert radar.wochen_frei_ab(stand, date(2026, 10, 6)) is None
|
||
|
||
|
||
def test_plan_testet_nur_im_fenster_mit_freier_woche_und_setzt_angefangene_fort():
|
||
stand = radar._leer()
|
||
stand["kandidaten"] = {"a": _kandidat("a", status="bestanden",
|
||
erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||
"b": _kandidat("b", rang=1)}
|
||
radar._ordne(stand)
|
||
assert stand["kandidaten"]["b"]["status"] == "wartet"
|
||
|
||
plan = radar.plane_test(stand, _zeit("2026-10-03T00:31:00"))
|
||
assert plan["kandidat"] is None and "06.10." in plan["grund"]
|
||
assert radar.plane_test(stand, _zeit("2026-10-06T00:31:00"))["kandidat"] == "b"
|
||
assert "außerhalb" in radar.plane_test(stand, _zeit("2026-10-06T14:00:00"))["grund"]
|
||
assert "Minuten" in radar.plane_test(stand, _zeit("2026-10-06T02:15:00"))["grund"]
|
||
|
||
# angefangen (z. B. Download unterbrochen): darf in der nächsten Nacht weiter, die Woche gilt ab dem 1. Start
|
||
stand["kandidaten"]["b"]["erster_start"] = _zeit("2026-10-06T00:31:00").timestamp()
|
||
assert radar.plane_test(stand, _zeit("2026-10-07T00:31:00"))["kandidat"] == "b"
|
||
|
||
|
||
def test_laufender_test_blockiert_zweiten_lauf_und_zeigt_sich(zustand):
|
||
stand = radar._leer()
|
||
stand["kandidaten"]["b"] = _kandidat("b", status="wartet", erster_start=time.time())
|
||
stand["lauf"] = {"pid": os.getpid(), "kandidat": "b", "schritt": "test", "seit": time.time(),
|
||
"frist": time.time() + 3600}
|
||
assert radar.plane_test(stand, _zeit("2026-10-06T00:40:00"))["grund"] == "es läuft schon ein Test"
|
||
radar._speichere(stand)
|
||
assert radar.uebersicht()["kandidaten"][0]["begruendung"].startswith("Läuft gerade seit")
|
||
stand["lauf"]["pid"] = 999_999_999 # Prozess tot → Lauf gilt als beendet
|
||
assert radar._aktiver_lauf(stand) is None
|
||
|
||
|
||
# --- Namen, Filter, Dateien, Speicher -----------------------------------------------------------
|
||
|
||
def test_namen_und_staemme():
|
||
assert radar.anzeigename("bartowski/Qwen_Qwen3.6-35B-A3B-GGUF") == "Qwen3.6-35B-A3B"
|
||
assert radar.anzeigename("unsloth/Qwen3.8-Flash-Next-GGUF") == "Qwen3.8-Flash-Next"
|
||
assert radar.stamm("Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf") == "qwen3.8-flash-next"
|
||
assert radar.stamm("Qwen3.6-35B-A3B") == radar.stamm("unsloth/Qwen3.6-35B-A3B-GGUF")
|
||
assert radar.generation("Qwen3.8-27B") == ("qwen", 3.8)
|
||
assert radar.kandidat_id("hirn", "ornith-ai/Ornith-1.5-35B-A3B-GGUF", "Q4_K_M") == \
|
||
"hirn-ornith-ai-ornith-1-5-35b-a3b-gguf-q4-k-m"
|
||
|
||
|
||
def test_entdeckung_filtert_alte_kleine_bekannte_und_dichte_hirne():
|
||
heute = {"hirn": "Qwen3.6-35B-A3B", "coder": "Qwen3.8-27B"}
|
||
bekannt = {"qwen3.6-35b-a3b", "qwen3.8-27b"}
|
||
|
||
def fund(repo, params, aktiv=None, kuratiert=False):
|
||
return {"repo": repo, "params_b": params, "caps": {"active_b": aktiv}, "curated": kuratiert}
|
||
|
||
def grund(m, rolle):
|
||
return radar.entdeckung_ungeeignet(m, rolle, bekannt, heute)
|
||
|
||
assert "Katalog" in grund(fund("unsloth/Qwen3-Coder-30B-A3B-Instruct-GGUF", 30, 3, kuratiert=True), "coder")
|
||
assert "auf der Box" in grund(fund("bartowski/Qwen_Qwen3.6-35B-A3B-GGUF", 35, 3), "hirn")
|
||
assert "zu klein" in grund(fund("lmstudio-community/Qwen2.5-Coder-14B-Instruct-GGUF", 14), "coder")
|
||
assert "dichte" in grund(fund("bartowski/Hermes-3-Llama-3.1-70B-GGUF", 70), "hirn")
|
||
assert "ältere Generation" in grund(fund("bartowski/Qwen_Qwen3.5-35B-A3B-GGUF", 35, 3), "hirn")
|
||
assert grund(fund("unsloth/Qwen-AgentWorld-35B-A3B-GGUF", 35, 3), "hirn") is None
|
||
|
||
|
||
BAUM_FLASH = [
|
||
{"type": "directory", "path": "UD-IQ3_XXS"},
|
||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf", "size": 10_000_000},
|
||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00002-of-00003.gguf", "size": 49_570_000_000},
|
||
{"path": "UD-IQ3_XXS/Qwen3.8-Flash-Next-UD-IQ3_XXS-00003-of-00003.gguf", "lfs": {"size": 32_380_000_000}},
|
||
{"path": "UD-IQ4_XS/Qwen3.8-Flash-Next-UD-IQ4_XS-00001-of-00003.gguf", "size": 10_000_000},
|
||
{"path": "MTP/mtp-Qwen3.8-Flash-Next-Q8_0.gguf", "size": 4_140_000_000},
|
||
{"path": "mmproj-BF16.gguf", "size": 910_000_000},
|
||
{"path": "mmproj-F16.gguf", "size": 900_000_000},
|
||
]
|
||
KOPF_FLASH = {"general.architecture": "qwen4exp", "qwen4exp.block_count": 48, "qwen4exp.attention.head_count": 24,
|
||
"qwen4exp.attention.head_count_kv": 2, "qwen4exp.attention.key_length": 256,
|
||
"qwen4exp.attention.value_length": 256, "qwen4exp.full_attention_interval": 4}
|
||
|
||
|
||
def test_dateiauswahl_nimmt_alle_teile_des_quants_und_das_f16_mmproj():
|
||
wahl = radar.waehle_dateien(BAUM_FLASH, "UD-IQ3_XXS")
|
||
assert [p.rsplit("-", 3)[-3] for p, _ in wahl["dateien"]] == ["00001", "00002", "00003"]
|
||
assert all(p.startswith("UD-IQ3_XXS/") for p, _ in wahl["dateien"])
|
||
assert wahl["mmproj"] == ["mmproj-F16.gguf", 900_000_000]
|
||
assert radar.waehle_dateien(BAUM_FLASH, "Q4_K_M") is None
|
||
assert radar.waehle_dateien(BAUM_FLASH, "Q8_0") is None # der MTP-Kopf ist kein Modell
|
||
ornith = [{"path": "Ornith-1.5-35B-Q4_K_M.gguf", "size": 21_710_000_000},
|
||
{"path": "Ornith-1.5-35B-Q8_0.gguf", "size": 37_800_000_000},
|
||
{"path": "mmproj-Ornith-1.5-35B-BF16.gguf", "size": 900_000_000}]
|
||
wahl = radar.waehle_dateien(ornith, "Q4_K_M")
|
||
assert wahl["dateien"] == [["Ornith-1.5-35B-Q4_K_M.gguf", 21_710_000_000]]
|
||
assert wahl["mmproj"][0] == "mmproj-Ornith-1.5-35B-BF16.gguf"
|
||
|
||
|
||
def _gguf_kopf_bytes(werte: dict) -> bytes:
|
||
"""Minimaler GGUF-Kopf wie von llama.cpp: Architektur-Schlüssel, danach der Tokenizer (abgeschnitten)."""
|
||
def s(text: str) -> bytes:
|
||
return struct.pack("<Q", len(text.encode())) + text.encode()
|
||
|
||
teile = [b"GGUF", struct.pack("<I", 3), struct.pack("<Q", 0), struct.pack("<Q", len(werte) + 1)]
|
||
for k, v in werte.items():
|
||
if isinstance(v, str):
|
||
teile += [s(k), struct.pack("<I", 8), s(v)]
|
||
elif isinstance(v, list):
|
||
teile += [s(k), struct.pack("<I", 9), struct.pack("<I", 5), struct.pack("<Q", len(v)),
|
||
struct.pack(f"<{len(v)}i", *v)]
|
||
else:
|
||
teile += [s(k), struct.pack("<I", 4), struct.pack("<I", v)]
|
||
teile += [s("tokenizer.ggml.tokens"), struct.pack("<I", 9)]
|
||
return b"".join(teile)
|
||
|
||
|
||
def test_kv_cache_aus_dem_gguf_kopf_kennt_hybrid_modelle():
|
||
daten = _gguf_kopf_bytes({"general.architecture": "qwen35moe", "qwen35moe.block_count": 41,
|
||
"qwen35moe.attention.head_count": 16, "qwen35moe.attention.head_count_kv": 2,
|
||
"qwen35moe.rope.dimension_sections": [11, 11, 10, 0],
|
||
"qwen35moe.attention.key_length": 256, "qwen35moe.attention.value_length": 256,
|
||
"qwen35moe.full_attention_interval": 4})
|
||
kopf = radar.gguf_kopf_aus_bytes(daten)
|
||
assert kopf["general.architecture"] == "qwen35moe" and kopf["qwen35moe.block_count"] == 41
|
||
# 11 von 41 Schichten mit KV-Cache × 2 KV-Köpfe × (256 + 256) × 1,0625 Byte (q8_0)
|
||
assert radar.kv_je_token_gb(kopf) == pytest.approx(11 * 2 * 512 * 1.0625 / 1e9)
|
||
assert radar.kv_je_token_gb(kopf) * 131072 == pytest.approx(1.57, abs=0.01)
|
||
ohne_hybrid = {k: v for k, v in kopf.items() if not k.endswith("full_attention_interval")}
|
||
assert radar.kv_je_token_gb(ohne_hybrid) == pytest.approx(41 * 2 * 512 * 1.0625 / 1e9)
|
||
assert radar.gguf_kopf_aus_bytes(b"kein gguf") is None
|
||
assert radar.gguf_kopf_aus_bytes(daten[:20]) is None
|
||
|
||
|
||
def test_speicher_passung_nach_der_115_gb_regel():
|
||
budget = radar.SPEICHER_GRENZE_GB - radar.WARMSET_GB
|
||
p = radar.speicher_passung("hirn", 22.6, 11968 / 1e9) # Ornith: bequem, voller Kontext
|
||
assert p["passt"] and not p["eng"] and p["ctx"] == 131072
|
||
p = radar.speicher_passung("coder", 82.9, 13056 / 1e9, eng_markiert=True) # Flash-Next: knapp
|
||
assert p["passt"] and p["eng"] and p["bedarf_gb"] <= budget
|
||
p = radar.speicher_passung("coder", 82.9, 52224 / 1e9) # großer KV-Cache → halber Kontext
|
||
assert p["passt"] and p["eng"] and p["ctx"] == 65536
|
||
p = radar.speicher_passung("coder", 94.0, 13056 / 1e9) # UD-IQ4_XS passt nicht
|
||
assert not p["passt"] and p["text"].startswith("Passt nicht")
|
||
p = radar.speicher_passung("hirn", 85.0, 1e-6) # Test ja, im Betrieb neben dem Coder knapp
|
||
assert p["passt"] and p["eng"] and "Im Betrieb" in p["text"]
|
||
assert radar.platte_nach_download(265, 1900, 82) == pytest.approx(18.26, abs=0.01)
|
||
|
||
|
||
def test_bewertung_ohne_netz(monkeypatch):
|
||
monkeypatch.setattr(radar, "_hf_baum", lambda repo: BAUM_FLASH)
|
||
monkeypatch.setattr(radar, "_gguf_kopf", lambda repo, datei: KOPF_FLASH)
|
||
monkeypatch.setattr(radar, "engine_kennt", lambda arch: True)
|
||
fund = {"name": "Qwen3.8-Flash-Next", "rolle": "coder", "repo": "unsloth/Qwen3.8-Flash-Next-GGUF",
|
||
"quant": "UD-IQ3_XXS", "eng": True, "quelle": "watchlist", "rang": 1}
|
||
k = radar._bewerte(fund)
|
||
assert k["tauglich"] and k["passt"] and k["eng"] and k["ctx"] == 131072
|
||
assert k["datei"] == "Qwen3.8-Flash-Next-UD-IQ3_XXS-00001-of-00003.gguf"
|
||
assert k["groesse_gb"] == pytest.approx(82.9, abs=0.05)
|
||
assert k["begruendung"].startswith("Aus der Merkliste. Passt knapp")
|
||
|
||
monkeypatch.setattr(radar, "engine_kennt", lambda arch: False)
|
||
k = radar._bewerte(fund)
|
||
assert not k["tauglich"] and "qwen4exp" in k["begruendung"]
|
||
|
||
# Kopf nicht lesbar: die grobe Schätzung sagt „passt nicht“ — das darf nicht zum Verwerfen führen
|
||
monkeypatch.setattr(radar, "_gguf_kopf", lambda repo, datei: None)
|
||
k = radar._bewerte(fund)
|
||
assert k["netzfehler"] and not k["tauglich"] and k["begruendung"].startswith("Speicherbedarf unklar")
|
||
|
||
|
||
def test_merkliste_im_repo():
|
||
funde, ok = radar._merkliste()
|
||
assert ok
|
||
nach_name = {f["name"]: f for f in funde}
|
||
assert nach_name["Ornith-1.5-35B-A3B"]["rolle"] == "hirn"
|
||
assert nach_name["Qwen3.8-Flash-Next"]["quant"] == "UD-IQ3_XXS" and nach_name["Qwen3.8-Flash-Next"]["eng"] is True
|
||
k2 = radar._bewerte(nach_name["K2-Horizon-MoVA-36B-A4B"]) # übersprungen, ganz ohne Netz
|
||
assert not k2["tauglich"] and k2["begruendung"].startswith("Übersprungen") and "llama.cpp" in k2["begruendung"]
|
||
|
||
|
||
# --- Zustandsübergänge ---------------------------------------------------------------------
|
||
|
||
def test_suche_uebernimmt_merkliste_und_laesst_geschichte_in_ruhe():
|
||
stand = radar._leer()
|
||
stand["kandidaten"] = {"alt": _kandidat("alt", status="durchgefallen", quelle="discover"),
|
||
"weg": _kandidat("weg", quelle="discover"),
|
||
"nein": _kandidat("nein", status="verworfen", verworfen_von="nutzer")}
|
||
bewertet = [
|
||
{**_kandidat("ornith", rang=0), "tauglich": True, "begruendung": "Aus der Merkliste. Passt."},
|
||
{**_kandidat("k2", rang=2), "tauglich": False, "passt": False,
|
||
"begruendung": "Übersprungen: läuft noch nicht im offiziellen llama.cpp."},
|
||
{**_kandidat("alt", quelle="discover"), "tauglich": True},
|
||
{**_kandidat("nein"), "tauglich": True},
|
||
{**_kandidat("klein", quelle="discover"), "tauglich": False, "begruendung": "Kein GGUF mit Q4_K_M im Repo."},
|
||
]
|
||
ergebnis = radar._suche_anwenden(stand, bewertet, {"watchlist": True, "discover": True}, 1000.0)
|
||
k = stand["kandidaten"]
|
||
assert k["ornith"]["status"] == "wartet" # Kopf der Warteschlange
|
||
assert k["k2"]["status"] == "verworfen" and k["k2"]["verworfen_von"] == "radar"
|
||
assert k["alt"]["status"] == "durchgefallen" # Geschichte bleibt
|
||
assert k["nein"]["status"] == "verworfen" and k["nein"]["verworfen_von"] == "nutzer"
|
||
assert "weg" not in k and "klein" not in k
|
||
assert "klein" in stand["abgelehnt"]
|
||
assert ergebnis["neu"] == ["ornith"] and stand["letzte_suche"] == 1000.0
|
||
|
||
stand["kandidaten"]["weg"] = _kandidat("weg", quelle="discover")
|
||
radar._suche_anwenden(stand, [], {"watchlist": True, "discover": False}, 2000.0) # Entdeckung offline
|
||
assert "weg" in stand["kandidaten"]
|
||
|
||
|
||
def test_bestanden_bleibt_liegen_durchgefallen_wird_geloescht(zustand):
|
||
stand = radar._leer()
|
||
for kid in ("gut", "schlecht"):
|
||
ordner = radar.RADAR_DIR / kid
|
||
ordner.mkdir()
|
||
(ordner / "m.gguf").write_bytes(b"x")
|
||
stand["kandidaten"][kid] = _kandidat(kid, status="wartet", ordner=str(ordner), erster_start=1.0)
|
||
test = radar._abschluss(stand, "gut", "bestanden", "Bestanden: versteht Bilder (heute nicht).",
|
||
werte={"tiefe": {"tg_tps": 95.0}}, vergleich={"Bilder": {"urteil": "besser"}})
|
||
radar._abschluss(stand, "schlecht", "durchgefallen", "Durchgefallen: Werkzeuge 3/6 (heute 6/6).")
|
||
assert stand["kandidaten"]["gut"]["status"] == "bestanden" and (radar.RADAR_DIR / "gut").exists()
|
||
assert "wartet auf deine Entscheidung" in stand["kandidaten"]["gut"]["begruendung"]
|
||
assert stand["kandidaten"]["schlecht"]["status"] == "durchgefallen" and not (radar.RADAR_DIR / "schlecht").exists()
|
||
assert [t["ergebnis"] for t in stand["tests"]] == ["bestanden", "durchgefallen"]
|
||
assert set(test) == set(radar.TEST_FELDER)
|
||
|
||
|
||
def test_abbrueche_zaehlen_nur_mit_schuld_des_kandidaten(zustand):
|
||
ordner = radar.RADAR_DIR / "zaeh"
|
||
ordner.mkdir()
|
||
stand = radar._leer()
|
||
stand["kandidaten"]["zaeh"] = _kandidat("zaeh", status="wartet", ordner=str(ordner), erster_start=1.0)
|
||
radar._abschluss(stand, "zaeh", "abgebrochen", "Port 5899 ist belegt.", zaehlt=False)
|
||
assert stand["kandidaten"]["zaeh"]["status"] == "wartet" and not stand["kandidaten"]["zaeh"].get("versuche")
|
||
for _ in range(radar.VERSUCHE_MAX):
|
||
radar._abschluss(stand, "zaeh", "abgebrochen", "Das Nachtfenster endete um 02:30, bevor der Test fertig war.")
|
||
k = stand["kandidaten"]["zaeh"]
|
||
assert k["status"] == "durchgefallen" and k["begruendung"].startswith(f"{radar.VERSUCHE_MAX}-mal abgebrochen")
|
||
assert not ordner.exists()
|
||
assert {t["ergebnis"] for t in stand["tests"]} == {"abgebrochen"}
|
||
|
||
|
||
def test_ohne_vergleichswerte_bleibt_der_kandidat_getestet(zustand):
|
||
stand = radar._leer()
|
||
stand["kandidaten"]["g"] = _kandidat("g", status="wartet", erster_start=1.0)
|
||
assert radar._abschluss(stand, "g", None, "", werte={"tiefe": {"tg_tps": 80.0}}) is None
|
||
assert stand["kandidaten"]["g"]["status"] == "getestet" and stand["kandidaten"]["g"]["messung"]
|
||
assert stand["tests"] == []
|
||
|
||
|
||
def test_verwerfen_und_uebernehmen_pruefen_den_status(zustand):
|
||
ordner = radar.RADAR_DIR / "b"
|
||
ordner.mkdir()
|
||
stand = radar._leer()
|
||
stand["kandidaten"] = {"b": _kandidat("b", status="bestanden", ordner=str(ordner)),
|
||
"u": _kandidat("u", status="uebernommen"), "n": _kandidat("n")}
|
||
radar._speichere(stand)
|
||
assert radar.uebernehmen("n")["status"] == 409 # nur bestandene
|
||
assert radar.uebernehmen("gibt-es-nicht")["status"] == 404
|
||
assert radar.verwerfen("u")["status"] == 409 # schon übernommen
|
||
antwort = radar.verwerfen("b")
|
||
assert antwort["ok"] and antwort["kandidat"]["status"] == "verworfen"
|
||
assert not ordner.exists()
|
||
assert radar._lade()["kandidaten"]["b"]["verworfen_von"] == "nutzer"
|
||
|
||
|
||
def test_uebersicht_hat_die_felder_der_oberflaeche(zustand):
|
||
stand = radar._leer()
|
||
stand["kandidaten"] = {"a": _kandidat("a", status="wartet", erster_start=_zeit("2026-09-29T00:31:00").timestamp()),
|
||
"b": _kandidat("b", rang=1)}
|
||
radar._abschluss(stand, "a", "bestanden", "Bestanden: Prefill bei 13k 18 % schneller.",
|
||
zeit=_zeit("2026-09-29T01:40:00").timestamp())
|
||
radar._speichere(stand)
|
||
u = radar.uebersicht(_zeit("2026-10-01T12:00:00"))
|
||
assert set(u) == {"naechster_test", "letzter_test", "kandidaten", "tests"}
|
||
assert set(u["kandidaten"][0]) == set(radar.KANDIDAT_FELDER)
|
||
assert [k["status"] for k in u["kandidaten"]] == ["wartet", "bestanden"]
|
||
assert u["naechster_test"] == "2026-10-06T00:30:00+02:00" # die Woche ab dem 29.09. ist belegt
|
||
assert u["letzter_test"].startswith("2026-09-29T01:40")
|
||
assert set(u["tests"][0]) == set(radar.TEST_FELDER)
|
||
|
||
|
||
def _uebernahme_vorbereiten(zustand, monkeypatch, rolle: str, alt: dict) -> tuple[list, dict]:
|
||
"""Bestandenen Kandidaten „o“ anlegen und llama-swap/Hirn-Wechsel durch Protokoll-Attrappen ersetzen."""
|
||
modelle = zustand / "models"
|
||
modelle.mkdir()
|
||
monkeypatch.setattr(radar, "MODELS_DIR", modelle)
|
||
ordner = radar.RADAR_DIR / "o"
|
||
ordner.mkdir()
|
||
(ordner / "o-Q4_K_M.gguf").write_bytes(b"x")
|
||
(ordner / "mmproj-F16.gguf").write_bytes(b"x")
|
||
stand = radar._leer()
|
||
stand["kandidaten"]["o"] = _kandidat("o", rolle=rolle, status="bestanden", ordner=str(ordner))
|
||
radar._speichere(stand)
|
||
aufrufe, cfg = [], {"models": {}}
|
||
|
||
def eintragen(pfad, **kw):
|
||
aufrufe.append(("eintragen", pfad, kw["role"], kw["set_alias"]))
|
||
cfg["models"]["o"] = {"cmd": "vorlage"}
|
||
return "o"
|
||
|
||
from services import agent
|
||
monkeypatch.setattr(radar, "_heutige_modelle", lambda: {rolle: alt})
|
||
monkeypatch.setattr(radar.llamaswap, "register_model", eintragen)
|
||
monkeypatch.setattr(radar.llamaswap, "read_config", lambda: cfg)
|
||
monkeypatch.setattr(radar.llamaswap, "write_config", lambda c: aufrufe.append(("schreiben",)))
|
||
monkeypatch.setattr(radar.llamaswap, "set_role", lambda mid, r: aufrufe.append(("rolle", mid, r)) or True)
|
||
monkeypatch.setattr(radar.llamaswap, "add_role", lambda mid, r: aufrufe.append(("dazu", mid, r)) or True)
|
||
monkeypatch.setattr(radar, "_steward_neu_starten", lambda: aufrufe.append(("steward",)) or None)
|
||
monkeypatch.setattr(radar.llamaswap, "set_ttl", lambda mid, ttl: aufrufe.append(("ttl", mid, ttl)) or True)
|
||
monkeypatch.setattr(agent, "set_agent_brain", lambda mid: aufrufe.append(("hirn", mid)) or {"ok": True})
|
||
monkeypatch.setattr(radar, "STEWARD_WARMSET", zustand / "warmset.conf")
|
||
return aufrufe, cfg
|
||
|
||
|
||
def test_uebernehmen_hirn_nutzt_den_hirn_wechsel_und_nimmt_fast_mit(zustand, monkeypatch):
|
||
aufrufe, cfg = _uebernahme_vorbereiten(zustand, monkeypatch, "hirn",
|
||
{"name": "Qwen3.6-35B-A3B", "aliases": ["hermes", "fast"], "ttl": 0})
|
||
(zustand / "warmset.conf").write_text('[Service]\nEnvironment="MC_WARMSET=Qwen3-Embedding-0.6B Qwen3.6-35B-A3B"\n',
|
||
encoding="utf-8")
|
||
antwort = radar.uebernehmen("o")
|
||
assert antwort["ok"] and antwort["modell_id"] == "o"
|
||
assert [a[0] for a in aufrufe] == ["eintragen", "schreiben", "hirn", "dazu", "steward"]
|
||
assert aufrufe[0][2:] == ("hermes", False) and aufrufe[3] == ("dazu", "o", "fast")
|
||
befehl = str(cfg["models"]["o"]["cmd"])
|
||
assert "--parallel 2" in befehl and "-ctk q8_0" in befehl and "--mmproj" in befehl
|
||
assert (zustand / "models" / "o-GGUF" / "o-Q4_K_M.gguf").exists() and not (radar.RADAR_DIR / "o").exists()
|
||
k = radar._lade()["kandidaten"]["o"]
|
||
assert k["status"] == "uebernommen" and "vorher Qwen3.6-35B-A3B" in k["begruendung"]
|
||
# Der Steward hält danach das neue Hirn warm, nicht mehr das alte — ohne Handgriff.
|
||
assert 'MC_WARMSET=Qwen3-Embedding-0.6B o"' in (zustand / "warmset.conf").read_text(encoding="utf-8")
|
||
assert antwort["text"] is None and "Noch zu tun" not in k["begruendung"]
|
||
|
||
|
||
def test_uebernehmen_coder_nimmt_heavy_mit(zustand, monkeypatch):
|
||
aufrufe, cfg = _uebernahme_vorbereiten(zustand, monkeypatch, "coder",
|
||
{"name": "Qwen3.8-27B", "aliases": ["coder", "heavy"], "ttl": 5400})
|
||
antwort = radar.uebernehmen("o")
|
||
assert antwort["ok"] and antwort["text"] is None
|
||
assert aufrufe[2:] == [("rolle", "o", "coder"), ("dazu", "o", "heavy"), ("ttl", "o", 5400)]
|
||
assert "--parallel 1" in str(cfg["models"]["o"]["cmd"])
|
||
|
||
|
||
def test_betriebsbefehl_uebernimmt_die_getesteten_flags():
|
||
k = _kandidat("f", rolle="coder", ctx=65536, draft={"typ": "draft-mtp", "n_max": 2, "datei": None}, flags=["-ub", "512"])
|
||
befehl = radar.betriebs_befehl("coder", {"gguf": "/srv/models/F-GGUF/f.gguf", "mmproj": "/srv/models/F-GGUF/mm.gguf",
|
||
"draft": None}, k)
|
||
assert "-c 65536" in befehl and "--parallel 1" in befehl and "-ctk q8_0 -ctv q8_0" in befehl
|
||
assert "--mmproj /srv/models/F-GGUF/mm.gguf" in befehl and "--load-mode none" in befehl
|
||
assert befehl.endswith("--spec-type draft-mtp --spec-draft-n-max 2 -ub 512")
|
||
assert "--parallel 2" in radar.betriebs_befehl("hirn", {"gguf": "x.gguf", "mmproj": None, "draft": None},
|
||
_kandidat("h"))
|
||
|
||
|
||
# --- Prüfstand: Urteil, Programmieraufgaben, Bild ----------------------------------------------------
|
||
|
||
@pytest.fixture(scope="module")
|
||
def ps():
|
||
return radar._pruefstand()
|
||
|
||
|
||
BILD_GUT = {"direkt": 1.0, "gefunden": ["Wort ZEBRA", "Zahl 4711"], "agentisch": True}
|
||
|
||
|
||
def _werte(rolle: str, decode: float = 100.0, prefill: float = 13.0, werkzeuge: int = 6, code: int = 8,
|
||
bild: dict | None = None, deutsch: bool = True) -> dict:
|
||
werte = {"rolle": rolle, "tiefe": {"tg_tps": decode, "pp_s": prefill}, "werkzeuge": {"ok": werkzeuge, "von": 6},
|
||
"bild": bild}
|
||
if rolle == "hirn":
|
||
werte.update(deutsch={"ok": deutsch}, json={"ok": True})
|
||
else:
|
||
werte["code"] = {"ok": code, "von": 10}
|
||
return werte
|
||
|
||
|
||
def test_urteil_hirn_mit_bildern_besteht_ohne_verschlechterung(ps):
|
||
u = ps.urteil("hirn", _werte("hirn", decode=95.0, bild=BILD_GUT), _werte("hirn", decode=100.8))
|
||
assert u["ergebnis"] == "bestanden"
|
||
assert u["zusammenfassung"] == "Bestanden: versteht Bilder (heute nicht)."
|
||
assert u["vergleich"]["Decode bei 13k"]["urteil"] == "gleich" # −6 % liegt in der Toleranz
|
||
assert list(u["vergleich"]) == ["Decode bei 13k", "Prefill bei 13k", "Werkzeug-Aufrufe", "Deutsch", "JSON",
|
||
"Bilder"]
|
||
|
||
|
||
def test_urteil_hirn_faellt_bei_schwaeche_oder_ohne_vorteil_durch(ps):
|
||
basis = _werte("hirn", decode=100.8)
|
||
u = ps.urteil("hirn", _werte("hirn", werkzeuge=5, bild=BILD_GUT), basis)
|
||
assert u["ergebnis"] == "durchgefallen" and "Werkzeuge 5/6 (heute 6/6)" in u["zusammenfassung"]
|
||
u = ps.urteil("hirn", _werte("hirn", decode=70.0, bild=BILD_GUT), basis)
|
||
assert u["ergebnis"] == "durchgefallen" and "Decode bei 13k 70 statt 100,8 t/s" in u["zusammenfassung"]
|
||
u = ps.urteil("hirn", _werte("hirn", bild=BILD_GUT), _werte("hirn", bild=BILD_GUT))
|
||
assert u["zusammenfassung"] == "Durchgefallen: kein Vorteil gegenüber dem heutigen Modell."
|
||
u = ps.urteil("hirn", _werte("hirn", decode=130.0), basis) # schneller, aber blind
|
||
assert u["ergebnis"] == "durchgefallen" and "versteht keine Bilder" in u["zusammenfassung"]
|
||
|
||
|
||
def test_urteil_coder_zaehlt_programmieraufgaben_und_bilder(ps):
|
||
basis = _werte("coder", decode=30.4, prefill=41.2, code=8, bild=BILD_GUT)
|
||
u = ps.urteil("coder", _werte("coder", decode=45.0, prefill=12.0, code=9, bild=BILD_GUT), basis)
|
||
assert u["ergebnis"] == "bestanden" and "Programmieraufgaben 9/10 statt 8/10" in u["zusammenfassung"]
|
||
u = ps.urteil("coder", _werte("coder", decode=45.0, prefill=12.0, code=7, bild=BILD_GUT), basis)
|
||
assert u["ergebnis"] == "durchgefallen" and "Programmieraufgaben 7/10 (heute 8/10)" in u["zusammenfassung"]
|
||
u = ps.urteil("coder", _werte("coder", code=9), basis) # der heutige Coder kann Bilder
|
||
assert u["ergebnis"] == "durchgefallen" and "versteht keine Bilder" in u["zusammenfassung"]
|
||
|
||
|
||
REFERENZ = {
|
||
"palindrom": "def ist_palindrom(text):\n z = [c.lower() for c in text if c.isalnum()]\n return z == z[::-1]\n",
|
||
"roemisch": (
|
||
"def roemisch(zahl):\n"
|
||
" if not isinstance(zahl, int) or not 1 <= zahl <= 3999:\n"
|
||
" raise ValueError(zahl)\n"
|
||
" out = ''\n"
|
||
" for wert, zeichen in ((1000, 'M'), (900, 'CM'), (500, 'D'), (400, 'CD'), (100, 'C'), (90, 'XC'),\n"
|
||
" (50, 'L'), (40, 'XL'), (10, 'X'), (9, 'IX'), (5, 'V'), (4, 'IV'), (1, 'I')):\n"
|
||
" while zahl >= wert:\n"
|
||
" out += zeichen\n"
|
||
" zahl -= wert\n"
|
||
" return out\n"),
|
||
"intervalle": (
|
||
"def zusammenfassen(intervalle):\n"
|
||
" out = []\n"
|
||
" for a, b in sorted(intervalle):\n"
|
||
" if out and a <= out[-1][1]:\n"
|
||
" out[-1][1] = max(out[-1][1], b)\n"
|
||
" else:\n"
|
||
" out.append([a, b])\n"
|
||
" return out\n"),
|
||
"flach": (
|
||
"def flach(liste):\n"
|
||
" out = []\n"
|
||
" for x in liste:\n"
|
||
" out.extend(flach(x) if isinstance(x, list) else [x])\n"
|
||
" return out\n"),
|
||
"iban": (
|
||
"def iban_gueltig(iban):\n"
|
||
" s = iban.replace(' ', '').upper()\n"
|
||
" if not 15 <= len(s) <= 34 or not s.isalnum() or not s.isascii():\n"
|
||
" return False\n"
|
||
" return int(''.join(str(int(c, 36)) for c in s[4:] + s[:4])) % 97 == 1\n"),
|
||
"lru": (
|
||
"from collections import OrderedDict\n"
|
||
"class LRUCache:\n"
|
||
" def __init__(self, kapazitaet):\n"
|
||
" self.k, self.d = kapazitaet, OrderedDict()\n"
|
||
" def get(self, schluessel):\n"
|
||
" if schluessel not in self.d:\n"
|
||
" return None\n"
|
||
" self.d.move_to_end(schluessel)\n"
|
||
" return self.d[schluessel]\n"
|
||
" def put(self, schluessel, wert):\n"
|
||
" self.d[schluessel] = wert\n"
|
||
" self.d.move_to_end(schluessel)\n"
|
||
" if len(self.d) > self.k:\n"
|
||
" self.d.popitem(last=False)\n"),
|
||
"klammern": (
|
||
"def klammern_ok(text):\n"
|
||
" paare, stapel = {')': '(', ']': '[', '}': '{'}, []\n"
|
||
" for c in text:\n"
|
||
" if c in '([{':\n"
|
||
" stapel.append(c)\n"
|
||
" elif c in paare and (not stapel or stapel.pop() != paare[c]):\n"
|
||
" return False\n"
|
||
" return not stapel\n"),
|
||
"wege": (
|
||
"import heapq\n"
|
||
"def kuerzeste_wege(graph, start):\n"
|
||
" dist, heap = {start: 0}, [(0, start)]\n"
|
||
" while heap:\n"
|
||
" d, u = heapq.heappop(heap)\n"
|
||
" if d > dist[u]:\n"
|
||
" continue\n"
|
||
" for v, w in graph.get(u, {}).items():\n"
|
||
" if v not in dist or d + w < dist[v]:\n"
|
||
" dist[v] = d + w\n"
|
||
" heapq.heappush(heap, (d + w, v))\n"
|
||
" return dist\n"),
|
||
"median": (
|
||
"def median(zahlen):\n"
|
||
" if not zahlen:\n"
|
||
" raise ValueError('leer')\n"
|
||
" s, n = sorted(zahlen), len(zahlen)\n"
|
||
" return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2\n"),
|
||
"csv": (
|
||
"def spaltensumme(csv_text, spalte):\n"
|
||
" zeilen = [z for z in csv_text.splitlines() if z.strip()]\n"
|
||
" kopf = zeilen[0].split(';')\n"
|
||
" if spalte not in kopf:\n"
|
||
" raise KeyError(spalte)\n"
|
||
" i, summe = kopf.index(spalte), 0.0\n"
|
||
" for z in zeilen[1:]:\n"
|
||
" zellen = z.split(';')\n"
|
||
" wert = zellen[i].strip() if i < len(zellen) else ''\n"
|
||
" if wert:\n"
|
||
" summe += float(wert.replace('.', '').replace(',', '.'))\n"
|
||
" return summe\n"),
|
||
}
|
||
|
||
|
||
def test_referenzloesungen_bestehen_alle_programmieraufgaben(ps):
|
||
assert {a["id"] for a in ps.AUFGABEN_CODE} == set(REFERENZ)
|
||
for aufgabe in ps.AUFGABEN_CODE:
|
||
ok, grund = ps.fuehre_aufgabe_aus(REFERENZ[aufgabe["id"]], aufgabe["tests"])
|
||
assert ok, f"{aufgabe['id']}: {grund}"
|
||
|
||
|
||
def test_falsche_verbotene_und_endlose_loesungen_fallen_durch(ps):
|
||
tests = next(a["tests"] for a in ps.AUFGABEN_CODE if a["id"] == "median")
|
||
ok, _ = ps.fuehre_aufgabe_aus("def median(z):\n z.sort()\n return z[len(z) // 2]\n", tests)
|
||
assert not ok
|
||
ok, grund = ps.fuehre_aufgabe_aus("import os\ndef median(z):\n return os.getpid()\n", tests)
|
||
assert not ok and grund.startswith("unerlaubter Aufruf")
|
||
ok, grund = ps.fuehre_aufgabe_aus("def median(z):\n while True:\n pass\n", tests, zeitlimit=2)
|
||
assert not ok and "Zeitlimit" in grund
|
||
|
||
|
||
def test_code_aus_antwort_nimmt_den_loesungsblock(ps):
|
||
antwort = "<think>kurz nachdenken</think>Hier:\n```python\ndef f():\n return 1\n```\nAufruf:\n```python\nf()\n```"
|
||
assert ps.code_aus_antwort(antwort) == "def f():\n return 1\n"
|
||
assert ps.code_aus_antwort("def g(): return 2") == "def g(): return 2"
|
||
|
||
|
||
def test_bild_ist_ein_gueltiges_png_mit_rotem_kreis(ps):
|
||
png = ps.bild_png()
|
||
assert png.startswith(b"\x89PNG\r\n\x1a\n")
|
||
breite, hoehe = struct.unpack(">II", png[16:24])
|
||
assert (breite, hoehe) == (448, 300)
|
||
laenge = struct.unpack(">I", png[33:37])[0]
|
||
roh = zlib.decompress(png[41:41 + laenge])
|
||
zeile = 1 + breite * 3
|
||
assert roh[95 * zeile + 1 + 80 * 3:95 * zeile + 1 + 81 * 3] == b"\xdc\x14\x14" # Mitte des Kreises
|
||
beschreibung = "Ein roter Kreis, ein blaues Quadrat, ein grünes Dreieck und darunter der Text ZEBRA 4711."
|
||
assert ps.bild_merkmale(beschreibung) == list(ps.BILD_MERKMALE)
|
||
assert ps.bild_data_url().startswith("data:image/png;base64,")
|
||
|
||
|
||
def test_werkzeug_treffer_liest_json_argumente(ps):
|
||
aufruf = [{"function": {"name": "send_telegram", "arguments": json.dumps({"text": "Prüfstand läuft."})}}]
|
||
assert ps.trifft(aufruf, [("send_telegram", r"pr(ü|ue)fstand")])
|
||
assert not ps.trifft(aufruf, [("terminal", r".*")])
|
||
assert len(ps.WERKZEUGE_HIRN) >= 90 # Lucy hat ~100 Werkzeuge
|
||
assert ps.tiefen_prompt("ab12").startswith("Messlauf ab12.")
|
||
|
||
|
||
def test_add_role_behaelt_die_uebrigen_aliase(monkeypatch):
|
||
"""add_role ohne Attrappe: coder bleibt, heavy wandert vom alten Coder herüber."""
|
||
from services import llamaswap
|
||
cfg = {"models": {"alt": {"cmd": "x", "aliases": ["coder", "heavy"]}, "neu": {"cmd": "y", "aliases": ["coder"]}}}
|
||
monkeypatch.setattr(llamaswap, "read_config", lambda: cfg)
|
||
monkeypatch.setattr(llamaswap, "write_config", lambda c: None)
|
||
assert llamaswap.add_role("neu", "heavy") is True
|
||
assert cfg["models"]["neu"]["aliases"] == ["coder", "heavy"]
|
||
assert cfg["models"]["alt"]["aliases"] == ["coder"]
|
||
assert llamaswap.add_role("gibt-es-nicht", "heavy") is False
|
||
|
||
|
||
def test_baseline_misst_neu_wenn_sich_die_proben_geaendert_haben(ps, tmp_path, monkeypatch):
|
||
"""Eine Baseline aus einem älteren Prüfstand ist nicht vergleichbar (23.09.: Ablenker geändert)."""
|
||
cache = tmp_path / "baseline.json"
|
||
cache.write_text(json.dumps({"hirn": {"kennung": "k", "version": ps.PRUEFSTAND_VERSION - 1,
|
||
"zeit": time.time(), "werte": {"alt": True}}}), encoding="utf-8")
|
||
gemessen = []
|
||
monkeypatch.setattr(ps, "pruefe", lambda *a, **kw: gemessen.append(1) or {"neu": True})
|
||
assert ps.baseline("hirn", str(cache), kennung="k", protokoll=lambda *_: None)["werte"] == {"neu": True}
|
||
assert gemessen == [1]
|
||
# Gleiche Version und Kennung: aus dem Cache, keine neue Messung.
|
||
assert ps.baseline("hirn", str(cache), kennung="k", protokoll=lambda *_: None)["werte"] == {"neu": True}
|
||
assert gemessen == [1]
|
||
|
||
|
||
def test_kein_ablenker_klingt_nach_plattenplatz(ps):
|
||
namen = {w["function"]["name"] for w in ps.WERKZEUGE_HIRN}
|
||
assert "system_speicher" not in namen and "system_arbeitsspeicher" in namen
|
||
|
||
|
||
def test_baseline_misst_bilder_ueber_den_zwilling(monkeypatch, ps):
|
||
"""Seit 24.09. sieht das heutige Hirn über seinen Bild-Zwilling (vision) — die Baseline muss das wissen."""
|
||
hirn = {"name": "Qwen3.6-35B-A3B", "aliases": ["hermes", "fast"], "gguf_path": "/m/hirn.gguf",
|
||
"cmd": "llama-server -m /m/hirn.gguf --spec-type draft-dflash"}
|
||
zwilling = {"name": "Qwen3.6-35B-A3B-Bild", "aliases": ["vision"], "gguf_path": "/m/hirn.gguf",
|
||
"cmd": "llama-server -m /m/hirn.gguf --mmproj /m/mmproj.gguf"}
|
||
fremd = {"name": "Qwen3-VL", "aliases": [], "gguf_path": "/m/vl.gguf", "cmd": "llama-server -m /m/vl.gguf --mmproj x"}
|
||
monkeypatch.setattr(radar.llamaswap, "list_models", lambda: [hirn, zwilling, fremd])
|
||
aufrufe = []
|
||
monkeypatch.setattr(ps, "baseline", lambda rolle, pfad, **kw: aufrufe.append(kw) or {"werte": {}})
|
||
radar._baseline("hirn", time.time() + 60)
|
||
assert aufrufe[0]["bild"] is True and aufrufe[0]["bild_modell"] == "vision"
|
||
|
||
# Ohne Zwilling (Projektor auf anderen Gewichten zählt nicht): keine Bild-Probe.
|
||
monkeypatch.setattr(radar.llamaswap, "list_models", lambda: [hirn, fremd])
|
||
radar._baseline("hirn", time.time() + 60)
|
||
assert aufrufe[1]["bild"] is False and aufrufe[1]["bild_modell"] is None
|
||
assert aufrufe[0]["kennung"] != aufrufe[1]["kennung"] # neuer Zwilling → neue Baseline
|