refactor: Pipeline-Stufen in Ordner mit ihren Prompts

Im Wurzelverzeichnis liegt nur noch TextToModel.py (Ablauf und CLI).
Jede Stufe liegt mit Code und Prompt in einem eigenen Ordner: core,
understanding, modeling, checks, solving, reporting, data, benchmark;
Tests unter tests/, Aufgaben unter tasks/. understand_task,
text_to_model/revise_model und review_assumptions/needs_work/select_best
sind aus TextToModel.py zu ihren Prompts gewandert, inhaltlich
unveraendert. Prompt-Kennung des Benchmarks bleibt 93744636095a.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-09-17 12:11:51 +02:00
commit 7459690bf5
12 changed files with 315 additions and 279 deletions

138
tests/test_charts.py Normal file
View file

@ -0,0 +1,138 @@
"""Inline-SVG-Diagramme: gültiges SVG, korrekte Geometrie, Einbindung in den Report."""
import xml.etree.ElementTree as ET
import pytest
from reporting import charts
from core.model_schema import PipelineResult, ResultTable, SolverRun, SolverStatus
from reporting.report import build_document, to_html, to_markdown
from test_model_checks import make_spec
from test_result_tables import PORTFOLIO_TABLE
from test_review import make_review
LABELS = ["A02", "A08", "A03"]
VALUES = [0.03433, 0.03093, 0.02587]
def parse(svg: str):
assert svg.startswith("<svg")
return ET.fromstring(svg)
def test_bar_chart_is_valid_svg_with_one_bar_per_label():
root = parse(charts.bar_chart(LABELS, VALUES, "Portfolio"))
assert len(root.findall(".//{http://www.w3.org/2000/svg}path")) == len(LABELS)
texts = [t.text for t in root.findall(".//{http://www.w3.org/2000/svg}text")]
for label in LABELS:
assert label in texts # Direktbeschriftung statt Legende (eine Serie)
assert root.get("role") == "img" and "Portfolio" in root.get("aria-label")
def test_bars_are_proportional_and_capped_in_thickness():
svg = charts.bar_chart(["groß", "klein"], [1.0, 0.5], "Test")
widths = [float(d.split("H")[1].split(" ")[0].split(",")[0]) for d in
[p.get("d") for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")]]
assert widths[0] > widths[1]
heights = {float(p.get("d").split("V")[1].split(" ")[0]) for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")}
assert charts.BAR_HEIGHT <= 24
def test_negative_values_use_the_other_colour_role():
svg = charts.bar_chart(["Ertrag", "Kosten"], [1.0, -0.4], "Test")
fills = [p.get("fill") for p in parse(svg).findall(".//{http://www.w3.org/2000/svg}path")]
assert fills == ["var(--viz-pos)", "var(--viz-neg)"]
def test_waterfall_has_step_per_term_plus_total():
svg = charts.waterfall_chart(["Rendite", "Risiko", "Kosten"], [0.1277, -0.0052, -0.0256],
"Zielwert", 0.0969, "Zielwert-Zerlegung", " ")
root = parse(svg)
fills = [p.get("fill") for p in root.findall(".//{http://www.w3.org/2000/svg}path")]
assert fills == ["var(--viz-pos)", "var(--viz-neg)", "var(--viz-neg)", "var(--viz-total)"]
assert "Zielwert" in [t.text for t in root.findall(".//{http://www.w3.org/2000/svg}text")]
def test_tooltip_per_mark():
root = parse(charts.bar_chart(LABELS, VALUES, "Portfolio"))
titles = [t.text for t in root.findall(".//{http://www.w3.org/2000/svg}title")]
assert len(titles) == len(LABELS) and titles[0].startswith("A02: ")
def make_result(tables):
return PipelineResult(
task="Aufgabe", specification=make_spec(), review=make_review(), revisions=0,
solver_run=SolverRun(status=SolverStatus.OPTIMAL, objective_value=0.09756, tables=tables,
headline="Das Portfolio erreicht 9,76 %.", solution={"w[A02]": 0.3}),
)
def test_html_report_embeds_chart_and_markdown_does_not():
doc = build_document(make_result([ResultTable.model_validate(PORTFOLIO_TABLE)]))
page = to_html(doc)
assert '<figure class="viz-root">' in page and "<svg" in page
assert "<svg" not in to_markdown(doc)
def test_decomposition_table_gets_waterfall():
table = ResultTable(title="Zielwert-Zerlegung", columns=["Bestandteil", "Wert"],
rows=[["Rendite", 0.1277], ["Risiko", -0.0052], ["Kosten", -0.0256], ["Zielwert", 0.0969]],
objective_column="Wert")
page = to_html(build_document(make_result([table])))
assert page.count("var(--viz-total)") == 1 # Gesamtbalken nur im Wasserfall
def test_result_comes_first_and_headline_is_lead():
doc = build_document(make_result([ResultTable.model_validate(PORTFOLIO_TABLE)]))
h2 = [b[1] for b in doc if b[0] == "h2"]
assert h2[0] == "Ergebnis" and h2.index("Ergebnis") < h2.index("Mathematisches Modell")
assert ("lead", "Das Portfolio erreicht 9,76 %.") in doc
assert '<p class="lead">' in to_html(doc)
@pytest.mark.parametrize("table", [
ResultTable(title="Ohne Zahlen", columns=["A", "B"], rows=[["x", "1,5 %"]], objective_column="B"),
ResultTable(title="Nur eine Zeile", columns=["A", "B"], rows=[["x", 1.0]], objective_column="B"),
ResultTable(title="Ohne Beitragsspalte", columns=["A", "B"], rows=[["x", 1.0], ["y", 2.0]]),
])
def test_no_chart_without_usable_numbers(table):
assert "<svg" not in to_html(build_document(make_result([table])))
# --- Graphenzeichnung ---
def test_graph_chart_is_valid_svg_and_marks_the_path():
import xml.etree.ElementTree as ET
from reporting.charts import graph_chart
svg = graph_chart(["0", "1", "2", "3"],
[("0", "1", 8), ("0", "2", 18), ("0", "3", 31), ("1", "2", 10), ("1", "3", 21), ("2", "3", 12)],
path=["0", "1", "3"], title="Graph")
ET.fromstring(svg)
assert svg.count("<circle") == 4 # jeder Knoten einmal
assert svg.count("marker-end") == 6 # jeder Bogen einmal, und gerichtet
assert svg.count("--viz-pos") >= 2 # Weg und Knoten des Weges hervorgehoben
def test_graph_chart_refuses_unknown_nodes_and_huge_graphs():
from reporting.charts import graph_chart
assert graph_chart(["a"], [("a", "b", 1)]) == "" # Bogen ins Nichts
assert graph_chart([str(i) for i in range(20)], [("0", "1", 1)]) == "" # zu groß zum Zeichnen
assert graph_chart([], []) == ""
def test_graph_section_explains_before_it_shows():
from core.model_schema import GraphSolution, SolverRun, SolverStatus
from reporting.report import _graph_section
run = SolverRun(status=SolverStatus.OPTIMAL, objective_value=29, graph=GraphSolution.model_validate({
"nodes": ["0", "1", "3"],
"arcs": [{"source": "0", "target": "1", "weight": 8}, {"source": "1", "target": "3", "weight": 21}],
"path": ["0", "1", "3"],
"steps": [{"node": "0", "label": 0}, {"node": "1", "label": 8}, {"node": "3", "label": 29}],
"node_meaning": "ein Zeitpunkt – das Ende von Jahr i.",
"arc_meaning": "Kauf am Ende von Jahr i und Nutzung bis Ende von Jahr j.",
}))
doc = _graph_section(run)
arten = [block[0] for block in doc]
assert arten[0] == "p" and "Ende von Jahr" in doc[0][1] # Erklärung steht vor der Zeichnung
assert "svg" in arten
assert any(b[0] == "table" and "Markierungsverlauf" in str(doc[arten.index("table") - 1]) for b in doc)
assert any("8 + 21 = 29" in str(b) for b in doc) # Rechenweg ausgeschrieben

View file

@ -0,0 +1,49 @@
"""Profil beigelegter Datendateien – deterministisch, ohne LLM."""
import pytest
from data.data_profile import profile_file, profile_files, to_prompt
@pytest.fixture
def kurse(tmp_path):
datei = tmp_path / "kurse.csv"
datei.write_text("Aktie;Rendite;Risiko;Gebuehr\nA01;6,9;2,04;0,68\nA02;14,4;1,63;0,74\nA03;11,5;;0,39\n",
encoding="utf-8")
return datei
def test_german_numbers_are_recognised(kurse):
# deutsche Daten sind der Normalfall: "6,9" ist eine Zahl, keine Zeichenkette
profil = profile_file(kurse)
assert profil.rows == 3 and len(profil.columns) == 4
rendite = next(s for s in profil.columns if s.name == "Rendite")
assert rendite.dtype == "Zahl"
assert rendite.minimum == pytest.approx(6.9) and rendite.maximum == pytest.approx(14.4)
def test_text_columns_and_gaps(kurse):
profil = profile_file(kurse)
assert next(s for s in profil.columns if s.name == "Aktie").dtype == "Text"
assert next(s for s in profil.columns if s.name == "Risiko").missing == 1
def test_separator_is_detected(tmp_path):
komma = tmp_path / "en.csv"
komma.write_text("name,value\nfoo,1.5\nbar,2.5\n")
profil = profile_file(komma)
assert [s.name for s in profil.columns] == ["name", "value"]
assert next(s for s in profil.columns if s.name == "value").mean == pytest.approx(2.0)
def test_unreadable_file_is_reported_not_raised(tmp_path):
profil = profile_file(tmp_path / "gibtsnicht.csv")
assert profil.error and "FileNotFoundError" in profil.error
assert profil.rows == 0
def test_prompt_names_columns_and_ranges(kurse):
text = to_prompt(profile_files([kurse]))
assert "kurse.csv: 3 Zeilen, 4 Spalten" in text
assert "Rendite (Zahl, von 6.9 bis 14.4" in text
assert "1 fehlend" in text
assert to_prompt([]) == ""

145
tests/test_feasibility.py Normal file
View file

@ -0,0 +1,145 @@
"""Deterministische Zulässigkeitsprüfung: Lösung einsetzen, Schranken, Ganzzahligkeit, Namensvertrag."""
import pytest
from core.model_schema import PipelineResult, Sensitivity, SolverRun, SolverStatus
from reporting.report import build_document, to_html
from solving.solver_stage import check_constraint_coverage, check_feasibility, check_sensitivity
from test_model_checks import make_spec
from test_review import make_review
SPEC = make_spec(
constraints=[{"identifier": "Budget", "latex_formula": "\\sum_j p_j n_j \\le B", "description": "-"},
{"identifier": "Mindestanzahl", "latex_formula": "\\sum_j y_j \\ge 5", "description": "-"}],
variables=[{"name": "n_j", "var_type": "integer", "domain": ">= 0", "description": "-"}],
)
def run_with(constraints=(), variables=(), status=SolverStatus.OPTIMAL):
return SolverRun(status=status, objective_value=1.0, constraints=list(constraints), variables=list(variables))
BUDGET_OK = {"name": "Budget", "lhs": 99999.99, "sense": "<=", "rhs": 100000.0}
BUDGET_8_CENT = {"name": "Budget", "lhs": 100000.08, "sense": "<=", "rhs": 100000.0}
ANZAHL_OK = {"name": "Mindestanzahl", "lhs": 5, "sense": ">=", "rhs": 5}
def test_clean_solution_has_no_findings():
run = run_with([BUDGET_OK, ANZAHL_OK], [{"name": "n[A02]", "value": 1037, "var_type": "integer", "lb": 0}])
assert check_feasibility(run, SPEC) == []
def test_eight_cent_budget_violation_is_found():
# genau der beobachtete Fall: SCIP meldet "optimal", 8 Cent über dem Budget (8e-7 relativ)
problems = check_feasibility(run_with([BUDGET_8_CENT, ANZAHL_OK]), SPEC)
assert len(problems) == 1 and "Budget" in problems[0] and "verletzt" in problems[0]
def test_floating_point_noise_is_tolerated():
noisy = {"name": "Budget", "lhs": 100000.0000000001, "sense": "<=", "rhs": 100000.0}
assert check_feasibility(run_with([noisy, ANZAHL_OK]), SPEC) == []
@pytest.mark.parametrize("constraint, expected", [
({"name": "Mindestanzahl", "lhs": 4, "sense": ">=", "rhs": 5}, "Mindestanzahl"),
({"name": "Budget", "lhs": 99999.0, "sense": "==", "rhs": 100000.0}, "Budget"),
])
def test_other_senses_are_checked(constraint, expected):
problems = check_feasibility(run_with([constraint, BUDGET_OK, ANZAHL_OK][:2] + [constraint]), SPEC)
assert any(expected in p for p in problems)
@pytest.mark.parametrize("variable, expected", [
({"name": "n[A02]", "value": 1037.4, "var_type": "integer", "lb": 0}, "nicht ganzzahlig"),
({"name": "n[A02]", "value": -1, "var_type": "integer", "lb": 0}, "Untergrenze"),
({"name": "n[A02]", "value": 11, "var_type": "integer", "lb": 0, "ub": 10}, "Obergrenze"),
({"name": "y[A02]", "value": 2, "var_type": "binary"}, "weder 0 noch 1"),
])
def test_variable_violations_are_found(variable, expected):
problems = check_feasibility(run_with([BUDGET_OK, ANZAHL_OK], [variable]), SPEC)
assert any(expected in p for p in problems)
def test_missing_constraints_report_is_a_finding():
assert "keine Restriktionen" in check_feasibility(run_with(), SPEC)[0]
def test_infeasible_run_is_not_checked():
assert check_feasibility(run_with(status=SolverStatus.INFEASIBLE), SPEC) == []
# --- Namensvertrag ---
class Named:
def __init__(self, name): self.name = name
def test_missing_constraint_in_program_is_found():
problems = check_constraint_coverage([Named("Budget")], SPEC)
assert problems == ["Restriktion 'Mindestanzahl' aus der Spezifikation fehlt im Programm"]
def test_invented_constraint_is_found():
problems = check_constraint_coverage([Named("Budget"), Named("Mindestanzahl"), Named("Erfunden")], SPEC)
assert any("'Erfunden'" in p and "passt zu keiner" in p for p in problems)
@pytest.mark.parametrize("name", ["Budget", "Budget_A02", "Budget[A02]"])
def test_indexed_names_are_accepted(name):
assert check_constraint_coverage([Named(name), Named("Mindestanzahl")], SPEC) == []
# --- Sensitivität (Punkt 3): Plausibilität der Neulösungen ---
KAPAZITAET = {"name": "Budget", "lhs": 100000.0, "sense": "<=", "rhs": 100000.0}
def sens_run(sensitivity, objective=12000.0, relaxation=None, constraints=(KAPAZITAET,)):
return SolverRun(status=SolverStatus.OPTIMAL, objective_value=objective, constraints=list(constraints),
sensitivity=[Sensitivity.model_validate(s) for s in sensitivity], relaxation_bound=relaxation)
def test_plausible_sensitivity_has_no_warning():
# Maximierung: mehr Budget (rhs+δ) darf nicht schlechter sein als weniger (rhs−δ)
run = sens_run([{"name": "Budget", "delta": 1000, "objective_plus": 12100.0, "objective_minus": 11900.0}])
assert check_sensitivity(run, "maximize") == []
def test_implausible_direction_is_warned():
run = sens_run([{"name": "Budget", "delta": 1000, "objective_plus": 11900.0, "objective_minus": 12100.0}])
warnings = check_sensitivity(run, "maximize")
assert len(warnings) == 1 and "unplausibel" in warnings[0]
def test_greater_equal_constraint_direction_is_handled():
# Minimierung mit ">=": Lockern heißt rhs senken; niedrigere Kosten dort sind plausibel
bedarf = {"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}
run = sens_run([{"name": "Bedarf", "delta": 1, "objective_plus": 11.0, "objective_minus": 9.0}],
objective=10.0, constraints=(bedarf,))
assert check_sensitivity(run, "minimize") == []
falsch = sens_run([{"name": "Bedarf", "delta": 1, "objective_plus": 9.0, "objective_minus": 11.0}],
objective=10.0, constraints=(bedarf,))
assert any("unplausibel" in w for w in check_sensitivity(falsch, "minimize"))
def test_unknown_constraint_and_bad_delta_are_warned():
run = sens_run([{"name": "Gibt es nicht", "delta": 1, "objective_plus": 1.0, "objective_minus": 1.0},
{"name": "Budget", "delta": 0, "objective_plus": 1.0, "objective_minus": 1.0}])
warnings = check_sensitivity(run, "maximize")
assert any("keine solche Restriktion" in w for w in warnings) and any("delta" in w for w in warnings)
def test_relaxation_bound_must_not_be_worse_than_integer_solution():
gut = sens_run([], objective=12000.0, relaxation=12500.0) # Relaxation ist Schranke nach oben
schlecht = sens_run([], objective=12000.0, relaxation=11500.0)
assert check_sensitivity(gut, "maximize") == []
assert any("Relaxationsschranke" in w for w in check_sensitivity(schlecht, "maximize"))
def test_report_shows_sensitivity_and_integrality_price():
run = sens_run([{"name": "Budget", "delta": 1000, "objective_plus": 12100.0, "objective_minus": 11900.0,
"unit": "€ je €"}], relaxation=12500.0)
result = PipelineResult(task="Aufgabe", specification=make_spec(), review=make_review(), revisions=0,
solver_run=run)
page = to_html(build_document(result))
assert "Sensitivität der bindenden Restriktionen" in page and "Preis der Ganzzahligkeit" in page
assert "0,1" in page.replace(".", ",") or "0.1" in page # (12100-12000)/1000 = 0.1

116
tests/test_llm.py Normal file
View file

@ -0,0 +1,116 @@
import threading
import pytest
from core import llm
from core.llm import is_fatal_error, is_transient_error, model_for_step, provider_for
from core.model_schema import LLMUsage
class FakeCompletions:
def __init__(self): self.kwargs = None
def create_with_completion(self, **kwargs):
self.kwargs = kwargs
return "ok", object()
def fake_session(model="deepseek/deepseek-v4.1-flash", step_models=None):
"""LLMSession ohne echten Client – der Konstruktor würde einen API-Key verlangen."""
session = llm.LLMSession.__new__(llm.LLMSession)
session.model, session.provider = model, provider_for(model)
session.step_models = step_models or {}
session.usage = LLMUsage(model=model, provider=session.provider)
session._lock = threading.Lock()
completions = {}
session._clients = {}
for provider in {session.provider, *(provider_for(m) for m in session.step_models.values())}:
completions[provider] = FakeCompletions()
session._clients[provider] = type("C", (), {"chat": type("Chat", (), {"completions": completions[provider]})()})()
session.client = session._clients[session.provider]
return session, completions
@pytest.mark.parametrize("message", [
"429 RESOURCE_EXHAUSTED. {'error': {'code': 429, 'message': 'You exceeded your current quota'}}", # Gemini-Kontingent
"Error code: 402 - {'error': {'message': 'Insufficient credits'}}", # OpenRouter-Guthaben
"Error code: 401 - {'error': {'message': 'No auth credentials found'}}",
])
def test_fatal_errors_are_not_retried(message):
assert is_fatal_error(message)
assert not is_transient_error(message)
@pytest.mark.parametrize("message", [
"503 UNAVAILABLE. {'error': {'code': 503, 'message': 'model is currently experiencing high demand'}}",
"Error code: 429 - {'error': {'message': 'Rate limit exceeded'}}",
"Error code: 502 - Bad Gateway",
"Connection error.",
])
def test_transient_errors_are_retried(message):
assert is_transient_error(message)
def test_validation_error_is_neither():
message = "1 validation error for OptimizationModelSpecification"
assert not is_fatal_error(message) and not is_transient_error(message)
def test_provider_detection():
assert provider_for("deepseek/deepseek-v4-flash-0731") == "openrouter"
assert provider_for("gemini-2.5-flash") == "gemini"
def test_reasoning_effort_is_sent_only_when_configured(monkeypatch):
def make_session(effort):
monkeypatch.setattr(llm, "REASONING_EFFORT", effort)
session, completions = fake_session()
session.create(str, [{"role": "user", "content": "x"}])
return completions["openrouter"].kwargs["extra_body"]
assert "reasoning" not in make_session("")
assert make_session("low")["reasoning"] == {"effort": "low"}
def test_truncated_answer_is_retried():
# beobachtet: ein Anbieter schnitt die Modellspezifikation ab (max_tokens-Vorgabe des Anbieters)
message = "IncompleteOutputException: The output is incomplete due to a max_tokens length limit."
assert is_transient_error(message) and not is_fatal_error(message)
def test_max_tokens_is_sent_to_openrouter():
session, completions = fake_session()
session.create(str, [{"role": "user", "content": "x"}])
assert completions["openrouter"].kwargs["max_tokens"] == llm.LLM_MAX_TOKENS >= 16000
# --- Modell je Pipeline-Stufe (LLM_MODEL_EXTRACTION, LLM_MODEL_SOLVER, ...) ---
def test_step_model_falls_back_to_session_model(monkeypatch):
monkeypatch.delenv("LLM_MODEL_SOLVER", raising=False)
assert model_for_step("Solver-Code", "a/b") == "a/b"
assert model_for_step("unbenannt", "a/b") == "a/b" # Stufe ohne Umgebungsvariable
def test_step_model_from_environment(monkeypatch):
monkeypatch.setenv("LLM_MODEL_SOLVER", "schnell/modell")
assert model_for_step("Solver-Code", "a/b") == "schnell/modell"
assert model_for_step("Solver-Reparatur", "a/b") == "schnell/modell" # dieselbe Variable
assert model_for_step("Extraktion", "a/b") == "a/b"
def test_create_uses_step_model_and_records_it():
session, completions = fake_session(step_models={"Extraktion": "stark/modell", "Solver-Code": "gemini-2.5-flash"})
session.create(str, [{"role": "user", "content": "x"}], step="Extraktion")
assert completions["openrouter"].kwargs["model"] == "stark/modell"
# anderer Anbieter je Stufe: Gemini bekommt keine OpenRouter-Parameter
session.create(str, [{"role": "user", "content": "x"}], step="Solver-Code")
assert completions["gemini"].kwargs["model"] == "gemini-2.5-flash"
assert "extra_body" not in completions["gemini"].kwargs
assert [r.model for r in session.usage.call_records] == ["stark/modell", "gemini-2.5-flash"]
def test_overrides_lists_only_deviating_steps():
session, _ = fake_session(model="a/b", step_models={"Extraktion": "a/b", "Solver-Code": "schnell/modell"})
assert session.overrides() == {"Solver-Code": "schnell/modell"}

448
tests/test_model_checks.py Normal file
View file

@ -0,0 +1,448 @@
from checks.model_checks import check_spec, has_errors
from core.model_schema import OptimizationModelSpecification
def make_spec(**overrides) -> OptimizationModelSpecification:
data = {
"problem_class": "Linear Programming (LP)",
"target_stack": "SciPy (scipy.optimize.linprog)",
"routing_rationale": "-",
"executive_summary": "-",
"indices": ["j \\in J (Produkte)"],
"parameters": [
{"symbol": "c_j", "dimension": "Vektor", "unit": "€/t", "description": "Kosten", "extracted_values": "[3, 2]"},
{"symbol": "b", "dimension": "Skalar", "unit": "t", "description": "Bedarf", "extracted_values": "4"},
],
"variables": [
{"name": "x_{j}", "var_type": "continuous", "domain": ">= 0", "description": "Menge"},
],
"objective": {"sense": "minimize", "latex_formula": "\\min \\sum_{j \\in J} c_j x_j", "description": "-"},
"constraints": [
{"identifier": "Bedarf", "latex_formula": "\\sum_{j \\in J} x_j \\ge b", "description": "-"},
],
"suggested_export_format": "Matrix-Dict",
}
data.update(overrides)
return OptimizationModelSpecification.model_validate(data)
def codes(spec):
return {issue.code for issue in check_spec(spec)}
def test_valid_spec_has_no_issues():
assert check_spec(make_spec()) == []
def test_binary_variable_in_lp_is_error():
spec = make_spec(variables=[{"name": "x_j", "var_type": "binary", "domain": "{0,1}", "description": "-"}])
assert {"LP_WITH_DISCRETE_VARS", "SCIPY_WITH_DISCRETE_VARS"} <= codes(spec)
assert has_errors(check_spec(spec))
def test_stack_mismatch_is_error():
spec = make_spec(problem_class="Quadratic Programming (QP)")
assert "STACK_MISMATCH" in codes(spec)
def test_continuous_variable_with_cpsat_is_error():
spec = make_spec(problem_class="Constraint Programming (CP)", target_stack="Google OR-Tools (CP-SAT)")
assert "CPSAT_WITH_CONTINUOUS_VARS" in codes(spec)
def test_quadratic_term_in_lp_is_warning():
spec = make_spec(objective={"sense": "minimize", "latex_formula": "\\min \\sum_j c_j x_j^2", "description": "-"})
issues = check_spec(spec)
assert "NONLINEAR_TERM" in codes(spec)
assert not has_errors(issues)
def test_undeclared_symbol_is_warning():
spec = make_spec(constraints=[
{"identifier": "Kapazitaet", "latex_formula": "\\sum_{j} a_j x_j \\le K", "description": "-"},
])
messages = [i.message for i in check_spec(spec) if i.code == "UNDECLARED_SYMBOL"]
assert any("'a'" in m for m in messages)
assert any("'K'" in m for m in messages)
def test_superscript_and_escaped_underscore_names_are_recognized():
spec = make_spec(
parameters=[
{"symbol": "C^{prod}_{j}", "dimension": "Vektor", "unit": "€", "description": "-", "extracted_values": "1"},
{"symbol": "min\\_staff_1", "dimension": "Skalar", "unit": "Personen", "description": "-", "extracted_values": "3"},
{"symbol": "\\Sigma", "dimension": "Matrix", "unit": "dimensionslos", "description": "-", "extracted_values": "I"},
],
objective={"sense": "minimize", "latex_formula": "\\min \\sum_{j} C^{prod}_{j} x_{j}", "description": "-"},
constraints=[
{"identifier": "A", "latex_formula": "\\sum_{j} x_j \\ge min\\_staff_1 \\quad \\forall j \\in J", "description": "-"},
{"identifier": "B", "latex_formula": "x^T \\Sigma x \\le 1 \\text{ (Risiko)}", "description": "-"},
],
problem_class="Quadratic Programming (QP)",
target_stack="CVXPY",
)
assert "UNDECLARED_SYMBOL" not in codes(spec)
def test_missing_parameter_values_is_warning():
spec = make_spec(parameters=[{"symbol": "c_j", "dimension": "Vektor", "unit": "€", "description": "-"}])
assert "MISSING_DATA" in codes(spec)
def test_no_constraints_is_error():
assert "NO_CONSTRAINTS" in codes(make_spec(constraints=[]))
def test_indices_with_inline_math_delimiters_are_declared():
# beobachtet: "Mitarbeitende: \(p \in P\)" führte zu Fehlalarmen für p, P, d, D, Mo, Di …
spec = make_spec(
indices=["Mitarbeitende: \\(p \\in P = \\{A,B\\}\\)", "Tage: \\(d \\in D = \\{Mo,Di\\}\\) (Folgetag d+1)"],
variables=[{"name": "x_{p,d}", "var_type": "continuous", "domain": ">= 0", "description": "-"}],
objective={"sense": "minimize", "latex_formula": "\\min \\sum_{p \\in P} \\sum_{d \\in D} x_{p,d}", "description": "-"},
constraints=[{"identifier": "Folgetag", "latex_formula": "x_{p,Mo} + x_{p,Di} \\le 1 \\quad \\forall p \\in P", "description": "-"}],
)
assert "UNDECLARED_SYMBOL" not in codes(spec)
def test_miqp_with_scip_is_consistent():
spec = make_spec(
problem_class="Mixed-Integer Quadratic Programming (MIQP)", target_stack="SCIP (pyscipopt)",
variables=[{"name": "w_j", "var_type": "continuous", "domain": "[0,1]", "description": "-"},
{"name": "y_j", "var_type": "binary", "domain": "{0,1}", "description": "-"}],
objective={"sense": "maximize", "latex_formula": "\\max \\sum_j c_j w_j - \\sum_j c_j w_j^2 - b \\sum_j y_j", "description": "-"},
constraints=[{"identifier": "Kopplung", "latex_formula": "w_j \\le y_j", "description": "-"}],
)
assert check_spec(spec) == [] # insbesondere kein NONLINEAR_TERM, kein Routing-Fehler
def test_quadratic_with_binary_on_cvxpy_is_error():
spec = make_spec(
problem_class="Quadratic Programming (QP)", target_stack="CVXPY",
variables=[{"name": "y_j", "var_type": "binary", "domain": "{0,1}", "description": "-"}],
)
assert "CVXPY_WITH_DISCRETE_VARS" in codes(spec)
assert has_errors(check_spec(spec))
def test_miqp_without_discrete_vars_is_warning():
spec = make_spec(problem_class="Mixed-Integer Quadratic Programming (MIQP)", target_stack="SCIP (pyscipopt)")
assert "MIQP_WITHOUT_DISCRETE_VARS" in codes(spec)
assert not has_errors(check_spec(spec))
def test_every_problem_class_has_allowed_stacks():
from checks.model_checks import ALLOWED_STACKS
from core.model_schema import ProblemClass
assert set(ALLOWED_STACKS) == set(ProblemClass)
def test_nlp_with_ipopt_is_consistent():
spec = make_spec(problem_class="Nonlinear Programming (NLP)", target_stack="Ipopt (cyipopt)")
assert not has_errors(check_spec(spec))
def test_ipopt_with_integer_vars_is_error():
spec = make_spec(problem_class="Nonlinear Programming (NLP)", target_stack="Ipopt (cyipopt)",
variables=[{"name": "n_j", "var_type": "integer", "domain": ">= 0", "description": "-"}])
assert "IPOPT_WITH_DISCRETE_VARS" in codes(spec)
# --- Einheiten der Zielfunktion (ParameterDefinition.unit, ohne LLM) ---
def portfolio_spec(risk_unit: str, objective: str):
"""Portfolio: Rendite und Kosten in Euro, Risiko wahlweise dimensionslos oder in Euro."""
return make_spec(
problem_class="Quadratic Programming (QP)", target_stack="CVXPY",
objective={"sense": "maximize", "latex_formula": objective, "description": "-"},
parameters=[
{"symbol": "\\mu_i", "dimension": "Vektor", "unit": "€", "description": "Ertrag", "extracted_values": "[1]"},
{"symbol": "\\lambda", "dimension": "Skalar", "unit": "dimensionslos", "description": "Risikoaversion",
"extracted_values": "0.5"},
{"symbol": "\\Sigma", "dimension": "Matrix", "unit": risk_unit, "description": "Kovarianz",
"extracted_values": "[[1]]"},
],
variables=[{"name": "w_i", "var_type": "continuous", "domain": "[0, 1]", "description": "Anteil"}],
constraints=[{"identifier": "Budget", "latex_formula": "\\sum_i w_i \\le 1", "description": "-"}],
)
def test_dimensionless_risk_next_to_euro_amounts_is_error():
# beobachtet: Risiko als Summe über Anteile ist um den Faktor Kapital zu klein und damit wirkungslos
spec = portfolio_spec("dimensionslos", "\\max \\sum_i \\mu_i w_i - \\lambda w^T \\Sigma w")
assert "MIXED_OBJECTIVE_UNITS" in codes(spec)
assert has_errors(check_spec(spec))
def test_risk_scaled_to_euro_is_accepted():
spec = portfolio_spec("€", "\\max \\sum_i \\mu_i w_i - \\lambda w^T \\Sigma w")
assert "MIXED_OBJECTIVE_UNITS" not in codes(spec)
def test_single_term_objective_is_never_mixed():
spec = portfolio_spec("dimensionslos", "\\min w^T \\Sigma w")
assert "MIXED_OBJECTIVE_UNITS" not in codes(spec)
def test_index_range_is_not_split_into_terms():
# \sum_{i \in I} enthält kein Minus, wohl aber Klammern – der Term darf nicht zerschnitten werden
from checks.model_checks import _additive_terms
assert len(_additive_terms("\\min \\sum_{j \\in J} c_j x_j")) == 1
assert len(_additive_terms("\\max \\sum_i \\mu_i w_i - \\lambda w^T \\Sigma w")) == 2
assert len(_additive_terms("\\min z = a x + b y")) == 2 # linke Seite zählt nicht mit
def test_unknown_units_produce_no_finding():
# Stunden neben Stück: kein Fehlalarm, geprüft werden nur Geld gegen dimensionslos
spec = make_spec(parameters=[
{"symbol": "c_j", "dimension": "Vektor", "unit": "h", "description": "Zeit", "extracted_values": "[3]"},
{"symbol": "b", "dimension": "Skalar", "unit": "Stück", "description": "Bedarf", "extracted_values": "4"},
], objective={"sense": "minimize", "latex_formula": "\\min \\sum_j c_j x_j + b", "description": "-"})
assert "MIXED_OBJECTIVE_UNITS" not in codes(spec)
def test_objective_label_is_not_an_undeclared_symbol():
# beobachtet: Zielfunktion "Z = 12\,x" meldete Z als undeklariertes Symbol
spec = make_spec(objective={"sense": "maximize", "latex_formula": "Z = 12\\,x_j", "description": "-"})
assert "UNDECLARED_SYMBOL" not in codes(spec)
def test_symbol_right_of_the_equals_sign_is_still_checked():
spec = make_spec(objective={"sense": "maximize", "latex_formula": "Z = q \\cdot x_j", "description": "-"})
assert "UNDECLARED_SYMBOL" in codes(spec)
def test_number_sets_are_not_symbols():
# beobachtet: Restriktion "x \in \mathbb{Z}" meldete Z als undeklariertes Symbol
spec = make_spec(constraints=[
{"identifier": "ganzzahligkeit", "latex_formula": "x_j \\in \\mathbb{Z}", "description": "-"},
{"identifier": "vorzeichen", "latex_formula": "x_j \\in \\mathbb{R}_{\\ge 0}", "description": "-"},
])
assert "UNDECLARED_SYMBOL" not in codes(spec)
def test_implicit_products_are_not_undeclared():
# beobachtet: "\sin(cx)" meinte c*x, der Check las 'cx' als eigenen Namen (ebenso 'xy')
spec = make_spec(
parameters=[{"symbol": "c", "dimension": "Skalar", "unit": "1/K", "description": "-", "extracted_values": "3"},
{"symbol": "q", "dimension": "Skalar", "unit": "dimensionslos", "description": "-", "extracted_values": "0.5"}],
variables=[{"name": "x", "var_type": "continuous", "domain": "[0, 4]", "description": "-"},
{"name": "y", "var_type": "continuous", "domain": "[0, 4]", "description": "-"}],
objective={"sense": "maximize", "latex_formula": "\\max x\\sin(cx) + y\\sin(cy) + q\\,xy", "description": "-"},
constraints=[{"identifier": "Sicherheit", "latex_formula": "x + y \\le 6", "description": "-"}],
)
assert "UNDECLARED_SYMBOL" not in codes(spec)
def test_real_names_are_still_reported():
# Gegenprobe: 'Kapazitaet' besteht nicht nur aus deklarierten Einzelsymbolen
spec = make_spec(constraints=[
{"identifier": "K1", "latex_formula": "x_j \\le Kapazitaet", "description": "-"},
])
assert "UNDECLARED_SYMBOL" in codes(spec)
# --- Fachbegriffe müssen erklärt sein (Block ERKLÄREN im Routing-Prompt) ---
def test_unexplained_jargon_is_reported():
# beobachtet: Restriktion "Flusserhaltung_Jahr1", Zusammenfassung erwähnt "Fluss" nirgends
spec = make_spec(
executive_summary="Der Bauer entscheidet, wann er die Maschine ersetzt.",
constraints=[{"identifier": "Flusserhaltung_Jahr1", "latex_formula": "\\sum_j x_j = 1", "description": "-"},
{"identifier": "Quelle_Startkauf", "latex_formula": "\\sum_j x_j = 1", "description": "-"}],
)
meldungen = [i.message for i in check_spec(spec) if i.code == "UNEXPLAINED_JARGON"]
assert meldungen and "Fluss" in meldungen[0]
assert not has_errors(check_spec(spec)) # Warnung, kein Fehler: sonst gälte ein richtiges Ergebnis als unzuverlässig
def test_explained_jargon_is_accepted():
spec = make_spec(
executive_summary="Ein Bogen i → j bedeutet Kauf am Ende von Jahr i. Die Flusserhaltung sorgt dafür, "
"dass die Kette nicht abreißt: zu jedem Knoten gehört genau ein Nachfolger.",
constraints=[{"identifier": "Flusserhaltung_Jahr1", "latex_formula": "\\sum_j x_j = 1", "description": "-"}],
indices=["Knoten j \\in J (Jahresenden)"],
)
assert "UNEXPLAINED_JARGON" not in codes(spec)
def test_everyday_words_are_not_jargon():
from checks.model_checks import unexplained_jargon
assert unexplained_jargon("Menge je Produkt und Lager, Kosten in Euro", "Eine schlichte Zusammenfassung.") == []
# --- Im Aufgabentext verlangtes Verfahren (check_requested_method) ---
DIJKSTRA_AUFGABE = "a) Formulieren Sie das Problem als Kürzeste-Wege-Problem!\nb) Lösen Sie es mit dem Dijkstra-Algorithmus!"
def test_requested_dijkstra_with_flow_model_is_error():
# beobachtet: Fall graph_ersatz_dijkstra wurde als MILP-Flussmodell geloest - richtig gerechnet, Aufgabe verfehlt
spec = make_spec(problem_class="Mixed-Integer Linear Programming (MILP)", target_stack="HiGHS",
variables=[{"name": "x_{ij}", "var_type": "binary", "domain": "{0,1}", "description": "-"}])
issues = check_spec(spec, DIJKSTRA_AUFGABE)
assert "REQUESTED_METHOD_IGNORED" in {i.code for i in issues}
assert has_errors(issues) # loest eine Korrekturrunde aus, die neu routen kann
def test_requested_dijkstra_with_networkx_is_accepted():
spec = make_spec(problem_class="Kürzeste Wege im Graphen (Shortest Path)", target_stack="NetworkX (Dijkstra)",
variables=[{"name": "x_{ij}", "var_type": "binary", "domain": "{0,1}", "description": "-"}])
assert "REQUESTED_METHOD_IGNORED" not in {i.code for i in check_spec(spec, DIJKSTRA_AUFGABE)}
def test_method_check_stays_quiet_without_a_named_method():
# ohne Aufgabentext und ohne genanntes Verfahren darf die Pruefung nichts melden
spec = make_spec()
assert "REQUESTED_METHOD_IGNORED" not in codes(spec)
assert "REQUESTED_METHOD_IGNORED" not in {i.code for i in check_spec(spec, "Minimiere die Kosten.")}
def test_every_checker_marker_in_the_prompts_exists():
"""Jede [→ X]-Kennung in den Prompts muss einem echten Check-Code oder einer Prüffunktion entsprechen.
Verhindert, dass eine Regel eine Prüfung behauptet, die es nicht (mehr) gibt.
"""
import re
from pathlib import Path
from checks import model_checks
from solving import solver_stage
quelle = Path(model_checks.__file__).read_text() + Path(solver_stage.__file__).read_text()
bekannt = set(re.findall(r'code="([A-Z_]+)"', quelle)) | set(re.findall(r"^def (\w+)", quelle, re.M))
marker = set()
projekt = Path(model_checks.__file__).parent.parent
for datei in projekt.glob("*/*-Prompt.txt"):
for treffer in re.findall(r"\[→ ([^\]]+)\]", datei.read_text(encoding="utf-8")):
marker |= {teil.strip() for teil in treffer.split(",") if teil.strip() != "…"} # Legende ausnehmen
assert marker, "keine Prüfer-Kennungen in den Prompts gefunden"
assert marker <= bekannt, f"Kennung ohne zugehörige Prüfung: {sorted(marker - bekannt)}"
# --- Verstehensstufe: Zahlen und Teilfragen des Aufgabentexts ---
def make_understanding(**overrides):
from core.model_schema import TaskUnderstanding
daten = {"is_optimization_task": True, "reason": "-", "facts": [], "ignored_numbers": [],
"sub_questions": [{"id": "1", "quote": "-", "kind": "rechnen", "needs_solver": True}]}
daten.update(overrides)
return TaskUnderstanding.model_validate(daten)
def test_numbers_are_normalised_once():
from checks.model_checks import normalize_number
assert normalize_number("1.234,56") == "1234.56" # deutsch: Punkt trennt Tausender
assert normalize_number("3.8") == "3.8" # Aufgabennummer, kein Tausendertrenner
assert normalize_number("1.234") == "1234"
assert normalize_number("0,05") == "0.05"
assert normalize_number("100.000") == "100000"
def test_numbers_inside_words_are_not_numbers():
from checks.model_checks import numbers_in
assert numbers_in("Aktie A02, Variable x1, Schicht 3") == {"3"}
def test_uncaptured_numbers_are_reported():
from checks.model_checks import check_understanding
# beobachtet: 0 von 14 Zahlen des Aufgabentexts landeten in der Spezifikation
aufgabe = "Der Deckungsbeitrag ist 12 € je Stück. Mindestens 100 Stück sind abzusetzen."
codes = {i.code for i in check_understanding(make_understanding(), aufgabe)}
assert "UNCAPTURED_NUMBERS" in codes
def test_numbers_quoted_in_a_fact_count_as_captured():
# eine Tabellenzeile als Zitat deckt alle Zahlen dieser Zeile ab
from checks.model_checks import check_understanding
aufgabe = "| A01 | 6,9 % | 0,68 % | 2,04 % |"
verstanden = make_understanding(facts=[{
"value": "6,9 %", "unit": "%", "meaning": "Rendite von A01", "quote": "| A01 | 6,9 % | 0,68 % | 2,04 % |"}])
assert "UNCAPTURED_NUMBERS" not in {i.code for i in check_understanding(verstanden, aufgabe)}
def test_explicitly_ignored_numbers_count_as_captured():
from checks.model_checks import check_understanding
aufgabe = "## Aufgabe 2.7: Der Preis ist 8 €."
verstanden = make_understanding(
facts=[{"value": "8", "unit": "€", "meaning": "Preis", "quote": "Der Preis ist 8 €."}],
ignored_numbers=[{"value": "2.7", "reason": "Aufgabennummer"}])
assert "UNCAPTURED_NUMBERS" not in {i.code for i in check_understanding(verstanden, aufgabe)}
def test_missing_sub_questions_are_reported():
from checks.model_checks import check_understanding
aufgabe = "- a) Formulieren Sie das Problem!\n- b) Lösen Sie es mit Dijkstra!"
verstanden = make_understanding(sub_questions=[{"id": "a", "quote": "-", "kind": "formulieren", "needs_solver": False}])
meldungen = [i.message for i in check_understanding(verstanden, aufgabe) if i.code == "MISSING_SUB_QUESTIONS"]
assert meldungen and "b" in meldungen[0]
def test_complete_sub_questions_are_accepted():
from checks.model_checks import check_understanding
aufgabe = "- a) Formulieren Sie das Problem!\n- b) Lösen Sie es mit Dijkstra!"
verstanden = make_understanding(sub_questions=[
{"id": "a", "quote": "-", "kind": "formulieren", "needs_solver": False},
{"id": "b", "quote": "-", "kind": "rechnen", "needs_solver": True}])
assert "MISSING_SUB_QUESTIONS" not in {i.code for i in check_understanding(verstanden, aufgabe)}
def test_split_sub_questions_cover_their_parent():
# beobachtet an "Portfolio Nr. 2": Teil 2 als 2a/2b erfasst, gemeldet wurde "es fehlen: 2"
from checks.model_checks import check_understanding
aufgabe = "* **Teil 1: Daten**\n* **Teil 2: Optimierung**\n* **Teil 3: Evaluation**"
verstanden = make_understanding(sub_questions=[
{"id": i, "quote": "-", "kind": "rechnen", "needs_solver": True} for i in ("Teil 1", "Teil 2a", "Teil 2b", "Teil 3")])
assert "MISSING_SUB_QUESTIONS" not in {i.code for i in check_understanding(verstanden, aufgabe)}
# "21" ist eine andere Teilfrage, keine Zerlegung von "2"
verstanden = make_understanding(sub_questions=[
{"id": i, "quote": "-", "kind": "rechnen", "needs_solver": True} for i in ("1", "21", "3")])
assert "MISSING_SUB_QUESTIONS" in {i.code for i in check_understanding(verstanden, aufgabe)}
def test_captured_facts_must_reach_the_specification():
# beobachtet: Verstehensstufe erfasste die ganze Kostenmatrix, extracted_values blieb None
from checks.model_checks import check_data_transfer
verstanden = make_understanding(facts=[
{"value": "8", "unit": "€", "meaning": "Kosten 0->1", "quote": "| 0 | 8 | 18 | 31 |"}])
ohne = make_spec(parameters=[{"symbol": "c_{ij}", "dimension": "Matrix", "unit": "€", "description": "Kosten"}])
meldungen = [i.message for i in check_data_transfer(ohne, verstanden)]
assert meldungen and "8" in meldungen[0]
mit = make_spec(parameters=[{"symbol": "c_{ij}", "dimension": "Matrix", "unit": "€", "description": "Kosten",
"extracted_values": "8, 18, 31"}])
assert check_data_transfer(mit, verstanden) == []
def test_data_transfer_tolerates_partially_transferred_values():
# Indexwerte und Zeilenbeschriftungen muessen nicht in extracted_values stehen - kein Fehlalarm
from checks.model_checks import check_data_transfer
verstanden = make_understanding(facts=[
{"value": "0,1,2", "unit": "Jahr", "meaning": "moegliche Kaufzeitpunkte", "quote": "i = 0,1,2"},
{"value": "8", "unit": "€", "meaning": "Kosten", "quote": "| 0 | 8 |"}])
spec = make_spec(parameters=[{"symbol": "c", "dimension": "Matrix", "unit": "€", "description": "Kosten",
"extracted_values": "8"}])
assert check_data_transfer(spec, verstanden) == []
def test_data_transfer_check_is_quiet_without_understanding():
from checks.model_checks import check_data_transfer
assert check_data_transfer(make_spec(), None) == []
def test_sum_limits_are_not_an_objective_label():
# beobachtet: "\sum_{i=0}^{2}..." wurde am "=" der Summengrenze zerschnitten -> NONLINEAR_TERM
spec = make_spec(
problem_class="Kürzeste Wege im Graphen (Shortest Path)", target_stack="NetworkX (Dijkstra)",
variables=[{"name": "x_{i,j}", "var_type": "binary", "domain": "{0,1}", "description": "-"}],
parameters=[{"symbol": "c_{i,j}", "dimension": "Matrix", "unit": "€", "description": "Kosten",
"extracted_values": "8"}],
objective={"sense": "minimize", "description": "-",
"latex_formula": "\\sum_{i=0}^{2}\\sum_{j=i+1}^{3} c_{i,j}\\, x_{i,j}"},
constraints=[{"identifier": "Weg", "latex_formula": "\\sum_j x_{0,j} = 1", "description": "-"}],
)
assert "NONLINEAR_TERM" not in codes(spec)
def test_objective_label_is_still_stripped():
from checks.model_checks import _objective_without_label
assert _objective_without_label("Z = 12 x").strip() == "12 x"
assert _objective_without_label("\\sum_{i=0}^{2} c_i x_i").startswith("\\sum")

545
tests/test_pipeline.py Normal file
View file

@ -0,0 +1,545 @@
"""run_pipeline ohne LLM: Neustart bei Fremdproblem und Kennzeichnung unzuverlässiger Ergebnisse."""
from benchmark.benchmark import result_correct
from core.model_schema import (
AssumptionReview,
LLMUsage,
OptimizationModelSpecification,
SolverCode,
SolverStatus,
SubQuestion,
TaskUnderstanding,
)
from TextToModel import run_pipeline
from test_model_checks import make_spec
from test_review import make_review
from test_script_export import STRUCTURED_CODE
# Standardverstaendnis: eine Frage, keine Zahlen - reicht fuer Tests, die das Verstehen nicht pruefen
EINFACHES_VERSTAENDNIS = TaskUnderstanding(
is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id="1", quote="-", kind="rechnen", needs_solver=True)])
class FakeSession:
"""Liefert vorgegebene Antworten je Antwort-Typ in fester Reihenfolge."""
def __init__(self, specs, reviews, codes=(), understandings=None):
self.queues = {
OptimizationModelSpecification: list(specs),
AssumptionReview: list(reviews),
SolverCode: list(codes),
TaskUnderstanding: list(understandings if understandings is not None else []),
}
self.usage = LLMUsage(model="fake", provider="fake")
self.calls = {kind: 0 for kind in self.queues}
self.steps = []
self.tasks = []
self.letzter_prompt = ""
def overrides(self):
return {}
def create(self, response_model, messages, step="unbenannt"):
self.letzter_prompt = messages[-1]["content"]
if step == "Extraktion":
self.tasks.append(messages[-1]["content"])
self.calls[response_model] += 1
self.usage.calls += 1
self.steps.append(step)
warteschlange = self.queues[response_model]
if response_model is TaskUnderstanding and not warteschlange:
return EINFACHES_VERSTAENDNIS # Standard, solange ein Test nichts anderes vorgibt
return warteschlange.pop(0)
HALLUCINATED = make_spec(executive_summary="Produktionsplanung mit Drehen und Fräsen") # Fremdproblem
GOOD = make_spec(executive_summary="Rohstoffeinkauf mit minimalen Kosten")
MISMATCH = make_review(matches=False)
OK = make_review()
def test_clean_first_attempt_needs_no_restart():
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=1)
assert result.reliable and result.restarts == 0 and result.revisions == 0
assert result.solver_run.status == SolverStatus.OPTIMAL
def test_restart_recovers_from_hallucinated_problem():
# Versuch 1: Fremdproblem, beide Korrekturrunden erfolglos; Neustart liefert das richtige Modell
session = FakeSession(
specs=[HALLUCINATED, HALLUCINATED, HALLUCINATED, GOOD],
reviews=[MISMATCH, MISMATCH, MISMATCH, OK],
codes=[SolverCode(code=STRUCTURED_CODE)],
)
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=1)
assert result.restarts == 1
assert result.reliable and result.reliability_issues == []
assert result.specification.executive_summary == GOOD.executive_summary
assert result.solver_run.status == SolverStatus.OPTIMAL and result.solver_run.objective_value == 9
def test_persistent_mismatch_is_unreliable_and_not_solved():
# beobachtet im Benchmark: Fremdproblem wurde als "optimal, 2550" ausgegeben
session = FakeSession(specs=[HALLUCINATED] * 6, reviews=[MISMATCH] * 6)
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=1)
assert result.restarts == 1
assert not result.reliable
assert any("Zielfunktion passt nicht" in issue for issue in result.reliability_issues)
assert result.solver_run is None and result.solver_code is None
assert session.calls[SolverCode] == 0 # kein Solver-Code für ein falsches Problem
assert not result_correct(result.solver_run, {"objective_value": 9, "tolerance": 1e-6})
def test_better_earlier_attempt_wins_over_worse_restart():
# Neustart ist schlechter (Zielgröße falsch) als Versuch 1 (nur Check-Error) -> Versuch 1 verwenden
stack_error = make_spec(executive_summary="richtig, aber Stack falsch", problem_class="Quadratic Programming (QP)")
session = FakeSession(
specs=[stack_error, stack_error, stack_error, HALLUCINATED, HALLUCINATED, HALLUCINATED],
reviews=[OK, OK, OK, MISMATCH, MISMATCH, MISMATCH],
codes=[SolverCode(code=STRUCTURED_CODE)],
)
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=1)
assert result.specification.executive_summary == "richtig, aber Stack falsch"
assert not result.reliable # STACK_MISMATCH bleibt bestehen
assert any("STACK_MISMATCH" in issue for issue in result.reliability_issues)
assert result.solver_run is not None # Zielgröße passt -> Solver läuft, Ergebnis aber als unzuverlässig markiert
def test_no_restart_when_disabled():
session = FakeSession(specs=[HALLUCINATED] * 3, reviews=[MISMATCH] * 3)
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=0)
assert result.restarts == 0 and not result.reliable
def test_report_shows_prominent_warning_for_unreliable_result():
from reporting.report import build_document, to_html, to_markdown
session = FakeSession(specs=[HALLUCINATED] * 3, reviews=[MISMATCH] * 3)
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=0)
doc = build_document(result)
assert doc[1][0] == "warn" # direkt unter der Überschrift
assert '<div class="warn">⚠ Ergebnis unzuverlässig' in to_html(doc)
assert "> **⚠ Ergebnis unzuverlässig" in to_markdown(doc)
assert "Der Solver wurde nicht gestartet." in doc[1][1]
def test_report_shows_the_rejected_reading_of_each_assumption():
from reporting.report import build_document, to_markdown
spec = make_spec(open_assumptions=[{
"unclear_point": "„aufeinanderfolgend“",
"assumption": "Kalendertage, auch über das Wochenende",
"alternative": "Nur Werktage; verworfen, weil der Text durchgehenden Betrieb nennt",
"impact": "Restriktion Ruhezeit greift auch am Wochenende",
}])
session = FakeSession([spec], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, solve_model=False)
markdown = to_markdown(build_document(result))
assert "Verworfene Lesart" in markdown
assert "verworfen, weil der Text durchgehenden Betrieb nennt" in markdown
def test_pipeline_labels_every_llm_call_for_the_timing_breakdown():
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)])
run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=1)
assert session.steps == ["Verstehen", "Extraktion", "Prüfung", "Solver-Code"]
def test_revision_and_repair_steps_are_labelled():
session = FakeSession(specs=[HALLUCINATED, GOOD], reviews=[MISMATCH, OK],
codes=[SolverCode(code="x = 1"), SolverCode(code=STRUCTURED_CODE)])
run_pipeline("Aufgabe", session=session, max_revisions=1, max_restarts=0)
assert session.steps == ["Verstehen", "Extraktion", "Prüfung", "Korrektur", "Prüfung", "Solver-Code", "Solver-Reparatur"]
# --- Zulässigkeitsprüfung im Zusammenspiel ---
VIOLATING_CODE = '''
def solve_model():
"""Liefert eine Lösung, die das Budget um 8 Cent überschreitet (beobachteter Fall)."""
return 100000.08
def main():
"""Meldet die Lösung samt Restriktionen zur Nachprüfung."""
invested = solve_model()
report("optimal", 12041.74, {"x": invested},
constraints=[{"name": "Bedarf", "lhs": invested, "sense": "<=", "rhs": 100000.0}],
variables=[{"name": "x", "value": invested, "var_type": "continuous", "lb": 0}])
if __name__ == "__main__":
main()
'''
def test_infeasible_solution_triggers_repair_and_is_marked_unreliable():
session = FakeSession([GOOD], [OK], codes=[SolverCode(code=VIOLATING_CODE)] * 3)
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0, solver_retries=2)
assert session.calls[SolverCode] == 3 # Erstversuch + 2 Reparaturen
assert result.solver_run.feasibility and "Bedarf" in result.solver_run.feasibility[0]
assert not result.reliable
assert any("Zulässigkeitsprüfung" in issue for issue in result.reliability_issues)
def test_feasible_solution_stays_reliable():
session = FakeSession([GOOD], [OK], codes=[SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert result.solver_run.feasibility == []
assert result.reliable and session.calls[SolverCode] == 1
# --- Gegenprobe bei "unlösbar" ---
INFEASIBLE_CODE = '''
def main():
"""Meldet das Modell als unlösbar."""
report("infeasible", None, {})
if __name__ == "__main__":
main()
'''
def test_infeasible_is_rechecked_and_can_be_refuted():
# beobachtet bei qwen3-235b: "unlösbar" war ein Modellierungsfehler, die Aufgabe ist lösbar
session = FakeSession(specs=[GOOD, GOOD], reviews=[OK, OK],
codes=[SolverCode(code=INFEASIBLE_CODE), SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert result.solver_run.status == SolverStatus.OPTIMAL and result.solver_run.objective_value == 9
assert not result.infeasible_confirmed and result.restarts == 1
assert "HINWEIS" in session.tasks[1] and "unabhängige Modellierung" in session.tasks[1]
def test_infeasible_confirmed_by_second_modelling():
session = FakeSession(specs=[GOOD, GOOD], reviews=[OK, OK],
codes=[SolverCode(code=INFEASIBLE_CODE)] * 2)
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert result.solver_run.status == SolverStatus.INFEASIBLE
assert result.infeasible_confirmed and result.reliable
def test_infeasible_without_recheck_is_flagged(monkeypatch):
import TextToModel
monkeypatch.setattr(TextToModel, "INFEASIBLE_RECHECK", 0)
session = FakeSession([GOOD], [OK], codes=[SolverCode(code=INFEASIBLE_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert not result.reliable and any("ohne Gegenprobe" in i for i in result.reliability_issues)
# --- Wiedereinstieg ab einer Stufe (continue_pipeline / CLI --von) ---
def stored_result(tmp_path, spec=GOOD, review=OK):
"""Ergebnis eines früheren Laufs als Datei – Grundlage für --von."""
session = FakeSession([spec], [review], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Rohstoffeinkauf", session=session, max_revisions=0, max_restarts=0, solve_model=False)
path = tmp_path / "ergebnis.json"
path.write_text(result.model_dump_json(indent=2), encoding="utf-8")
return path
def test_continue_from_solver_skips_extraction_and_review(tmp_path):
from TextToModel import continue_pipeline
path = stored_result(tmp_path)
session = FakeSession([], [], [SolverCode(code=STRUCTURED_CODE)])
result = continue_pipeline(str(path), "solver", session=session)
assert session.steps == ["Solver-Code"] # kein Extraktions-, kein Prüfaufruf
assert result.specification.executive_summary == GOOD.executive_summary
assert result.task == "Rohstoffeinkauf" # Aufgabentext kommt aus dem Ergebnis
assert result.solver_run.status == SolverStatus.OPTIMAL
def test_continue_from_review_keeps_specification_but_checks_again(tmp_path):
from TextToModel import continue_pipeline
path = stored_result(tmp_path)
session = FakeSession([], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = continue_pipeline(str(path), "pruefung", session=session)
assert session.steps == ["Prüfung", "Solver-Code"] # Extraktion entfällt
assert result.specification.executive_summary == GOOD.executive_summary
def test_continue_from_review_can_still_revise(tmp_path):
from TextToModel import continue_pipeline
path = stored_result(tmp_path)
session = FakeSession([GOOD], [MISMATCH, OK], [SolverCode(code=STRUCTURED_CODE)])
result = continue_pipeline(str(path), "pruefung", session=session, max_revisions=1)
assert session.steps == ["Prüfung", "Korrektur", "Prüfung", "Solver-Code"]
assert result.revisions == 1
def test_continue_from_solver_refuses_when_objective_does_not_match(tmp_path):
# gespeicherte Zielgrößen-Abweichung bleibt bindend: kein Solver, kein LLM-Aufruf
from TextToModel import continue_pipeline
path = stored_result(tmp_path, review=MISMATCH)
session = FakeSession([], [], [])
result = continue_pipeline(str(path), "solver", session=session)
assert session.steps == []
assert result.solver_run is None and not result.reliable
def test_unknown_stage_is_rejected(tmp_path):
import pytest
from TextToModel import continue_pipeline
path = stored_result(tmp_path)
with pytest.raises(ValueError, match="Unbekannte Stufe"):
continue_pipeline(str(path), "extraktion", session=FakeSession([], [], []))
def test_warnings_reach_the_revision_prompt():
# Warnungen loesen keine Runde aus, muessen aber mitkorrigiert werden, wenn ohnehin eine stattfindet
from core.model_schema import CheckIssue, CheckSeverity
from modeling.extraction import revise_model
session = FakeSession([GOOD], [], [])
revise_model(session, "Aufgabe", GOOD, OK, issues=[
CheckIssue(severity=CheckSeverity.WARNING, code="UNEXPLAINED_JARGON", message="Fachbegriffe: Fluss"),
CheckIssue(severity=CheckSeverity.ERROR, code="STACK_MISMATCH", message="passt nicht"),
])
prompt = session.letzter_prompt
assert "STACK_MISMATCH" in prompt.split("HINWEISE")[0] # Fehler im Fehlerteil
assert "UNEXPLAINED_JARGON" in prompt.split("HINWEISE")[1] # Warnung im Hinweisteil
# --- Benchmark haelt fest, womit gemessen wurde ---
def test_benchmark_records_the_prompt_state():
from benchmark.benchmark import PROMPT_FILES, prompt_state
stand = prompt_state()
assert {"prompts", "commit", "dirty"} <= set(stand)
assert len(stand["prompts"]) == 12 and stand["commit"]
assert "modeling/Routing-und-Extraktions-Prompt.txt" in PROMPT_FILES
assert "core/model_schema.py" in PROMPT_FILES # Feldbeschreibungen wirken wie Prompt-Anweisungen
def test_deviating_settings_are_recorded(monkeypatch):
# Ohne die Umgebung liesse sich ein A/B-Lauf (Stufe abgeschaltet, anderes Modell) nicht zuordnen
from benchmark import benchmark
benchmark.prompt_state.cache_clear()
monkeypatch.setenv("UNDERSTAND_RETRIES", "-1")
monkeypatch.setenv("LLM_MODEL_SOLVER", "billig/modell")
stand = benchmark.prompt_state()
assert stand["config"] == {"UNDERSTAND_RETRIES": "-1", "LLM_MODEL_SOLVER": "billig/modell"}
benchmark.prompt_state.cache_clear()
def test_prompt_digest_changes_with_the_prompt(tmp_path):
from benchmark.benchmark import prompt_digest
a, b = tmp_path / "eins.txt", tmp_path / "zwei.txt"
a.write_text("Regel A"), b.write_text("Regel B")
vorher = prompt_digest([a, b])
assert prompt_digest([a, b]) == vorher # gleicher Inhalt, gleiche Kennung
b.write_text("Regel B, nun schaerfer")
assert prompt_digest([a, b]) != vorher # eine geaenderte Regel aendert die Kennung
assert prompt_digest([b, a]) != prompt_digest([a, b]) # Reihenfolge ist Teil der Kennung
def test_unexplained_jargon_triggers_a_revision_but_stays_reliable():
# Erklaeren ist Kernziel: die Warnung ist eine Runde wert, macht das Ergebnis aber nicht unzuverlaessig
from checks.model_checks import check_spec
unerklaert = make_spec(
executive_summary="Der Betrieb entscheidet ueber Mengen.",
constraints=[{"identifier": "Flusserhaltung", "latex_formula": "\\sum_j x_j = 1", "description": "-"}])
assert "UNEXPLAINED_JARGON" in {i.code for i in check_spec(unerklaert)}
session = FakeSession(specs=[unerklaert, GOOD], reviews=[OK, OK], codes=[SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=1, max_restarts=0)
assert session.steps == ["Verstehen", "Extraktion", "Prüfung", "Korrektur", "Prüfung", "Solver-Code"]
assert result.revisions == 1
assert result.reliable # Warnung, kein Fehler
def test_clean_model_still_needs_no_revision():
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=2, max_restarts=0)
assert result.revisions == 0
# --- Verstehensstufe im Ablauf ---
def test_understanding_runs_first_and_is_kept_in_the_result():
from core.model_schema import Fact
verstanden = TaskUnderstanding(
is_optimization_task=True, reason="Kostenminimierung", ignored_numbers=[],
sub_questions=[SubQuestion(id="a", quote="Wie viel?", kind="rechnen", needs_solver=True)],
facts=[Fact(value="12", unit="€", meaning="Deckungsbeitrag je Stueck", quote="12 € je Stueck")])
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)], understandings=[verstanden])
result = run_pipeline("Der Deckungsbeitrag ist 12 € je Stueck.", session=session,
max_revisions=0, max_restarts=0)
assert session.steps[0] == "Verstehen"
assert result.understanding.facts[0].value == "12"
assert [q.id for q in result.understanding.sub_questions] == ["a"]
def test_incomplete_understanding_is_asked_again():
# erste Antwort uebersieht die Zahl, zweite erfasst sie
from core.model_schema import Fact
luecke = TaskUnderstanding(is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id="1", quote="-", kind="rechnen", needs_solver=True)])
vollstaendig = luecke.model_copy(update={
"facts": [Fact(value="12", unit="€", meaning="Deckungsbeitrag", quote="12 € je Stueck")]})
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)],
understandings=[luecke, vollstaendig])
result = run_pipeline("12 € je Stueck.", session=session, max_revisions=0, max_restarts=0)
assert session.steps[:2] == ["Verstehen", "Verstehen"] # genau ein Nachfassen
assert result.understanding.facts[0].value == "12"
def test_report_shows_questions_and_numbers():
from core.model_schema import Fact
from reporting.report import build_document, to_markdown
verstanden = TaskUnderstanding(
is_optimization_task=True, reason="-",
sub_questions=[SubQuestion(id="a", quote="Formulieren Sie das Problem!", kind="formulieren", needs_solver=False),
SubQuestion(id="b", quote="Loesen Sie es!", kind="rechnen", needs_solver=True)],
facts=[Fact(value="8", unit="€", meaning="Kosten im ersten Jahr", quote="Zeile 0: 8 | 18 | 31")],
ignored_numbers=[])
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)], understandings=[verstanden])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
markdown = to_markdown(build_document(result))
assert "Gestellte Fragen" in markdown and "Formulieren Sie das Problem!" in markdown
assert "Gegebene Zahlen" in markdown and "Kosten im ersten Jahr" in markdown
def test_report_puts_the_answers_before_the_numbers():
from reporting.report import build_document, to_markdown
code = '''
def main():
"""Loest und beantwortet."""
report("optimal", 9.0, {"x": 1.0},
headline="Die Kosten betragen 9 Geldeinheiten.",
constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}],
answers=[{"question_id": "a", "text": "Das Modell ist ein lineares Programm."},
{"question_id": "b", "text": "Die minimalen Kosten betragen 9 Geldeinheiten."}])
if __name__ == "__main__":
main()
'''
verstanden = TaskUnderstanding(
is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id="a", quote="Formulieren Sie das Modell!", kind="formulieren", needs_solver=False),
SubQuestion(id="b", quote="Berechnen Sie die Kosten!", kind="rechnen", needs_solver=True)])
session = FakeSession([GOOD], [OK], [SolverCode(code=code)], understandings=[verstanden])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
markdown = to_markdown(build_document(result))
assert "Antworten auf die Aufgabe" in markdown
assert markdown.index("Formulieren Sie das Modell!") < markdown.index("| Status |")
assert result.solver_run.warnings == [] # Abnahme ohne Befund
def test_unanswered_sub_question_is_a_warning_not_a_wrong_number():
code = '''
def main():
"""Beantwortet nur die zweite Frage."""
report("optimal", 9.0, {"x": 1.0},
constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}],
answers=[{"question_id": "b", "text": "Die minimalen Kosten betragen 9 Geldeinheiten."}])
if __name__ == "__main__":
main()
'''
verstanden = TaskUnderstanding(
is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id="a", quote="Zeichnen Sie den Graphen!", kind="zeichnen", needs_solver=False),
SubQuestion(id="b", quote="Berechnen Sie die Kosten!", kind="rechnen", needs_solver=True)])
session = FakeSession([GOOD], [OK], [SolverCode(code=code)] * 3, understandings=[verstanden])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
assert any("Teilfragen ohne Antwort" in w for w in result.solver_run.warnings)
assert result.solver_run.objective_value == 9.0 # die Zahl bleibt richtig
def test_report_shows_data_profile_and_scenarios(tmp_path):
from core.model_schema import Scenario
from reporting.report import build_document, to_markdown
datei = tmp_path / "werte.csv"
datei.write_text("name,menge\nA,3\nB,4\n")
code = '''
DATA = {"faktor": 1.0}
def main():
"""Meldet zwei Datenvarianten."""
report("optimal", 7.0, {"x": 1.0},
constraints=[{"name": "Bedarf", "lhs": 4.0, "sense": ">=", "rhs": 4.0}],
variables=[{"name": "x", "value": 1.0, "var_type": "continuous", "lb": 0}],
scenarios=[{"name": "Doppelt", "description": "Faktor verdoppelt",
"data_override": {"faktor": 2.0}, "objective_value": 7.0}])
if __name__ == "__main__":
main()
'''
session = FakeSession([GOOD], [OK], [SolverCode(code=code)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0, data_files=[datei])
markdown = to_markdown(build_document(result))
assert "Beigelegte Daten" in markdown and "werte.csv (2 Zeilen)" in markdown
assert "Verglichene Konfigurationen" in markdown and "Doppelt" in markdown
assert result.data_files[0].rows == 2
assert result.solver_run.scenarios[0].verified_value == 7.0 # ohne LLM nachgerechnet
# --- Benchmark misst "Aufgabe erfuellt", nicht nur "Zahl richtig" ---
def test_benchmark_counts_unanswered_questions_as_not_fulfilled(monkeypatch):
from benchmark import benchmark
from core.model_schema import Answer, SubQuestion, TaskUnderstanding
def ergebnis(antworten, warnungen=()):
session = FakeSession([GOOD], [OK], [SolverCode(code=STRUCTURED_CODE)], understandings=[TaskUnderstanding(
is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id=f, quote="-", kind="rechnen", needs_solver=True) for f in ("a", "b")])])
r = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
r.solver_run.answers = [Answer(question_id=k, text="-") for k in antworten]
r.solver_run.warnings = list(warnungen)
return r
fall = {"name": "test", "task": "Aufgabe", "expected": {"objective_value": 9, "tolerance": 1e-6}}
for antworten, warnungen, erwartet in [
(["a", "b"], [], True), # alles beantwortet
(["b"], ["Teilfragen ohne Antwort: a"], False), # eine Frage offen
(["a", "b"], ["Antwort zu 'b' nennt Zahlen …"], False), # unbelegte Zahl
]:
monkeypatch.setattr(benchmark, "run_pipeline", lambda *a, **k: ergebnis(antworten, warnungen))
zeile = benchmark.run_case(fall, "fake", 1)
assert zeile["objective_ok"] is True # die Zahl stimmt immer
assert zeile["task_ok"] is erwartet, (antworten, warnungen)
def test_benchmark_case_data_files_are_collected(tmp_path, monkeypatch):
from benchmark import benchmark
fall = tmp_path / "cases" / "mit_daten"
fall.mkdir(parents=True)
(fall / "task.txt").write_text("Aufgabe")
(fall / "expected.json").write_text('{"objective_value": 1, "tolerance": 1e-6}')
(fall / "werte.csv").write_text("a,b\n1,2\n")
monkeypatch.setattr(benchmark, "CASES_DIR", tmp_path / "cases")
faelle = benchmark.load_cases()
assert [p.split("/")[-1] for p in faelle[0]["data_files"]] == ["werte.csv"]
# Spezifikation ohne Kostendaten, Programm erfindet sie per Zufallsgenerator (Muster "Portfolio Nr. 2")
OHNE_DATEN = make_spec(parameters=[
{"symbol": "c_j", "dimension": "Vektor", "unit": "€/t", "description": "Kosten", "extracted_values": ""},
{"symbol": "b", "dimension": "Skalar", "unit": "t", "description": "Bedarf", "extracted_values": "4"}])
ERFUNDENE_DATEN = STRUCTURED_CODE.replace(
"COSTS = [3, 2]", "import numpy as np\nCOSTS = [float(v) for v in np.random.default_rng(0).integers(2, 4, 2)]")
def test_undeclared_generated_data_is_repaired():
session = FakeSession([OHNE_DATEN], [OK], [
SolverCode(code=ERFUNDENE_DATEN),
SolverCode(code=ERFUNDENE_DATEN, data_assumptions=["c_j: zufällig erzeugt (Seed 0), keine Daten im Text"])])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0, solver_retries=1)
assert session.calls[SolverCode] == 2 # der Befund hat eine Reparatur ausgelöst
assert result.reliable and result.solver_run.undeclared_data == []
def test_persistently_undeclared_data_is_unreliable():
# Nachmessung "Portfolio Nr. 2": data_assumptions leer, Antwort behauptete geladene Kursdaten
session = FakeSession([OHNE_DATEN], [OK], [SolverCode(code=ERFUNDENE_DATEN)] * 2)
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0, solver_retries=1)
assert not result.reliable
assert any(p.startswith("Datenherkunft") for p in result.reliability_issues)

