claude-werkzeuge/.vault-tools/plan-kopie/mutanten_lauf_kopie_treiber.py

#!/usr/bin/env python3
"""Treiber fuer mutanten_lauf_kopie.sh: Faelle W1 bis W13 (lo im Netz-Namensraum).

Zweck: Baut je Fall unter /tmp ein kleines Projekt (kopie/) und einen
Planordner (plan/) mit mutanten.toml, ruft mutanten_lauf_kopie.sh mit beiden
Ordnern als Windows-Pfaden auf, wie plankritik.js es tut, und vergleicht
Exit-Code und Ausgabezeilen mit der Erwartung aus dem Plan
docs/plans/2026-10-09-mutanten-kopie-loopback.md. Jeder Fall raeumt seine
Ordner selbst auf. Nach jedem Fall prueft der Treiber, dass die uebergebene
Kopie byte-gleich ist und kein Ordner /tmp/plankritik-mut-* liegen blieb.

Aufruf (in WSL): python mutanten_lauf_kopie_treiber.py [--skript <pfad>]
Ohne --skript faehrt er mutanten_lauf_kopie.sh neben sich, mit --skript eine
andere Fassung (Regressionsnachweis: Stand aus git show HEAD, kaputte
Varianten). W7 ist die Grundlinie und laeuft zuerst; ist sie rot, messen die
anderen Faelle wenig.

Pfade: Das Skript wandelt seine Argumente mit wslpath -u um. Der Treiber gibt
ihm deshalb wslpath -w seiner /tmp-Ordner und prueft vor dem ersten Fall,
dass die Rundreise denselben Pfad ergibt.

Netz: Die Tests verbinden sich nur mit 127.0.0.1 und, nur im Netz-Namensraum
des Skripts, mit 192.0.2.1 (TEST-NET-1). Der Test dazu prueft zuerst, dass es
keine Route ausser ueber lo gibt; ohne Namensraum scheitert er dort, bevor er
eine Verbindung versucht.

Faelle W10 bis W13 (Nachbesserung 1, Plan
docs/plans/2026-10-09-mutanten-kopie-loopback-nachbesserung-1.md) ersetzen im
Attrappen-Ordner einzelne Werkzeuge durch kleine sh-Skripte, nach den
Angriffen A1, A3 und A4 des Pruefers vom 2026-10-09: ip scheitert in der
Probe (W10), ip meldet Erfolg und lo bleibt aus (W11), ip scheitert erst im
Hauptaufruf (W12), setpriv wirft keine Faehigkeit ab (W13).

Nicht im Treiber: ob plankritik.js das Skript so aufruft wie hier
(MSYS_NO_PATHCONV, cygpath -m -l); ob die Probe bei Exit 0 mit falscher
Ausgabe scheitert.
"""

import argparse
import hashlib
import os
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import uuid
from pathlib import Path

HIER = Path(__file__).resolve().parent
SKRIPT_STANDARD = HIER / "mutanten_lauf_kopie.sh"
INTERPRETER = "/home/claude/venvs/trading/bin/python"
BASH = shutil.which("bash")
# Was mutanten_lauf_kopie.sh ueber PATH aufruft (Stand 2026-10-09); daraus
# bauen W4, W5, W9 und W10 bis W13 ihre Attrappen-Ordner.
WERKZEUGE = (
    "unshare", "wslpath", "mktemp", "cp", "chmod", "find", "rm", "timeout",
    "sh", "true", "ip", "setpriv", "python3",
)

FEHLERLISTE = []


def melde(fallname, ok, zusatz=""):
    if ok:
        print("{0} ok".format(fallname))
    else:
        print("{0} FEHLER: {1}".format(fallname, zusatz))
        FEHLERLISTE.append(fallname)


def kurz(ergebnis):
    return "exit={0}\n{1}{2}".format(ergebnis.returncode, ergebnis.stdout, ergebnis.stderr)


def sha_von(pfad):
    return hashlib.sha256(pfad.read_bytes()).hexdigest()


def plankritik_ordner():
    return set(Path("/tmp").glob("plankritik-mut-*"))


def windows_pfad(pfad):
    ergebnis = subprocess.run(["wslpath", "-w", str(pfad)], capture_output=True, text=True, timeout=30)
    return ergebnis.stdout.strip()


