Spaces:
Running
Running
File size: 1,411 Bytes
16f323d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """Exercise worker cleanup without requiring Linux ELF compilers."""
import os
from time import sleep
import pytest
from demo.execution import verification_reports
def successful_worker(task, connection, directory):
os.setsid()
connection.send(("report", task))
connection.send(("done", None))
connection.close()
def failing_worker(task, connection, directory):
os.setsid()
connection.send(("error", "invalid candidate"))
connection.close()
def stalled_worker(task, connection, directory):
os.setsid()
if task:
connection.send(("report", os.getpid()))
sleep(60)
def test_reports_and_errors():
assert list(verification_reports("report", _worker=successful_worker)) == ["report"]
with pytest.raises(ValueError, match="invalid candidate"):
list(verification_reports(None, _worker=failing_worker))
def test_closing_stream_stops_worker():
reports = verification_reports(True, _worker=stalled_worker)
pid = next(reports)
reports.close()
with pytest.raises(ProcessLookupError):
os.kill(pid, 0)
def test_verifier_worker_reports_build_errors(tmp_path):
from verifier import Task
task = Task(binary=tmp_path / "missing.elf", function=0x401000,
candidate="unsigned target(void) { return 0; }")
with pytest.raises(ValueError, match="missing.elf"):
list(verification_reports(task))
|