163 lines
6.2 KiB
Python
163 lines
6.2 KiB
Python
"""Training-Harness: Registry↔config, ENV-Override, ACO-Trainer (Stub-Runner), Soll-Abgleich."""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import config
|
|
import train_params
|
|
from train import AmeisenTrainer, score
|
|
from train_lauf import soll_abgleich
|
|
|
|
BACKEND = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def test_registry_spiegelt_config():
|
|
"""Jeder Registry-Parameter existiert in config mit identischem Default, flow-sicheren
|
|
Rändern und einer Fidelity-Zuordnung — sonst optimiert der Trainer Phantome."""
|
|
for name, p in train_params.PARAMS.items():
|
|
assert getattr(config, name, None) == p["default"], name
|
|
assert p["min"] <= p["default"] <= p["max"], name
|
|
assert p["step"] > 0, name
|
|
assert p["fidelity"] in ("board2", "voll"), name
|
|
|
|
|
|
def test_creator_params_override_wirkt_im_subprozess():
|
|
out = subprocess.run(
|
|
[sys.executable, "-c", "import config; print(config.FACTS_CHUNK_SUBS, config.TIMEOUTS['subblock_check'][0])"],
|
|
capture_output=True, text=True, cwd=BACKEND,
|
|
env={"PATH": "/usr/bin:/bin", "CREATOR_PARAMS": '{"FACTS_CHUNK_SUBS": 6, "TIMEOUT_subblock_check_base": 77}'})
|
|
assert out.stdout.split() == ["6", "77"], out.stderr
|
|
|
|
|
|
def test_creator_params_unbekannter_name_bricht_ab():
|
|
out = subprocess.run([sys.executable, "-c", "import config"],
|
|
capture_output=True, text=True, cwd=BACKEND,
|
|
env={"PATH": "/usr/bin:/bin", "CREATOR_PARAMS": '{"GIBT_ES_NICHT": 1}'})
|
|
assert out.returncode != 0 and "GIBT_ES_NICHT" in out.stderr
|
|
|
|
|
|
def _metrics(note=8.0, dauer=10.0, tokens=1_000_000, **quoten):
|
|
return {"note": note, "quoten": quoten, "quoten_artefakte": {},
|
|
"dauer_min": dauer, "tokens": {"input": tokens, "output": 0}, "agents": {}}
|
|
|
|
|
|
def _stub(score_fn):
|
|
"""Runner-Paar (F1/F2 + F0) für Tests: score_fn(params) → Metriken."""
|
|
calls = []
|
|
|
|
async def runner(params, fidelity, suffix=""):
|
|
calls.append((dict(params), fidelity))
|
|
return score_fn(params)
|
|
|
|
async def f0(params):
|
|
return {"ok": True, "invarianten_fehler": [], "calls": 100}
|
|
|
|
runner.calls = calls
|
|
return runner, f0
|
|
|
|
|
|
def _trainer(tmp_path, runner, f0, **kw):
|
|
args = dict(max_trials=999, max_stunden=1, ameisen=3, seed=7, f2_intervall=1000)
|
|
args.update(kw)
|
|
return AmeisenTrainer(tmp_path / "s", runner=runner, runner_f0=f0, **args)
|
|
|
|
|
|
async def test_aco_konvergiert_auf_optimum(tmp_path):
|
|
"""Gepflanztes Optimum (GEN_PANEL=3) wird gefunden und bestätigt übernommen;
|
|
die Pheromon-Spur konzentriert sich dort."""
|
|
def bewertung(params):
|
|
return _metrics(note=9.5, dauer=7.0) if params.get("GEN_PANEL") == 3 else _metrics()
|
|
|
|
runner, f0 = _stub(bewertung)
|
|
t = _trainer(tmp_path, runner, f0, max_trials=120)
|
|
best = await t.run()
|
|
assert best.get("GEN_PANEL") == 3
|
|
taus = t.pheromon["GEN_PANEL"]
|
|
assert max(taus, key=lambda k: taus[k]) == "3"
|
|
|
|
|
|
async def test_uebernahme_braucht_bestaetigung(tmp_path):
|
|
"""Einmaliger Glückstreffer ohne bestätigten Zweitlauf wird nicht Bester."""
|
|
zustand = {"mal": 0}
|
|
|
|
def bewertung(params):
|
|
if params.get("GEN_PANEL") == 3:
|
|
zustand["mal"] += 1
|
|
return _metrics(note=9.5) if zustand["mal"] == 1 else _metrics(note=8.0)
|
|
return _metrics()
|
|
|
|
runner, f0 = _stub(bewertung)
|
|
t = _trainer(tmp_path, runner, f0, max_trials=40)
|
|
best = await t.run()
|
|
assert best.get("GEN_PANEL") != 3
|
|
|
|
|
|
async def test_f0_filter_verwirft_kaputte_kandidaten(tmp_path):
|
|
"""Kandidaten mit Invarianten-Fehlern erreichen nie einen bezahlten Lauf."""
|
|
runner, _f0 = _stub(lambda p: _metrics())
|
|
|
|
async def f0_kaputt(params):
|
|
if params: # nur Nicht-Baseline
|
|
return {"ok": True, "invarianten_fehler": ["kaputt"], "calls": 100}
|
|
return {"ok": True, "invarianten_fehler": [], "calls": 100}
|
|
|
|
t = _trainer(tmp_path, runner, f0_kaputt, max_trials=20)
|
|
await t.run()
|
|
bezahlt_mit_params = [c for c, _f in runner.calls if c]
|
|
assert bezahlt_mit_params == [] # nur Baselines liefen
|
|
|
|
|
|
async def test_resume_laedt_pheromon_und_cache(tmp_path):
|
|
def bewertung(params):
|
|
return _metrics(note=9.5) if params.get("GEN_PANEL") == 3 else _metrics()
|
|
|
|
runner, f0 = _stub(bewertung)
|
|
t = _trainer(tmp_path, runner, f0, max_trials=60)
|
|
await t.run()
|
|
best, tau = t.best_params, dict(t.pheromon["GEN_PANEL"])
|
|
runner2, f02 = _stub(bewertung)
|
|
t2 = _trainer(tmp_path, runner2, f02, max_trials=0) # kein Budget: alles aus Persistenz
|
|
assert t2.best_params == best
|
|
assert t2.pheromon["GEN_PANEL"] == tau
|
|
|
|
|
|
async def test_budget_stoppt(tmp_path):
|
|
runner, f0 = _stub(lambda p: _metrics())
|
|
t = _trainer(tmp_path, runner, f0, max_trials=4)
|
|
await t.run()
|
|
assert len(runner.calls) <= 4
|
|
|
|
|
|
def test_score_richtungen():
|
|
basis = _metrics()
|
|
assert score(_metrics(note=9.0), basis) > score(basis, basis)
|
|
assert score(_metrics(dauer=20.0, tokens=2_000_000), basis) < score(basis, basis)
|
|
assert score(_metrics(fremd=0.2, luecken=0.1), basis) < score(basis, basis)
|
|
mit_soll = dict(_metrics(), soll={"f1": 1.0})
|
|
ohne_soll = dict(_metrics(), soll={"f1": 0.5})
|
|
assert score(mit_soll, basis) > score(ohne_soll, basis)
|
|
|
|
|
|
def test_soll_abgleich():
|
|
soll = {"bloecke": [{"titel": "Symmetrische Verschlüsselung"},
|
|
{"titel": "Asymmetrische Verschlüsselung",
|
|
"alternativen": ["Public-Key-Kryptographie"]},
|
|
{"titel": "Digitale Signaturen"}]}
|
|
r = soll_abgleich(["Symmetrische Verschlüsselung", "Public-Key-Kryptographie", "Quantencomputer"], soll)
|
|
assert r["fehlend"] == ["Digitale Signaturen"]
|
|
assert r["extra"] == ["Quantencomputer"]
|
|
assert 0 < r["f1"] < 1
|
|
|
|
|
|
async def test_copy_topic_dupliziert_karten_und_bloecke(testdb):
|
|
db = testdb
|
|
await db.kanban_upsert_card("q", "inventory", "b1", "block", "done_block", {"title": "Alpha"})
|
|
await db.upsert_block("q", "alpha", "Alpha", "Beschreibung", ["s1"], "r1")
|
|
await db.copy_topic("q", "z")
|
|
karten = await db.kanban_cards("z")
|
|
assert [c["card_id"] for c in karten] == ["b1"]
|
|
bloecke = await db.list_blocks("z")
|
|
assert [b["title"] for b in bloecke] == ["Alpha"]
|