def linux_pfad(pfad):
    ergebnis = subprocess.run(["wslpath", "-u", pfad], capture_output=True, text=True, timeout=30)
    return ergebnis.stdout.strip()


# --- Projektbausteine -------------------------------------------------------

GRUNDDATEIEN = {
    "pytest.ini": "[pytest]\n",
    "paket/__init__.py": "",
    "tests/__init__.py": "",
}

MODUL = (
    "def f1(x):\n"
    "    if x > 0:\n"
    "        return True\n"
    "    return False\n"
)

TEST_MODUL = (
    "from paket.modul import f1\n"
    "\n"
    "\n"
    "def test_f1_positiv():\n"
    "    assert f1(5) is True\n"
)

DIENST = (
    "import socket\n"
    "\n"
    "\n"
    "def antwort_ueber_lo(nachricht):\n"
    "    server = socket.socket()\n"
    "    server.bind((\"127.0.0.1\", 0))\n"
    "    server.listen(1)\n"
    "    client = socket.create_connection(server.getsockname(), timeout=5)\n"
    "    verbindung, _ = server.accept()\n"
    "    client.sendall(nachricht)\n"
    "    daten = verbindung.recv(1024)\n"
    "    for sock in (client, verbindung, server):\n"
    "        sock.close()\n"
    "    return daten\n"
)

TEST_DIENST = (
    "from paket.dienst import antwort_ueber_lo\n"
    "\n"
    "\n"
    "def test_echo_ueber_lo():\n"
    "    assert antwort_ueber_lo(b\"hallo\") == b\"hallo\"\n"
)

NETZ = (
    "AUSSEN = (\"192.0.2.1\", 9)\n"
    "\n"
    "\n"
    "def aussen_adresse():\n"
    "    return AUSSEN\n"
)

TEST_NETZ = (
    "import errno\n"
    "import socket\n"
    "\n"
    "import pytest\n"
    "\n"
    "from paket.netz import aussen_adresse\n"
    "\n"
    "\n"
    "def test_keine_route_nach_aussen():\n"
    "    with open(\"/proc/net/route\", encoding=\"ascii\") as datei:\n"
    "        zeilen = datei.read().splitlines()[1:]\n"
    "    schnittstellen = [zeile.split()[0] for zeile in zeilen if zeile.strip()]\n"
    "    assert all(name == \"lo\" for name in schnittstellen), \"Routen ausser lo: {0}\".format(schnittstellen)\n"
    "    sock = socket.socket()\n"
    "    sock.settimeout(5)\n"
    "    try:\n"
    "        sock.connect(aussen_adresse())\n"
    "    except OSError as exc:\n"
    "        assert exc.errno == errno.ENETUNREACH, \"erwartet ENETUNREACH: {0!r}\".format(exc)\n"
    "    else:\n"
    "        pytest.fail(\"Verbindung nach aussen gelungen\")\n"
    "    finally:\n"
    "        sock.close()\n"
)

RECHTE = (
    "def status_feld(name):\n"
    "    with open(\"/proc/self/status\", encoding=\"utf-8\", errors=\"replace\") as datei:\n"
    "        for zeile in datei:\n"
    "            schluessel, _, wert = zeile.partition(\":\")\n"
    "            if schluessel == name:\n"
    "                return wert.strip()\n"
    "    raise KeyError(name)\n"
)

HAENGEN = (
    "def schritt_a(n):\n"
    "    return n + 1\n"
    "\n"
    "\n"
    "def schritt_b(n):\n"
    "    return n + 2\n"
)

TEST_HAENGEN = (
    "from paket.haengen import schritt_a, schritt_b\n"
    "\n"
    "\n"
    "def test_schritt_a():\n"
    "    assert schritt_a(1) == 2\n"
    "\n"
    "\n"
    "def test_schritt_b():\n"
    "    assert schritt_b(1) == 3\n"
)

SCHLEIFE = "    while True:\n        pass\n"

PRUEFZEILE_AUSSEN = "    assert info.value.errno in (errno.ECONNREFUSED, errno.ENETUNREACH), repr(info.value)\n"