54
tests/test_report_math.py Normal file
View file

@ -0,0 +1,54 @@
import pytest
from reporting.report import Math, _html_inline, _md_inline, split_math
def marked(text: str) -> str:
"""Formelteile mit «…» markieren – kompakt vergleichbar."""
return "".join(f"«{part}»" if is_math else part for is_math, part in split_math(text))
@pytest.mark.parametrize("text, expected", [
# ohne Begrenzer (häufigster Fall in LLM-Texten, beobachtet in gespeicherten Läufen)
("U_i = 0.40 für alle i \\in I", "«U_i = 0.40» für alle «i \\in I»"),
("R_{\\min} = 0.08", "«R_{\\min} = 0.08»"),
("r_A=0.07, r_B=0.10", "«r_A=0.07, r_B=0.10»"),
("lautet \\sqrt{\\mathbf{w}^\\top \\Sigma \\mathbf{w}}. Das ist konvex.",
"lautet «\\sqrt{\\mathbf{w}^\\top \\Sigma \\mathbf{w}}». Das ist konvex."),
("j \\in I = \\{A,B,C,D\\} (für Kovarianzterme)", "«j \\in I = \\{A,B,C,D\\}» (für Kovarianzterme)"),
("x_{p,d,s} = 1, wenn Person p arbeitet", "«x_{p,d,s} = 1», wenn Person p arbeitet"),
# Mehrbuchstaben-Index klammern, % maskieren
("Mindestrendite r_min = 0.08 gilt", "Mindestrendite «r_{min} = 0.08» gilt"),
("bei w_A = 23.21%.", "bei «w_A = 23.21\\%»."),
# explizite Begrenzer
("Mitarbeitende: \\(p \\in P\\)", "Mitarbeitende: «p \\in P»"),
("Fonds $i \\in I$ gesamt", "Fonds «i \\in I» gesamt"),
("\\begin{pmatrix} 1 & 2 \\\\ 3 & 4 \\end{pmatrix}", "«\\begin{pmatrix} 1 & 2 \\\\ 3 & 4 \\end{pmatrix}»"),
])
def test_formulas_are_detected(text, expected):
assert marked(text) == expected
@pytest.mark.parametrize("text", [
"Erwartete Jahresrendite von Fonds i",
"Kapazitaet_M1 ist ausgelastet",
"WARNING [UNDECLARED_SYMBOL] Symbol 'h' fehlt",
"Kosten $5 bzw. $7",
"1, wenn Person p am Tag d in Schicht s arbeitet, sonst 0.",
"(F = Frühschicht, S = Spätschicht)",
"Second-Order Cone Programming (SOCP)",
])
def test_plain_text_stays_text(text):
assert split_math(text) == [(False, text)]
def test_html_free_text_wraps_formulas_and_escapes_text():
rendered = _html_inline("Anteil w_i <= 0.4 & mehr")
assert rendered == "Anteil \\(w_i &lt;= 0.4\\) &amp; mehr"
def test_markdown_table_cell_escapes_pipes_outside_and_inside_math():
rendered = _md_inline("Norm \\(\\|w\\|\\) | Rest", in_table=True)
assert rendered == "Norm $\\Vert w\\Vert $ \\| Rest"
assert _md_inline(Math("|x|"), in_table=True) == "$\\vert x\\vert $"
assert _md_inline(Math("x"), in_table=False) == "$x$"

