Spaces:
Running
Running
| """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)) | |