def text_test_rechte(uid):
    return (
        "from paket.rechte import status_feld\n"
        "\n"
        "UID_AUSSEN = \"" + str(uid) + "\"\n"
        "\n"
        "\n"
        "def test_keine_faehigkeiten_gleiche_uid():\n"
        "    for feld in (\"CapInh\", \"CapPrm\", \"CapEff\", \"CapAmb\"):\n"
        "        wert = status_feld(feld)\n"
        "        assert int(wert, 16) == 0, \"{0} ist {1}\".format(feld, wert)\n"
        "    assert status_feld(\"Uid\").split()[0] == UID_AUSSEN\n"
    )


def text_test_aussen(port):
    return (
        "import errno\n"
        "import socket\n"
        "\n"
        "import pytest\n"
        "\n"
        "PORT_AUSSEN = " + str(port) + "\n"
        "\n"
        "\n"
        "def test_dienst_des_rechners_unerreichbar():\n"
        "    sock = socket.socket()\n"
        "    sock.settimeout(5)\n"
        "    try:\n"
        "        with pytest.raises(OSError) as info:\n"
        "            sock.connect((\"127.0.0.1\", PORT_AUSSEN))\n"
        "    finally:\n"
        "        sock.close()\n"
        + PRUEFZEILE_AUSSEN
    )


def kopf():
    return "format = 1\npython = \"{0}\"\n\n".format(INTERPRETER)


def mutant(id_, datei, aus, wird, rot_bei):
    return (
        "[[mutant]]\n"
        "id = \"" + id_ + "\"\n"
        "datei = \"" + datei + "\"\n"
        "aus = '''\n" + aus + "'''\n"
        "wird = '''\n" + wird + "'''\n"
        "rot_bei = [\"" + rot_bei + "\"]\n"
        "\n"
    )


class Fall:
    """Ein Fall unter /tmp: kopie/ mit den Projektdateien, plan/ mit mutanten.toml."""

    def __init__(self, dateien, mutanten):
        self.basis = Path(tempfile.mkdtemp(prefix="mkopie_treiber_"))
        self.kopie = self.basis / "kopie"
        self.plan = self.basis / "plan"
        self.plan.mkdir()
        alle = dict(GRUNDDATEIEN)
        alle.update(dateien)
        self.snap = {}
        for relpfad, inhalt in alle.items():
            pfad = self.kopie / relpfad
            pfad.parent.mkdir(parents=True, exist_ok=True)
            pfad.write_text(inhalt, encoding="utf-8")
            self.snap[pfad] = sha_von(pfad)
        (self.plan / "mutanten.toml").write_text(kopf() + mutanten, encoding="utf-8")

    def pruefe_kopie(self, fallname):
        for pfad, sha in self.snap.items():
            if not pfad.exists() or sha_von(pfad) != sha:
                melde(fallname + " (kopie)", False, "uebergebene Kopie veraendert: {0}".format(pfad))
                return

    def aufraeumen(self):
        shutil.rmtree(self.basis, ignore_errors=True)


def starte_skript(skript, fall, modus="voll", pfad=None):
    env = dict(os.environ)
    if pfad is not None:
        env["PATH"] = pfad
    cmd = [BASH, str(skript), windows_pfad(fall.kopie), windows_pfad(fall.plan), modus]
    return subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=180)


# --- Faelle ----------------------------------------------------------------

def fahre_einfach(name, skript, dateien, mutanten, mutant_id):
    fall = Fall(dateien, mutanten)
    try:
        ergebnis = starte_skript(skript, fall)
        ok = (
            ergebnis.returncode == 0
            and "{0} OK".format(mutant_id) in ergebnis.stdout
            and "EXIT mutanten_lauf: 0" in ergebnis.stdout
        )
        melde(name, ok, kurz(ergebnis))
        fall.pruefe_kopie(name)
    finally:
        fall.aufraeumen()


def fall_einfach():
    return Fall(
        {"paket/modul.py": MODUL, "tests/test_modul.py": TEST_MODUL},
        mutant("MW7", "paket/modul.py", "        return True\n", "        return False\n",
               "tests/test_modul.py::test_f1_positiv"),
    )