98
tests/test_resolve.py Normal file
View file

@ -0,0 +1,98 @@
"""Punkt 1: dasselbe Programm mit geänderten Daten neu rechnen – ohne LLM-Aufruf."""
import pytest
from core.model_schema import SolverStatus
from solving.solver_stage import run_solver_code
# Programm nach der Vorgabe des Solver-Prompts: alle Zahlen stehen in DATA
DATA_CODE = '''
from scipy.optimize import linprog
DATA = {
"kosten": [3, 2], # Kosten je Tonne [Tsd. €]
"bedarf": 4, # Mindestmenge [t]
"obergrenze": 3, # je Rohstoff lieferbar [t]
}
def solve_model():
"""Baut das LP aus DATA auf und löst es."""
grenze = DATA["obergrenze"]
return linprog(DATA["kosten"], A_ub=[[-1, -1]], b_ub=[-DATA["bedarf"]],
bounds=[(0, grenze), (0, grenze)], method="highs")
def main():
"""Löst das Modell und meldet Ergebnis samt Restriktionen."""
res = solve_model()
x, y = float(res.x[0]), float(res.x[1])
report("optimal" if res.status == 0 else "error", float(res.fun), {"x": x, "y": y},
constraints=[{"name": "Bedarf", "lhs": x + y, "sense": ">=", "rhs": DATA["bedarf"]}],
variables=[{"name": "x", "value": x, "var_type": "continuous", "lb": 0, "ub": DATA["obergrenze"]}])
if __name__ == "__main__":
main()
'''
def test_run_without_override_uses_own_data():
run = run_solver_code(DATA_CODE)
assert run.status == SolverStatus.OPTIMAL
assert run.objective_value == pytest.approx(9) # 1t * 3 + 3t * 2
assert run.data["bedarf"] == 4
def test_changed_data_is_used_without_touching_the_code():
run = run_solver_code(DATA_CODE, data={"bedarf": 5})
assert run.status == SolverStatus.OPTIMAL
assert run.objective_value == pytest.approx(12) # 2t * 3 + 3t * 2
assert run.data["bedarf"] == 5 and run.data["kosten"] == [3, 2] # nur der geänderte Wert wird ersetzt
def test_constraints_follow_the_changed_data():
run = run_solver_code(DATA_CODE, data={"bedarf": 5})
assert run.constraints[0].rhs == 5
def test_infeasible_data_is_reported_not_crashing():
run = run_solver_code(DATA_CODE, data={"bedarf": 99})
assert run.status in (SolverStatus.ERROR, SolverStatus.INFEASIBLE)
# --- CLI-Modus: gespeichertes Ergebnis neu rechnen ---
def gespeichertes_ergebnis(tmp_path):
import json
from core.model_schema import PipelineResult, SolverCode
from test_model_checks import make_spec
from test_review import make_review
result = PipelineResult(task="Aufgabe", specification=make_spec(), review=make_review(), revisions=0,
solver_code=SolverCode(code=DATA_CODE))
pfad = tmp_path / "ergebnis.json"
pfad.write_text(result.model_dump_json(indent=2), encoding="utf-8")
return pfad
def test_resolve_without_data_reproduces_the_result(tmp_path):
from TextToModel import resolve_result
neu = resolve_result(str(gespeichertes_ergebnis(tmp_path)))
assert neu.solver_run.objective_value == pytest.approx(9)
assert neu.reliable and neu.usage is None # kein LLM-Aufruf
def test_resolve_with_changed_data(tmp_path):
import json
from TextToModel import resolve_result
daten = tmp_path / "daten.json"
daten.write_text(json.dumps({"bedarf": 5}), encoding="utf-8")
neu = resolve_result(str(gespeichertes_ergebnis(tmp_path)), str(daten))
assert neu.solver_run.objective_value == pytest.approx(12)
assert neu.solver_run.data["bedarf"] == 5
def test_resolve_keeps_specification_and_checks_feasibility(tmp_path):
from TextToModel import resolve_result
neu = resolve_result(str(gespeichertes_ergebnis(tmp_path)))
assert neu.specification.problem_class.value.startswith("Linear")
assert neu.solver_run.feasibility == []

