Automatischer Push: 11 Datei(en) geändert
- .claude/agents/planer.md - .claude/workflows/plankritik.js - .vault-tools/plan-kopie/LIESMICH.md - .vault-tools/plan-kopie/mutanten_lauf.py - .vault-tools/plan-kopie/mutanten_lauf_ENTWURF.md - .vault-tools/plan-kopie/mutanten_lauf_kopie.sh - .vault-tools/plan-kopie/mutanten_lauf_treiber.py - .vault-tools/workflow-pruefung/faelle_workflows.mjs - .vault-tools/workflow-pruefung/mutanten_workflows.py - AI/Regelwerk/codeworkflow.md - CLAUDE.md Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5 committed
1 parent
3b7128c9e1
commit
1be4f913aa
11 files changed
+2214
-10
No files matched your search
@@ -0,0 +1,853 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Allgemeiner Mutantenlaeufer aus einer mutanten.toml.
|
||||
|
||||
Zweck: Fuer eine Liste von Mutanten (Textersetzungen in einer Datei) prueft
|
||||
dieses Skript, ob die dazu genannten Tests unter jedem Mutanten wie erwartet
|
||||
rot werden ("getoetet"). Es liest dazu genau eine Datei im Format aus dem
|
||||
Entwurf (Abschnitt 2), nicht Prosa aus einem Plan.
|
||||
|
||||
Aufruf (in WSL, Python ab 3.11, nur Standardbibliothek):
|
||||
python mutanten_lauf.py --wurzel <kopie> --mutanten <mutanten.toml>
|
||||
[--nur ID,ID,...] [--nur-muster] [--bericht <datei.json>]
|
||||
[--zeitlimit <sekunden>]
|
||||
|
||||
Geltungsbereich: Dieses Werkzeug ist ein Fangnetz, kein Nachweis. Es sagt,
|
||||
ob die genannten Tests unter einer Aenderung fallen - nicht, ob sie an der
|
||||
gemeinten Stelle fallen. Das zeigt nur die mitgelieferte Fehlermeldung, das
|
||||
Urteil bleibt beim Menschen.
|
||||
|
||||
Benannte Luecken der Abdichtung (Entwurf, Abschnitt 5): Ein Unterprozess,
|
||||
den ein Test mit absolutem Pfad startet, ist nicht abgedichtet. Ein Socket
|
||||
ohne Proxy-Nutzung ist nicht abgedichtet. Ein HTTP-Client mit
|
||||
trust_env=False oder eigenem Transport ist nicht abgedichtet. Dateien mit
|
||||
absolutem Pfad ausserhalb von HOME sind nicht abgedichtet. Die eigentliche
|
||||
Sperre bleibt die Netzsperre der Testdateien selbst und eine Kopie ohne
|
||||
Geheimnisdateien (der Schluessel "ohne" in mutanten.toml).
|
||||
|
||||
Spezifikation: .vault-tools/plan-kopie/mutanten_lauf_ENTWURF.md
|
||||
|
||||
Entscheidungen, die der Entwurf offen liess (siehe Bericht des Executers):
|
||||
Verhalten bei einer Zieldatei, die kein gueltiges UTF-8 ist, wird wie ein
|
||||
Formatbefund behandelt (Abbruch mit Exit 2), weil das Skript sonst keine
|
||||
verlaessliche Aussage ueber die Trefferzahl treffen kann.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import tomllib
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
|
||||
PLUGIN_NAME = "_mutanten_lauf_plugin"
|
||||
|
||||
PLUGIN_QUELLTEXT = '''"""Pytest-Plugin des Mutantenlaeufers, wird in einen Temp-Ordner geschrieben.
|
||||
|
||||
Schreibt je Bericht eine JSON-Zeile in die Datei aus der Umgebungsvariable
|
||||
MUTANTEN_LAUF_ERGEBNIS. Siehe mutanten_lauf_ENTWURF.md, Abschnitt 3.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
|
||||
def _ergebnis_pfad():
|
||||
return os.environ.get("MUTANTEN_LAUF_ERGEBNIS")
|
||||
|
||||
|
||||
def _schreiben(datensatz):
|
||||
pfad = _ergebnis_pfad()
|
||||
if not pfad:
|
||||
return
|
||||
with open(pfad, "a", encoding="utf-8") as datei:
|
||||
datei.write(json.dumps(datensatz, ensure_ascii=True))
|
||||
datei.write("\\n")
|
||||
datei.flush()
|
||||
|
||||
|
||||
def _erste_zeile(text):
|
||||
for zeile in str(text).splitlines():
|
||||
zeile = zeile.strip()
|
||||
if zeile:
|
||||
return zeile[:300]
|
||||
return ""
|
||||
|
||||
|
||||
def _meldung_aus_longrepr(longrepr):
|
||||
if longrepr is None:
|
||||
return ""
|
||||
crash = getattr(longrepr, "reprcrash", None)
|
||||
meldung = getattr(crash, "message", None)
|
||||
if meldung:
|
||||
return str(meldung)[:300]
|
||||
return _erste_zeile(longrepr)
|
||||
|
||||
|
||||
def pytest_runtest_logreport(report):
|
||||
meldung = ""
|
||||
if report.outcome == "failed":
|
||||
meldung = _meldung_aus_longrepr(getattr(report, "longrepr", None))
|
||||
datensatz = {
|
||||
"art": "test",
|
||||
"knoten": report.nodeid,
|
||||
"phase": report.when,
|
||||
"ergebnis": report.outcome,
|
||||
"xfail": hasattr(report, "wasxfail"),
|
||||
"meldung": meldung,
|
||||
}
|
||||
_schreiben(datensatz)
|
||||
|
||||
|
||||
def pytest_collectreport(report):
|
||||
if report.outcome == "failed":
|
||||
meldung = _meldung_aus_longrepr(getattr(report, "longrepr", None))
|
||||
_schreiben({"art": "sammeln", "knoten": report.nodeid, "meldung": meldung})
|
||||
|
||||
|
||||
def pytest_deselected(items):
|
||||
for item in items:
|
||||
_schreiben({"art": "abgewaehlt", "knoten": item.nodeid})
|
||||
'''
|
||||
|
||||
TOP_LEVEL_SCHLUESSEL = {"format", "python", "ohne", "kuerzel", "python_je_datei", "mutant"}
|
||||
MUTANT_SCHLUESSEL = {"id", "datei", "aus", "wird", "rot_bei", "hinweis"}
|
||||
|
||||
|
||||
class RuecksetzenGescheitert(Exception):
|
||||
def __init__(self, mutant_id, datei):
|
||||
super().__init__("Ruecksetzen gescheitert bei Mutant {0} ({1})".format(mutant_id, datei))
|
||||
|
||||
|
||||
def ist_relativer_pfad(pfad):
|
||||
if not isinstance(pfad, str) or pfad == "":
|
||||
return False
|
||||
if pfad.startswith("/") or pfad.startswith("~"):
|
||||
return False
|
||||
if len(pfad) >= 2 and pfad[1] == ":":
|
||||
return False
|
||||
teile = pfad.split("/")
|
||||
if ".." in teile:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def zeilen_aus_text(text):
|
||||
if text == "":
|
||||
return []
|
||||
if text.endswith("\n"):
|
||||
text = text[:-1]
|
||||
return text.split("\n")
|
||||
|
||||
|
||||
def positionen_finden(datei_zeilen, muster_zeilen):
|
||||
treffer = []
|
||||
n = len(muster_zeilen)
|
||||
if n == 0:
|
||||
return treffer
|
||||
obergrenze = len(datei_zeilen) - n + 1
|
||||
for i in range(max(obergrenze, 0)):
|
||||
if datei_zeilen[i:i + n] == muster_zeilen:
|
||||
treffer.append(i)
|
||||
return treffer
|
||||
|
||||
|
||||
def kuerzel_aufloesen(eintrag, kuerzel_tabelle):
|
||||
if "::" not in eintrag:
|
||||
return None
|
||||
vorderteil, rest = eintrag.split("::", 1)
|
||||
if vorderteil in kuerzel_tabelle:
|
||||
return kuerzel_tabelle[vorderteil] + "::" + rest
|
||||
if ist_relativer_pfad(vorderteil) and vorderteil.endswith(".py"):
|
||||
return vorderteil + "::" + rest
|
||||
return None
|
||||
|
||||
|
||||
def interpreter_gueltig(pfad):
|
||||
return isinstance(pfad, str) and pfad != "" and "~" not in pfad and pfad.startswith("/") \
|
||||
and os.path.isfile(pfad) and os.access(pfad, os.X_OK)
|
||||
|
||||
|
||||
def lade_und_validiere(args):
|
||||
"""Schritt 1 des Ablaufs: Vorbedingungen und Format (Entwurf, Abschnitt 3.1)."""
|
||||
befunde = []
|
||||
wurzel = Path(args.wurzel)
|
||||
wurzel_ok = wurzel.exists() and wurzel.is_dir()
|
||||
if not wurzel_ok:
|
||||
befunde.append("Wurzel fehlt oder ist kein Ordner: {0}".format(wurzel))
|
||||
|
||||
if wurzel_ok:
|
||||
git_pfad = wurzel / ".git"
|
||||
if git_pfad.exists() or git_pfad.is_symlink():
|
||||
befunde.append("Wurzel enthaelt .git: {0}".format(git_pfad))
|
||||
|
||||
mutanten_pfad = Path(args.mutanten)
|
||||
daten = None
|
||||
try:
|
||||
with open(mutanten_pfad, "rb") as datei:
|
||||
daten = tomllib.load(datei)
|
||||
except FileNotFoundError:
|
||||
befunde.append("Mutanten-Datei fehlt: {0}".format(mutanten_pfad))
|
||||
except tomllib.TOMLDecodeError as exc:
|
||||
befunde.append("TOML nicht lesbar: {0}".format(exc))
|
||||
except OSError as exc:
|
||||
befunde.append("TOML nicht lesbar: {0}".format(exc))
|
||||
|
||||
kontext = {
|
||||
"wurzel": wurzel,
|
||||
"mutanten_pfad": mutanten_pfad,
|
||||
"python": None,
|
||||
"ohne": [],
|
||||
"kuerzel": {},
|
||||
"python_je_datei": {},
|
||||
"mutanten": [],
|
||||
}
|
||||
|
||||
if daten is None:
|
||||
return befunde, kontext
|
||||
|
||||
if not isinstance(daten, dict):
|
||||
befunde.append("Mutanten-Datei ist auf oberster Ebene keine Tabelle")
|
||||
return befunde, kontext
|
||||
|
||||
unbekannt = set(daten.keys()) - TOP_LEVEL_SCHLUESSEL
|
||||
for schluessel in sorted(unbekannt):
|
||||
befunde.append("unbekannter Schluessel auf oberster Ebene: {0}".format(schluessel))
|
||||
|
||||
if "format" not in daten:
|
||||
befunde.append("format fehlt")
|
||||
else:
|
||||
format_wert = daten.get("format")
|
||||
if not isinstance(format_wert, int) or isinstance(format_wert, bool) or format_wert != 1:
|
||||
befunde.append("format muss die Ganzzahl 1 sein, ist {0!r}".format(format_wert))
|
||||
|
||||
if "python" not in daten:
|
||||
befunde.append("python fehlt")
|
||||
else:
|
||||
python_wert = daten.get("python")
|
||||
if not isinstance(python_wert, str) or python_wert == "":
|
||||
befunde.append("python ist kein nicht-leerer Text")
|
||||
elif "~" in python_wert:
|
||||
befunde.append("python enthaelt ~: {0}".format(python_wert))
|
||||
elif not python_wert.startswith("/"):
|
||||
befunde.append("python ist kein absoluter Pfad: {0}".format(python_wert))
|
||||
elif not (os.path.isfile(python_wert) and os.access(python_wert, os.X_OK)):
|
||||
befunde.append("python fehlt oder ist nicht ausfuehrbar: {0}".format(python_wert))
|
||||
else:
|
||||
kontext["python"] = python_wert
|
||||
|
||||
ohne_wert = []
|
||||
if "ohne" in daten:
|
||||
roh = daten.get("ohne")
|
||||
if not isinstance(roh, list):
|
||||
befunde.append("ohne ist keine Liste")
|
||||
else:
|
||||
for eintrag in roh:
|
||||
if not ist_relativer_pfad(eintrag):
|
||||
befunde.append("ohne enthaelt keinen relativen Pfad: {0!r}".format(eintrag))
|
||||
else:
|
||||
ohne_wert.append(eintrag)
|
||||
kontext["ohne"] = ohne_wert
|
||||
|
||||
if wurzel_ok:
|
||||
for eintrag in ohne_wert:
|
||||
ziel = wurzel / eintrag
|
||||
if ziel.exists() or ziel.is_symlink():
|
||||
befunde.append("Pfad aus ohne existiert: {0}".format(eintrag))
|
||||
|
||||
kuerzel_wert = {}
|
||||
if "kuerzel" in daten:
|
||||
roh = daten.get("kuerzel")
|
||||
if not isinstance(roh, dict):
|
||||
befunde.append("kuerzel ist keine Tabelle")
|
||||
else:
|
||||
for name, pfad in roh.items():
|
||||
if not ist_relativer_pfad(pfad) or not str(pfad).endswith(".py"):
|
||||
befunde.append("kuerzel {0} zeigt auf keinen gueltigen relativen Pfad: {1!r}".format(name, pfad))
|
||||
else:
|
||||
kuerzel_wert[name] = pfad
|
||||
kontext["kuerzel"] = kuerzel_wert
|
||||
|
||||
pjd_wert = {}
|
||||
if "python_je_datei" in daten:
|
||||
roh = daten.get("python_je_datei")
|
||||
if not isinstance(roh, dict):
|
||||
befunde.append("python_je_datei ist keine Tabelle")
|
||||
else:
|
||||
for pfad, interpreter in roh.items():
|
||||
if not ist_relativer_pfad(pfad) or not str(pfad).endswith(".py"):
|
||||
befunde.append("python_je_datei-Schluessel ist kein gueltiger relativer Pfad: {0!r}".format(pfad))
|
||||
continue
|
||||
if not interpreter_gueltig(interpreter):
|
||||
befunde.append("Interpreter in python_je_datei fehlt oder ist nicht ausfuehrbar: {0!r}".format(interpreter))
|
||||
continue
|
||||
pjd_wert[pfad] = interpreter
|
||||
kontext["python_je_datei"] = pjd_wert
|
||||
|
||||
if "mutant" not in daten:
|
||||
befunde.append("mutant fehlt")
|
||||
mutant_liste_roh = []
|
||||
else:
|
||||
mutant_liste_roh = daten.get("mutant")
|
||||
if not isinstance(mutant_liste_roh, list) or len(mutant_liste_roh) == 0:
|
||||
befunde.append("mutant ist keine nicht-leere Liste")
|
||||
mutant_liste_roh = []
|
||||
|
||||
gesehene_ids = set()
|
||||
mutanten_normalisiert = []
|
||||
for index, eintrag in enumerate(mutant_liste_roh):
|
||||
vorspann = "Mutant Nr. {0}".format(index + 1)
|
||||
if not isinstance(eintrag, dict):
|
||||
befunde.append("{0} ist keine Tabelle".format(vorspann))
|
||||
continue
|
||||
|
||||
unbekannt = set(eintrag.keys()) - MUTANT_SCHLUESSEL
|
||||
for schluessel in sorted(unbekannt):
|
||||
befunde.append("{0}: unbekannter Schluessel {1}".format(vorspann, schluessel))
|
||||
|
||||
mid = eintrag.get("id")
|
||||
if not isinstance(mid, str) or mid == "":
|
||||
befunde.append("{0}: id fehlt oder ist leer".format(vorspann))
|
||||
mid = None
|
||||
elif mid in gesehene_ids:
|
||||
befunde.append("doppelte id: {0}".format(mid))
|
||||
else:
|
||||
gesehene_ids.add(mid)
|
||||
|
||||
bezeichner = mid if mid else vorspann
|
||||
|
||||
datei = eintrag.get("datei")
|
||||
datei_ok = False
|
||||
if not isinstance(datei, str) or datei == "":
|
||||
befunde.append("{0}: datei fehlt oder ist leer".format(bezeichner))
|
||||
elif not ist_relativer_pfad(datei):
|
||||
befunde.append("{0}: datei ist kein gueltiger relativer Pfad: {1!r}".format(bezeichner, datei))
|
||||
elif wurzel_ok:
|
||||
ziel = wurzel / datei
|
||||
if ziel.is_symlink():
|
||||
befunde.append("{0}: datei ist eine Verknuepfung: {1}".format(bezeichner, datei))
|
||||
elif not ziel.is_file():
|
||||
befunde.append("{0}: datei existiert nicht: {1}".format(bezeichner, datei))
|
||||
else:
|
||||
datei_ok = True
|
||||
|
||||
aus = eintrag.get("aus")
|
||||
if not isinstance(aus, str) or aus == "":
|
||||
befunde.append("{0}: aus fehlt oder ist leer".format(bezeichner))
|
||||
aus = None
|
||||
|
||||
wird = eintrag.get("wird")
|
||||
if not isinstance(wird, str):
|
||||
befunde.append("{0}: wird fehlt oder ist kein Text".format(bezeichner))
|
||||
wird = None
|
||||
|
||||
if aus is not None and wird is not None and aus == wird:
|
||||
befunde.append("{0}: aus ist gleich wird (wirkungslos)".format(bezeichner))
|
||||
|
||||
rot_bei_roh = eintrag.get("rot_bei")
|
||||
rot_bei = []
|
||||
if not isinstance(rot_bei_roh, list) or len(rot_bei_roh) == 0:
|
||||
befunde.append("{0}: rot_bei ist keine nicht-leere Liste".format(bezeichner))
|
||||
else:
|
||||
for roheintrag in rot_bei_roh:
|
||||
if not isinstance(roheintrag, str) or roheintrag == "":
|
||||
befunde.append("{0}: rot_bei enthaelt keinen nicht-leeren Text: {1!r}".format(bezeichner, roheintrag))
|
||||
continue
|
||||
aufgeloest = kuerzel_aufloesen(roheintrag, kuerzel_wert)
|
||||
if aufgeloest is None:
|
||||
befunde.append("{0}: unbekanntes Kuerzel oder ungueltiger Eintrag in rot_bei: {1!r}".format(bezeichner, roheintrag))
|
||||
else:
|
||||
rot_bei.append((roheintrag, aufgeloest))
|
||||
|
||||
hinweis = eintrag.get("hinweis")
|
||||
if "hinweis" in eintrag and not isinstance(hinweis, str):
|
||||
befunde.append("{0}: hinweis ist kein Text".format(bezeichner))
|
||||
hinweis = None
|
||||
|
||||
if mid is not None and datei_ok and aus is not None and wird is not None and rot_bei:
|
||||
mutanten_normalisiert.append({
|
||||
"id": mid,
|
||||
"datei": datei,
|
||||
"aus": aus,
|
||||
"wird": wird,
|
||||
"rot_bei": rot_bei,
|
||||
"hinweis": hinweis,
|
||||
})
|
||||
|
||||
kontext["mutanten"] = mutanten_normalisiert
|
||||
return befunde, kontext
|
||||
|
||||
|
||||
def muster_pruefen(mutant, wurzel):
|
||||
"""Schritt 2 des Ablaufs (Entwurf, Abschnitt 3.2)."""
|
||||
ziel = wurzel / mutant["datei"]
|
||||
rohbytes = ziel.read_bytes()
|
||||
try:
|
||||
text = rohbytes.decode("utf-8", errors="strict")
|
||||
except UnicodeDecodeError as exc:
|
||||
return {"kodierungsfehler": str(exc)}
|
||||
|
||||
datei_zeilen = text.split("\n")
|
||||
aus_zeilen = zeilen_aus_text(mutant["aus"])
|
||||
wird_zeilen = zeilen_aus_text(mutant["wird"])
|
||||
treffer_positionen = positionen_finden(datei_zeilen, aus_zeilen)
|
||||
|
||||
return {
|
||||
"kodierungsfehler": None,
|
||||
"treffer": len(treffer_positionen),
|
||||
"start": treffer_positionen[0] if len(treffer_positionen) == 1 else None,
|
||||
"aus_zeilen": aus_zeilen,
|
||||
"wird_zeilen": wird_zeilen,
|
||||
"cr_in_datei": "\r" in text,
|
||||
}
|
||||
|
||||
|
||||
def gruppiere_nach_interpreter(knoten_liste, standard_python, python_je_datei):
|
||||
gruppen = {}
|
||||
for knoten in knoten_liste:
|
||||
dateipfad = knoten.split("::", 1)[0]
|
||||
interpreter = python_je_datei.get(dateipfad, standard_python)
|
||||
gruppen.setdefault(interpreter, []).append(knoten)
|
||||
return gruppen
|
||||
|
||||
|
||||
def umgebung_abdichten(home_dir, plugin_dir, ergebnis_pfad):
|
||||
"""Abdichtung nach Entwurf, Abschnitt 5. Siehe Kopfkommentar fuer die Luecken."""
|
||||
env = os.environ.copy()
|
||||
env["HOME"] = str(home_dir)
|
||||
for schluessel in ("http_proxy", "https_proxy", "all_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"):
|
||||
env[schluessel] = "http://127.0.0.1:9"
|
||||
for schluessel in ("no_proxy", "NO_PROXY"):
|
||||
env.pop(schluessel, None)
|
||||
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
||||
env["COLUMNS"] = "300"
|
||||
env["MUTANTEN_LAUF_ERGEBNIS"] = ergebnis_pfad
|
||||
vorhanden = env.get("PYTHONPATH", "")
|
||||
env["PYTHONPATH"] = str(plugin_dir) + (os.pathsep + vorhanden if vorhanden else "")
|
||||
return env
|
||||
|
||||
|
||||
def pytest_lauf(python_exe, knoten_liste, wurzel, home_dir, plugin_dir, zeitlimit):
|
||||
ergebnis_datei = tempfile.NamedTemporaryFile(prefix="mutanten_lauf_ergebnis_", suffix=".jsonl", delete=False)
|
||||
ergebnis_datei.close()
|
||||
ergebnis_pfad = ergebnis_datei.name
|
||||
env = umgebung_abdichten(home_dir, plugin_dir, ergebnis_pfad)
|
||||
cmd = [python_exe, "-m", "pytest", "-q", "-p", "no:cacheprovider", "-p", PLUGIN_NAME] + list(knoten_liste)
|
||||
|
||||
zeit_ueberschritten = False
|
||||
absturz = None
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
cmd, cwd=str(wurzel), env=env,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
start_new_session=True,
|
||||
)
|
||||
try:
|
||||
proc.communicate(timeout=zeitlimit)
|
||||
except subprocess.TimeoutExpired:
|
||||
zeit_ueberschritten = True
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
||||
except (ProcessLookupError, PermissionError):
|
||||
pass
|
||||
proc.communicate()
|
||||
except OSError as exc:
|
||||
absturz = "pytest-Aufruf gescheitert ({0}): {1}".format(python_exe, exc)
|
||||
|
||||
meldungen = []
|
||||
try:
|
||||
with open(ergebnis_pfad, "r", encoding="utf-8") as datei:
|
||||
for zeile in datei:
|
||||
zeile = zeile.strip()
|
||||
if not zeile:
|
||||
continue
|
||||
try:
|
||||
meldungen.append(json.loads(zeile))
|
||||
except ValueError:
|
||||
pass
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
try:
|
||||
os.remove(ergebnis_pfad)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return zeit_ueberschritten, meldungen, absturz
|
||||
|
||||
|
||||
def gruppen_ausfuehren(gruppen, wurzel, home_dir, plugin_dir, zeitlimit):
|
||||
reports = {}
|
||||
deselected = set()
|
||||
sammelfehler = {}
|
||||
zeit_ueberschritten = False
|
||||
absturzmeldungen = []
|
||||
for python_exe, knoten_liste in gruppen.items():
|
||||
zeit, meldungen, absturz = pytest_lauf(python_exe, knoten_liste, wurzel, home_dir, plugin_dir, zeitlimit)
|
||||
if zeit:
|
||||
zeit_ueberschritten = True
|
||||
if absturz:
|
||||
absturzmeldungen.append(absturz)
|
||||
for m in meldungen:
|
||||
art = m.get("art")
|
||||
if art == "test":
|
||||
reports.setdefault(m.get("knoten"), []).append({
|
||||
"phase": m.get("phase"),
|
||||
"ergebnis": m.get("ergebnis"),
|
||||
"xfail": bool(m.get("xfail")),
|
||||
"meldung": m.get("meldung", ""),
|
||||
})
|
||||
elif art == "sammeln":
|
||||
sammelfehler[m.get("knoten")] = m.get("meldung", "")
|
||||
elif art == "abgewaehlt":
|
||||
deselected.add(m.get("knoten"))
|
||||
return reports, deselected, sammelfehler, zeit_ueberschritten, absturzmeldungen
|
||||
|
||||
|
||||
def ergebnis_bestimmen(knoten, reports, deselected):
|
||||
if knoten in deselected:
|
||||
return "ABGEWAEHLT", ""
|
||||
phasen = reports.get(knoten, [])
|
||||
for p in phasen:
|
||||
if p["ergebnis"] == "failed" and not p["xfail"]:
|
||||
return "ROT", p["meldung"]
|
||||
for p in phasen:
|
||||
if p["ergebnis"] == "skipped" or p["xfail"]:
|
||||
return "UEBERSPRUNGEN", p["meldung"]
|
||||
for p in phasen:
|
||||
if p["phase"] == "call" and p["ergebnis"] == "passed":
|
||||
return "GRUEN", ""
|
||||
return "FEHLT", ""
|
||||
|
||||
|
||||
def sammelfehler_fuer(knoten, sammelfehler_tabelle):
|
||||
dateipfad = knoten.split("::", 1)[0]
|
||||
for cf_knoten, meldung in sammelfehler_tabelle.items():
|
||||
cf_datei = cf_knoten.split("::", 1)[0]
|
||||
if cf_knoten == dateipfad or cf_datei == dateipfad:
|
||||
return meldung
|
||||
if dateipfad == cf_knoten or dateipfad.startswith(cf_knoten + "::") or dateipfad.startswith(cf_knoten + "/"):
|
||||
return meldung
|
||||
return None
|
||||
|
||||
|
||||
def baseline_pruefen(kontext, ausgewaehlte_mutanten, home_dir, plugin_dir, zeitlimit):
|
||||
"""Schritt 3 des Ablaufs (Entwurf, Abschnitt 3.3)."""
|
||||
alle_tests = []
|
||||
gesehen = set()
|
||||
for mutant in ausgewaehlte_mutanten:
|
||||
for _, knoten in mutant["rot_bei"]:
|
||||
if knoten not in gesehen:
|
||||
gesehen.add(knoten)
|
||||
alle_tests.append(knoten)
|
||||
|
||||
gruppen = gruppiere_nach_interpreter(alle_tests, kontext["python"], kontext["python_je_datei"])
|
||||
reports, deselected, sammelfehler, zeit_ueberschritten, absturzmeldungen = gruppen_ausfuehren(
|
||||
gruppen, kontext["wurzel"], home_dir, plugin_dir, zeitlimit,
|
||||
)
|
||||
|
||||
befunde = list(absturzmeldungen)
|
||||
ergebnis_je_knoten = {}
|
||||
for knoten in alle_tests:
|
||||
if zeit_ueberschritten:
|
||||
status, meldung = "ZEIT", "Zeitlimit in der Grundlinie ueberschritten"
|
||||
else:
|
||||
status, meldung = ergebnis_bestimmen(knoten, reports, deselected)
|
||||
ergebnis_je_knoten[knoten] = {"ergebnis": status, "meldung": meldung}
|
||||
if status != "GRUEN":
|
||||
zusatz = " ({0})".format(meldung) if meldung else ""
|
||||
befunde.append("Grundlinie: {0} ist {1}{2}".format(knoten, status, zusatz))
|
||||
|
||||
return befunde, ergebnis_je_knoten
|
||||
|
||||
|
||||
def mutant_anwenden_und_pruefen(mutant, treffer_start, aus_zeilen, wird_zeilen, wurzel,
|
||||
home_dir, plugin_dir, python_je_datei, standard_python, zeitlimit):
|
||||
"""Schritt 4 des Ablaufs, ein Mutant (Entwurf, Abschnitt 3.4)."""
|
||||
ziel = wurzel / mutant["datei"]
|
||||
original_bytes = ziel.read_bytes()
|
||||
sha_vorher = hashlib.sha256(original_bytes).hexdigest()
|
||||
|
||||
datei_zeilen = original_bytes.decode("utf-8").split("\n")
|
||||
neue_zeilen = datei_zeilen[:treffer_start] + wird_zeilen + datei_zeilen[treffer_start + len(aus_zeilen):]
|
||||
neuer_inhalt = "\n".join(neue_zeilen).encode("utf-8")
|
||||
|
||||
status = "UNKLAR"
|
||||
test_ergebnisse = []
|
||||
try:
|
||||
ziel.write_bytes(neuer_inhalt)
|
||||
|
||||
knoten_liste = [knoten for _, knoten in mutant["rot_bei"]]
|
||||
gruppen = gruppiere_nach_interpreter(knoten_liste, standard_python, python_je_datei)
|
||||
reports, deselected, sammelfehler, zeit_ueberschritten, absturzmeldungen = gruppen_ausfuehren(
|
||||
gruppen, wurzel, home_dir, plugin_dir, zeitlimit,
|
||||
)
|
||||
|
||||
if zeit_ueberschritten:
|
||||
status = "ZEIT"
|
||||
for eintrag, knoten in mutant["rot_bei"]:
|
||||
test_ergebnisse.append({"eintrag": eintrag, "knoten": knoten, "ergebnis": "ZEIT", "meldung": ""})
|
||||
else:
|
||||
unklar = bool(absturzmeldungen)
|
||||
luecke = False
|
||||
if not reports and not sammelfehler and not deselected and not absturzmeldungen:
|
||||
unklar = True
|
||||
for eintrag, knoten in mutant["rot_bei"]:
|
||||
ergebnis, meldung = ergebnis_bestimmen(knoten, reports, deselected)
|
||||
if ergebnis == "FEHLT":
|
||||
sammel_meldung = sammelfehler_fuer(knoten, sammelfehler)
|
||||
if sammel_meldung is not None:
|
||||
ergebnis = "ROT"
|
||||
meldung = "Sammelfehler: " + sammel_meldung
|
||||
else:
|
||||
unklar = True
|
||||
if ergebnis in ("GRUEN", "UEBERSPRUNGEN", "ABGEWAEHLT"):
|
||||
luecke = True
|
||||
test_ergebnisse.append({"eintrag": eintrag, "knoten": knoten, "ergebnis": ergebnis, "meldung": meldung})
|
||||
|
||||
if unklar:
|
||||
status = "UNKLAR"
|
||||
elif luecke:
|
||||
status = "LUECKE"
|
||||
else:
|
||||
status = "OK"
|
||||
finally:
|
||||
ziel.write_bytes(original_bytes)
|
||||
sha_nachher = hashlib.sha256(ziel.read_bytes()).hexdigest()
|
||||
if sha_nachher != sha_vorher:
|
||||
raise RuecksetzenGescheitert(mutant["id"], mutant["datei"])
|
||||
|
||||
return status, test_ergebnisse
|
||||
|
||||
|
||||
def format_meldung(meldung, laenge):
|
||||
meldung = (meldung or "").replace("\n", " ")
|
||||
if len(meldung) > laenge:
|
||||
meldung = meldung[:max(laenge - 3, 0)] + "..."
|
||||
return meldung
|
||||
|
||||
|
||||
def format_mutant_zeile(mutant, status, test_ergebnisse, muster_ergebnis):
|
||||
stuecke = []
|
||||
for eintrag in test_ergebnisse:
|
||||
stuecke.append("{0}={1}({2})".format(eintrag["eintrag"], eintrag["ergebnis"], format_meldung(eintrag["meldung"], 120)))
|
||||
if status == "MUSTER":
|
||||
stuecke.append("Trefferzahl={0}".format(muster_ergebnis["treffer"]))
|
||||
if muster_ergebnis.get("cr_in_datei"):
|
||||
stuecke.append("Datei hat CR")
|
||||
zeile = "{0} {1} {2}".format(mutant["id"], status, " ".join(stuecke))
|
||||
return zeile.rstrip()
|
||||
|
||||
|
||||
def python_version_ermitteln(interpreter):
|
||||
try:
|
||||
ergebnis = subprocess.run([interpreter, "--version"], capture_output=True, text=True, timeout=30)
|
||||
text = (ergebnis.stdout or "") + (ergebnis.stderr or "")
|
||||
return text.strip()
|
||||
except OSError as exc:
|
||||
return "nicht ermittelbar ({0})".format(exc)
|
||||
|
||||
|
||||
def kopfzeilen_ausgeben(kontext, ausgewaehlte_mutanten):
|
||||
print("wurzel: {0}".format(kontext["wurzel"]))
|
||||
print("mutanten-datei: {0}".format(kontext["mutanten_pfad"]))
|
||||
print("mutanten gesamt: {0}, ausgewaehlt: {1}".format(len(kontext["mutanten"]), len(ausgewaehlte_mutanten)))
|
||||
interpreter_menge = {kontext["python"]}
|
||||
interpreter_menge.update(kontext["python_je_datei"].values())
|
||||
for interpreter in sorted(interpreter_menge):
|
||||
print("python {0}: {1}".format(interpreter, python_version_ermitteln(interpreter)))
|
||||
|
||||
|
||||
def zusammenfassung_rechnen(mutant_liste):
|
||||
zaehler = {"OK": 0, "LUECKE": 0, "UNKLAR": 0, "ZEIT": 0, "MUSTER": 0}
|
||||
for m in mutant_liste:
|
||||
if m["status"] in zaehler:
|
||||
zaehler[m["status"]] += 1
|
||||
zaehler["gesamt"] = len(mutant_liste)
|
||||
return zaehler
|
||||
|
||||
|
||||
def zusammenfassung_zeile(z):
|
||||
return "ergebnis: {0} Mutanten, {1} OK, {2} LUECKE, {3} UNKLAR, {4} ZEIT, {5} MUSTER".format(
|
||||
z["gesamt"], z["OK"], z["LUECKE"], z["UNKLAR"], z["ZEIT"], z["MUSTER"],
|
||||
)
|
||||
|
||||
|
||||
def parse_args(argv=None):
|
||||
parser = argparse.ArgumentParser(description="Allgemeiner Mutantenlaeufer aus mutanten.toml")
|
||||
parser.add_argument("--wurzel", required=True)
|
||||
parser.add_argument("--mutanten", required=True)
|
||||
parser.add_argument("--nur")
|
||||
parser.add_argument("--nur-muster", dest="nur_muster", action="store_true")
|
||||
parser.add_argument("--bericht")
|
||||
parser.add_argument("--zeitlimit", type=float, default=900.0)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def ablauf(args, bericht):
|
||||
befunde, kontext = lade_und_validiere(args)
|
||||
bericht["wurzel"] = str(kontext["wurzel"])
|
||||
|
||||
alle_ids = [m["id"] for m in kontext["mutanten"]]
|
||||
ausgewaehlte_ids = None
|
||||
if args.nur:
|
||||
ausgewaehlte_ids = [x.strip() for x in args.nur.split(",") if x.strip() != ""]
|
||||
for x in ausgewaehlte_ids:
|
||||
if x not in alle_ids:
|
||||
befunde.append("--nur nennt eine unbekannte id: {0}".format(x))
|
||||
|
||||
bericht["befunde"] = list(befunde)
|
||||
if befunde:
|
||||
return 2
|
||||
|
||||
if ausgewaehlte_ids is not None:
|
||||
ausgewaehlte_mutanten = [m for m in kontext["mutanten"] if m["id"] in ausgewaehlte_ids]
|
||||
else:
|
||||
ausgewaehlte_mutanten = list(kontext["mutanten"])
|
||||
|
||||
wurzel = kontext["wurzel"]
|
||||
kopfzeilen_ausgeben(kontext, ausgewaehlte_mutanten)
|
||||
|
||||
muster_ergebnisse = {}
|
||||
for mutant in ausgewaehlte_mutanten:
|
||||
ergebnis = muster_pruefen(mutant, wurzel)
|
||||
muster_ergebnisse[mutant["id"]] = ergebnis
|
||||
if ergebnis.get("kodierungsfehler"):
|
||||
befunde.append("{0}: Zieldatei ist kein UTF-8: {1}".format(mutant["id"], ergebnis["kodierungsfehler"]))
|
||||
|
||||
bericht["befunde"] = list(befunde)
|
||||
if befunde:
|
||||
return 2
|
||||
|
||||
if args.nur_muster:
|
||||
mutant_ergebnisliste = []
|
||||
alle_ok = True
|
||||
for mutant in ausgewaehlte_mutanten:
|
||||
ergebnis = muster_ergebnisse[mutant["id"]]
|
||||
if ergebnis["treffer"] == 1:
|
||||
status = "OK"
|
||||
else:
|
||||
status = "MUSTER"
|
||||
alle_ok = False
|
||||
eintrag = {
|
||||
"id": mutant["id"],
|
||||
"datei": mutant["datei"],
|
||||
"status": status,
|
||||
"hinweis": mutant.get("hinweis"),
|
||||
"treffer": ergebnis["treffer"],
|
||||
"tests": [],
|
||||
}
|
||||
mutant_ergebnisliste.append(eintrag)
|
||||
bericht["mutanten"] = mutant_ergebnisliste
|
||||
print(format_mutant_zeile(mutant, status, [], ergebnis))
|
||||
if mutant.get("hinweis"):
|
||||
print(" " + mutant["hinweis"])
|
||||
zusammenfassung = zusammenfassung_rechnen(mutant_ergebnisliste)
|
||||
bericht["zusammenfassung"] = zusammenfassung
|
||||
print(zusammenfassung_zeile(zusammenfassung))
|
||||
return 0 if alle_ok else 1
|
||||
|
||||
plugin_dir = Path(tempfile.mkdtemp(prefix="mutanten_lauf_plugin_"))
|
||||
home_dir = Path(tempfile.mkdtemp(prefix="mutanten_lauf_home_"))
|
||||
try:
|
||||
(plugin_dir / (PLUGIN_NAME + ".py")).write_text(PLUGIN_QUELLTEXT, encoding="utf-8")
|
||||
|
||||
baseline_befunde, baseline_ergebnisse = baseline_pruefen(
|
||||
kontext, ausgewaehlte_mutanten, home_dir, plugin_dir, args.zeitlimit,
|
||||
)
|
||||
bericht["grundlinie"] = baseline_ergebnisse
|
||||
if baseline_befunde:
|
||||
bericht["befunde"] = list(befunde) + baseline_befunde
|
||||
return 2
|
||||
|
||||
bericht["mutanten"] = []
|
||||
for mutant in ausgewaehlte_mutanten:
|
||||
muster = muster_ergebnisse[mutant["id"]]
|
||||
if muster["treffer"] != 1:
|
||||
eintrag = {
|
||||
"id": mutant["id"],
|
||||
"datei": mutant["datei"],
|
||||
"status": "MUSTER",
|
||||
"hinweis": mutant.get("hinweis"),
|
||||
"treffer": muster["treffer"],
|
||||
"tests": [],
|
||||
}
|
||||
bericht["mutanten"].append(eintrag)
|
||||
print(format_mutant_zeile(mutant, "MUSTER", [], muster))
|
||||
if mutant.get("hinweis"):
|
||||
print(" " + mutant["hinweis"])
|
||||
continue
|
||||
|
||||
status, test_ergebnisse = mutant_anwenden_und_pruefen(
|
||||
mutant, muster["start"], muster["aus_zeilen"], muster["wird_zeilen"],
|
||||
wurzel, home_dir, plugin_dir,
|
||||
kontext["python_je_datei"], kontext["python"], args.zeitlimit,
|
||||
)
|
||||
eintrag = {
|
||||
"id": mutant["id"],
|
||||
"datei": mutant["datei"],
|
||||
"status": status,
|
||||
"hinweis": mutant.get("hinweis"),
|
||||
"treffer": muster["treffer"],
|
||||
"tests": test_ergebnisse,
|
||||
}
|
||||
bericht["mutanten"].append(eintrag)
|
||||
print(format_mutant_zeile(mutant, status, test_ergebnisse, muster))
|
||||
if mutant.get("hinweis"):
|
||||
print(" " + mutant["hinweis"])
|
||||
|
||||
zusammenfassung = zusammenfassung_rechnen(bericht["mutanten"])
|
||||
bericht["zusammenfassung"] = zusammenfassung
|
||||
print(zusammenfassung_zeile(zusammenfassung))
|
||||
|
||||
alle_ok = all(m["status"] == "OK" for m in bericht["mutanten"])
|
||||
return 0 if alle_ok else 1
|
||||
except RuecksetzenGescheitert as exc:
|
||||
bericht["befunde"] = list(befunde) + [str(exc)]
|
||||
return 2
|
||||
finally:
|
||||
shutil.rmtree(plugin_dir, ignore_errors=True)
|
||||
shutil.rmtree(home_dir, ignore_errors=True)
|
||||
|
||||
|
||||
def ausgabe_schreiben(bericht, args):
|
||||
if args.bericht:
|
||||
try:
|
||||
with open(args.bericht, "w", encoding="utf-8") as datei:
|
||||
json.dump(bericht, datei, ensure_ascii=True, indent=2)
|
||||
except OSError as exc:
|
||||
print("Bericht konnte nicht geschrieben werden: {0}".format(exc), file=sys.stderr)
|
||||
if bericht.get("befunde"):
|
||||
print("befunde:")
|
||||
for b in bericht["befunde"]:
|
||||
print(" - {0}".format(b))
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
args = parse_args(argv)
|
||||
bericht = {
|
||||
"wurzel": args.wurzel,
|
||||
"mutanten_datei": args.mutanten,
|
||||
"befunde": [],
|
||||
"grundlinie": {},
|
||||
"mutanten": [],
|
||||
"zusammenfassung": {},
|
||||
}
|
||||
try:
|
||||
exit_code = ablauf(args, bericht)
|
||||
except Exception:
|
||||
bericht["befunde"] = bericht.get("befunde", []) + ["unerwarteter Abbruch:\n" + traceback.format_exc()]
|
||||
exit_code = 3
|
||||
|
||||
ausgabe_schreiben(bericht, args)
|
||||
return exit_code
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user