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"