"""Prüfung des Anhang-A-Codes. Läuft ohne Desktop und ohne Netz. Zweck: Das Buch behauptet, dieser Code funktioniere. Diese Datei belegt es. Aufruf: python3 test_anhang_a.py """ import decimal import os import sqlite3 import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) BERICHT = [] def pruefe(name, bedingung, detail=""): BERICHT.append((name, bool(bedingung), detail)) if not bedingung: print(f" [FEHLER] {name}: {detail}") # ========================================================================== # 1) betrag_extrahieren # ========================================================================== from helfer import betrag_extrahieren, schaetze_kosten # noqa: E402 LIEFERSCHEIN = """ Muster GmbH · Lieferschein LS-2026-0417 Datum: 14.08.2026 Pos Artikel Menge Einzelpreis Betrag 1 Kabelbinder 200mm 500 0,04 € 20,00 € 2 Montageschiene 2m 40 18,50 € 740,00 € 3 Dübel S8 1200 0,09 € 108,00 € Zwischensumme 868,00 € zzgl. 19 % MwSt. 164,92 € Gesamtbetrag 1.032,92 € """ pruefe( "betrag_extrahieren · deutscher Lieferschein", betrag_extrahieren(LIEFERSCHEIN) == decimal.Decimal("1032.92"), f"ergab {betrag_extrahieren(LIEFERSCHEIN)}, erwartet 1032.92", ) pruefe( "betrag_extrahieren · Zwischensumme gewinnt NICHT", betrag_extrahieren(LIEFERSCHEIN) > decimal.Decimal("868.00"), "Zwischensumme wurde als Gesamtbetrag genommen", ) pruefe( "betrag_extrahieren · Tausenderpunkt", betrag_extrahieren("Summe: 12.345,67 EUR") == decimal.Decimal("12345.67"), f"ergab {betrag_extrahieren('Summe: 12.345,67 EUR')}", ) pruefe( "betrag_extrahieren · ohne Nachkommastellen", betrag_extrahieren("Gesamtbetrag 940 EUR") == decimal.Decimal("940.00"), f"ergab {betrag_extrahieren('Gesamtbetrag 940 EUR')}", ) pruefe( "betrag_extrahieren · unklarer Beleg gibt None, rät nicht", betrag_extrahieren("Lieferschein ohne erkennbaren Betrag") is None, "hat geraten statt None zurückzugeben", ) pruefe( "betrag_extrahieren · leerer Text", betrag_extrahieren(None) is None and betrag_extrahieren("") is None, "leerer Text nicht abgefangen", ) # ========================================================================== # 2) schaetze_kosten — die 272K-Schwelle ist die teuerste Falle des Buches # ========================================================================== # 100.000 Input, 10.000 Output, kein Cache, unter der Schwelle: # 100_000/1e6 * 10 = 1,00 # 10_000/1e6 * 50 = 0,50 => 1,50 k = schaetze_kosten({"input_tokens": 100_000, "output_tokens": 10_000}) pruefe( "schaetze_kosten · Normalfall", k == decimal.Decimal("1.500000"), f"ergab {k}, erwartet 1.50", ) # 300.000 Input (> 272.000) => 2x Input, 1,5x Output # 300_000/1e6 * 10 * 2 = 6,00 # 10_000/1e6 * 50 * 1,5 = 0,75 => 6,75 k = schaetze_kosten({"input_tokens": 300_000, "output_tokens": 10_000}) pruefe( "schaetze_kosten · über der 272K-Schwelle", k == decimal.Decimal("6.750000"), f"ergab {k}, erwartet 6.75", ) # Genau an der Schwelle: 272.000 ist NICHT "mehr als 272K" -> kein Aufschlag k_grenze = schaetze_kosten({"input_tokens": 272_000, "output_tokens": 0}) k_drueber = schaetze_kosten({"input_tokens": 272_001, "output_tokens": 0}) pruefe( "schaetze_kosten · Schwelle greift erst DARÜBER", k_grenze == decimal.Decimal("2.720000") and k_drueber > k_grenze * decimal.Decimal("1.9"), f"an der Grenze {k_grenze}, darüber {k_drueber}", ) # Cache: 90.000 zwischengespeichert, 10.000 frisch # 10_000/1e6 * 10 = 0,10 ; 90_000/1e6 * 1 = 0,09 => 0,19 k = schaetze_kosten( { "input_tokens": 100_000, "output_tokens": 0, "input_tokens_details": {"cached_tokens": 90_000}, } ) pruefe( "schaetze_kosten · Cache wird billiger gerechnet", k == decimal.Decimal("0.190000"), f"ergab {k}, erwartet 0.19", ) pruefe( "schaetze_kosten · Fast mode verdoppelt", schaetze_kosten({"input_tokens": 100_000, "output_tokens": 10_000}, fast_mode=True) == decimal.Decimal("3.000000"), "Fast mode nicht verdoppelt", ) # ========================================================================== # 3) datenbank_summe — die Abnahme, die dem Agenten nicht glaubt # ========================================================================== with tempfile.TemporaryDirectory() as tmp: pfad = os.path.join(tmp, "erfassung.sqlite") with sqlite3.connect(pfad) as con: con.execute("CREATE TABLE lieferscheine (id INTEGER PRIMARY KEY, betrag_cent INTEGER)") con.executemany( "INSERT INTO lieferscheine (betrag_cent) VALUES (?)", [(103292,), (25000,), (1,)], ) os.environ["ZIEL_DB"] = pfad import importlib import helfer importlib.reload(helfer) anzahl, summe = helfer.datenbank_summe() pruefe( "datenbank_summe · Anzahl und Summe", anzahl == 3 and summe == decimal.Decimal("1282.93"), f"ergab {anzahl} Datensätze, Summe {summe}", ) with sqlite3.connect(pfad) as con: con.execute("DELETE FROM lieferscheine") anzahl, summe = helfer.datenbank_summe() pruefe( "datenbank_summe · leere Tabelle stürzt nicht ab", anzahl == 0 and summe == decimal.Decimal("0.00"), f"ergab {anzahl}, {summe}", ) # ========================================================================== # 4) Sandbox: persistenter Namensraum, Fehler, Zeitlimit # ========================================================================== os.environ.setdefault("SANDBOX_TIMEOUT", "3") import sandbox_server as sbx # noqa: E402 s = sbx.get_session("test-1") items, err = sbx.run_code(s, "zaehler = 41", 5) pruefe("sandbox · einfache Zuweisung", err is None, str(err)) items, err = sbx.run_code(s, "zaehler += 1\nlog(zaehler)", 5) pruefe( "sandbox · Variablen überleben zwischen Aufrufen", err is None and items and items[0]["text"] == "42", f"items={items}, err={err}", ) items2, _ = sbx.run_code(s, "log('zweiter')", 5) pruefe( "sandbox · Ausgabe wird je Aufruf geleert", len(items2) == 1 and items2[0]["text"] == "zweiter", f"items={items2}", ) items, err = sbx.run_code(s, "1/0", 5) pruefe( "sandbox · Fehler kommt als Text zurück, kein Absturz", err is not None and "ZeroDivisionError" in err, f"err={err}", ) items, err = sbx.run_code(s, "x = 1", 5) pruefe("sandbox · nach einem Fehler weiter benutzbar", err is None, str(err)) items, err = sbx.run_code(s, "import time as _t\n_t.sleep(10)", 1) pruefe( "sandbox · Zeitlimit greift", err is not None and "Zeitlimit" in err, f"err={err}", ) s2 = sbx.get_session("test-2") items, err = sbx.run_code(s2, "log('zaehler' in dir())", 5) pruefe( "sandbox · Sitzungen sind voneinander getrennt", err is None and items[0]["text"] == "False", f"items={items}", ) # ========================================================================== # Bericht # ========================================================================== print("\n=== PRÜFBERICHT ANHANG A ===") for name, ok, detail in BERICHT: print(f" [{'OK ' if ok else 'FEHLER'}] {name}") fehler = sum(1 for _, ok, _ in BERICHT if not ok) print(f"\n{len(BERICHT) - fehler}/{len(BERICHT)} bestanden.") sys.exit(1 if fehler else 0)