def fall_w7(skript):
    # Grundlinie wie bisher: kein Server-Test, Modus voll und muster.
    fall = fall_einfach()
    try:
        voll = starte_skript(skript, fall)
        muster = starte_skript(skript, fall, modus="muster")
        ok = (
            voll.returncode == 0
            and "modus: voll" in voll.stdout
            and "MW7 OK" in voll.stdout
            and "EXIT mutanten_lauf: 0" in voll.stdout
            and muster.returncode == 0
            and "modus: muster" in muster.stdout
            and "EXIT mutanten_lauf: 0" in muster.stdout
        )
        melde("W7", ok, "voll " + kurz(voll) + "\nmuster " + kurz(muster))
        fall.pruefe_kopie("W7")
    finally:
        fall.aufraeumen()


def fall_w1(skript):
    # Ein Test startet einen Server auf 127.0.0.1 und tauscht Daten aus.
    fahre_einfach(
        "W1", skript,
        {"paket/dienst.py": DIENST, "tests/test_dienst.py": TEST_DIENST},
        mutant("MW1", "paket/dienst.py", "    return daten\n", "    return daten[::-1]\n",
               "tests/test_dienst.py::test_echo_ueber_lo"),
        "MW1",
    )


def fall_w2(skript):
    # Verbindung nach aussen scheitert mit ENETUNREACH; der Mutant zielt auf
    # 127.0.0.1:9 und faellt nur, wenn lo oben ist (ECONNREFUSED).
    fahre_einfach(
        "W2", skript,
        {"paket/netz.py": NETZ, "tests/test_netz.py": TEST_NETZ},
        mutant("MW2", "paket/netz.py", "AUSSEN = (\"192.0.2.1\", 9)\n", "AUSSEN = (\"127.0.0.1\", 9)\n",
               "tests/test_netz.py::test_keine_route_nach_aussen"),
        "MW2",
    )


def fall_w3(skript):
    # CapInh, CapPrm, CapEff, CapAmb gleich 0 und dieselbe uid wie aussen.
    fahre_einfach(
        "W3", skript,
        {"paket/rechte.py": RECHTE, "tests/test_rechte.py": text_test_rechte(os.getuid())},
        mutant("MW3", "paket/rechte.py", "                return wert.strip()\n", "                return \"1\"\n",
               "tests/test_rechte.py::test_keine_faehigkeiten_gleiche_uid"),
        "MW3",
    )


def baue_attrappen(ordner, ohne):
    ordner.mkdir()
    for name in WERKZEUGE:
        if name != ohne:
            os.symlink(shutil.which(name), ordner / name)


def fahre_ohne_werkzeug(name, skript, werkzeug):
    # Erst ein Kontrolllauf mit allen Attrappen (muss gruen sein), dann einer
    # ohne das eine Werkzeug (muss vor dem Kopieren mit Exit 2 enden).
    fall = fall_einfach()
    try:
        alle = fall.basis / "attrappen_alle"
        ohne = fall.basis / "attrappen_ohne"
        baue_attrappen(alle, None)
        baue_attrappen(ohne, werkzeug)
        kontrolle = starte_skript(skript, fall, pfad=str(alle))
        vorher = plankritik_ordner()
        ergebnis = starte_skript(skript, fall, pfad=str(ohne))
        neu = sorted(str(p) for p in plankritik_ordner() - vorher)
        ok = (
            kontrolle.returncode == 0
            and "MW7 OK" in kontrolle.stdout
            and ergebnis.returncode == 2
            and "MUTANTEN_LAUF_FEHLER: {0} fehlt".format(werkzeug) in ergebnis.stdout
            and "modus:" not in ergebnis.stdout
            and "EXIT mutanten_lauf" not in ergebnis.stdout
            and not neu
        )
        melde(name, ok, "kontrolle exit={0}, ohne {1} exit={2}, neue Kopien={3}\nkontrolle {4}\nohne {1} {5}".format(
            kontrolle.returncode, werkzeug, ergebnis.returncode, neu, kurz(kontrolle), kurz(ergebnis)))
        fall.pruefe_kopie(name)
    finally:
        fall.aufraeumen()


def fall_w4(skript):
    fahre_ohne_werkzeug("W4", skript, "ip")


def fall_w5(skript):
    fahre_ohne_werkzeug("W5", skript, "setpriv")


def fall_w9(skript):
    fahre_ohne_werkzeug("W9", skript, "python3")


