896 lines
33 KiB
Python
896 lines
33 KiB
Python
#!/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))
|
|
|
|
|
|
class SigtermUnterbrechung(Exception):
|
|
"""Wird aus dem SIGTERM-Handler ausgeloest, damit die finally-Bloecke
|
|
(Ruecksetzen der Zieldatei, Aufraeumen der Temp-Ordner) laufen. Siehe
|
|
Nachbesserung 2026-09-28, N4."""
|
|
|
|
|
|
def _sigterm_werfen(signum, frame):
|
|
raise SigtermUnterbrechung()
|
|
|
|
|
|
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
|
|
proc = None
|
|
try:
|
|
try:
|
|
proc = subprocess.Popen(
|
|
cmd, cwd=str(wurzel), env=env,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
except OSError as exc:
|
|
absturz = "pytest-Aufruf gescheitert ({0}): {1}".format(python_exe, exc)
|
|
|
|
if proc is not None:
|
|
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 BaseException:
|
|
# Auch ein SIGTERM (SigtermUnterbrechung) darf keinen pytest-Prozess
|
|
# zurueck lassen: Prozessgruppe beenden, dann weiterwerfen, damit die
|
|
# finally-Bloecke der Aufrufer (Ruecksetzen der Zieldatei, Aufraeumen
|
|
# der Temp-Ordner) laufen. Siehe Nachbesserung 2026-09-28, N4.
|
|
try:
|
|
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
|
except (ProcessLookupError, PermissionError):
|
|
pass
|
|
try:
|
|
proc.communicate()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
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
|
|
|
|
return zeit_ueberschritten, meldungen, absturz
|
|
finally:
|
|
# Muss auch bei einem SIGTERM mitten im Lauf greifen (N4): sonst bleibt die
|
|
# Ergebnisdatei liegen, obwohl der pytest-Prozess schon beendet ist.
|
|
try:
|
|
os.remove(ergebnis_pfad)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
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_"))
|
|
aktueller_mutant_id = None
|
|
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
|
|
|
|
aktueller_mutant_id = mutant["id"]
|
|
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,
|
|
)
|
|
aktueller_mutant_id = None
|
|
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 SigtermUnterbrechung:
|
|
zeile = (
|
|
"ABBRUCH: SIGTERM, Mutant {0} zurueckgesetzt".format(aktueller_mutant_id)
|
|
if aktueller_mutant_id else "ABBRUCH: SIGTERM"
|
|
)
|
|
print(zeile)
|
|
bericht["befunde"] = list(befunde) + [zeile]
|
|
return 3
|
|
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):
|
|
signal.signal(signal.SIGTERM, _sigterm_werfen)
|
|
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())
|