Notebook-Tests: nbmake + Subprocess-Zellen + inkrementeller Test
Alle 25 Notebooks sind nun fehlerfrei testbar (25 passed). Dafür drei
Änderungen im Build und eine neue Test-Infrastruktur:
1. pyproject.toml: nbmake>=1.5 in der dev-Gruppe.
2. build_version_04.py: Notebooks bekommen zwei Setup-Zellen und
Subprocess-Zellen für Programme, die nicht direkt im Kernel laufen:
- NOTEBOOK_SETUP_2: __file__ definieren, sys.path für or_kern,
multiprocessing auf fork (Spawn findet Kernel-Funktionen nicht).
- _braucht_subprocess: erkennt highspy (Solver-Konflikt), pytest
(SystemExit), get_context('spawn'), subprocess+__file__ (sucht
.py-Dateien), sys.exit (SystemExit). Diese Programme erscheinen
als Markdown-Codeblock (sichtbar, nicht ausführbar) plus einer
Subprocess-Zelle mit NO_COLOR=1 (verhindert ANSI-Codes, die der
Parser von Mutationstest.py nicht verarbeitet).
- Solver-Konflikt-Erkennung: wenn ein Kapitel sowohl ortools als
auch highspy importiert, laufen auch reine ortools-Programme als
Subprocess (highspy schon im Kernel).
3. teste_code_04.py: inkrementeller Test — nur geänderte/neue Programme
und Notebooks (via git status). Netzabhängige Dateien (yfinance)
werden automatisch erkannt und nur getestet, wenn Internet verfügbar.
4. PLAN.md Abschnitt 10: neuer Verifikationsschritt 4 (pytest --nbmake).
PROGRESS.md und CLAUDE.md aktualisiert.
Gefundene und behobene Probleme:
- or_kern-Import: sys.path im Notebook erweitert.
- __file__ nicht definiert: zweite Setup-Zelle setzt es.
- multiprocessing spawn: fork erzwungen (Ein_System_Vier_Ansaetze.py).
- ortools/highspy-Konflikt: Subprocess für beide Solver.
- pytest SystemExit: Subprocess für test_or_kern.py und Mutationstest.py.
- sys.exit(0): Subprocess für Installationstest.py.
- ANSI-Codes in pytest-Ausgabe: NO_COLOR=1 in Subprocess-Zelle.
This commit is contained in:
dschlueter
parent
0e2c011732
commit
a4fc247116
70 changed files with 4073 additions and 305 deletions
|
|
@ -157,4 +157,4 @@ Details zu den Lernpfaden stehen in [`00_Vorwort_und_Lesehilfe.md`](00_Vorwort_u
|
|||
|
||||
---
|
||||
|
||||
*Autor / Herausgeber: Dieter Schlüter · Stand: 10. September 2026 v17.51*
|
||||
*Autor / Herausgeber: Dieter Schlüter · Stand: 11. September 2026 v22.38*
|
||||
|
|
|
|||
|
|
@ -2293,6 +2293,59 @@ NOTEBOOK_SETUP = (
|
|||
" scikit-learn matplotlib plotly pyomo linopy pymoo pydantic openpyxl\n"
|
||||
)
|
||||
|
||||
# Zweite Setup-Zelle — wird NICHT von nbmake übersprungen, damit __file__,
|
||||
# sys.path und multiprocessing-Startmodus beim Testen aktiv sind.
|
||||
# __file__ : Programme nutzen es für OUTPUT_DIR (s.o.).
|
||||
# sys.path : 'from or_kern import …' findet or_kern.py im
|
||||
# Programmverzeichnis (lokal und im Docker-Image).
|
||||
# multiprocessing : ProcessPoolExecutor (Ein_System_Vier_Ansaetze.py u.a.)
|
||||
# braucht den 'fork'-Startmodus, weil der Spawn-Modus
|
||||
# die Funktionen in __main__ des Kernels nicht findet.
|
||||
NOTEBOOK_SETUP_2 = (
|
||||
"# Notebook-spezifisches Setup: __file__ definieren (Programme nutzen es\n"
|
||||
"# fuer OUTPUT_DIR), Programmverzeichnis in den Suchpfad aufnehmen (or_kern)\n"
|
||||
"# und multiprocessing auf 'fork' stellen (Spawn findet Kernel-Funktionen nicht).\n"
|
||||
"import os, sys\n"
|
||||
"__file__ = os.path.join(os.getcwd(), '_notebook.py')\n"
|
||||
"for _p in [os.path.join(os.path.dirname(os.getcwd()),\n"
|
||||
" 'Operations_Research_mit_Python_Version_04_Programme'),\n"
|
||||
" os.path.dirname(os.getcwd()), os.getcwd()]:\n"
|
||||
" if os.path.isfile(os.path.join(_p, 'or_kern.py')):\n"
|
||||
" sys.path.insert(0, _p); break\n"
|
||||
"import multiprocessing as _mp\n"
|
||||
"try: _mp.set_start_method('fork', force=True)\n"
|
||||
"except (RuntimeError, ValueError): pass\n"
|
||||
)
|
||||
|
||||
|
||||
def _braucht_subprocess(quelltext: str) -> bool:
|
||||
"""True, wenn das Programm in einem Notebook nicht direkt ausfuehrbar ist.
|
||||
|
||||
Gruende:
|
||||
- import highspy: kollidiert mit ortools im selben Kernel (beide bringen eine
|
||||
eigene HiGHS-Kopie mit). Die Programme loesen das durch getrennte Prozesse;
|
||||
im Notebook muss das Subprocess uebernehmen.
|
||||
- import pytest: pytest.main() beendet den Kernel mit SystemExit.
|
||||
- ProcessPoolExecutor mit explizitem 'spawn': der geforkte Prozess findet
|
||||
die Funktionen in __main__ (dem IPython-Kernel) nicht.
|
||||
"""
|
||||
if "import highspy" in quelltext or "from highspy" in quelltext:
|
||||
return True
|
||||
if "import pytest" in quelltext or "from pytest" in quelltext:
|
||||
return True
|
||||
if 'get_context("spawn")' in quelltext:
|
||||
return True
|
||||
# Programme, die __file__ nutzen, um andere .py-Dateien zu finden
|
||||
# (z.B. Mutationstest.py sucht or_kern.py und test_or_kern.py im selben
|
||||
# Verzeichnis — im Notebook liegen sie nicht dort).
|
||||
if "import subprocess" in quelltext and "__file__" in quelltext:
|
||||
return True
|
||||
# Programme, die sys.exit() nutzen — im Notebook wirft das SystemExit,
|
||||
# das nbmake als Fehler wertet (auch bei Exit-Code 0).
|
||||
if "sys.exit(" in quelltext:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _notebook_zellen(text: str, titel: str) -> list[dict]:
|
||||
"""Baut die Zellenliste eines Kapitel-Notebooks: Titel, Setup-Zelle und
|
||||
|
|
@ -2303,17 +2356,35 @@ def _notebook_zellen(text: str, titel: str) -> list[dict]:
|
|||
return {"cell_type": "markdown", "metadata": {},
|
||||
"source": quelle.splitlines(keepends=True)}
|
||||
|
||||
def code(quelle: str) -> dict:
|
||||
return {"cell_type": "code", "execution_count": None, "metadata": {},
|
||||
"outputs": [], "source": quelle.rstrip("\n").splitlines(keepends=True)}
|
||||
def code(quelle: str, tags: list[str] | None = None) -> dict:
|
||||
zelle = {"cell_type": "code", "execution_count": None, "metadata": {},
|
||||
"outputs": [], "source": quelle.rstrip("\n").splitlines(keepends=True)}
|
||||
if tags:
|
||||
zelle["metadata"]["tags"] = tags
|
||||
return zelle
|
||||
|
||||
zellen = [
|
||||
md(f"# {titel}\n\n"
|
||||
f"Begleitnotebook zu *{SITE_TITEL}*. Die Codezellen sind identisch "
|
||||
f"mit den im Buch abgedruckten Programmen.\n"),
|
||||
code(NOTEBOOK_SETUP),
|
||||
# Setup-Zelle per Tag markieren, damit nbmake sie beim Testen
|
||||
# überspringt - sonst wuerde jeder Testlauf die Pakete neu
|
||||
# installieren (langsam und in der CI netzabhaengig).
|
||||
code(NOTEBOOK_SETUP, tags=["nbmake-skip"]),
|
||||
# Zweite Setup-Zelle (wird beim Testen ausgefuehrt): __file__,
|
||||
# sys.path und multiprocessing-Startmodus.
|
||||
code(NOTEBOOK_SETUP_2),
|
||||
]
|
||||
|
||||
# Pre-Scan: enthaelt das Kapitel sowohl ortools als auch highspy?
|
||||
# Dann beiessen sich die Imports im selben Kernel — alle Solver-Programme
|
||||
# muessen als Subprocess laufen, nicht nur die highspy-Importe.
|
||||
hat_ortools = any("import ortools" in t.group(1) or "from ortools" in t.group(1)
|
||||
for t in CODEBLOCK_RE.finditer(text))
|
||||
hat_highspy = any("import highspy" in t.group(1) or "from highspy" in t.group(1)
|
||||
for t in CODEBLOCK_RE.finditer(text))
|
||||
solver_konflikt = hat_ortools and hat_highspy
|
||||
|
||||
# Zu jedem Codeblock die letzte Ueberschrift davor merken, damit die
|
||||
# Markdown-Zelle sagt, wo im Kapitel man sich befindet.
|
||||
ueberschriften = [(m.start(), m.group(1)) for m in NOTEBOOK_UEBERSCHRIFT_RE.finditer(text)]
|
||||
|
|
@ -2334,7 +2405,43 @@ def _notebook_zellen(text: str, titel: str) -> list[dict]:
|
|||
if programm:
|
||||
kopf += f"\n`{programm.group(1)}`\n"
|
||||
zellen.append(md(kopf or "---\n"))
|
||||
zellen.append(code(quelltext))
|
||||
braucht_sub = _braucht_subprocess(quelltext)
|
||||
# Bei Solver-Konflikt im Kapitel: auch reine ortools-Programme als
|
||||
# Subprocess, weil highspy schon im Kernel geladen sein kann.
|
||||
if solver_konflikt and ("import ortools" in quelltext or "from ortools" in quelltext):
|
||||
braucht_sub = True
|
||||
if braucht_sub:
|
||||
# Code als Markdown-Codeblock zeigen — sichtbar für den Leser,
|
||||
# aber nicht ausführbar (nbmake überspringt Markdown-Zellen).
|
||||
code_block = f"```python\n{quelltext.rstrip()}\n```\n"
|
||||
zellen.append(md((kopf or "---\n") + "\n" + code_block))
|
||||
programm_name = programm.group(1) if programm else "programm.py"
|
||||
subproc = (
|
||||
f"# In einem Notebook kann '{programm_name}' nicht direkt ausgefuehrt werden\n"
|
||||
f"# (ortools/highspy-Konflikt im selben Kernel, pytest oder multiprocessing).\n"
|
||||
f"# Deshalb als Subprocess — so laeuft es genauso wie im Terminal.\n"
|
||||
f"import subprocess, sys, os\n"
|
||||
f"# NO_COLOR/TERM=dumb verhindern ANSI-Codes in der Ausgabe (pytest\n"
|
||||
f"# im Jupyter-Kernel wuerde sonst farbige Escape-Sequenzen ausgeben).\n"
|
||||
f"_env = dict(os.environ, NO_COLOR='1', TERM='dumb')\n"
|
||||
f"for _d in [os.path.join(os.path.dirname(os.getcwd()),\n"
|
||||
f" 'Operations_Research_mit_Python_Version_04_Programme'),\n"
|
||||
f" os.path.dirname(os.getcwd()), os.getcwd()]:\n"
|
||||
f" _p = os.path.join(_d, '{programm_name}')\n"
|
||||
f" if os.path.isfile(_p):\n"
|
||||
f" _r = subprocess.run([sys.executable, _p], cwd=_d, env=_env,\n"
|
||||
f" capture_output=True, text=True)\n"
|
||||
f" if _r.stdout: print(_r.stdout)\n"
|
||||
f" if _r.stderr: print(_r.stderr)\n"
|
||||
f" _r.check_returncode()\n"
|
||||
f" break\n"
|
||||
f"else:\n"
|
||||
f" print('{programm_name} nicht gefunden')\n"
|
||||
)
|
||||
zellen.append(code(subproc))
|
||||
else:
|
||||
zellen.append(md(kopf or "---\n"))
|
||||
zellen.append(code(quelltext))
|
||||
|
||||
return zellen
|
||||
|
||||
|
|
@ -2363,7 +2470,7 @@ def baue_notebooks(struktur: list[dict]) -> dict[str, str]:
|
|||
titel = (f'{eintrag["nummer"]}: {eintrag["titel"]}'
|
||||
if eintrag["nummer"] else eintrag["titel"])
|
||||
zellen = _notebook_zellen(text, titel)
|
||||
if len(zellen) <= 2: # nur Titel + Setup => kein Code im Kapitel
|
||||
if len(zellen) <= 3: # nur Titel + 2 Setup-Zellen => kein Code im Kapitel
|
||||
continue
|
||||
|
||||
name = eintrag["seite"].replace(".html", ".ipynb")
|
||||
|
|
|
|||
235
Operations_Research_mit_Python_Version_04/teste_code_04.py
Normal file
235
Operations_Research_mit_Python_Version_04/teste_code_04.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
#!/usr/bin/env python3
|
||||
# teste_code_04.py
|
||||
"""
|
||||
Inkrementeller Test der Programme und Notebooks.
|
||||
|
||||
Testet nur, was sich seit dem letzten Commit geaendert hat oder neu
|
||||
hinzugekommen ist. Netzabhaengige Dateien (yfinance-Downloads) werden
|
||||
automatisch erkannt und nur getestet, wenn Internet verfuegbar ist;
|
||||
scheitern sie am Netz, gilt das nicht als Fehler.
|
||||
|
||||
Aufruf:
|
||||
python3 teste_code_04.py # nur geaenderte/neue
|
||||
python3 teste_code_04.py --alles # alle Programme und Notebooks
|
||||
python3 teste_code_04.py --check # nur anzeigen, was getestet wuerde
|
||||
|
||||
Voraussetzung: pip install -e ".[dev,notebook]"
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
HIER = os.path.dirname(os.path.abspath(__file__))
|
||||
BASIS = os.path.dirname(HIER)
|
||||
PROG_DIR = os.path.join(BASIS, "Operations_Research_mit_Python_Version_04_Programme")
|
||||
NB_DIR = os.path.join(BASIS, "Notebooks_04")
|
||||
TIMEOUT = 900 # 15 Minuten pro Programm/Notebook (Solver können dauern)
|
||||
|
||||
|
||||
# --- Erkennung netzabhaengiger Dateien ---------------------------------------
|
||||
|
||||
def netz_abhaengig(datei: str) -> bool:
|
||||
"""True, wenn die Datei yfinance importiert (Programm oder Notebook)."""
|
||||
try:
|
||||
with open(datei, encoding="utf-8") as f:
|
||||
inhalt = f.read()
|
||||
except OSError:
|
||||
return False
|
||||
return "import yfinance" in inhalt or "from yfinance" in inhalt
|
||||
|
||||
|
||||
def internet_verfuegbar() -> bool:
|
||||
"""Prüft, ob yfinance tatsächlich Daten laden kann (nicht nur Port-Erreichbarkeit)."""
|
||||
try:
|
||||
import yfinance as yf
|
||||
h = yf.Ticker("AAPL").history(period="5d")
|
||||
return len(h) > 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# --- Geaenderte Dateien via Git ermitteln ------------------------------------
|
||||
|
||||
def geaenderte_dateien() -> tuple[set[str], set[str]]:
|
||||
"""Liefert (geänderte Programme, geänderte Notebooks) relativ zu BASIS.
|
||||
|
||||
'Geändert' = modifiziert oder neu (untracked) gegenüber HEAD.
|
||||
"""
|
||||
# --porcelain liefert zeilenweise 'XY datei'; X=Y=?? bedeutet untracked.
|
||||
ergebnis = subprocess.run(
|
||||
["git", "status", "--porcelain"],
|
||||
cwd=BASIS, capture_output=True, text=True, check=True,
|
||||
)
|
||||
programme: set[str] = set()
|
||||
notebooks: set[str] = set()
|
||||
for zeile in ergebnis.stdout.splitlines():
|
||||
status, datei = zeile[:2], zeile[3:]
|
||||
# Umbenennungen werden als 'XY alt -> neu' geschrieben.
|
||||
if " -> " in datei:
|
||||
datei = datei.split(" -> ", 1)[1]
|
||||
if datei.startswith("Operations_Research_mit_Python_Version_04_Programme/") \
|
||||
and datei.endswith(".py"):
|
||||
programme.add(os.path.basename(datei))
|
||||
elif datei.startswith("Notebooks_04/") and datei.endswith(".ipynb"):
|
||||
notebooks.add(os.path.basename(datei))
|
||||
return programme, notebooks
|
||||
|
||||
|
||||
def alle_dateien() -> tuple[list[str], list[str]]:
|
||||
"""Alle vorhandenen Programme und Notebooks."""
|
||||
programme = sorted(
|
||||
f for f in os.listdir(PROG_DIR) if f.endswith(".py") and f != "or_kern.py"
|
||||
)
|
||||
notebooks = sorted(f for f in os.listdir(NB_DIR) if f.endswith(".ipynb"))
|
||||
return programme, notebooks
|
||||
|
||||
|
||||
# --- Tests -------------------------------------------------------------------
|
||||
|
||||
def test_programm(name: str, netz_ok: bool) -> str:
|
||||
"""Führt ein Programm aus. Rückgabe: 'ok', 'fehler', 'netz-übersprungen'."""
|
||||
pfad = os.path.join(PROG_DIR, name)
|
||||
if netz_abhaengig(pfad) and not netz_ok:
|
||||
print(f" {name}: ÜBERSPRUNGEN (braucht Internet, nicht verfügbar)")
|
||||
return "netz"
|
||||
print(f" {name}: läuft …")
|
||||
umgebung = dict(os.environ, MPLBACKEND="Agg")
|
||||
try:
|
||||
subprocess.run(
|
||||
[sys.executable, pfad],
|
||||
env=umgebung, timeout=TIMEOUT,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
check=True,
|
||||
)
|
||||
print(f" {name}: OK")
|
||||
return "ok"
|
||||
except subprocess.TimeoutExpired:
|
||||
print(f" {name}: FEHLER (Timeout nach {TIMEOUT}s)")
|
||||
return "fehler"
|
||||
except subprocess.CalledProcessError as e:
|
||||
ausgabe = (e.stderr or e.stdout or b"").decode("utf-8", errors="replace")
|
||||
# Netzfehler bei einem netzabhängigen Programm → kein harter Fehler
|
||||
if netz_abhaengig(pfad) and "Download fehlgeschlagen" in ausgabe:
|
||||
print(f" {name}: ÜBERSPRUNGEN (Download fehlgeschlagen — kein Internet)")
|
||||
return "netz"
|
||||
print(f" {name}: FEHLER")
|
||||
for zeile in ausgabe.splitlines()[-15:]:
|
||||
print(f" {zeile}")
|
||||
return "fehler"
|
||||
|
||||
|
||||
def test_notebooks(names: list[str], netz_ok: bool) -> int:
|
||||
"""Führt Notebooks via pytest --nbmake aus. Rückgabe: Anzahl Fehler."""
|
||||
if not names:
|
||||
return 0
|
||||
# Netzabhängige Notebooks ohne Internet herausfiltern
|
||||
testen: list[str] = []
|
||||
for name in names:
|
||||
pfad = os.path.join(NB_DIR, name)
|
||||
if netz_abhaengig(pfad) and not netz_ok:
|
||||
print(f" {name}: ÜBERSPRUNGEN (braucht Internet, nicht verfügbar)")
|
||||
else:
|
||||
testen.append(pfad)
|
||||
if not testen:
|
||||
return 0
|
||||
umgebung = dict(os.environ, MPLBACKEND="Agg")
|
||||
cmd = [sys.executable, "-m", "pytest", "--nbmake",
|
||||
f"--nbmake-timeout={TIMEOUT}", "-q", *testen]
|
||||
print(f" pytest --nbmake für {len(testen)} Notebook(s) …")
|
||||
ergebnis = subprocess.run(cmd, env=umgebung, cwd=BASIS)
|
||||
return 0 if ergebnis.returncode == 0 else 1
|
||||
|
||||
|
||||
# --- Hauptprogramm -----------------------------------------------------------
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--alles", action="store_true",
|
||||
help="Alle Programme und Notebooks testen, nicht nur geänderte.")
|
||||
parser.add_argument("--check", action="store_true",
|
||||
help="Nur anzeigen, was getestet würde, nichts ausführen.")
|
||||
args = parser.parse_args()
|
||||
|
||||
print("=== Inkrementeller Test der Programme und Notebooks ===\n")
|
||||
|
||||
# 1. Programme und Notebooks regenerieren
|
||||
print("1. Programme extrahieren …")
|
||||
subprocess.run(
|
||||
[sys.executable, os.path.join(HIER, "extract_programme_04.py")],
|
||||
cwd=BASIS, check=True,
|
||||
)
|
||||
print(" Notebooks bauen …")
|
||||
subprocess.run(
|
||||
[sys.executable, os.path.join(HIER, "build_version_04.py"), "--html"],
|
||||
cwd=BASIS, capture_output=True, check=True,
|
||||
)
|
||||
|
||||
# 2. Geänderte Dateien ermitteln
|
||||
if args.alles:
|
||||
programme, notebooks = alle_dateien()
|
||||
modus = "alle"
|
||||
else:
|
||||
programme_set, notebooks_set = geaenderte_dateien()
|
||||
programme = sorted(programme_set)
|
||||
notebooks = sorted(notebooks_set)
|
||||
modus = "geänderte/neue"
|
||||
|
||||
# or_kern.py ist eine Bibliothek, kein Programm — nie direkt ausführen
|
||||
if "or_kern.py" in programme:
|
||||
programme.remove("or_kern.py")
|
||||
|
||||
print(f"\n2. {modus.capitalize()} Dateien: "
|
||||
f"{len(programme)} Programm(e), {len(notebooks)} Notebook(s)")
|
||||
|
||||
if not programme and not notebooks:
|
||||
print(" Nichts zu testen — alles aktuell.")
|
||||
return 0
|
||||
|
||||
if programme:
|
||||
print(" Programme:")
|
||||
for p in programme:
|
||||
print(f" {p}")
|
||||
if notebooks:
|
||||
print(" Notebooks:")
|
||||
for nb in notebooks:
|
||||
print(f" {nb}")
|
||||
|
||||
if args.check:
|
||||
print("\n--check: Nur Anzeige, kein Testlauf.")
|
||||
return 0
|
||||
|
||||
# 3. Internet-Check
|
||||
netz_ok = internet_verfuegbar()
|
||||
print(f"\n3. Internet: {'verfügbar' if netz_ok else 'nicht verfügbar'}"
|
||||
f" (yfinance-Programme werden "
|
||||
f"{'getestet' if netz_ok else 'übersprungen'})")
|
||||
|
||||
# 4. Programme testen
|
||||
fehler_zaehler = 0
|
||||
if programme:
|
||||
print("\n4. Programme testen:")
|
||||
for p in programme:
|
||||
ergebnis = test_programm(p, netz_ok)
|
||||
if ergebnis == "fehler":
|
||||
fehler_zaehler += 1
|
||||
|
||||
# 5. Notebooks testen
|
||||
if notebooks:
|
||||
print("\n5. Notebooks testen:")
|
||||
nb_fehler = test_notebooks(notebooks, netz_ok)
|
||||
fehler_zaehler += nb_fehler
|
||||
|
||||
# 6. Zusammenfassung
|
||||
print(f"\n=== Zusammenfassung ===")
|
||||
if fehler_zaehler > 0:
|
||||
print(f" {fehler_zaehler} Fehler insgesamt.")
|
||||
return 1
|
||||
print(" Alle getesteten Programme und Notebooks fehlerfrei.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Reference in a new issue