def baue_attrappen_mit_ersatz(ordner, ersatz):
    # Wie baue_attrappen, aber jedes Werkzeug aus ersatz ist ein kleines
    # sh-Skript statt eines Verweises auf das echte Werkzeug (Bauweise aus den
    # Angriffen des Pruefers vom 2026-10-09).
    ordner.mkdir()
    for name in WERKZEUGE:
        pfad = ordner / name
        if name in ersatz:
            pfad.write_text("#!/bin/sh\n" + ersatz[name] + "\n", encoding="utf-8")
            pfad.chmod(0o755)
        else:
            os.symlink(shutil.which(name), pfad)


def fahre_probe_scheitert(name, skript, ersatz, merkmal):
    # Die Vorpruefung muss scheitern: Exit 2 mit der Meldung der Probe und dem
    # Merkmal des Angriffs, ohne Modus-Zeile, ohne EXIT-Zeile, ohne Kopie.
    fall = fall_einfach()
    try:
        ordner = fall.basis / "attrappen"
        baue_attrappen_mit_ersatz(ordner, ersatz)
        vorher = plankritik_ordner()
        ergebnis = starte_skript(skript, fall, pfad=str(ordner))
        neu = sorted(str(p) for p in plankritik_ordner() - vorher)
        ok = (
            ergebnis.returncode == 2
            and "MUTANTEN_LAUF_FEHLER: lo im Netz-Namensraum nicht nutzbar" in ergebnis.stdout
            and merkmal in ergebnis.stdout
            and "modus:" not in ergebnis.stdout
            and "EXIT mutanten_lauf" not in ergebnis.stdout
            and not neu
        )
        melde(name, ok, "neue Kopien={0}, {1}".format(neu, kurz(ergebnis)))
        fall.pruefe_kopie(name)
    finally:
        fall.aufraeumen()


def fall_w10(skript):
    # ip scheitert schon in der Probe (Angriff A1 des Pruefers).
    fahre_probe_scheitert("W10", skript, {"ip": "echo ip-attrappe scheitert >&2; exit 1"}, "ip-attrappe scheitert")


def fall_w11(skript):
    # ip meldet Erfolg, lo bleibt aus (A3): Die Verbindung der Probe scheitert
    # mit Errno 101.
    fahre_probe_scheitert("W11", skript, {"ip": "exit 0"}, "Errno 101")


def fall_w12(skript):
    # ip klappt in der Probe und scheitert erst im Hauptaufruf (A4); eine
    # Zaehlerdatei haelt fest, der wievielte Aufruf es ist. Erwartet:
    # Modus-Zeile, Meldung des Hauptaufrufs, EXIT-Zeile mit 2, kein Laeufer,
    # genau zwei Aufrufe von ip, keine Kopie unter /tmp.
    fall = fall_einfach()
    try:
        zaehler = fall.basis / "ip_zaehler"
        z = shlex.quote(str(zaehler))
        ip_echt = shlex.quote(shutil.which("ip"))
        ersatz = {"ip": (
            "n=0\n"
            "if [ -f " + z + " ]; then read n < " + z + "; fi\n"
            "n=$((n + 1))\n"
            "echo \"$n\" > " + z + "\n"
            "if [ \"$n\" -ge 2 ]; then echo \"ip-attrappe zweiter Aufruf\" >&2; exit 1; fi\n"
            "exec " + ip_echt + " \"$@\""
        )}
        ordner = fall.basis / "attrappen"
        baue_attrappen_mit_ersatz(ordner, ersatz)
        vorher = plankritik_ordner()
        ergebnis = starte_skript(skript, fall, pfad=str(ordner))
        neu = sorted(str(p) for p in plankritik_ordner() - vorher)
        try:
            aufrufe = zaehler.read_text(encoding="ascii").strip()
        except OSError:
            aufrufe = "keine"
        ok = (
            ergebnis.returncode == 2
            and "modus: voll" in ergebnis.stdout
            and "MUTANTEN_LAUF_FEHLER: lo im Netz-Namensraum nicht hochgefahren" in ergebnis.stdout
            and "EXIT mutanten_lauf: 2" in ergebnis.stdout
            and "nicht nutzbar" not in ergebnis.stdout
            and "MW7" not in ergebnis.stdout
            and "ip-attrappe zweiter Aufruf" in ergebnis.stderr
            and aufrufe == "2"
            and not neu
        )
        melde("W12", ok, "ip-aufrufe={0}, neue Kopien={1}, {2}".format(aufrufe, neu, kurz(ergebnis)))
        fall.pruefe_kopie("W12")
    finally:
        fall.aufraeumen()


