m4
d4b53bfb08
Docker / build (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.12) (push) Has been cancelled
Test / pytest (windows-latest, 3.11) (push) Has been cancelled
Test / pytest (windows-latest, 3.12) (push) Has been cancelled
Lint / ruff (push) Has been cancelled
Test / pytest (ubuntu-latest, 3.11) (push) Has been cancelled
Build / build (push) Has been cancelled
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""Cancellation regressions using disposable local process groups only."""
|
|
|
|
import asyncio
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
import EvoScientist.native_sandbox as sandbox
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups")
|
|
async def test_async_execute_cancellation_waits_for_worker_cleanup(tmp_path, monkeypatch):
|
|
files = tmp_path / "files"
|
|
runtime = tmp_path / "runtime"
|
|
files.mkdir()
|
|
runtime.mkdir()
|
|
backend = sandbox.NativeWorkspaceBackend(files, runtime, timeout=3)
|
|
started = threading.Event()
|
|
cleaned = threading.Event()
|
|
|
|
def execute(_command, *, timeout, cancel_event):
|
|
process = subprocess.Popen(
|
|
["/bin/sh", "-c", "sleep 30 & wait"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True,
|
|
)
|
|
started.set()
|
|
try:
|
|
sandbox._collect_process(process, timeout=3, output_limit=1024,
|
|
cancel_event=cancel_event)
|
|
assert process.poll() is not None
|
|
cleaned.set()
|
|
finally:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait(timeout=3)
|
|
|
|
monkeypatch.setattr(backend._executor, "execute", execute)
|
|
task = asyncio.create_task(backend.aexecute("sleep 30"))
|
|
try:
|
|
assert await asyncio.to_thread(started.wait, 3)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert cleaned.is_set()
|
|
finally:
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups")
|
|
def test_collector_read_failure_reaps_process(monkeypatch):
|
|
process = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(30)"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
|
|
def failed_read(*_args):
|
|
raise OSError("collector read failed")
|
|
|
|
try:
|
|
with monkeypatch.context() as patcher:
|
|
patcher.setattr(sandbox.os, "read", failed_read)
|
|
with pytest.raises(OSError, match="collector read failed"):
|
|
sandbox._collect_process(process, timeout=3, output_limit=1024)
|
|
assert process.poll() is not None, "collector failure left command running"
|
|
finally:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait(timeout=3)
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix", reason="POSIX process groups")
|
|
@pytest.mark.parametrize("cancel", [True, False])
|
|
def test_collector_cancel_and_timeout_reap_process_group(cancel):
|
|
process = subprocess.Popen(
|
|
["/bin/sh", "-c", "sleep 30 & wait"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
event = threading.Event()
|
|
if cancel:
|
|
event.set()
|
|
try:
|
|
_, code, _, timed_out, cancelled = sandbox._collect_process(
|
|
process, timeout=1, output_limit=1024, cancel_event=event
|
|
)
|
|
assert code == (130 if cancel else 124)
|
|
assert (timed_out, cancelled) == (not cancel, cancel)
|
|
assert process.poll() is not None
|
|
with pytest.raises(ProcessLookupError):
|
|
os.killpg(process.pid, 0)
|
|
finally:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait(timeout=3) |