HitonabiandClaude Opus 5.5
133a491aca
Ampel / ampel (push) Failing after 20s
Ohne den Key ist eine llama-swap-Gruppe exklusiv: JEDE Anfrage ans Hirn entlud den Coder und die Bild-Zwillinge (live gemessen 24.09.). Fuer OpenChamber hiess das nach jeder Lucy-, NerdQuiz- oder Job-Anfrage: Coder neu laden und den ganzen Vorlauf nachrechnen. set_group setzt den Key fuer immer-warme Gruppen jetzt selbst, damit ein Hirn-Tausch ihn nicht wieder verliert. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
670 lines
36 KiB
Python
670 lines
36 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
|
||
|
||
|
||
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"]
|