feat: configure synthetic operation mix and measure burst restart recovery
All checks were successful
Application acceptance / application-tests (push) Successful in 10s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
This commit is contained in:
tegwick 2026-10-19 15:36:17 +02:00
parent bfd1a2d46d
commit 44e5458b17
11 changed files with 862 additions and 102 deletions

View file

@ -18,10 +18,20 @@ if v["app_runtime"]["nano_cpus"] != args.expected_cpu_m * 1000000:
raise ValueError("actual CPU limit does not match expected limit")
if v["app_runtime"]["memory_bytes"] != args.expected_memory_mib * 1024**2:
raise ValueError("actual memory limit does not match expected limit")
samples = [s for s in v["runtime_samples"] if s.get("phase") == "ordinary" and "cpu_seconds" in s]
load_names = {p["name"] for p in v.get("phase_results", []) if p["kind"] == "load"} or {"ordinary"}
samples = [s for s in v["runtime_samples"] if s.get("phase") in load_names and "cpu_seconds" in s]
pairs = [
(a, b)
for a, b in zip(samples, samples[1:])
if a["phase"] == b["phase"]
and b["cpu_seconds"] >= a["cpu_seconds"]
and b["monotonic_seconds"] > a["monotonic_seconds"]
]
if not pairs:
raise ValueError("no valid CPU sample intervals")
rates = [
(b["cpu_seconds"] - a["cpu_seconds"]) / (b["monotonic_seconds"] - a["monotonic_seconds"]) * 1000
for a, b in zip(samples, samples[1:])
for a, b in pairs
]
@ -29,11 +39,16 @@ def p95(values):
return sorted(values)[math.ceil(0.95 * len(values)) - 1]
start, end = samples[0]["cpu_stat"], samples[-1]["cpu_stat"]
db = [s["docker_stats"] for s in v["database_samples"] if s["phase"] == "ordinary"]
def counter_delta(key):
return sum(max(0, b["cpu_stat"][key] - a["cpu_stat"][key]) for a, b in pairs)
db = [s["docker_stats"] for s in v["database_samples"] if s["phase"] in load_names]
report = {
"status": "provisional-existing-limit-comparison",
"ordinary": v["ordinary"],
"startup_seconds": v.get("startup_seconds"),
"phase_results": v.get("phase_results", []),
"document_transfer": v["document_transfer"],
"complete": v["complete"],
"provisional_pass": v["provisional_pass"],
@ -45,13 +60,13 @@ report = {
"application_measured": {
"sample_intervals": len(rates),
"cpu_p95_m": p95(rates),
"cpu_mean_m": (samples[-1]["cpu_seconds"] - samples[0]["cpu_seconds"])
/ (samples[-1]["monotonic_seconds"] - samples[0]["monotonic_seconds"])
"cpu_mean_m": sum(b["cpu_seconds"] - a["cpu_seconds"] for a, b in pairs)
/ sum(b["monotonic_seconds"] - a["monotonic_seconds"] for a, b in pairs)
* 1000,
"cgroup_memory_peak_mib": max(s["rss_bytes"] for s in samples) / 1024**2,
"throttled_periods": end["nr_throttled"] - start["nr_throttled"],
"periods": end["nr_periods"] - start["nr_periods"],
"throttled_usec": end["throttled_usec"] - start["throttled_usec"],
"throttled_periods": counter_delta("nr_throttled"),
"periods": counter_delta("nr_periods"),
"throttled_usec": counter_delta("throttled_usec"),
},
"database_separate": {
"samples": len(db),

View file

@ -33,6 +33,130 @@ def summarize(rows):
}
LEGACY_OPERATIONS = [
"list",
"search",
"tender-create",
"detail",
"lot-create",
"task-create",
"document-upload",
"document-download",
]
DEPENDENCIES = {
"list": None,
"search": None,
"tender-create": None,
"detail": "tender-create",
"tender-edit": "tender-create",
"lot-create": "tender-create",
"lot-edit": "lot-create",
"task-create": "tender-create",
"task-edit": "task-create",
"document-upload": "tender-create",
"document-download": "document-upload",
}
def operation_plan(operations):
if not isinstance(operations, list) or not operations:
raise ValueError("operations must be a nonempty ordered list")
seen = set()
for operation in operations:
if operation not in DEPENDENCIES:
raise ValueError(f"unknown operation: {operation}")
dependency = DEPENDENCIES[operation]
if dependency and dependency not in seen:
raise ValueError(f"{operation} requires earlier {dependency}")
if operation in seen and (operation.endswith("-create") or operation == "document-upload"):
raise ValueError(f"only one {operation} per cycle; repeat reads/edits to tune mix")
seen.add(operation)
return list(operations)
def phase_plan(config):
phases = config.get(
"phases",
[
{
"name": "ordinary",
"kind": "load",
"duration_seconds": config["duration_seconds"],
"think_seconds": config["think_seconds"],
}
],
)
if not isinstance(phases, list) or not phases:
raise ValueError("phases must be a nonempty list")
names = set()
result = []
for item in phases:
item = dict(item)
name = item.get("name")
if not isinstance(name, str) or not name or name in names or name in ("idle", "startup"):
raise ValueError("phase names must be unique and not idle/startup")
names.add(name)
if item.get("kind") == "restart":
if config.get("runtime") != "container":
raise ValueError("restart phase requires container runtime")
elif item.get("kind") == "load":
for key in ("duration_seconds", "think_seconds"):
value = item.get(key)
if type(value) not in (int, float) or not math.isfinite(value) or value < 0:
raise ValueError(f"invalid phase {key}")
if not item["duration_seconds"]:
raise ValueError("load duration must be positive")
item["operations"] = operation_plan(
item.get("operations", config.get("operations", LEGACY_OPERATIONS))
)
else:
raise ValueError("phase kind must be load or restart")
result.append(item)
if not any(p["kind"] == "load" for p in result):
raise ValueError("at least one load phase required")
return result
def expected_requests(completed, operations, users):
return sum(completed) * len(operations) + users * 2
def wait_ready(base_url, timeout=30):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
with urllib.request.urlopen(base_url + "/health/", timeout=1) as response:
if response.status == 200:
return
except OSError:
pass
time.sleep(0.1)
raise RuntimeError("application readiness timeout")
def container_url(name):
binding = subprocess.check_output(
["docker", "port", name, "8000/tcp"], text=True, timeout=10
).strip()
host, port = binding.rsplit(":", 1)
if host != "127.0.0.1" or not port.isdigit():
raise ValueError("synthetic runtime must have one loopback HTTP binding")
return f"http://127.0.0.1:{port}"
def restart_container(name):
subprocess.run(
["docker", "restart", "--time", "5", name],
check=True,
timeout=20,
stdout=subprocess.DEVNULL,
)
# Docker can allocate a different random published port after restart.
endpoint = container_url(name)
wait_ready(endpoint)
return endpoint
def validate(config):
for key in (
"users",
@ -68,6 +192,7 @@ def validate(config):
raise ValueError("app_cpu_limit must be positive and at most 8")
if not config.get("app_image") or not config.get("app_memory_limit"):
raise ValueError("container image and memory limit required")
phase_plan(config)
return config
@ -103,12 +228,16 @@ def sampling_gaps(samples, db_samples, errors):
good = [s for s in samples if "monotonic_seconds" in s]
if len(good) < 2 or not db_samples:
gaps.append({"sampler": "coverage", "reason": "insufficient samples"})
if any(b["monotonic_seconds"] - a["monotonic_seconds"] > 3 for a, b in zip(good, good[1:])):
if any(
b["monotonic_seconds"] - a["monotonic_seconds"] > 3
for a, b in zip(good, good[1:])
if a.get("phase") == b.get("phase")
):
gaps.append({"sampler": "application", "reason": "sampling interval exceeds 3 seconds"})
return gaps
def run(config, root):
def run(config, root, failure_output=None):
# Never consume a deployment database, media tree, session or credential.
os.environ["DATABASE_URL"] = os.environ["SYNTHETIC_DATABASE_URL"]
os.environ["SECRET_KEY"] = "synthetic-disposable-not-a-deployment-key"
@ -133,8 +262,10 @@ def run(config, root):
from django.db import connections
from vergabe_teilnahme.apps.accounts.models import Mitarbeiter
from vergabe_teilnahme.apps.aufgaben.models import Aufgabe
from vergabe_teilnahme.apps.ausschreibungen.models import Ausschreibung
from vergabe_teilnahme.apps.dokumente.models import Dokument
from vergabe_teilnahme.apps.lose.models import Los
call_command("migrate", verbosity=0)
for i in range(config["tenders"]):
@ -155,7 +286,9 @@ def run(config, root):
sampling_done = threading.Event()
sampler_threads = []
sampler_errors = []
sampling_lock = threading.Lock()
try:
startup_started = time.monotonic()
if container_mode:
app_container = os.environ["SYNTHETIC_CONTAINER"] + "-app"
root.chmod(0o755)
@ -204,14 +337,7 @@ def run(config, root):
timeout=30,
stdout=subprocess.DEVNULL,
)
port = (
subprocess.check_output(
["docker", "port", app_container, "8000/tcp"], text=True, timeout=10
)
.strip()
.split(":")[-1]
)
base_url = f"http://127.0.0.1:{port}"
base_url = container_url(app_container)
else:
server = subprocess.Popen(
[
@ -231,40 +357,41 @@ def run(config, root):
stdout=server_log,
stderr=server_log,
)
for _ in range(100):
try:
with urllib.request.urlopen(base_url + "/health/", timeout=1) as response:
if response.status == 200:
break
except OSError:
if server is not None and server.poll() is not None:
raise RuntimeError("Gunicorn startup failed; see isolated server log")
time.sleep(0.1)
else:
raise RuntimeError("Gunicorn readiness timeout")
samples = []
db_samples = []
phase = {"name": "idle"}
phase = {"name": "startup", "kind": "startup"}
def sample():
ticks_per_second = os.sysconf("SC_CLK_TCK")
while not sampling_done.is_set():
if container_mode:
value = subprocess.check_output(
[
"docker",
"exec",
app_container,
"sh",
"-c",
(
"cat /sys/fs/cgroup/cpu.stat; echo MEMORY; "
"cat /sys/fs/cgroup/memory.current"
),
],
text=True,
timeout=5,
if phase.get("kind") == "restart":
samples.append(
{
"at": datetime.now(UTC).isoformat(),
"phase": phase["name"],
"expected_unavailable": True,
"reason": "planned restart",
}
)
sampling_done.wait(0.5)
continue
if container_mode:
with sampling_lock:
value = subprocess.check_output(
[
"docker",
"exec",
app_container,
"sh",
"-c",
(
"cat /sys/fs/cgroup/cpu.stat; echo MEMORY; "
"cat /sys/fs/cgroup/memory.current"
),
],
text=True,
timeout=5,
)
cpu_text, memory_text = value.split("MEMORY\n")
counters = {
k: int(v) for k, v in (line.split() for line in cpu_text.splitlines())
@ -342,11 +469,15 @@ def run(config, root):
sampler_threads.extend([sampler, db_sampler])
sampler.start()
db_sampler.start()
wait_ready(base_url)
startup_seconds = time.monotonic() - startup_started
phase.update(name="idle", kind="idle")
time.sleep(config["idle_seconds"])
phase["name"] = "ordinary"
deadline = time.monotonic() + config["duration_seconds"]
completed = [0] * config["users"]
barrier = threading.Barrier(config["users"])
plan = phase_plan(config)
active_plan = []
phase_results = []
transport_rebinds = []
phase_completed = {}
def worker(i):
jar = http.cookiejar.CookieJar()
@ -400,11 +531,14 @@ def run(config, root):
received = len(response.read())
status = response.status
ok = status == expected
if transfer and method == "get":
ok = ok and received == config["document_bytes"]
response.close()
except Exception:
status, ok, received = None, False, 0
rows.append(
{
"phase": phase["name"],
"user": i,
"operation": operation,
"transfer": transfer,
@ -428,59 +562,172 @@ def run(config, root):
for n in range(config["iterations"]):
if time.monotonic() >= deadline:
break
request("list", "get", "/ausschreibungen/")
request("search", "get", "/ausschreibungen/?q=Synthetic")
title = f"Workload-{i}-{n}"
created = request(
"tender-create",
"post",
"/ausschreibungen/neu/",
{"titel": title, "ausschreiber": "Synthetic"},
302,
)
tender = Ausschreibung.objects.filter(titel=title).first() if created else None
if tender:
base = f"/ausschreibungen/{tender.pk}/"
request("detail", "get", base)
request(
"lot-create",
"post",
base + "lose/neu/",
{"losnummer": "1", "lostitel": "Synthetic lot"},
302,
)
request(
"task-create",
"post",
base + "aufgaben/neu/",
{"titel": "Synthetic task", "typ": "fachlich", "prioritaet": 2},
302,
)
request(
"document-upload",
"post",
base + "dokumente/hochladen/",
{"kategorie": "intern", "version": "1.0"},
302,
True,
)
document = Dokument.objects.filter(ausschreibung=tender).first()
if document:
request("document-download", "get", document.datei.url, transfer=True)
title = f"Workload-{phase['name']}-{i}-{n}"
tender = lot = task = document = None
base = None
for index, operation in enumerate(active_plan):
dependency = DEPENDENCIES[operation]
entity = {
"tender-create": tender,
"lot-create": lot,
"task-create": task,
"document-upload": document,
}
if dependency and entity[dependency] is None:
rows.append(
{
"phase": phase["name"],
"user": i,
"operation": operation,
"transfer": operation.startswith("document-"),
"seconds": 0,
"status": None,
"ok": False,
"response_bytes": 0,
"reason": "dependency operation failed",
}
)
continue
if operation == "list":
request(operation, "get", "/ausschreibungen/")
elif operation == "search":
request(operation, "get", "/ausschreibungen/?q=Synthetic")
elif operation == "tender-create":
if request(
operation,
"post",
"/ausschreibungen/neu/",
{"titel": title, "ausschreiber": "Synthetic"},
302,
):
tender = Ausschreibung.objects.filter(titel=title).first()
base = f"/ausschreibungen/{tender.pk}/" if tender else None
elif operation == "detail":
request(operation, "get", base)
elif operation == "tender-edit":
changed = f"{title} edited-{index}"
request(
operation,
"post",
base + "bearbeiten/",
{"titel": changed, "ausschreiber": "Synthetic edited"},
302,
)
tender.refresh_from_db()
rows[-1]["ok"] &= tender.titel == changed
elif operation == "lot-create":
request(
operation,
"post",
base + "lose/neu/",
{"losnummer": "1", "lostitel": "Synthetic lot"},
302,
)
lot = Los.objects.filter(ausschreibung=tender).first()
elif operation == "lot-edit":
changed = f"Edited lot-{index}"
request(
operation,
"post",
base + f"lose/{lot.pk}/bearbeiten/",
{"losnummer": "1", "lostitel": changed},
302,
)
lot.refresh_from_db()
rows[-1]["ok"] &= lot.lostitel == changed
elif operation == "task-create":
request(
operation,
"post",
base + "aufgaben/neu/",
{"titel": "Synthetic task", "typ": "fachlich", "prioritaet": 2},
302,
)
task = Aufgabe.objects.filter(ausschreibung=tender).first()
elif operation == "task-edit":
changed = f"Edited task-{index}"
request(
operation,
"post",
base + f"aufgaben/{task.pk}/bearbeiten/",
{"titel": changed, "typ": "fachlich", "prioritaet": 1},
302,
)
task.refresh_from_db()
rows[-1]["ok"] &= task.titel == changed and task.prioritaet == 1
elif operation == "document-upload":
request(
operation,
"post",
base + "dokumente/hochladen/",
{"kategorie": "intern", "version": "1.0"},
302,
True,
)
document = Dokument.objects.filter(ausschreibung=tender).first()
elif operation == "document-download":
request(operation, "get", document.datei.url, transfer=True)
completed[i] += 1
time.sleep(config["think_seconds"])
time.sleep(active_phase["think_seconds"])
connections.close_all()
return rows
started = datetime.now(UTC).isoformat()
before = time.perf_counter()
with concurrent.futures.ThreadPoolExecutor(max_workers=config["users"]) as pool:
rows = [r for group in pool.map(worker, range(config["users"])) for r in group]
rows = []
for active_phase in plan:
phase.update(name=active_phase["name"], kind=active_phase["kind"])
phase_started = time.monotonic()
if active_phase["kind"] == "restart":
old_url = base_url
with sampling_lock:
base_url = restart_container(app_container)
transport_rebinds.append(
{"phase": active_phase["name"], "before": old_url, "after": base_url}
)
phase_results.append(
{
"name": active_phase["name"],
"kind": "restart",
"seconds": time.monotonic() - phase_started,
"ready": True,
}
)
continue
active_plan = active_phase["operations"]
deadline = time.monotonic() + active_phase["duration_seconds"]
completed = [0] * config["users"]
barrier = threading.Barrier(config["users"])
with concurrent.futures.ThreadPoolExecutor(max_workers=config["users"]) as pool:
phase_rows = [
r for group in pool.map(worker, range(config["users"])) for r in group
]
phase_completed[active_phase["name"]] = completed
phase_ordinary = summarize([r for r in phase_rows if not r["transfer"]])
phase_transfers = summarize([r for r in phase_rows if r["transfer"]])
expected = expected_requests(completed, active_plan, config["users"])
phase_results.append(
{
"name": active_phase["name"],
"kind": "load",
"seconds": time.monotonic() - phase_started,
"operations": active_plan,
"expected_requests": expected,
"actual_requests": len(phase_rows),
"complete": all(completed) and len(phase_rows) == expected,
"ordinary": phase_ordinary,
"document_transfer": phase_transfers,
}
)
rows.extend(phase_rows)
ordinary = summarize([r for r in rows if not r["transfer"]])
transfers = summarize([r for r in rows if r["transfer"]])
complete = all(completed) and len(rows) == sum(completed) * 8 + config["users"] * 2
complete = all(p.get("complete", p.get("ready", False)) for p in phase_results)
result = {
"schema_version": 1,
"schema_version": 2,
"startup_seconds": startup_seconds,
"transport_rebinds": transport_rebinds,
"phase_results": phase_results,
"started_at": started,
"finished_at": datetime.now(UTC).isoformat(),
"duration_seconds": time.perf_counter() - before,
@ -499,16 +746,20 @@ def run(config, root):
"provisional_pass": complete
and ordinary["errors"] == 0
and transfers["errors"] == 0
and ordinary["p95_seconds"] <= config["ordinary_p95_seconds"],
and all(
p["ordinary"]["p95_seconds"] <= config["ordinary_p95_seconds"]
for p in phase_results
if p["kind"] == "load"
),
"limitations": [
"No browser, ingress, TLS or SSO",
"No CPU/memory allocation recommendation or human acceptance",
"No timed startup/restart or distinct burst; idle and ordinary phases only",
"No deployed-release hardware or shared-service contention acceptance",
],
"operations": rows,
"runtime_samples": samples,
"database_samples": db_samples,
"completed_iterations": completed,
"completed_iterations": phase_completed,
}
sampling_done.set()
for thread in sampler_threads:
@ -537,6 +788,26 @@ def run(config, root):
"dockerfile_sha256": hashlib.sha256(Path("Dockerfile").read_bytes()).hexdigest(),
}
return result
except Exception as exc:
if failure_output is not None:
failure = {
"schema_version": 2,
"status": "failed",
"complete": False,
"provisional_pass": False,
"measurement_complete": False,
"error_type": type(exc).__name__,
"phase": locals().get("phase", {"name": "startup"}),
"config": config,
"phase_results": locals().get("phase_results", []),
"operations": locals().get("rows", []),
"runtime_samples": locals().get("samples", []),
"database_samples": locals().get("db_samples", []),
"measurement_gaps": [{"reason": "runtime failure; partial evidence only"}],
"driver_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
}
failure_output.write_text(json.dumps(failure, indent=2) + "\n")
raise
finally:
sampling_done.set()
for thread in sampler_threads:
@ -618,7 +889,7 @@ if __name__ == "__main__":
f"postgres://synthetic:synthetic-local-only@127.0.0.1:{port}/synthetic"
)
with tempfile.TemporaryDirectory(prefix="vergabe-synthetic-") as directory:
result = run(config, Path(directory))
result = run(config, Path(directory), args.output)
result["postgres_image_id"] = subprocess.check_output(
["docker", "inspect", "--format", "{{.Image}}", name], text=True
).strip()

View file

@ -1,8 +1,11 @@
import importlib.util
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
spec = importlib.util.spec_from_file_location(
"pilot", Path(__file__).with_name("synthetic_pilot.py")
@ -54,7 +57,111 @@ class SyntheticContractTests(unittest.TestCase):
self.assertEqual(self.config()["ordinary_p95_seconds"], 2)
class PlanTests(unittest.TestCase):
def test_edit_dependency_and_unknown_operation_fail_before_runtime(self):
for operations in [
["task-edit"],
["tender-create", "lot-edit"],
["unknown"],
["tender-create", "tender-create"],
]:
with self.subTest(operations=operations), self.assertRaises(ValueError):
pilot.operation_plan(operations)
def test_repeated_reads_change_derived_expectation(self):
plan = pilot.operation_plan(["list", "list", "tender-create", "detail", "tender-edit"])
self.assertEqual(pilot.expected_requests([2, 3], plan, 2), 29)
def test_restart_requires_container_and_unique_names(self):
base = dict(duration_seconds=1, think_seconds=0)
with self.assertRaises(ValueError):
pilot.phase_plan(base | {"phases": [{"name": "restart", "kind": "restart"}]})
with self.assertRaises(ValueError):
pilot.phase_plan(
base
| {
"phases": [
dict(name="x", kind="load", duration_seconds=1, think_seconds=0),
dict(name="x", kind="load", duration_seconds=1, think_seconds=0),
]
}
)
def test_planned_restart_gap_does_not_hide_same_phase_gap(self):
rows = [
dict(monotonic_seconds=0, phase="ordinary"),
dict(expected_unavailable=True, phase="restart"),
dict(monotonic_seconds=10, phase="recovery"),
]
self.assertEqual(pilot.sampling_gaps(rows, [{}], []), [])
rows.append(dict(monotonic_seconds=15, phase="recovery"))
self.assertTrue(pilot.sampling_gaps(rows, [{}], []))
class ComparisonTests(unittest.TestCase):
def test_restart_counter_reset_is_not_averaged_across_phase_boundary(self):
samples = []
for phase, start in [("ordinary", 0), ("recovery", 10)]:
for offset in range(2):
samples.append(
dict(
phase=phase,
monotonic_seconds=start + offset,
cpu_seconds=1 + offset,
rss_bytes=1024**2,
cpu_stat=dict(
nr_throttled=offset, nr_periods=offset, throttled_usec=offset
),
)
)
record = dict(
measurement_complete=True,
measurement_gaps=[],
app_runtime=dict(nano_cpus=1000000000, memory_bytes=1073741824),
runtime_samples=samples,
phase_results=[dict(name=n, kind="load") for n in ["ordinary", "recovery"]],
database_samples=[dict(phase="ordinary", docker_stats={"CPUPerc": "1%"})],
ordinary={},
document_transfer={},
complete=True,
provisional_pass=True,
config={},
completed_iterations={},
)
with tempfile.TemporaryDirectory() as tmp:
source, output = Path(tmp) / "input.json", Path(tmp) / "output.json"
source.write_text(json.dumps(record))
subprocess.run(
[
sys.executable,
str(Path(__file__).with_name("compare_synthetic_limits.py")),
str(source),
"--output",
str(output),
],
check=True,
stdout=subprocess.DEVNULL,
)
measured = json.loads(output.read_text())["application_measured"]
self.assertEqual(measured["cpu_mean_m"], 1000)
self.assertEqual(measured["throttled_periods"], 2)
class LifecycleTests(unittest.TestCase):
def test_restart_checks_new_loopback_binding(self):
with (
mock.patch.object(pilot.subprocess, "run"),
mock.patch.object(pilot, "container_url", return_value="http://127.0.0.1:49002"),
mock.patch.object(pilot, "wait_ready") as ready,
):
self.assertEqual(pilot.restart_container("synthetic"), "http://127.0.0.1:49002")
ready.assert_called_once_with("http://127.0.0.1:49002")
def test_public_binding_is_rejected(self):
with mock.patch.object(pilot.subprocess, "check_output", return_value="0.0.0.0:49002"):
with self.assertRaises(ValueError):
pilot.container_url("synthetic")
def test_process_group_cleanup_stops_real_child(self):
process = subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(60)"], start_new_session=True