SETPRIV_OHNE_WIRKUNG = (
    "while [ \"$#\" -gt 0 ] && [ \"$1\" != \"--\" ]; do shift; done\n"
    "shift\n"
    "exec \"$@\""
)


def fall_w13(skript):
    # setpriv wirft keine Faehigkeit ab und startet nur den Befehl hinter --:
    # Die Faehigkeitspruefung der Probe muss anschlagen.
    fahre_probe_scheitert("W13", skript, {"setpriv": SETPRIV_OHNE_WIRKUNG}, "Faehigkeiten nicht abgeworfen")


def prozesse_mit_marke(marke):
    """pid -> argv aller Prozesse, deren Umgebung die Marke traegt."""
    gefunden = {}
    for envdatei in Path("/proc").glob("[0-9]*/environ"):
        try:
            if marke.encode("ascii") not in envdatei.read_bytes():
                continue
            argv = (envdatei.parent / "cmdline").read_bytes().split(b"\0")
        except OSError:
            continue
        gefunden[envdatei.parent.name] = [teil.decode("utf-8", "replace") for teil in argv if teil]
    return gefunden


def elternname(pid):
    try:
        stat = Path("/proc/{0}/stat".format(pid)).read_text()
        ppid = stat.rsplit(")", 1)[1].split()[1]
        return Path("/proc/{0}/comm".format(ppid)).read_text().strip()
    except (OSError, IndexError):
        return None


def fall_w6(skript):
    # Zeitgrenze: zwei haengende Mutanten, BUDGET_S 10, ZEITLIMIT_S 6. Waehrend
    # des Laufs muss timeout der Elternprozess des Laeufers sein (keine
    # Zwischenschale); danach lebt kein Prozess des Falls mehr.
    fall = Fall(
        {"paket/haengen.py": HAENGEN, "tests/test_haengen.py": TEST_HAENGEN},
        mutant("MW6a", "paket/haengen.py", "    return n + 1\n", SCHLEIFE, "tests/test_haengen.py::test_schritt_a")
        + mutant("MW6b", "paket/haengen.py", "    return n + 2\n", SCHLEIFE, "tests/test_haengen.py::test_schritt_b"),
    )
    try:
        wert = uuid.uuid4().hex
        marke = "KOPIE_TREIBER_MARKE=" + wert
        eigenes_tmpdir = fall.basis / "tmpdir"
        eigenes_tmpdir.mkdir()
        env = dict(os.environ)
        env.update({
            "BUDGET_S": "10",
            "ZEITLIMIT_S": "6",
            "TMPDIR": str(eigenes_tmpdir),
            "KOPIE_TREIBER_MARKE": wert,
        })
        cmd = [BASH, str(skript), windows_pfad(fall.kopie), windows_pfad(fall.plan), "voll"]
        proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        eltern = None
        pytest_gesehen = False
        for _ in range(12):
            time.sleep(0.5)
            for pid, argv in prozesse_mit_marke(marke).items():
                if (argv and Path(argv[0]).name.startswith("python")
                        and any(teil.endswith("mutanten_lauf.py") for teil in argv)):
                    eltern = elternname(pid)
                if "pytest" in argv:
                    pytest_gesehen = True
            if eltern is not None and pytest_gesehen:
                break
        try:
            stdout, stderr = proc.communicate(timeout=90)
        except subprocess.TimeoutExpired:
            proc.kill()
            stdout, stderr = proc.communicate()
            stderr += "\nTREIBER: Skript nach 90 s nicht beendet"
        reste = prozesse_mit_marke(marke)
        tmp_reste = sorted(p.name for p in eigenes_tmpdir.iterdir())
        ok = (
            proc.returncode == 124
            and "ZEITGRENZE" in stdout
            and "EXIT mutanten_lauf: 124" in stdout
            and "ABBRUCH: SIGTERM" in stdout
            and eltern == "timeout"
            and pytest_gesehen
            and not reste
            and not tmp_reste
        )
        melde("W6", ok, "exit={0}, eltern des laeufers={1}, pytest gesehen={2}, reste={3}, tmp={4}\n{5}{6}".format(
            proc.returncode, eltern, pytest_gesehen, reste, tmp_reste, stdout, stderr))
        fall.pruefe_kopie("W6")
    finally:
        fall.aufraeumen()


