87 lines
4.3 KiB
Python
87 lines
4.3 KiB
Python
"""Exercise process-level cancellation and deadlines with an unresponsive worker."""
|
|
from contextlib import ExitStack
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
from uuid import uuid4
|
|
|
|
from app.main import _mark_simulation_task_result, _register_simulation_task
|
|
from app.simulation.config import SolveIVPConfig, SolverActivityTracker
|
|
from app.simulation.native_codegen import runner
|
|
|
|
|
|
class NativeWorkerControlTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory(prefix="native-worker-control-")
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
self.helper = self.root / "worker.py"
|
|
self.helper.write_text(
|
|
"import sys,time\n"
|
|
"from pathlib import Path\n"
|
|
"Path(sys.argv[1]).write_text('{incomplete result')\n"
|
|
"print('{\"phase\":\"integrating\",\"time\":0.1,\"nfev\":7,\"acceptedSteps\":2}', file=sys.stderr,flush=True)\n"
|
|
"time.sleep(30)\n", encoding="utf-8")
|
|
self.build = SimpleNamespace(executable=self.root / "model", manifest={"buildKey": "test"},
|
|
cache_hit=False, seconds=0, details={}, close=Mock())
|
|
self.config = SolveIVPConfig(t_stop=1)
|
|
self.stack = ExitStack()
|
|
self.addCleanup(self.stack.close)
|
|
real_popen = subprocess.Popen
|
|
|
|
def launch(command, **kwargs):
|
|
output = command[command.index("--output") + 1]
|
|
return real_popen([sys.executable, str(self.helper), output], **kwargs)
|
|
|
|
self.stack.enter_context(patch.object(runner.subprocess, "Popen", side_effect=launch))
|
|
self.stack.enter_context(patch.object(runner, "_WORKER_SHUTDOWN_GRACE_SECONDS", .3))
|
|
|
|
def test_forced_cancel_retains_only_reported_progress_in_both_transports(self):
|
|
for raw in (False, True):
|
|
with self.subTest(raw_series=raw):
|
|
tracker = SolverActivityTracker()
|
|
result = runner.execute_native(self.build, self.config, .01,
|
|
run_dir=self.root / str(raw), raw_series=raw, activity_tracker=tracker,
|
|
cancel_check=lambda: tracker.snapshot().rhs_call_count >= 7)
|
|
self.assertEqual(result["status"], "cancelled")
|
|
self.assertEqual(result["simulatedUntil"], .1)
|
|
self.assertEqual(result["nfev"], 7)
|
|
self.assertEqual(result["acceptedSteps"], 2)
|
|
self.assertEqual(result["series"], {})
|
|
self.assertEqual(result["final"], {})
|
|
self.assertNotIn("solveSeconds", result)
|
|
self.assertNotIn("njev", result)
|
|
self.assertFalse(result["solverControl"]["statisticsComplete"])
|
|
self.assertTrue(result["solverControl"]["forcedTermination"])
|
|
|
|
def test_forced_cancel_is_stopped_in_the_web_task_contract(self):
|
|
self.stack.enter_context(patch.object(runner, "compile_native_program",
|
|
return_value=SimpleNamespace(variables=(), state_keys=())))
|
|
self.stack.enter_context(patch.object(runner, "build_native", return_value=self.build))
|
|
result = runner.simulate_native(object(), self.config, sample_step=.01,
|
|
cancel_check=lambda: True, raw_series=True).as_dict(raw_series=True)
|
|
task = _register_simulation_task(uuid4().hex)
|
|
result = _mark_simulation_task_result(task, result)
|
|
self.assertEqual(result["status"], "stopped")
|
|
self.assertTrue(result["partial"])
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["diagnostics"]["sampleCount"], 0)
|
|
json.dumps(result, allow_nan=False)
|
|
self.build.close.assert_called_once()
|
|
|
|
def test_worker_deadline_and_crash_are_not_cancellation(self):
|
|
with self.assertRaisesRegex(RuntimeError, "worker time limit expired"):
|
|
runner.execute_native(self.build, self.config, .01, run_dir=self.root / "timeout", timeout=.1)
|
|
self.helper.write_text("import sys\nsys.exit(17)\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(RuntimeError, "exited with code 17 without results"):
|
|
runner.execute_native(self.build, self.config, .01, run_dir=self.root / "crash")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|