207
tests/test_result_tables.py Normal file
View file

@ -0,0 +1,207 @@
"""Verständliche Ergebnistabellen: Ausgabe im Programm, Übernahme ins Ergebnis, Summenprüfung, Report."""
import pytest
from core.model_schema import PipelineResult, SolverRun, SolverStatus
from reporting.report import build_document, to_html
from solving.solver_stage import parse_result_tables, run_solver_code
from test_model_checks import make_spec
from test_review import make_review
# Portfolio-Ergebnis (hard_portfolio_kardinalitaet); Beiträge summieren sich zum Zielwert 0.09756
PORTFOLIO_TABLE = {
"title": "Portfolio",
"columns": ["Aktie", "Anteil", "Rendite", "Beitrag zum Ziel"],
"rows": [
["A02", "30,00 %", "14,4 %", 0.03433],
["A08", "30,00 %", "13,3 %", 0.03093],
["A03", "28,43 %", "11,5 %", 0.02587],
["A04", "6,20 %", "9,8 %", 0.00326],
["A09", "5,37 %", "9,8 %", 0.00317],
["Summe", "100 %", "", 0.09756],
],
"objective_column": "Beitrag zum Ziel",
"note": "Anteil = Kapitalanteil je gewählter Aktie.",
}
def test_table_is_printed_and_parsed_without_warnings():
code = f"report('optimal', 0.09756, {{'w[A02]': 0.3}}, tables=[{PORTFOLIO_TABLE!r}])"
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL
assert run.warnings == []
assert run.tables[0].title == "Portfolio" and len(run.tables[0].rows) == 6
def test_standalone_output_shows_box_table(tmp_path):
import subprocess, sys
from solving.solver_stage import assemble_script
path = tmp_path / "solve.py"
path.write_text(assemble_script(f"report('optimal', 0.09756, {{}}, tables=[{PORTFOLIO_TABLE!r}])"), encoding="utf-8")
out = subprocess.run([sys.executable, str(path)], capture_output=True, text=True, timeout=60).stdout
assert "┌" in out and "│ A02 │ 30,00 % │" in out and "Anteil = Kapitalanteil" in out
def test_sum_mismatch_is_warned():
tables, warnings = parse_result_tables([PORTFOLIO_TABLE], objective_value=0.2)
assert len(tables) == 1
assert any("weicht vom Zielwert" in w for w in warnings)
def test_inconsistent_sum_row_is_warned():
table = dict(PORTFOLIO_TABLE, rows=PORTFOLIO_TABLE["rows"][:-1] + [["Summe", "100 %", "", 0.5]])
_, warnings = parse_result_tables([table], objective_value=0.09756)
assert any("Summenzeile" in w for w in warnings)
@pytest.mark.parametrize("table, expected", [
(dict(PORTFOLIO_TABLE, objective_column="Gibt es nicht"), "existiert nicht"),
(dict(PORTFOLIO_TABLE, rows=[["A02", "30 %", "14,4 %", "0,034"]]), "nicht nur Zahlen"),
({"title": "kaputt"}, "ungültig"),
])
def test_faulty_tables_are_warnings_not_errors(table, expected):
_, warnings = parse_result_tables([table], objective_value=0.09756)
assert any(expected in w for w in warnings)
def test_non_additive_objective_without_objective_column_is_fine():
table = dict(PORTFOLIO_TABLE, objective_column=None)
_, warnings = parse_result_tables([table], objective_value=0.117) # z. B. Volatilität
assert warnings == []
def test_report_shows_tables_and_grouped_variables_with_meaning():
spec = make_spec(variables=[
{"name": "w_i", "var_type": "continuous", "domain": "[0,1]", "description": "Kapitalanteil in Aktie i"},
{"name": "y_i", "var_type": "binary", "domain": "{0,1}", "description": "1, falls Aktie i ausgewählt wird"},
])
tables, warnings = parse_result_tables([PORTFOLIO_TABLE], 0.09756)
result = PipelineResult(
task="Aufgabe", specification=spec, review=make_review(), revisions=0,
solver_run=SolverRun(status=SolverStatus.OPTIMAL, objective_value=0.09756, tables=tables, warnings=warnings,
solution={"w[A02]": 0.6, "y[A02]": 1.0, "w[A08]": 0.4, "y[A08]": 1.0}),
)
doc = build_document(result)
headings = [b[1] for b in doc if b[0] in ("h3", "h4")]
assert headings.index("Portfolio") < headings.index("Alle Variablenwerte (≠ 0)")
assert "w – Kapitalanteil in Aktie i" in headings and "y – 1, falls Aktie i ausgewählt wird" in headings
group_tables = [b for b in doc if b[0] == "table" and b[1] == ["Variable", "Wert"]]
assert [t[2][-1] for t in group_tables] == [["Summe", "1"], ["Summe", "2"]]
assert "Anteil = Kapitalanteil" in to_html(doc)
def test_report_shows_table_warning_prominently():
tables, warnings = parse_result_tables([PORTFOLIO_TABLE], 0.2)
result = PipelineResult(
task="Aufgabe", specification=make_spec(), review=make_review(), revisions=0,
solver_run=SolverRun(status=SolverStatus.OPTIMAL, objective_value=0.2, tables=tables, warnings=warnings),
)
assert '<div class="warn">⚠ Tabelle' in to_html(build_document(result))
DECOMPOSITION = { # beobachtet: Summenzeile heißt "Zielwert", nicht "Summe"
"title": "Zielwert-Zerlegung",
"columns": ["Bestandteil", "Wert"],
"rows": [["Erwartete Rendite", 0.127133], ["Risiko (lambda * Varianz)", -0.006376],
["Fixkosten (Transaktion)", -0.0232], ["Zielwert", 0.097557]],
"objective_column": "Wert",
}
def test_sum_row_recognised_by_label_zielwert():
tables, warnings = parse_result_tables([DECOMPOSITION], objective_value=0.097557)
assert warnings == [] and len(tables) == 1
def test_sum_row_recognised_by_value_when_label_is_unusual():
table = dict(DECOMPOSITION, rows=DECOMPOSITION["rows"][:-1] + [["Netto nach Abzügen", 0.097557]])
_, warnings = parse_result_tables([table], objective_value=0.097557)
assert warnings == []
def test_real_sum_error_is_still_found():
table = dict(DECOMPOSITION, rows=[["Rendite", 0.2], ["Kosten", -0.05], ["Zielwert", 0.15]])
_, warnings = parse_result_tables([table], objective_value=0.097557)
assert any("weicht vom Zielwert" in w for w in warnings)
def test_negligible_term_hints_at_mixed_units():
# beobachtet: Risiko 0,0066 (dimensionslos) neben Rendite 12.808 € und Gebühren 186 €
table = {"title": "Zielwert-Zerlegung", "columns": ["Bestandteil", "Wert"],
"rows": [["Erwartete Rendite", 12808.80], ["Risiko", -0.0066], ["Gebühren", -185.62],
["Zielwert", 12623.17]],
"objective_column": "Wert"}
_, warnings = parse_result_tables([table], objective_value=12623.17)
assert any("Größenordnungen kleiner" in w and "Risiko" in w for w in warnings)
def test_normal_decomposition_has_no_magnitude_warning():
_, warnings = parse_result_tables([DECOMPOSITION], objective_value=0.097557)
assert warnings == []
def test_report_shows_multistart_stability():
from core.model_schema import Multistart, SolverRun
from reporting.report import build_document, to_markdown
from test_pipeline import FakeSession
from TextToModel import run_pipeline
from test_review import make_review
from test_script_export import STRUCTURED_CODE
from core.model_schema import SolverCode
session = FakeSession([make_spec()], [make_review()], [SolverCode(code=STRUCTURED_CODE)])
result = run_pipeline("Aufgabe", session=session, max_revisions=0, max_restarts=0)
result.solver_run.multistart = Multistart(starts=5, converged=4, best=1.5, worst=9.0, mean=3.2, stddev=2.8)
markdown = to_markdown(build_document(result))
assert "Stabilität über mehrere Startpunkte" in markdown
assert "nicht nachweislich das globale Optimum" in markdown
def test_transliterated_umlauts_are_reported():
# beobachtet: "Der guenstigste Ersatzplan", Tabelle "Gewaehlte Nutzungszeitraeume"
from core.model_schema import ResultTable, SolverRun, SolverStatus
from solving.solver_stage import check_text_quality
run = SolverRun(status=SolverStatus.OPTIMAL, headline="Der guenstigste Plan kostet 29 €.",
tables=[ResultTable(title="Gewaehlte Zeitraeume", columns=["Kauf"], rows=[])])
befunde = check_text_quality(run)
assert any("guenstigste" in b for b in befunde)
assert any("Gewaehlte" in b and "Zeitraeume" in b for b in befunde)
def test_correct_german_is_not_flagged():
# Gegenprobe: echte Wörter mit ue/oe/ae dürfen keinen Fehlalarm auslösen
from core.model_schema import ResultTable, SolverRun, SolverStatus
from solving.solver_stage import check_text_quality
run = SolverRun(status=SolverStatus.OPTIMAL,
headline="Die Quelle liefert neue Koeffizienten für tägliche Werte.",
tables=[ResultTable(title="Gewählte Zeiträume", columns=["Menge", "Beitrag"], rows=[])])
assert check_text_quality(run) == []
def test_transliterations_are_repaired_after_being_reported():
# Die Warnung dokumentiert den Prompt-Verstoss, der Report soll trotzdem lesbar sein
from core.model_schema import Answer, ResultTable, SolverRun, SolverStatus
from solving.solver_stage import check_text_quality, fix_transliterations, repair_texts
run = SolverRun(status=SolverStatus.OPTIMAL, headline="Der guenstigste Plan kostet 29 €.",
answers=[Answer(question_id="a", text="Die Loesung waehlt zwei Aktien.")],
tables=[ResultTable(title="Gewaehlte Zeitraeume", columns=["Naechste Periode"],
rows=[["gewaehlt", 3]], note="Gemaess Aufgabe.")])
assert check_text_quality(run) # erst melden
repair_texts(run) # dann beheben
assert run.headline == "Der günstigste Plan kostet 29 €."
assert run.answers[0].text == "Die Lösung wählt zwei Aktien."
assert run.tables[0].title == "Gewählte Zeiträume"
assert run.tables[0].columns == ["Nächste Periode"]
assert run.tables[0].rows == [["gewählt", 3]]
assert run.tables[0].note == "Gemäß Aufgabe."
def test_repair_leaves_names_and_english_alone():
from solving.solver_stage import fix_transliterations
assert fix_transliterations("Mueller AG, Quelle, value, Koeffizient") == "Mueller AG, Quelle, value, Koeffizient"
assert fix_transliterations("") == ""
def test_repair_covers_stems_seen_in_the_lot_sizing_case():
# beobachtet in wagner_whitin_dijkstra: "110 Baelle", "Das heisst:" im Antworttext
from solving.solver_stage import fix_transliterations
assert fix_transliterations("Das heisst: 110 Baelle, der Ball") == "Das heißt: 110 Bälle, der Ball"