def fall_w8(skript):
    # Ein Dienst des Treibers auf 127.0.0.1 darf aus dem Namensraum nicht
    # erreichbar sein, und bei ihm darf keine Verbindung ankommen.
    dienst = socket.socket()
    dienst.bind(("127.0.0.1", 0))
    dienst.listen(5)
    dienst.setblocking(False)
    fall = None
    try:
        fall = Fall(
            {"tests/test_aussen.py": text_test_aussen(dienst.getsockname()[1])},
            mutant("MW8", "tests/test_aussen.py", PRUEFZEILE_AUSSEN, "    assert False\n",
                   "tests/test_aussen.py::test_dienst_des_rechners_unerreichbar"),
        )
        ergebnis = starte_skript(skript, fall)
        try:
            verbindung, _ = dienst.accept()
            verbindung.close()
            angekommen = True
        except BlockingIOError:
            angekommen = False
        ok = (
            ergebnis.returncode == 0
            and "MW8 OK" in ergebnis.stdout
            and "EXIT mutanten_lauf: 0" in ergebnis.stdout
            and not angekommen
        )
        melde("W8", ok, "beim Dienst des Rechners angekommen={0}, {1}".format(angekommen, kurz(ergebnis)))
        fall.pruefe_kopie("W8")
    finally:
        dienst.close()
        if fall is not None:
            fall.aufraeumen()


FAELLE = [
    ("W7", fall_w7),
    ("W1", fall_w1),
    ("W2", fall_w2),
    ("W3", fall_w3),
    ("W4", fall_w4),
    ("W5", fall_w5),
    ("W6", fall_w6),
    ("W8", fall_w8),
    ("W9", fall_w9),
    ("W10", fall_w10),
    ("W11", fall_w11),
    ("W12", fall_w12),
    ("W13", fall_w13),
]


def main():
    zerleger = argparse.ArgumentParser(description="Treiber fuer mutanten_lauf_kopie.sh (Faelle W1 bis W13)")
    zerleger.add_argument("--skript", default=str(SKRIPT_STANDARD), help="zu pruefende Fassung des Skripts")
    argumente = zerleger.parse_args()
    skript = Path(argumente.skript)
    print("skript: {0}".format(skript))

    fehlend = [name for name in WERKZEUGE if shutil.which(name) is None]
    if BASH is None:
        fehlend.append("bash")
    if fehlend:
        print("TREIBER_FEHLER: Werkzeuge fehlen: {0}".format(", ".join(fehlend)))
        return 2
    if not skript.is_file():
        print("TREIBER_FEHLER: Skript fehlt: {0}".format(skript))
        return 2
    if not Path(INTERPRETER).is_file():
        print("TREIBER_FEHLER: Interpreter fehlt: {0}".format(INTERPRETER))
        return 2

    probe = Path(tempfile.mkdtemp(prefix="mkopie_treiber_rundreise_"))
    try:
        windows = windows_pfad(probe)
        zurueck = linux_pfad(windows) if windows else ""
    finally:
        probe.rmdir()
    if not windows or zurueck != str(probe):
        print("TREIBER_FEHLER: wslpath-Rundreise: {0} -> {1} -> {2}".format(probe, windows, zurueck))
        return 2
    print("rundreise: {0} -> {1}".format(probe, windows))

    for name, funktion in FAELLE:
        vorher = plankritik_ordner()
        try:
            funktion(skript)
        except Exception as exc:
            melde(name, False, "Ausnahme: {0!r}".format(exc))
        liegen_geblieben = sorted(str(p) for p in plankritik_ordner() - vorher)
        if liegen_geblieben:
            melde(name + " (aufraeumen)", False, "Kopien unter /tmp blieben liegen: {0}".format(liegen_geblieben))

    print("gerechnete Fehler: {0}".format(len(FEHLERLISTE)))
    return 0 if not FEHLERLISTE else 1


if __name__ == "__main__":
    sys.exit(main())