HitonabiandClaude Opus 5.5
e9f488b56c
Ampel / ampel (push) Successful in 26s
Ballast raus: - 35 Routen ohne Nutzer entfernt (agent/*, fit, roles, ctx, drafts, groups, routing/policy, system/history, system/self-update, maintenance/reboot, zeitmaschine/inhalt, zeitplan, voice/health|metrics|trace|voices|reference|tts). Von 95 auf 60. - Tote Module geloescht: agent-Router, roles, agent_aktivitaet, metrics_history (samt 10-s-Sampler), voice_metrics, migrate_config, parse_mc2_timeout, scripts/. - Unbenutzte Funktionen und Konstanten entfernt (Modell-Upgrade-Empfehlung, Draft-/Kontext- Setzer, Konsole, PC-Ausfuehrer-Probe, Routing-Policy-Editor ...). Robuster: - Jobs in eigener Prozessgruppe (Abbrechen beendet wirklich alles), Zeitlimit je Job-Art, start_job_exklusiv: zwei Klicks starten kein doppeltes Update mehr; alte Jobs raeumen sich auf. - Update-Pruefung meldet Fehler (pruef_fehler, Lampe "Pruefung unklar") statt "aktuell". - Nach jedem Update sofort neu pruefen (update_stand) statt 10 Minuten alten Stand zeigen. - llama-swap-Config: Sperre (RLock + flock) fuer UI, Radar, Aufraeumen und Hirn-Umstellung. - Hermes-Config: bei Lesefehler nichts schreiben, atomar, mit Sicherung. - Live-Strom und Gateway-Warnung blockieren den Event-Loop nicht mehr (Lucy, OpenChamber). - Gateway antwortet bei Engine-Ausfall im OpenAI-Fehlerformat (502) statt nacktem 500. - Abgestuerzte Waechter-Pruefung wird ein gelber Hinweis statt still zu verschwinden. - Download laedt nur den gewuenschten Quant (vorher bei Fehlen alle Teile aller Varianten), Download-Jobs in Gruppe "download"; HF-Suche kodiert den Suchbegriff. - Herkunftspruefung: schreibende /api-Aufrufe fremder Webseiten werden abgelehnt (keine Anmeldung, User-Entscheid); Skripte, Desktop-Lucy und /v1 unveraendert. - Modellpfade: Eintragen und Loeschen nur innerhalb von MODELS_DIR. - Dienste-Liste fragt keine abgebauten Dienste mehr ab (PC-Ausfuehrer haette 3 s gekostet). - SSE-Fehlerzeilen von /api/voice/chat als gueltiges JSON. - mission-control-2.service: --timeout-graceful-shutdown 3 (Neustart ohne 10-s-Haenger). Tests: 92 gruen (neu: Herkunft, Quant-Auswahl, abgestuerzte Pruefung). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
778 lines
43 KiB
Python
778 lines
43 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": {"alter-zwilling": {"cmd": "llama-server -m /alt.gguf --mmproj /alt-mm.gguf",
|
||
"aliases": ["vision" if rolle == "hirn" else "coder-bild"]}},
|
||
"groups": {"bild": {"swap": True, "exclusive": False, "members": ["alter-zwilling"]}}}
|
||
|
||
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" not in befehl
|
||
# Wie im Betrieb seit 24.09.: Bilder sieht der neue Bild-Zwilling (Projektor, ohne Draft), der alte fliegt raus.
|
||
zwilling = cfg["models"]["o-Bild"]
|
||
assert zwilling["aliases"] == ["vision"] and "--mmproj" in str(zwilling["cmd"]) and "--spec" not in str(zwilling["cmd"])
|
||
assert "alter-zwilling" not in cfg["models"] and cfg["groups"]["bild"]["members"] == ["o-Bild"]
|
||
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():
|
||
pfade = {"gguf": "/srv/models/F-GGUF/f.gguf", "mmproj": "/srv/models/F-GGUF/mm.gguf", "draft": None}
|
||
k = _kandidat("f", rolle="coder", ctx=65536, draft={"typ": "draft-mtp", "n_max": 2, "datei": None}, flags=["-ub", "512"])
|
||
befehl = radar.betriebs_befehl("coder", pfade, k)
|
||
assert "-c 65536" in befehl and "--parallel 1" in befehl and "-ctk q8_0 -ctv q8_0" in befehl
|
||
assert "--mmproj" not in befehl and "--load-mode none" in befehl # Bilder sieht der Zwilling
|
||
assert befehl.endswith("--spec-type draft-mtp --spec-draft-n-max 2 -ub 512")
|
||
zwilling = radar.zwilling_befehl("coder", pfade, k)
|
||
assert "--mmproj /srv/models/F-GGUF/mm.gguf" in zwilling and "--spec" not in zwilling and "-c 131072" in zwilling
|
||
assert zwilling.endswith("-ub 512")
|
||
assert "--parallel 2" in radar.betriebs_befehl("hirn", {"gguf": "x.gguf", "mmproj": None, "draft": None},
|
||
_kandidat("h"))
|
||
|
||
|
||
def test_betriebsbefehl_nimmt_den_im_test_gewaehlten_draft():
|
||
"""Gewählt wurde der Draft des heutigen Modells: fester Pfad, nicht der Kandidaten-Ordner."""
|
||
pfade = {"gguf": "/srv/models/O-GGUF/o.gguf", "mmproj": "/srv/models/O-GGUF/mm.gguf", "draft": None}
|
||
k = _kandidat("o", draft_betrieb={"typ": "draft-dflash", "ort": "fremd", "pfad": "/srv/models/D/d.gguf",
|
||
"n_max": None})
|
||
assert radar.betriebs_befehl("hirn", pfade, k).endswith("--spec-type draft-dflash --spec-draft-model /srv/models/D/d.gguf")
|
||
k["draft_betrieb"] = None # im Test war „ohne Draft“ am schnellsten
|
||
assert "--spec" not in radar.betriebs_befehl("hirn", pfade, k)
|
||
|
||
|
||
def _gguf_datei(pfad, arch: str) -> None:
|
||
"""Winzige GGUF-Datei mit general.architecture (genug für _lokale_architektur)."""
|
||
wert = arch.encode()
|
||
schluessel = b"general.architecture"
|
||
daten = (b"GGUF" + struct.pack("<I", 3) + struct.pack("<Q", 0) + struct.pack("<Q", 1)
|
||
+ struct.pack("<Q", len(schluessel)) + schluessel + struct.pack("<I", 8)
|
||
+ struct.pack("<Q", len(wert)) + wert)
|
||
pfad.write_bytes(daten)
|
||
|
||
|
||
def test_draft_varianten_probieren_den_draft_des_heutigen_modells_bei_gleicher_architektur(tmp_path, monkeypatch):
|
||
hirn_gguf, draft = tmp_path / "hirn.gguf", tmp_path / "dflash.gguf"
|
||
_gguf_datei(hirn_gguf, "qwen35moe")
|
||
draft.write_bytes(b"x" * 1000)
|
||
heute = {"name": "Hirn", "gguf_path": str(hirn_gguf),
|
||
"cmd": f"llama-server -m {hirn_gguf} --spec-type draft-dflash --spec-draft-model {draft}"}
|
||
monkeypatch.setattr(radar, "_heutige_modelle", lambda: {"hirn": heute})
|
||
pfade = {"gguf": "/x/k.gguf", "mmproj": "/x/mm.gguf", "draft": None}
|
||
|
||
gleich = radar._draft_varianten(_kandidat("k", arch="qwen35moe", bedarf_gb=24.2, mtp=True), pfade)
|
||
assert [v["name"] for v in gleich] == ["ohne Draft", "eingebauter MTP-Kopf", "Draft des heutigen Modells"]
|
||
assert gleich[2]["flags"] == ["--spec-type", "draft-dflash", "--spec-draft-model", str(draft)]
|
||
assert gleich[2]["draft"]["ort"] == "fremd"
|
||
|
||
fremd = radar._draft_varianten(_kandidat("k", arch="qwen3next", bedarf_gb=24.2), pfade)
|
||
assert [v["name"] for v in fremd] == ["ohne Draft"] # andere Architektur: nicht probieren
|
||
zu_gross = radar._draft_varianten(_kandidat("k", arch="qwen35moe", bedarf_gb=88.0), pfade)
|
||
assert [v["name"] for v in zu_gross] == ["ohne Draft"] # Draft passt nicht mehr in den Speicher
|
||
merkliste = radar._draft_varianten(_kandidat("k", arch="qwen35moe", draft={"typ": "draft-simple"}),
|
||
{**pfade, "draft": "/x/d.gguf"})
|
||
assert [v["name"] for v in merkliste] == ["Draft aus der Merkliste"]
|
||
|
||
|
||
def test_waehle_draft_nimmt_die_schnellste_variante_die_startet(ps, monkeypatch):
|
||
tempi = {"": 67.4, "draft-mtp": 81.0, "draft-dflash": 98.2}
|
||
|
||
def mit_server(gguf, arbeit, *, flags=(), **kw):
|
||
typ = flags[1] if flags else ""
|
||
if typ == "draft-mtp":
|
||
raise ps.ServerFehler("vocab mismatch")
|
||
return {"tiefe": {"tg_tps": tempi[typ], "acc": 0.8 if typ else None}}, None
|
||
|
||
monkeypatch.setattr(ps, "mit_server", mit_server)
|
||
varianten = [{"name": "ohne Draft", "flags": []},
|
||
{"name": "eingebauter MTP-Kopf", "flags": ["--spec-type", "draft-mtp"]},
|
||
{"name": "Draft des heutigen Modells", "flags": ["--spec-type", "draft-dflash"]}]
|
||
wahl = ps.waehle_draft("/x/k.gguf", varianten, protokoll=lambda *_: None)
|
||
assert wahl["name"] == "Draft des heutigen Modells"
|
||
assert wahl["messungen"]["ohne Draft"] == 67.4 and "geht nicht" in wahl["messungen"]["eingebauter MTP-Kopf"]
|
||
assert "Gemessen mit Draft des heutigen Modells" in radar._draft_text({**wahl, "draft": {"typ": "draft-dflash"}})
|
||
|
||
|
||
def test_mit_draft_laeuft_die_bild_probe_auf_einem_zwilling_ohne_draft(ps, monkeypatch):
|
||
"""Draft und Bild gehen in llama.cpp nicht zusammen (HTTP 500) — der Prüfstand trennt sie wie der Betrieb."""
|
||
starts = []
|
||
|
||
def mit_server(gguf, arbeit, *, mmproj=None, flags=(), **kw):
|
||
starts.append((mmproj, list(flags)))
|
||
return ({"bild": None, "tiefe": {}} if len(starts) == 1 else {"direkt": 1.0, "agentisch": True}), \
|
||
type("S", (), {"befehl": ["llama-server"], "ladezeit": 4})()
|
||
|
||
monkeypatch.setattr(ps, "mit_server", mit_server)
|
||
lauf = ps.pruefe_kandidat("hirn", "/x/k.gguf", mmproj="/x/mm.gguf",
|
||
flags=["--spec-type", "draft-dflash", "--spec-draft-model", "/d.gguf", "-ub", "512"])
|
||
assert starts[0] == (None, ["--spec-type", "draft-dflash", "--spec-draft-model", "/d.gguf", "-ub", "512"])
|
||
assert starts[1] == ("/x/mm.gguf", ["-ub", "512"])
|
||
assert lauf["werte"]["bild"] == {"direkt": 1.0, "agentisch": True}
|
||
|
||
|
||
# --- 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
|
||
|
||
|
||
def test_immer_warme_gruppe_verdraengt_nichts(monkeypatch):
|
||
"""24.09.: ohne exclusive: false entlud jede Anfrage ans Hirn den Coder — set_group muss das setzen."""
|
||
from services import llamaswap
|
||
cfg = {"models": {"h": {"cmd": "x", "ttl": 300}}, "groups": {}}
|
||
monkeypatch.setattr(llamaswap, "read_config", lambda: cfg)
|
||
monkeypatch.setattr(llamaswap, "write_config", lambda c: None)
|
||
llamaswap.set_group("brains", ["h"], swap=False, persist=True)
|
||
assert cfg["groups"]["brains"]["exclusive"] is False and cfg["groups"]["brains"]["persistent"] is True
|
||
assert cfg["models"]["h"]["ttl"] == 0
|
||
llamaswap.set_group("andere", ["h"], swap=True, persist=False)
|
||
assert "exclusive" not in cfg["groups"]["andere"]
|
||
|
||
|
||
def test_hf_download_nimmt_nur_den_gewuenschten_quant(monkeypatch):
|
||
"""24.09.2026: Fehlt der Quant, lädt der Download NICHTS (vorher: alle Teile aller Varianten)."""
|
||
from services import hf
|
||
|
||
baum = [
|
||
{"path": "Q8_0/Modell-Q8_0-00001-of-00002.gguf", "size": 50},
|
||
{"path": "Q8_0/Modell-Q8_0-00002-of-00002.gguf", "size": 50},
|
||
{"path": "Q4_K_M/Modell-Q4_K_M-00001-of-00002.gguf", "size": 30},
|
||
{"path": "Q4_K_M/Modell-Q4_K_M-00002-of-00002.gguf", "size": 30},
|
||
{"path": "mmproj-F16.gguf", "size": 5},
|
||
]
|
||
monkeypatch.setattr(hf, "_tree", lambda repo: baum)
|
||
leer = hf.resolve_gguf("x/y", "IQ3_XXS")
|
||
assert leer["first"] is None and leer["files"] == []
|
||
q4 = hf.resolve_gguf("x/y", "Q4_K_M")
|
||
assert q4["files"] == ["Q4_K_M/Modell-Q4_K_M-00001-of-00002.gguf", "Q4_K_M/Modell-Q4_K_M-00002-of-00002.gguf"]
|
||
assert q4["total_bytes"] == 65 and q4["mmproj"] == "mmproj-F16.gguf"
|