66
tests/test_review.py Normal file
View file

@ -0,0 +1,66 @@
from core.model_schema import AssumptionReview, CheckIssue, CheckSeverity
from checks.review import select_best
def make_review(matches=True, verdict="justified", missing=(), units_consistent=True):
return AssumptionReview.model_validate({
"objective_check": {
"task_evidence": "Bestimme die Aufteilung mit der geringsten Volatilität",
"asked_quantity": "Standardabweichung der Portfoliorendite",
"modeled_quantity": "Standardabweichung" if matches else "Varianz",
"matches": matches,
"units_consistent": units_consistent,
"unit_explanation": "Rendite €, Risiko €, Kosten €" if units_consistent else "Rendite €, Risiko dimensionslos, Kosten €",
"explanation": "-",
},
"checks": [{"unclear_point": "-", "verdict": verdict, "task_evidence": "-", "explanation": "-"}],
"missing_assumptions": list(missing),
})
def test_clean_review_needs_no_revision():
assert not make_review().needs_revision
def test_objective_mismatch_triggers_revision():
assert make_review(matches=False).needs_revision
def test_faulty_assumption_triggers_revision():
assert make_review(verdict="stricter_than_task").needs_revision
def test_missing_assumption_triggers_revision():
missing = [{"unclear_point": "-", "assumption": "-", "alternative": "-", "impact": "-"}]
assert make_review(missing=missing).needs_revision
# --- Auswahl der besten Korrekturrunde (checks.review.select_best) ---
ERROR = CheckIssue(severity=CheckSeverity.ERROR, code="CPSAT_WITH_CONTINUOUS_VARS", message="-")
WARNING = CheckIssue(severity=CheckSeverity.WARNING, code="UNDECLARED_SYMBOL", message="-")
def test_later_hallucinated_round_is_discarded():
# beobachtet in portfolio_qp: Runde 1 sauber, Runde 2 beschreibt ein Fremdproblem mit Check-Error
candidates = [
(None, [], make_review(matches=False)),
(None, [WARNING], make_review()),
(None, [ERROR, WARNING], make_review(matches=False)),
]
assert select_best(candidates) == 1
def test_clean_later_round_wins_tie():
candidates = [(None, [], make_review()), (None, [WARNING], make_review())]
assert select_best(candidates) == 1
def test_fewer_errors_preferred_when_all_have_errors():
candidates = [(None, [ERROR], make_review(matches=False)), (None, [ERROR, ERROR], make_review(matches=False))]
assert select_best(candidates) == 0
def test_mixed_units_trigger_revision():
# beobachtet: Rendite/Kosten in Euro, Risiko als dimensionslose Summe w_i^2 sigma_i^2 (Faktor 100.000 zu klein)
assert make_review(units_consistent=False).needs_revision

