text_to_solver_program/tests/test_result_tables.py

"""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_no_longer_reported():
    # Umschreibungen im Code sind erlaubt; Lesertexte werden still berichtigt statt gemeldet
    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=[])])
    assert check_text_quality(run) == []
    befunde = check_text_quality(run, summary="Es wird ein Plan gewählt.")
    assert not any("Umlaut" 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_silently():
    from core.model_schema import Answer, ResultTable, SolverRun, SolverStatus
    from solving.solver_stage import 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.")])
    repair_texts(run)
    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"


def test_words_from_the_run_repair_what_the_fixed_list_misses():
    # beobachtet in "Portfolio Nr. 2": Volatilitaet, Kardinalitaet, vollstaendigen, binaere im Report
    from core.model_schema import Answer, ConstraintValue, SolverRun, SolverStatus
    from solving.solver_stage import repair_texts, umlaut_map
    aufgabe = "Datenlieferant: Mueller GmbH."
    spezifikation = ("Minimiere die Volatilität unter der Kardinalität k; nur Titel mit vollständigen Reihen, "
                     "binäre Auswahl. Restriktion: Kardinalitaet")      # ASCII-Bezeichner des LLM
    run = SolverRun(status=SolverStatus.OPTIMAL,
                    headline="Volatilitaet 12 %, Kardinalitaet eingehalten.",
                    answers=[Answer(question_id="1", text="Nur Titel mit vollstaendigen Reihen, binaere Variablen; Quelle: Mueller GmbH.")],
                    constraints=[ConstraintValue(name="Kardinalitaet", lhs=20, sense="==", rhs=20)])
    repair_texts(run, umlaut_map(aufgabe, spezifikation))
    assert run.headline == "Volatilität 12 %, Kardinalität eingehalten."
    assert run.answers[0].text == "Nur Titel mit vollständigen Reihen, binäre Variablen; Quelle: Mueller GmbH."
    assert run.constraints[0].name == "Kardinalitaet"      # Bezeichner bleibt: Namensabgleich hängt daran


def test_ascii_form_present_in_the_task_is_not_replaced():
    # steht die ASCII-Form selbst im Text (z. B. ein Name), ist sie echt
    from solving.solver_stage import fix_transliterations, umlaut_map
    zuordnung = umlaut_map("Herr Müller und die Firma Mueller liefern.")
    assert "mueller" not in zuordnung
    assert fix_transliterations("Firma Mueller", zuordnung) == "Firma Mueller"


# beobachtet in Portfolio Nr. 2: eine Zeile je Konfiguration, die gewählte trägt den Zielwert
COMPARISON = {
    "title": "Vergleich der Konfigurationen",
    "columns": ["Konfiguration", "lambda1", "lambda2", "Zielfunktionswert"],
    "rows": [["Reine Varianzminimierung", "1.0", "0.0", 0.18217355],
             ["Reine Korrelationsminimierung", "0.0", "1.0", -187.29514112],
             ["Ausgewogene Skalarisierung", "0.5", "0.5", -93.56364364]],
    "objective_column": "Zielfunktionswert",
}


def test_comparison_table_is_not_a_sum_mismatch():
    tables, warnings = parse_result_tables([COMPARISON], objective_value=0.18217353985482126)
    assert warnings == [] and len(tables) == 1


def test_decomposition_with_one_term_equal_to_objective_is_still_checked():
    # eine Summenzeile macht daraus eine Zerlegung – deren Fehler bleibt sichtbar
    table = dict(DECOMPOSITION, rows=[["Rendite", 0.097557], ["Kosten", -0.05], ["Zielwert", 0.2]])
    _, warnings = parse_result_tables([table], objective_value=0.097557)
    assert any("weicht vom Zielwert" in w for w in warnings)


# --- Einheiten in Antworten: nur, was der Text hergibt ---

def _run_with_answer(text):
    from core.model_schema import Answer
    return SolverRun(status=SolverStatus.OPTIMAL, objective_value=459, answers=[Answer(question_id="1", text=text)])


def test_invented_unit_in_answer_is_found():
    # beobachtet an tasks/nur_formeln.md: reines ILP ohne jede Einheit
    from solving.solver_stage import check_answer_units
    befunde = check_answer_units(_run_with_answer("Der minimale Zielfunktionswert beträgt 459 Geldeinheiten (GE)."),
                                 "min z = 17*x1 + 23*x2  s.t.  6*x1 + 4*x2 >= 120")
    assert len(befunde) == 1 and "Geldeinheiten" in befunde[0]


def test_currency_is_invented_when_the_text_only_says_costs():
    from solving.solver_stage import check_answer_units
    aufgabe = "Die Kostenmatrix C enthält die Kosten je Nutzungsdauer."
    assert check_answer_units(_run_with_answer("Die Gesamtkosten betragen 29 €."), aufgabe)
    assert check_answer_units(_run_with_answer("Die Gesamtkosten betragen 29 GE."), aufgabe) == []  # neutral


def test_unit_of_the_same_kind_in_the_text_is_covered():
    from solving.solver_stage import check_answer_units
    aufgabe = "Kosten in Tausend Euro; Rüstzeit 2 Stunden."
    run = _run_with_answer("Die Kosten betragen 29 Tsd. € bei 90 min Rüstzeit, Auslastung 85 %.")
    assert check_answer_units(run, aufgabe) == []


def test_unit_from_attached_data_is_covered():
    from solving.solver_stage import check_answer_units
    run = _run_with_answer("Der Transport kostet 120 kg CO2.")
    assert check_answer_units(run, "Minimiere die Emissionen.") != []
    assert check_answer_units(run, "Minimiere die Emissionen.", data_summary="Spalte emission_kg") == []


def test_numbers_and_words_are_not_mistaken_for_units():
    from solving.solver_stage import check_answer_units
    run = _run_with_answer("Teil 2 zeigt: x1 = 6, 3 Maschinen, Stufe 2a, am Tag 3 sind alle 5 Aufträge erledigt.")
    assert check_answer_units(run, "Plane die Aufträge.") == []


def test_repair_reaches_compounds_and_inflections():
    # beobachtet in Portfolio Nr. 2 und nur_formeln (17.09.2026), Stämme gegen ngerman geprüft
    from solving.solver_stage import fix_transliterations
    text = "Aufgefuehrt: Beitraege, Jahresvolatilitaet, Gesamtstueckzahl, zulaessige Kardinalitaet, vollstaendig."
    assert fix_transliterations(text) == ("Aufgeführt: Beiträge, Jahresvolatilität, Gesamtstückzahl, "
                                          "zulässige Kardinalität, vollständig.")


def test_repair_leaves_correct_words_and_identifiers_alone():
    from solving.solver_stage import fix_transliterations
    richtig = "Die neue Dauer, Quelle, aktuell, Kostenkoeffizient, Bauer, streuen, blue."
    assert fix_transliterations(richtig) == richtig
    namen = "Bindend: Spaet_zu_Frueh_Verbot und KeinFruehNachSpaet sowie Kapazitaet_M1."
    assert fix_transliterations(namen) == namen


def _dijkstra_run(title, note, steps=True):
    from core.model_schema import GraphSolution, ResultTable
    graph = GraphSolution.model_validate({
        "nodes": ["0", "1"], "arcs": [{"source": "0", "target": "1", "weight": 8}], "path": ["0", "1"],
        "steps": [{"node": "0", "label": 0}, {"node": "1", "label": 8}] if steps else [],
        "node_meaning": "Zeitpunkt", "arc_meaning": "Kauf und Nutzung"})
    return SolverRun(status=SolverStatus.OPTIMAL, objective_value=8, graph=graph,
                     tables=[ResultTable(title=title, columns=["Knoten", "Marke"], rows=[["0", 0]], note=note)])


def test_dijkstra_marks_explained_by_the_report_are_not_flagged():
    # beobachtet: „Dijkstra-Markierungen“ gemeldet, obwohl der Report den Markierungsverlauf erklärt
    from solving.solver_stage import check_text_quality
    run = _dijkstra_run("Dijkstra-Markierungen", "Die Knoten in der Reihenfolge ihrer Markierung.")
    assert check_text_quality(run, "Ersatzplanung als kürzester Weg.") == []


def test_marks_without_dijkstra_steps_are_still_flagged():
    from solving.solver_stage import check_text_quality
    run = _dijkstra_run("Marken je Knoten", None, steps=False)
    assert any("Markierung" in b for b in check_text_quality(run, "Ersatzplanung als kürzester Weg."))


def test_everyday_marked_is_no_jargon():
    from checks.model_checks import unexplained_jargon
    assert unexplained_jargon("nur die mit 'ja' markierten Kombinationen gehören zum Plan", "Zusammenfassung") == []


def test_english_unit_words_cover_german_answers():
    # beobachtet: lp_klein auf Englisch – „1 Tonne“ galt als erfunden, weil die Aufgabe „tons“ schreibt
    from solving.solver_stage import check_answer_units
    aufgabe = "One ton of raw material X costs 3 thousand euros; at least 4 tons are required."
    assert check_answer_units(_run_with_answer("Es sollten 1 Tonne X und 3 Tonnen Y gekauft werden, Kosten 9 Tausend Euro."), aufgabe) == []
    assert check_answer_units(_run_with_answer("It takes 12 hours."), "Minimize the cost.") != []   # erfunden bleibt erfunden