142
tests/test_script_export.py Normal file
View file

@ -0,0 +1,142 @@
import ast
import subprocess
import sys
import warnings
import pytest
from core.model_schema import PipelineResult, SolverCode, SolverRun, SolverStatus
from reporting.report import build_document, to_html
from solving.script_export import build_solver_script
from solving.solver_stage import CodeRejected, assemble_script, run_solver_code, validate_structure
from test_model_checks import make_spec
from test_review import make_review
# min 3x + 2y s.t. x + y >= 4, 0 <= x, y <= 3 -> Zielwert 9
STRUCTURED_CODE = '''
from scipy.optimize import linprog
# Kosten je Tonne Rohstoff X und Y [Tsd. €]
COSTS = [3, 2]
def solve_model():
"""Baut das LP auf und löst es."""
# Bedarf: x + y >= 4 -> -x - y <= -4
return linprog(COSTS, A_ub=[[-1, -1]], b_ub=[-4], bounds=[(0, 3), (0, 3)], method="highs")
def main():
"""Löst das Modell und gibt das Ergebnis aus."""
res = solve_model()
x, y = float(res.x[0]), float(res.x[1])
report(
"optimal" if res.status == 0 else "error", res.fun, {"x": x, "y": y},
constraints=[{"name": "Bedarf", "lhs": x + y, "sense": ">=", "rhs": 4}],
variables=[{"name": "x", "value": x, "var_type": "continuous", "lb": 0, "ub": 3},
{"name": "y", "value": y, "var_type": "continuous", "lb": 0, "ub": 3}],
)
if __name__ == "__main__":
main()
'''
def make_result(task="Minimiere die Kosten für mindestens 4 t Rohstoff.", code=STRUCTURED_CODE):
return PipelineResult(
task=task,
specification=make_spec(),
review=make_review(),
revisions=0,
solver_code=SolverCode(code=code, data_assumptions=["Lieferfrist: keine (nicht im Text)"]),
solver_run=SolverRun(status=SolverStatus.OPTIMAL, objective_value=9.0),
)
def test_script_layout_order():
script = build_solver_script(make_result())
assert script.startswith('#!/usr/bin/env python3\nr"""\n')
tree = ast.parse(script)
docstring = ast.get_docstring(tree)
positions = [docstring.index(h) for h in ("AUFGABE", "MATHEMATISCHES MODELL", "ALGORITHMUS")]
assert positions == sorted(positions)
assert "Minimiere die Kosten" in docstring
assert "\\sum_{j \\in J} x_j \\ge b" in docstring # Restriktion in LaTeX
assert "Branch-and-Cut" not in docstring # SciPy-Stack, nicht HiGHS-Text
assert "linprog" in docstring
assert "Lieferfrist" in docstring # angenommene Daten
# Imports vor report(), report() vor dem abschließenden main-Guard
names = [type(n).__name__ + ":" + getattr(n, "name", "") for n in tree.body]
guard_index = len(names) - 1
assert names.index("ImportFrom:") < names.index("FunctionDef:main") < names.index("FunctionDef:report") < guard_index
assert isinstance(tree.body[guard_index], ast.If) and script.rstrip().endswith("main()")
def test_script_is_standalone_runnable(tmp_path):
path = tmp_path / "solve.py"
path.write_text(build_solver_script(make_result()), encoding="utf-8")
proc = subprocess.run([sys.executable, str(path)], capture_output=True, text=True, timeout=60)
assert proc.returncode == 0, proc.stderr
assert "Status: optimal" in proc.stdout
assert "Zielwert: 9.0" in proc.stdout
assert proc.stdout.strip().splitlines()[-1].startswith("RESULT_JSON:")
def test_latex_backslashes_and_quotes_do_not_break_script():
task = 'Text mit \\sum, \\( x \\), """ Anführungszeichen """ und Backslash am Ende \\'
script = build_solver_script(make_result(task=task))
with warnings.catch_warnings():
warnings.simplefilter("error") # SyntaxWarning (ungültige Escapes) wäre ein Fehler
compile(script, "solve.py", "exec")
def test_run_solver_code_with_structured_code():
run = run_solver_code(STRUCTURED_CODE)
assert run.status == SolverStatus.OPTIMAL
assert run.objective_value == pytest.approx(9)
def test_assemble_without_guard_prepends_helpers():
script = assemble_script("x = 1")
assert "def report(" in script and "_apply_data_override(globals())" in script
assert script.index("def report(") < script.index("x = 1")
def test_valid_structure_passes():
validate_structure(STRUCTURED_CODE)
@pytest.mark.parametrize("code, reason", [
("def solve():\n pass\n\nif __name__ == '__main__':\n solve()\n", "main()"),
("def main():\n pass\n\nmain()\n", "__main__"),
("def main():\n pass\n\nif __name__ == '__main__':\n main()\n\nx = 1\n", "__main__"),
])
def test_invalid_structure_is_rejected(code, reason):
with pytest.raises(CodeRejected, match=reason.replace("(", r"\(").replace(")", r"\)")):
validate_structure(code)
def test_html_report_has_copy_button_with_full_script():
page = to_html(build_document(make_result()))
assert page.count('class="copy-btn"') == 1
assert "navigator.clipboard" in page and "execCommand('copy')" in page
assert "#!/usr/bin/env python3" in page
assert "if __name__ == &quot;__main__&quot;:" in page
def test_inline_math_delimiters_are_removed_from_docstring():
spec = make_spec(indices=["Tage: \\(d \\in D = \\{Mo,Di\\}\\)"])
result = make_result()
result.specification = spec
docstring = ast.get_docstring(ast.parse(build_solver_script(result)))
assert "Tage: d \\in D = \\{Mo,Di\\}" in docstring
assert "\\(" not in docstring and "\\)" not in docstring
def test_every_stack_and_class_has_algorithm_text():
# Synchron-Regel: neuer TargetStack/ProblemClass braucht Texte in script_export
from core.model_schema import ProblemClass, TargetStack
from solving.script_export import ALGORITHMS, CLASS_FIT
assert set(ALGORITHMS) == set(TargetStack)
assert set(CLASS_FIT) == set(ProblemClass)

755
tests/test_solver_stage.py Normal file
View file

@ -0,0 +1,755 @@
import sys
import pytest
from core.model_schema import SolverStatus
from solving.solver_stage import CodeRejected, clean_generated_code, run_solver_code, validate_imports
# min 3x + 2y s.t. x + y >= 4, 0 <= x, y <= 3 -> x=1, y=3, Zielwert 9
HIGHS_CODE = """
import highspy
h = highspy.Highs(); h.silent()
x = h.addVariable(lb=0, ub=3)
y = h.addVariable(lb=0, ub=3)
h.addConstr(x + y >= 4)
h.minimize(3 * x + 2 * y)
ok = h.getModelStatus() == highspy.HighsModelStatus.kOptimal
report("optimal" if ok else "error", h.getInfo().objective_function_value, {"x": h.val(x), "y": h.val(y)})
"""
CPSAT_CODE = """
from ortools.sat.python import cp_model
m = cp_model.CpModel()
x = m.new_int_var(0, 3, "x"); y = m.new_int_var(0, 3, "y")
m.add(x + y >= 4)
m.minimize(3 * x + 2 * y)
s = cp_model.CpSolver(); st = s.solve(m)
report("optimal" if st == cp_model.OPTIMAL else "error", s.objective_value, {"x": s.value(x), "y": s.value(y)})
"""
CVXPY_CODE = """
import cvxpy as cp
x = cp.Variable(2)
prob = cp.Problem(cp.Minimize(3 * x[0] + 2 * x[1]), [cp.sum(x) >= 4, x >= 0, x <= 3])
prob.solve()
report("optimal" if prob.status == cp.OPTIMAL else "error", prob.value, {"x": x.value[0], "y": x.value[1]})
"""
SCIPY_CODE = """
from scipy.optimize import linprog
res = linprog([3, 2], A_ub=[[-1, -1]], b_ub=[-4], bounds=[(0, 3), (0, 3)], method="highs")
report("optimal" if res.status == 0 else "error", res.fun, {"x": res.x[0], "y": res.x[1]})
"""
@pytest.mark.parametrize("code", [HIGHS_CODE, CPSAT_CODE, CVXPY_CODE, SCIPY_CODE], ids=["highs", "cpsat", "cvxpy", "scipy"])
def test_all_stacks_report_optimum(code):
run = run_solver_code(code)
assert run.error is None
assert run.status == SolverStatus.OPTIMAL
assert run.objective_value == pytest.approx(9, abs=1e-6)
VOLATILITY_SETUP = """
import math
import numpy as np
import cvxpy as cp
mu = np.array([0.07, 0.10, 0.04, 0.12])
Sigma = np.array([[0.0225, 0.0090, 0.0012, 0.0150], [0.0090, 0.0400, 0.0032, 0.0250],
[0.0012, 0.0032, 0.0064, 0.0020], [0.0150, 0.0250, 0.0020, 0.0625]])
w = cp.Variable(4, nonneg=True)
cons = [cp.sum(w) == 1, w <= 0.4, w[0] + w[1] <= 0.5, mu @ w >= 0.08]
"""
@pytest.mark.parametrize("formulation", [
"p = cp.Problem(cp.Minimize(cp.quad_form(w, Sigma)), cons); p.solve()\nreport('optimal', math.sqrt(p.value), {})",
"L = np.linalg.cholesky(Sigma)\np = cp.Problem(cp.Minimize(cp.norm(L.T @ w, 2)), cons); p.solve()\nreport('optimal', p.value, {})",
], ids=["quad_form_sqrt", "cholesky_LT"])
def test_volatility_formulations_from_solver_prompt(formulation):
# Beide im Solver-Prompt empfohlenen Varianten liefern das verifizierte Optimum (hard_portfolio_volatilitaet)
run = run_solver_code(VOLATILITY_SETUP + formulation)
assert run.objective_value == pytest.approx(0.1172905, abs=1e-5)
def test_cholesky_without_transpose_is_wrong():
# Dokumentiert den im Benchmark beobachteten Fehler: ||L w|| ist nicht die Standardabweichung
code = VOLATILITY_SETUP + "L = np.linalg.cholesky(Sigma)\np = cp.Problem(cp.Minimize(cp.norm(L @ w, 2)), cons); p.solve()\nreport('optimal', p.value, {})"
assert run_solver_code(code).objective_value != pytest.approx(0.1172905, abs=1e-3)
def test_infeasible_is_reported_not_error():
code = "import highspy\nh = highspy.Highs(); h.silent()\nx = h.addVariable(lb=0, ub=1)\nh.addConstr(x >= 2)\nh.minimize(x)\n" \
"st = h.getModelStatus()\nreport('infeasible' if st == highspy.HighsModelStatus.kInfeasible else str(st), None, {})"
assert run_solver_code(code).status == SolverStatus.INFEASIBLE
@pytest.mark.parametrize("code", [
"import os\nos.listdir('.')",
"from subprocess import run",
"open('x.txt', 'w')",
"def report(*a): pass",
"def broken(:",
])
def test_forbidden_code_is_rejected(code):
with pytest.raises(CodeRejected):
validate_imports(code)
assert run_solver_code(code).status == SolverStatus.ERROR
def test_runtime_error_is_captured():
run = run_solver_code("import math\nmath.sqrt(-1)")
assert run.status == SolverStatus.ERROR
assert "ValueError" in run.error
def test_missing_report_is_error():
assert run_solver_code("x = 1").status == SolverStatus.ERROR
def test_timeout():
run = run_solver_code("while True:\n pass", timeout=2)
assert run.status == SolverStatus.ERROR
assert "Timeout" in run.error
# --- SCIP (pyscipopt): Muster aus Solver-Prompt.txt ---
SCIP_PORTFOLIO_SETUP = """
from pyscipopt import Model, quicksum
MU = [x / 100 for x in [6.9, 14.4, 11.5, 9.8, 4.0, 4.0, 2.8, 13.3, 9.8, 11.2]]
FEE = [x / 100 for x in [0.68, 0.74, 0.39, 0.27, 0.34, 0.46, 0.69, 0.72, 0.20, 0.51]]
VAR = [x / 100 for x in [2.04, 1.63, 3.62, 3.01, 3.30, 3.33, 3.24, 1.97, 3.26, 2.52]]
m = Model(); m.hideOutput(); m.setParam("limits/time", 30)
w = [m.addVar(f"w{i}", vtype="C", lb=0, ub=1) for i in range(10)]
y = [m.addVar(f"y{i}", vtype="B") for i in range(10)]
m.addCons(quicksum(w) == 1); m.addCons(quicksum(y) >= 5); m.addCons(quicksum(y) <= 10)
for i in range(10):
m.addCons(w[i] >= 0.05 * y[i]); m.addCons(w[i] <= 0.30 * y[i])
utility = quicksum(MU[i] * w[i] - VAR[i] * w[i] * w[i] - FEE[i] * y[i] for i in range(10))
"""
def test_scip_miqp_with_epigraph_matches_verified_optimum():
# Optimum 0.097558 unabhängig per Enumeration aller Auswahlen verifiziert (benchmark hard_portfolio_kardinalitaet)
code = SCIP_PORTFOLIO_SETUP + (
"z = m.addVar('z', lb=None, ub=None)\nm.addCons(z <= utility)\nm.setObjective(z, 'maximize')\nm.optimize()\n"
"report(m.getStatus(), m.getObjVal(), {f'y{i}': round(m.getVal(y[i])) for i in range(10) if m.getVal(y[i]) > 0.5})"
)
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == pytest.approx(0.097558, abs=1e-4)
assert sorted(run.solution) == ["y1", "y2", "y3", "y7", "y8"] # A02, A03, A04, A08, A09
def test_scip_rejects_quadratic_objective_directly():
# Dokumentiert, warum der Solver-Prompt die Epigraph-Formulierung vorschreibt
code = SCIP_PORTFOLIO_SETUP + "m.setObjective(utility, 'maximize')\nreport('optimal', 0.0, {})"
run = run_solver_code(code)
assert run.status == SolverStatus.ERROR and "nonlinear objective" in run.error
def test_scip_volatility_as_quadratic_constraint():
code = VOLATILITY_SETUP.replace("import cvxpy as cp", "from pyscipopt import Model, quicksum").split("w = cp.Variable")[0] + (
"m = Model(); m.hideOutput()\n"
"w = [m.addVar(f'w{i}', lb=0, ub=0.4) for i in range(4)]\n"
"m.addCons(quicksum(w) == 1); m.addCons(w[0] + w[1] <= 0.5)\n"
"m.addCons(quicksum(float(mu[i]) * w[i] for i in range(4)) >= 0.08)\n"
"t = m.addVar('t', lb=0)\n"
"m.addCons(t * t >= quicksum(float(Sigma[i][j]) * w[i] * w[j] for i in range(4) for j in range(4)))\n"
"m.setObjective(t, 'minimize'); m.optimize()\n"
"report(m.getStatus(), m.getObjVal(), {})"
)
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == pytest.approx(0.1172905, abs=1e-4)
# --- Bereinigung von Markdown-Resten im generierten Code ---
CLEAN = 'import math\n\n\ndef main():\n report("optimal", math.pi, {})\n\n\nif __name__ == "__main__":\n main()\n'
@pytest.mark.parametrize("raw", [
CLEAN,
"python\n" + CLEAN, # beobachtet im Benchmark (Rest von ```python)
"```python\n" + CLEAN + "```",
"```\n" + CLEAN + "\n```\n",
"\n\nPython\n\n" + CLEAN,
], ids=["sauber", "python-zeile", "fence-python", "fence-leer", "gross-leerzeilen"])
def test_markdown_remnants_are_removed(raw):
assert clean_generated_code(raw) == CLEAN
assert run_solver_code(clean_generated_code(raw)).status == SolverStatus.OPTIMAL
def test_code_mentioning_python_later_is_untouched():
code = "# python-Kommentar\nx = 'python'\n"
assert clean_generated_code(code) == code
@pytest.mark.parametrize("rest", ["}", "}\n", '"}', "}\n]", "```"], ids=["klammer", "klammer-nl", "quote", "zwei", "fence"])
def test_trailing_json_artifacts_are_removed(rest):
# beobachtet im Benchmark (lp_klein, beide Läufe): schließende JSON-Klammer nach dem main-Guard
assert clean_generated_code(CLEAN + rest) == CLEAN
assert run_solver_code(clean_generated_code(CLEAN + rest)).status == SolverStatus.OPTIMAL
def test_broken_code_keeps_its_real_syntax_error():
# kein blindes Abschneiden: der Reparaturversuch soll den echten Fehler sehen
broken = 'def main(:\n pass\n}\n'
assert clean_generated_code(broken).strip().endswith("}")
def test_data_with_tuple_keys_still_reports_a_result():
# beobachtet (hard_transport_lagerwahl): Transportkosten je Werk-Lager-Paar als Tupel-Schlüssel
code = (
'DATA = {"c": {("W1", "L1"): 4}}\n\n\n'
'def main():\n'
' report("optimal", 4.0, {"x": 1})\n\n\n'
'if __name__ == "__main__":\n main()\n'
)
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == 4.0
assert run.data == {} # nicht übernommen, dafür als Warnung gemeldet
assert any("nicht JSON-fähig" in w for w in run.warnings)
# --- Ipopt (cyipopt): Muster aus Solver-Prompt.txt ---
IPOPT_CONVEX = """
import numpy as np
from cyipopt import minimize_ipopt
# min (x-1)^2 + (y-2.5)^2 u.d.N. x + y <= 2, x,y >= 0 -> konvex, Optimum (0.25, 1.75), Wert 1.125
res = minimize_ipopt(lambda v: (v[0] - 1) ** 2 + (v[1] - 2.5) ** 2, x0=np.array([0.0, 0.0]),
bounds=[(0, None), (0, None)],
constraints=[{"type": "ineq", "fun": lambda v: 2 - v[0] - v[1]}],
options={"print_level": 0, "sb": "yes"})
report("optimal" if res.success else "error", float(res.fun), {"x": float(res.x[0]), "y": float(res.x[1])})
"""
IPOPT_MULTISTART = """
import numpy as np
from cyipopt import minimize_ipopt
# min x*y auf dem Einheitskreis, x,y >= 0: nichtkonvex, globales Minimum 0 an den Achsen
best = None
for x0 in ([0.5, 0.5], [0.9, 0.1], [0.1, 0.9], [0.0, 1.0], [1.0, 0.0]):
res = minimize_ipopt(lambda v: v[0] * v[1], x0=np.array(x0, dtype=float), bounds=[(0, None)] * 2,
constraints=[{"type": "eq", "fun": lambda v: v[0] ** 2 + v[1] ** 2 - 1}],
options={"print_level": 0, "sb": "yes"})
if res.success and (best is None or res.fun < best):
best = float(res.fun)
report("feasible", best, {})
"""
def test_ipopt_convex_problem_reports_optimum():
run = run_solver_code(IPOPT_CONVEX)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == pytest.approx(1.125, abs=1e-5)
def test_ipopt_multistart_finds_global_minimum_of_nonconvex_problem():
# ein einzelner Startpunkt (0.5, 0.5) liefert 0.5 statt 0 – daher verlangt der Prompt mehrere Startpunkte
run = run_solver_code(IPOPT_MULTISTART)
assert run.status == SolverStatus.FEASIBLE
assert run.objective_value == pytest.approx(0.0, abs=1e-6)
# --- Sandbox: der erzeugte Code läuft ohne Netz und ohne Schreibrechte außerhalb seines Laufs ---
import shutil
from solving.solver_stage import sandbox_command
ESCAPE_CODE = """
import numpy as np
def main():
ziel = "{ziel}"
try:
np.savetxt(ziel, np.zeros(1))
report("optimal", 1.0, {{"geschrieben": ziel}})
except OSError as exc:
report("optimal", 0.0, {{"blockiert": type(exc).__name__}})
if __name__ == "__main__":
main()
"""
@pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap nicht installiert")
def test_sandbox_blocks_writing_outside_the_run(tmp_path):
# numpy ist erlaubt und savetxt kein verbotener Aufruf – validate_imports sieht das nicht,
# die Sandbox schon. Genau dafür ist sie da.
ziel = tmp_path / "ausbruch.txt"
run = run_solver_code(ESCAPE_CODE.format(ziel=ziel))
assert run.status == SolverStatus.OPTIMAL, run.error
assert "blockiert" in run.solution, f"Schreiben nach {ziel} war möglich"
assert not ziel.exists()
@pytest.mark.skipif(not shutil.which("bwrap"), reason="bubblewrap nicht installiert")
def test_sandbox_allows_writing_inside_the_run():
# das Programm darf in seinem eigenen Arbeitsverzeichnis arbeiten (relativer Pfad)
run = run_solver_code(ESCAPE_CODE.format(ziel="zwischenergebnis.txt"))
assert run.status == SolverStatus.OPTIMAL, run.error
assert "geschrieben" in run.solution
def test_sandbox_command_uses_bwrap_when_available(monkeypatch, tmp_path):
from solving import solver_stage
monkeypatch.setattr(solver_stage, "SANDBOX", "auto")
monkeypatch.setattr(solver_stage.shutil, "which", lambda name: "/usr/bin/bwrap")
befehl = sandbox_command(tmp_path / "solve.py", str(tmp_path))
assert befehl[0] == "bwrap"
assert "--unshare-all" in befehl # kein Netz
assert befehl[befehl.index("--ro-bind") + 1:][:2] == ["/", "/"] # Dateisystem nur lesbar
assert str(tmp_path) in befehl # Ausnahme: das Arbeitsverzeichnis
def test_sandbox_can_be_switched_off(monkeypatch, tmp_path):
from solving import solver_stage
monkeypatch.setattr(solver_stage, "SANDBOX", "aus")
assert sandbox_command(tmp_path / "solve.py", str(tmp_path))[0] == sys.executable
def test_missing_bwrap_is_reported_when_required(monkeypatch, tmp_path):
from solving import solver_stage
monkeypatch.setattr(solver_stage, "SANDBOX", "bwrap")
monkeypatch.setattr(solver_stage.shutil, "which", lambda name: None)
with pytest.raises(CodeRejected, match="bubblewrap"):
sandbox_command(tmp_path / "solve.py", str(tmp_path))
def test_missing_bwrap_falls_back_when_automatic(monkeypatch, tmp_path):
from solving import solver_stage
monkeypatch.setattr(solver_stage, "SANDBOX", "auto")
monkeypatch.setattr(solver_stage.shutil, "which", lambda name: None)
assert sandbox_command(tmp_path / "solve.py", str(tmp_path))[0] == sys.executable
# --- Mehrstart-Streuung (Ipopt): check_multistart ---
from core.model_schema import Multistart
from solving.solver_stage import check_multistart
def multistart_run(**felder):
from core.model_schema import SolverRun
werte = {"starts": 5, "converged": 5, "best": 1.125, "worst": 1.125, "mean": 1.125, "stddev": 0.0}
werte.update(felder)
return SolverRun(status=SolverStatus.FEASIBLE, objective_value=werte["best"],
multistart=Multistart(**werte))
def test_identical_starts_give_no_warning():
assert check_multistart(multistart_run(), "minimize") == []
def test_scattered_starts_are_reported_as_local():
warnungen = check_multistart(multistart_run(worst=4.0, mean=2.0, stddev=1.2), "minimize")
assert any("nicht nachweislich global" in w for w in warnungen)
def test_mean_better_than_best_is_impossible():
# bei Minimierung darf kein Mittelwert unter dem gemeldeten besten Wert liegen
warnungen = check_multistart(multistart_run(mean=0.5, stddev=0.0), "minimize")
assert any("mean" in w and "unmöglich" in w for w in warnungen)
def test_maximization_direction_is_respected():
assert check_multistart(multistart_run(worst=0.5, mean=0.8, stddev=0.0), "maximize") == []
assert any("worst" in w for w in check_multistart(multistart_run(worst=2.0, mean=1.5, stddev=0.0), "maximize"))
def test_single_successful_start_cannot_show_stability():
warnungen = check_multistart(multistart_run(converged=1, worst=None, mean=None, stddev=None), "minimize")
assert any("nicht beurteilbar" in w for w in warnungen)
def test_no_multistart_means_no_check():
from core.model_schema import SolverRun
assert check_multistart(SolverRun(status=SolverStatus.OPTIMAL), "minimize") == []
def test_ipopt_program_reports_its_multistart():
code = """
import numpy as np
from cyipopt import minimize_ipopt
def main():
# min (x-1)^2 u.d.N. x >= 0 – konvex, jeder Startpunkt landet bei x = 1
werte = []
for start in [0.0, 0.5, 2.0, 5.0, 10.0]:
res = minimize_ipopt(lambda v: (v[0] - 1) ** 2, x0=np.array([start]), bounds=[(0, 20)],
options={"print_level": 0, "sb": "yes"})
if res.success:
werte.append(float(res.fun))
report("optimal", min(werte), {"x": 1.0},
multistart={"starts": 5, "converged": len(werte), "best": min(werte), "worst": max(werte),
"mean": float(np.mean(werte)), "stddev": float(np.std(werte))})
if __name__ == "__main__":
main()
"""
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.multistart is not None and run.multistart.converged == 5
assert run.multistart.stddev == pytest.approx(0.0, abs=1e-6)
assert check_multistart(run, "minimize") == []
# --- Graphenprobleme (NetworkX/Dijkstra): API aus Solver-Prompt.txt und check_graph ---
from core.model_schema import GraphSolution
from solving.solver_stage import check_graph
DIJKSTRA_CODE = '''
import networkx as nx
DATA = {
# Kostenmatrix c[i][j]: Kauf am Ende von Jahr i, Nutzung bis Ende Jahr j
"kosten": {"0": {"1": 8, "2": 18, "3": 31}, "1": {"2": 10, "3": 21}, "2": {"3": 12}},
"quelle": "0",
"ziel": "3",
}
def loesen():
"""Kuerzester Weg von der Quelle zur Senke mit dem Dijkstra-Algorithmus."""
G = nx.DiGraph()
for i, ziele in DATA["kosten"].items():
for j, c in ziele.items():
G.add_edge(i, j, weight=c)
laenge, pfad = nx.single_source_dijkstra(G, DATA["quelle"], DATA["ziel"], weight="weight")
marken = nx.single_source_dijkstra_path_length(G, DATA["quelle"], weight="weight")
return G, laenge, pfad, marken
def main():
G, laenge, pfad, marken = loesen()
boegen = [{"source": u, "target": v, "weight": d["weight"]} for u, v, d in G.edges(data=True)]
report("optimal", float(laenge), {f"x[{u},{v}]": 1.0 for u, v in zip(pfad, pfad[1:])},
graph={"nodes": sorted(G.nodes()), "arcs": boegen, "path": list(pfad),
"steps": [{"node": k, "label": float(w)} for k, w in marken.items()],
"node_meaning": "ein Zeitpunkt - das Ende von Jahr i",
"arc_meaning": "Kauf am Ende von Jahr i und Nutzung bis Ende von Jahr j"})
if __name__ == "__main__":
main()
'''
def test_dijkstra_program_from_solver_prompt():
# Fall graph_ersatz_dijkstra: verifiziertes Optimum 29 ueber den Weg 0 -> 1 -> 3
run = run_solver_code(DIJKSTRA_CODE)
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == 29
assert run.graph is not None
assert run.graph.path == ["0", "1", "3"]
assert [s.node for s in run.graph.steps] == ["0", "1", "2", "3"] # Reihenfolge der Markierung
assert [s.label for s in run.graph.steps] == [0, 8, 18, 29]
assert check_graph(run, "minimize") == []
def graph_run(**felder):
from core.model_schema import SolverRun
werte = {"nodes": ["0", "1", "2"],
"arcs": [{"source": "0", "target": "1", "weight": 4},
{"source": "1", "target": "2", "weight": 5},
{"source": "0", "target": "2", "weight": 12}],
"path": ["0", "1", "2"], "steps": [{"node": "0", "label": 0}, {"node": "1", "label": 4},
{"node": "2", "label": 9}],
"node_meaning": "-", "arc_meaning": "-"}
werte.update(felder)
return SolverRun(status=SolverStatus.OPTIMAL, objective_value=felder.pop("objective_value", 9),
graph=GraphSolution.model_validate(werte))
def test_consistent_graph_needs_no_warning():
assert check_graph(graph_run(), "minimize") == []
def test_path_must_use_declared_arcs():
warnungen = check_graph(graph_run(path=["0", "2", "1"]), "minimize")
assert any("nicht deklariert" in w for w in warnungen)
def test_path_length_must_match_objective():
from core.model_schema import SolverRun
run = graph_run()
run.objective_value = 7.0 # 4 + 5 = 9
assert any("weicht vom Zielwert" in w for w in check_graph(run, "minimize"))
def test_negative_weights_are_reported_for_dijkstra():
# geprueft: networkx meldet hier keinen Fehler, liefert aber ein falsches Ergebnis
arcs = [{"source": "0", "target": "1", "weight": 4}, {"source": "1", "target": "2", "weight": -5},
{"source": "0", "target": "2", "weight": 12}]
run = graph_run(arcs=arcs)
run.objective_value = -1.0
assert any("negative Bogengewichte" in w for w in check_graph(run, "minimize"))
def test_falling_labels_contradict_dijkstra():
steps = [{"node": "0", "label": 0}, {"node": "1", "label": 9}, {"node": "2", "label": 4}]
assert any("fallen zwischendurch" in w for w in check_graph(graph_run(steps=steps), "minimize"))
# --- Abnahme: jede Teilfrage beantwortet, jede Zahl belegt (check_answers) ---
from solving.solver_stage import check_answers
def verstaendnis(*fragen):
from core.model_schema import SubQuestion, TaskUnderstanding
return TaskUnderstanding(
is_optimization_task=True, reason="-", facts=[], ignored_numbers=[],
sub_questions=[SubQuestion(id=f, quote=f"Frage {f}", kind="rechnen", needs_solver=True) for f in fragen])
def geloest(antworten, **felder):
from core.model_schema import Answer, SolverRun, VariableValue
werte = {"status": SolverStatus.OPTIMAL, "objective_value": 29.0,
"variables": [VariableValue(name="x", value=1.0)],
"answers": [Answer(question_id=k, text=t) for k, t in antworten]}
werte.update(felder)
return SolverRun(**werte)
def test_all_questions_answered_is_clean():
run = geloest([("a", "Der Graph hat vier Knoten."), ("b", "Die Kosten betragen 29 Geldeinheiten.")])
assert check_answers(run, verstaendnis("a", "b")) == []
def test_missing_answer_is_reported():
# Fall graph_ersatz_dijkstra: Zielwert richtig, Teil a) unbeantwortet
run = geloest([("b", "Die Kosten betragen 29 Geldeinheiten.")])
befunde = check_answers(run, verstaendnis("a", "b"))
assert befunde and "a" in befunde[0] and "nicht vollständig beantwortet" in befunde[0]
def test_invented_numbers_in_the_answer_are_caught():
# 47 wurde nie berechnet und steht nicht in der Aufgabe
run = geloest([("a", "Die Ersparnis liegt bei 47 Geldeinheiten.")])
befunde = check_answers(run, verstaendnis("a"), task="Die Kosten betragen 8, 18 und 31.")
assert any("47" in b for b in befunde)
def test_comparison_numbers_belong_in_a_table():
# beobachtet: die Antwort nannte den Vergleichsweg "kostet 30" - richtig gerechnet,
# aber in keinem gemeldeten Feld, also fuer den Leser nicht nachvollziehbar
from core.model_schema import ResultTable
run = geloest([("b", "Der beste Weg kostet 29, die Alternative 30.")])
assert any("30" in b for b in check_answers(run, verstaendnis("b")))
mit_tabelle = geloest([("b", "Der beste Weg kostet 29, die Alternative 30.")],
tables=[ResultTable(title="Alle Wege", columns=["Weg", "Kosten"],
rows=[["0-1-3", 29], ["0-2-3", 30]])])
assert check_answers(mit_tabelle, verstaendnis("b")) == []
def test_numbers_from_the_task_are_allowed_in_the_answer():
run = geloest([("a", "Die Maschine wird am Ende von Jahr 1 ersetzt, Kosten 8 + 21 = 29.")])
assert check_answers(run, verstaendnis("a"), task="| 0 | 8 | 18 | 31 |\n| 1 | - | 10 | 21 |") == []
def test_no_understanding_means_no_acceptance_check():
assert check_answers(geloest([]), None) == []
def test_program_reports_answers_end_to_end():
code = '''
def main():
"""Loest die Aufgabe und beantwortet beide Teilfragen."""
report("optimal", 29.0, {"x": 1.0},
answers=[{"question_id": "a", "text": "Der Graph hat vier Knoten und sechs Boegen."},
{"question_id": "b", "text": "Der kuerzeste Weg kostet 29 Geldeinheiten."}])
if __name__ == "__main__":
main()
'''
run = run_solver_code(code)
assert run.status == SolverStatus.OPTIMAL, run.error
assert [a.question_id for a in run.answers] == ["a", "b"]
assert check_answers(run, verstaendnis("a", "b")) == []
def test_attached_data_file_is_readable_in_the_sandbox(tmp_path):
# Die Datei muss im Laufverzeichnis landen - es ist das einzige, das die Sandbox beschreibbar bindet
datei = tmp_path / "werte.csv"
datei.write_text("name,menge\nA,3\nB,4\n")
code = '''
import pandas as pd
DATA = {"dateien": {"werte": "werte.csv"}}
def main():
"""Liest die beigelegte Datei und meldet ihre Summe."""
tabelle = pd.read_csv(DATA["dateien"]["werte"])
report("optimal", float(tabelle["menge"].sum()), {"zeilen": float(len(tabelle))})
if __name__ == "__main__":
main()
'''
run = run_solver_code(code, data_files=[datei])
assert run.status == SolverStatus.OPTIMAL, run.error
assert run.objective_value == 7.0 and run.solution["zeilen"] == 2.0
def test_without_the_attachment_the_program_fails_cleanly(tmp_path):
code = '''
import pandas as pd
def main():
"""Liest eine Datei, die nicht beigelegt wurde."""
report("optimal", float(pd.read_csv("werte.csv")["menge"].sum()), {})
if __name__ == "__main__":
main()
'''
run = run_solver_code(code)
assert run.status == SolverStatus.ERROR and "FileNotFound" in (run.error or "")
# --- Szenarien: Datenvarianten desselben Programms, ohne LLM nachgerechnet ---
SZENARIO_CODE = '''
DATA = {"preis": 10.0, "menge": 3.0}
def loesen(daten):
"""Zielwert dieser Datenvariante."""
return daten["preis"] * daten["menge"]
def main():
"""Rechnet das Grundszenario und zwei Varianten."""
basis = loesen(DATA)
szenarien = []
for name, abweichung in [("doppelter Preis", {"preis": 20.0}), ("halbe Menge", {"menge": 1.5})]:
szenarien.append({"name": name, "description": "-", "data_override": abweichung,
"objective_value": loesen({**DATA, **abweichung})})
report("optimal", basis, {"wert": basis}, scenarios=szenarien)
if __name__ == "__main__":
main()
'''
def test_scenarios_are_recomputed_and_confirmed():
run = run_solver_code(SZENARIO_CODE)
assert run.objective_value == 30.0
assert [s.name for s in run.scenarios] == ["doppelter Preis", "halbe Menge"]
from solving.solver_stage import verify_scenarios
assert verify_scenarios(run, SZENARIO_CODE) == [] # gemeldet = nachgerechnet
assert [s.verified_value for s in run.scenarios] == [60.0, 15.0]
def test_wrong_scenario_value_is_caught():
from solving.solver_stage import verify_scenarios
run = run_solver_code(SZENARIO_CODE)
run.scenarios[0].objective_value = 99.0 # das Programm hat sich verrechnet
befunde = verify_scenarios(run, SZENARIO_CODE)
assert any("doppelter Preis" in b and "99" in b for b in befunde)
def test_scenario_without_override_is_reported():
from core.model_schema import Scenario
from solving.solver_stage import verify_scenarios
run = run_solver_code(SZENARIO_CODE)
run.scenarios = [Scenario(name="leer", description="-", data_override={}, objective_value=1.0)]
assert any("keine Datenabweichung" in b for b in verify_scenarios(run, SZENARIO_CODE))
def test_rounded_numbers_in_the_answer_are_accepted():
# beobachtet: "9.65" fuer den berechneten Wert 9.647868... - Antworten runden, Ergebnisse nicht
run = geloest([("b", "Der Nutzen betraegt 9.65 Prozentpunkte.")], objective_value=9.647868721241633)
assert check_answers(run, verstaendnis("b")) == []
def test_rounding_tolerance_does_not_hide_invented_numbers():
run = geloest([("b", "Der Nutzen betraegt 47 Prozentpunkte.")], objective_value=9.647868721241633)
assert any("47" in b for b in check_answers(run, verstaendnis("b")))
def test_unit_conversion_in_the_answer_is_accepted():
# beobachtet: "9,00 Tsd. € (also 9000 €)" fuer den Zielwert 9 - eine Umrechnung, keine neue Zahl
run = geloest([("1", "Die Kosten betragen 9,00 Tsd. € (also 9000 €).")], objective_value=9.0)
assert check_answers(run, verstaendnis("1")) == []
def test_percent_and_fraction_are_the_same_number():
run = geloest([("1", "Der Nutzen betraegt 9.76 % des Kapitals.")], objective_value=0.0975577)
assert check_answers(run, verstaendnis("1")) == []
def test_invented_number_survives_the_scale_tolerance():
run = geloest([("1", "Die Ersparnis betraegt 47 €.")], objective_value=9.0)
assert any("47" in b for b in check_answers(run, verstaendnis("1"), task="Kosten 3 und 2."))
def test_numbers_from_the_attached_data_are_traceable():
# Bei beigelegten Dateien stehen die Eingabewerte NICHT im Aufgabentext - trotzdem sind sie belegt
run = geloest([("a", "Aktie A02 hat die hoechste Rendite mit 14.4 Prozentpunkten.")], objective_value=12.03)
run.data = {"dateien": {"aktien": "aktien.csv"}}
befunde = check_answers(run, verstaendnis("a"), task="Die Kennzahlen stehen in aktien.csv.",
data_summary="Rendite (Zahl, von 2.8 bis 14.4, Mittel 8.77): 6,9, 14,4, 11,5")
assert befunde == []
def test_duplicate_question_ids_still_need_one_answer_each():
from core.model_schema import SubQuestion, TaskUnderstanding
u = TaskUnderstanding(is_optimization_task=True, reason="-", facts=[], ignored_numbers=[], sub_questions=[
SubQuestion(id="a1", quote="Formulieren Sie …", kind="formulieren", needs_solver=False),
SubQuestion(id="a2", quote="Zeichnen Sie …", kind="zeichnen", needs_solver=False)])
run = geloest([("a1", "Das Modell ist ein Graph.")])
assert any("a2" in b for b in check_answers(run, u))
def test_self_generated_data_must_be_declared():
# beobachtet an "Portfolio Nr. 2": synthetischer Markt mit festem Seed, data_assumptions leer
from core.model_schema import SolverCode
from solving.solver_stage import check_generated_data
code = "import numpy as np\nrng = np.random.default_rng(42)\nmu = rng.normal(0.1, 0.05, 25)\n"
assert check_generated_data(code, SolverCode(code=code, data_assumptions=[]))
assert check_generated_data(code, SolverCode(code=code, data_assumptions=["mu, Sigma: synthetisch, Seed 42"])) == []
def test_fixed_seed_with_complete_data_is_not_generated_data():
# Ipopt-Mehrstart mit festem Zufallsstartwert: alle Werte stehen in der Spezifikation
from core.model_schema import SolverCode
from solving.solver_stage import check_generated_data
from test_model_checks import make_spec
code = "import numpy as np\nrng = np.random.default_rng(42)\nstarts = rng.uniform(0, 1, (5, 2))\n"
ipopt = make_spec(problem_class="Nonlinear Programming (NLP)", target_stack="Ipopt (cyipopt)")
assert check_generated_data(code, SolverCode(code=code, data_assumptions=[]), ipopt) == []
def test_prose_instead_of_values_does_not_hide_generated_data():
# Nachmessung "Portfolio Nr. 2": extracted_values war nicht leer, sondern Text ohne Werte
from core.model_schema import SolverCode
from solving.solver_stage import check_generated_data
from test_model_checks import make_spec
spec = make_spec(parameters=[{"symbol": "\\mu_i", "dimension": "Vektor", "unit": "1/Jahr", "description": "Rendite",
"extracted_values": "aus Kursdaten berechnet (Teil 1); im Text nicht numerisch vorgegeben"}])
code = "import numpy as np\nrng = np.random.default_rng(0)\nmu = rng.normal(0.1, 0.05, 25)\n"
assert check_generated_data(code, SolverCode(code=code, data_assumptions=[]), spec)
def test_ordinary_programs_are_not_suspected():
from core.model_schema import SolverCode
from solving.solver_stage import check_generated_data
code = "import highspy\nh = highspy.Highs()\nDATA = {'kosten': [3, 2]}\n"
assert check_generated_data(code, SolverCode(code=code, data_assumptions=[])) == []