From 458c4f342c0b97bab2f5fa55e30076255fad7feb Mon Sep 17 00:00:00 2001 From: Tomislav Greblicki Date: Thu, 1 Oct 2026 14:55:08 +0200 Subject: [PATCH] add worker thread for task execution, support SIGABRT for aborting tasks, update unit tests and docs --- digitalai/release/integration/base_task.py | 30 +- digitalai/release/integration/wrapper.py | 125 ++++++-- docs/classes/base_task.md | 39 ++- docs/examples.md | 20 +- tests/release/integration/blocking_task.py | 14 + tests/release/integration/test_base_task.py | 40 +++ .../release/integration/test_wrapper_abort.py | 290 ++++++++++++++++++ 7 files changed, 527 insertions(+), 31 deletions(-) create mode 100644 tests/release/integration/blocking_task.py create mode 100644 tests/release/integration/test_wrapper_abort.py diff --git a/digitalai/release/integration/base_task.py b/digitalai/release/integration/base_task.py index d75eda9..07efa82 100644 --- a/digitalai/release/integration/base_task.py +++ b/digitalai/release/integration/base_task.py @@ -1,4 +1,5 @@ import sys +import threading from abc import ABC, abstractmethod from typing import Any, Dict, Optional @@ -20,7 +21,8 @@ def __init__(self): self.release_context = None self.release_server_url = None self.input_properties = None - self.output_context = None + self.output_context = OutputContext(0, "", {}, []) + self._abort_event = threading.Event() def execute_task(self) -> None: """ @@ -29,7 +31,6 @@ def execute_task(self) -> None: is raised, the task's exit code is also set to 1. """ try: - self.output_context = OutputContext(0, "", {}, []) self.execute() except AbortException: dai_logger.info("Abort requested") @@ -51,10 +52,31 @@ def execute(self) -> None: def abort(self) -> None: """ - Sets the task's exit code to 1 and exits the program with a status code of 1. + Marks the task as aborted (exit code 1, "Abort requested") and raises + AbortException. When called from within execute(), this is caught by + execute_task()'s existing AbortException handling, stopping the task + immediately. When called from the SDK's signal handler (outside of + execute_task()), the caller is responsible for catching this exception + and performing its own cleanup/exit. """ self.set_exit_code(1) - sys.exit(1) + self.set_error_message("Abort requested") + raise AbortException() + + def is_abort_requested(self) -> bool: + """ + Returns True once the SDK has signaled that an abort was requested + (e.g. the process received SIGABRT/SIGTERM). Tasks may poll this from + within execute() to cooperatively stop long-running work promptly. + """ + return self._abort_event.is_set() + + def _request_abort(self) -> None: + """ + Internal SDK hook: signals is_abort_requested() to start returning True. + Not intended to be called by task authors. + """ + self._abort_event.set() def get_output_context(self) -> OutputContext: """ diff --git a/digitalai/release/integration/wrapper.py b/digitalai/release/integration/wrapper.py index a9898c5..5077867 100644 --- a/digitalai/release/integration/wrapper.py +++ b/digitalai/release/integration/wrapper.py @@ -7,6 +7,7 @@ import os import signal import sys +import threading import time from typing import Any, Dict, Optional, Tuple @@ -14,6 +15,7 @@ from digitalai.release.integration import k8s, watcher from .base_task import BaseTask +from .exceptions import AbortException from .input_context import InputContext from .job_data_encryptor import AESJobDataEncryptor, NoOpJobDataEncryptor from .logger import dai_logger @@ -45,6 +47,10 @@ HTTP_CONNECT_TIMEOUT = float(os.getenv('HTTP_CONNECT_TIMEOUT', '10')) HTTP_READ_TIMEOUT = float(os.getenv('HTTP_READ_TIMEOUT', '60')) +# How often the main thread polls the worker thread (running the task's +# execute_task()) via join(), so it stays responsive to SIGABRT/SIGTERM. +WORKER_POLL_INTERVAL_SECONDS = 0.5 + # A single Session reused across all callback requests so the underlying urllib3 # connection pool is shared (instead of opening a fresh connection per call). _http_session: requests.Session = requests.Session() @@ -63,24 +69,88 @@ def get_encryptor(): # Initialize the global task object dai_task_object: Optional[BaseTask] = None +# Single-writer guard so the output context is flushed exactly once, whether +# the task finishes normally (worker thread's completion path) or is aborted +# (main thread's signal handler). Reset at the start of each execute_task() +# call so daemon-mode iterations each get a fresh guard. +_report_lock = threading.Lock() +_result_reported = False + + +def _reset_report_guard() -> None: + """Clears the single-writer flush guard ahead of a new task execution.""" + global _result_reported + with _report_lock: + _result_reported = False + + +def _report_once() -> bool: + """ + Flushes the current dai_task_object's output context exactly once. The + first caller (either the worker thread's normal completion path or the + main thread's abort handler) wins; subsequent calls are no-ops. + + Returns True if this call performed the flush, False if it was already + reported by another caller. + """ + global _result_reported + with _report_lock: + if _result_reported: + return False + _result_reported = True + + if dai_task_object is not None and dai_task_object.get_output_context() is not None: + update_output_context(dai_task_object.get_output_context()) + else: + dai_logger.error("No output context available to report task result") + update_output_context(OutputContext(1, "Task produced no output context", {}, [])) + return True + def abort_handler(signum, frame): """ - This function handles the abort request by calling the abort method on the global task object, if it exists. - If the task object does not exist, it logs a message and exits with a status code of 1. + Handles SIGTERM/SIGABRT by requesting a cooperative abort and running the + task's abort() cleanup hook (if a task object exists), then flushing the + (aborted) output context exactly once via _report_once(), and finally + terminating the process immediately with os._exit(1). + + os._exit() (rather than sys.exit()) is used deliberately: it bypasses all + Python finally/atexit machinery, which matters in daemon mode where run()'s + finally block would otherwise restart a blocking input-context watcher and + prevent the process from actually exiting. + + The worker thread that may still be running the task's execute() is a + daemon thread and is not waited on; process exit here is sufficient to + end it. """ - dai_logger.info("Received SIGTERM to gracefully stop the process") + signal_name = signal.Signals(signum).name + dai_logger.info(f"Received {signal_name} to gracefully stop the process") global dai_task_object if dai_task_object: - dai_task_object.abort() + dai_task_object._request_abort() + try: + dai_task_object.abort() + except AbortException: + # Expected: the default abort() raises this after marking the + # output context as aborted. + pass + except SystemExit: + # A task's overridden abort() might still call sys.exit()/exit(); + # ignore it so we can still flush and exit deliberately below. + pass + except Exception: + dai_logger.error("Unexpected error occurred in abort()", exc_info=True) else: dai_logger.info("Abort requested") - sys.exit(1) + + _report_once() + os._exit(1) # Register abort handler signal.signal(signal.SIGTERM, abort_handler) +signal.signal(signal.SIGABRT, abort_handler) def _post_callback(url: str, encrypted_json) -> requests.Response: @@ -224,26 +294,35 @@ def should_retry_callback_request(encrypted_data) -> bool: def execute_task(task_object: BaseTask): """ - Execute the given task object by setting it as the global task object and starting its execution. - If an exception is raised during execution, log the error. Finally, update the output context file - using the output context of the task object. + Execute the given task object on a dedicated daemon worker thread, while the + main thread waits responsively (polling worker.join with a short timeout) + so it stays free to handle SIGABRT/SIGTERM even if the task is blocked in + an uninterruptible call. + + The worker thread is a daemon thread: if the process exits (e.g. via the + signal handler's os._exit(1) during an abort), it is not waited on. + + After the worker finishes on its own (the non-aborted path), the output + context is flushed exactly once via _report_once() -- a no-op if the + abort handler already reported it first. """ global dai_task_object - try: - dai_task_object = task_object - dai_logger.info("Starting task execution") - dai_task_object.execute_task() - except Exception: - dai_logger.error("Unexpected error occurred.", exc_info=True) - finally: - # Guard against a task object that was never constructed or that failed - # before producing an output context, so the finally block does not raise - # a second exception that masks the original one. - if dai_task_object is not None and dai_task_object.get_output_context() is not None: - update_output_context(dai_task_object.get_output_context()) - else: - dai_logger.error("No output context available to report task result") - update_output_context(OutputContext(1, "Task produced no output context", {}, [])) + _reset_report_guard() + dai_task_object = task_object + dai_logger.info("Starting task execution") + + def _run_task(): + try: + task_object.execute_task() + except Exception: + dai_logger.error("Unexpected error occurred.", exc_info=True) + + worker = threading.Thread(target=_run_task, name="dai-task-worker", daemon=True) + worker.start() + while worker.is_alive(): + worker.join(timeout=WORKER_POLL_INTERVAL_SECONDS) + + _report_once() def find_class_file(root_dir, class_name): diff --git a/docs/classes/base_task.md b/docs/classes/base_task.md index 7acfdfe..cba29e9 100644 --- a/docs/classes/base_task.md +++ b/docs/classes/base_task.md @@ -9,7 +9,7 @@ | **Class** | `BaseTask` | | **Module** | `digitalai.release.integration.base_task` | | **Source** | [`digitalai/release/integration/base_task.py`](../../digitalai/release/integration/base_task.py) | -| **Methods** | 18 | +| **Methods** | 20 | Subclass `BaseTask` to build a container-based integration task. Implement the abstract [`execute`](#execute) method with your task logic; the Release runtime @@ -46,7 +46,8 @@ class Hello(BaseTask): |--------|---------|-------------| | [`execute`](#execute) | `None` | Abstract method holding the task's main logic; implemented by subclasses. | | [`execute_task`](#execute_task) | `None` | Runs `execute` with error handling and exit-code management; called by the runtime. | -| [`abort`](#abort) | `None` | Sets the exit code to 1 and exits the program. | +| [`abort`](#abort) | `None` | Marks the task as aborted (exit code 1, "Abort requested") and raises `AbortException`. | +| [`is_abort_requested`](#is_abort_requested) | `bool` | Returns `True` once the SDK has signaled that an abort was requested. | | [`get_input_properties`](#get_input_properties) | `dict[str, Any]` | Returns the task's input properties. | | [`set_output_property`](#set_output_property) | `None` | Sets a named output property of the task. | | [`get_output_properties`](#get_output_properties) | `dict[str, Any]` | Returns the output properties of the task. | @@ -102,12 +103,44 @@ _No parameters._ ### `abort` -Sets the task's exit code to `1` and exits the program with a status code of `1`. +Marks the task as aborted by setting the exit code to `1` and the error message to +`"Abort requested"`, then raises `AbortException`. The default implementation is +the user-overridable hook invoked by the SDK when the process receives `SIGABRT` +or `SIGTERM` (e.g. when a Release job is aborted). Override it to perform custom +cleanup; call `super().abort()` (or raise `AbortException` yourself) to preserve +the exit code / error message contract, or call `self.set_exit_code(...)` / +`self.set_error_message(...)` directly if you need different semantics. + +For long-running tasks, prefer polling [`is_abort_requested`](#is_abort_requested) +from within `execute` to stop cooperatively and promptly, rather than relying +solely on `abort` being called. _No parameters._ **Returns:** _None._ +**Raises:** `AbortException`. + +[↑ Method index](#method-index) + +### `is_abort_requested` + +Returns `True` once the SDK has signaled that an abort was requested (the process +received `SIGABRT`/`SIGTERM`). Tasks may poll this from within `execute` to +cooperatively stop long-running work promptly instead of waiting to be +forcibly terminated. + +```python +class MyTask(BaseTask): + def execute(self) -> None: + while not self.is_abort_requested(): + do_one_unit_of_work() +``` + +_No parameters._ + +**Returns:** `bool` — `True` if an abort has been requested, `False` otherwise. + [↑ Method index](#method-index) ### `get_input_properties` diff --git a/docs/examples.md b/docs/examples.md index c75201e..700eb61 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -104,7 +104,8 @@ class Deploy(BaseTask): Raising an exception from `execute` fails the task: `execute_task` records the message and sets a non-zero exit code. To stop immediately, call `abort`, which -sets the exit code to `1` and exits the process. +marks the task as aborted (exit code `1`, "Abort requested") and raises +`AbortException`, unwinding out of `execute`. **Classes:** [`BaseTask`](classes/base_task.md) @@ -126,6 +127,23 @@ class GuardedTask(BaseTask): self.add_comment("Proceeding") ``` +For long-running tasks, the SDK also handles `SIGABRT`/`SIGTERM` (e.g. when a +Release job is aborted) by running `abort` on your task automatically — you +don't need to install a signal handler yourself. If your `execute` performs +work in a loop or in chunks, poll `is_abort_requested` to stop promptly instead +of waiting for the whole task to finish: + +```python +class LongRunningTask(BaseTask): + + def execute(self) -> None: + for item in self.input_properties.get("items", []): + if self.is_abort_requested(): + self.add_comment("Stopping early: abort requested") + return + process(item) +``` + [↑ Contents](#contents) ## Call the Release API with ApiBaseTask diff --git a/tests/release/integration/blocking_task.py b/tests/release/integration/blocking_task.py new file mode 100644 index 0000000..de10ce6 --- /dev/null +++ b/tests/release/integration/blocking_task.py @@ -0,0 +1,14 @@ +import time + +from digitalai.release.integration import BaseTask + + +class BlockingTask(BaseTask): + """ + Fixture task used by test_wrapper_abort.py to verify that SIGABRT interrupts + a task promptly even while it is blocked in an uninterruptible call (i.e. + without polling is_abort_requested()). + """ + + def execute(self) -> None: + time.sleep(30) diff --git a/tests/release/integration/test_base_task.py b/tests/release/integration/test_base_task.py index f75987f..b6ef8fc 100644 --- a/tests/release/integration/test_base_task.py +++ b/tests/release/integration/test_base_task.py @@ -9,6 +9,7 @@ AutomatedTaskAsUserContext, ReleaseContext, ) +from digitalai.release.integration.output_context import OutputContext # Realistic ids, matching the sample input context. FOLDER_ID = 'Applications/Folder1f65c7220b394afbb941154342fd9fc6' @@ -97,6 +98,45 @@ def execute(self) -> None: self.assertEqual(ctx.exit_code, 1) self.assertEqual(ctx.job_error_message, "Abort requested") + def test_abort_raises_abort_exception_and_marks_output_context(self): + task = self._task() + task.output_context = OutputContext(0, "", {}, []) + + with self.assertRaises(AbortException): + task.abort() + + ctx = task.get_output_context() + self.assertEqual(ctx.exit_code, 1) + self.assertEqual(ctx.job_error_message, "Abort requested") + + def test_output_context_is_never_none_immediately_after_construction(self): + # Guards against the window between task construction and the worker + # thread starting execute_task(), during which the main thread's + # SIGABRT/SIGTERM handler may already call abort(). + task = self._task() + self.assertIsNotNone(task.get_output_context()) + + def test_abort_immediately_after_construction_does_not_crash(self): + # Simulates an abort signal arriving before execute_task()/execute() + # has ever run on the worker thread. + task = self._task() + + with self.assertRaises(AbortException): + task.abort() + + ctx = task.get_output_context() + self.assertEqual(ctx.exit_code, 1) + self.assertEqual(ctx.job_error_message, "Abort requested") + + def test_is_abort_requested_defaults_to_false(self): + task = self._task() + self.assertFalse(task.is_abort_requested()) + + def test_is_abort_requested_true_after_request_abort(self): + task = self._task() + task._request_abort() + self.assertTrue(task.is_abort_requested()) + def test_set_output_property_rejects_empty_name(self): task = self._task() task.execute_task() diff --git a/tests/release/integration/test_wrapper_abort.py b/tests/release/integration/test_wrapper_abort.py new file mode 100644 index 0000000..31b160a --- /dev/null +++ b/tests/release/integration/test_wrapper_abort.py @@ -0,0 +1,290 @@ +""" +Concurrency-aware tests for the SIGABRT worker-thread abort model +(docs/SPEC-abort-worker-thread.md). + +Most tests here exercise wrapper.py's internals directly, in-process, with +os._exit() and update_output_context() mocked out so a simulated abort does +not tear down the test runner itself or touch real files/secrets/callbacks. + +One test (`test_real_sigabrt_interrupts_blocking_task`) sends a genuine +SIGABRT to a subprocess running the real wrapper end-to-end, mirroring how the +release-runner aborts a task, to prove the full integration works. +""" +import json +import os +import signal +import subprocess +import sys +import threading +import time +import unittest +from unittest import mock + +import digitalai.release.integration.wrapper as wrapper +from digitalai.release.integration.base_task import BaseTask +from digitalai.release.integration.output_context import OutputContext + +THIS_DIR = os.path.dirname(os.path.abspath(__file__)) + + +class _RecordingAbortTask(BaseTask): + """Stub task that counts abort() calls and lets tests set a canned output_context.""" + + def __init__(self): + super().__init__() + self.abort_call_count = 0 + + def execute(self) -> None: # pragma: no cover - not used directly + pass + + def abort(self) -> None: + self.abort_call_count += 1 + super().abort() + + +class WrapperAbortHandlerTests(unittest.TestCase): + """In-process tests for wrapper.abort_handler() and the single-writer flush guard.""" + + def setUp(self): + wrapper._reset_report_guard() + wrapper.dai_task_object = None + + def tearDown(self): + wrapper._reset_report_guard() + wrapper.dai_task_object = None + + def test_abort_handler_calls_abort_flushes_once_and_exits(self): + task = _RecordingAbortTask() + task.output_context = OutputContext(0, "", {}, []) + wrapper.dai_task_object = task + + with mock.patch.object(wrapper, "update_output_context") as mock_update, \ + mock.patch.object(wrapper.os, "_exit") as mock_exit: + wrapper.abort_handler(signal.SIGABRT, None) + + self.assertEqual(task.abort_call_count, 1) + mock_update.assert_called_once() + (reported_ctx,), _ = mock_update.call_args + self.assertEqual(reported_ctx.exit_code, 1) + self.assertEqual(reported_ctx.job_error_message, "Abort requested") + mock_exit.assert_called_once_with(1) + + def test_abort_handler_does_not_double_flush(self): + task = _RecordingAbortTask() + task.output_context = OutputContext(0, "", {}, []) + wrapper.dai_task_object = task + + with mock.patch.object(wrapper, "update_output_context") as mock_update, \ + mock.patch.object(wrapper.os, "_exit"): + # Simulate the worker's own normal-completion path reporting first. + self.assertTrue(wrapper._report_once()) + # Then the abort handler fires; abort() cleanup still runs, but the + # flush must not happen a second time. + wrapper.abort_handler(signal.SIGABRT, None) + + self.assertEqual(task.abort_call_count, 1) + mock_update.assert_called_once() + + def test_abort_handler_with_no_task_object_exits_cleanly(self): + wrapper.dai_task_object = None + + with mock.patch.object(wrapper, "update_output_context") as mock_update, \ + mock.patch.object(wrapper.os, "_exit") as mock_exit: + wrapper.abort_handler(signal.SIGTERM, None) + + mock_update.assert_called_once() + (reported_ctx,), _ = mock_update.call_args + self.assertEqual(reported_ctx.exit_code, 1) + mock_exit.assert_called_once_with(1) + + def test_abort_handler_before_worker_starts_still_reports_abort_requested(self): + # Simulates SIGABRT arriving right after dai_task_object is assigned, + # before the worker thread has had a chance to run execute_task()'s + # body -- i.e. before output_context was ever touched post-construction. + task = _RecordingAbortTask() + wrapper.dai_task_object = task + + with mock.patch.object(wrapper, "update_output_context") as mock_update, \ + mock.patch.object(wrapper.os, "_exit") as mock_exit: + wrapper.abort_handler(signal.SIGABRT, None) + + self.assertEqual(task.abort_call_count, 1) + mock_update.assert_called_once() + (reported_ctx,), _ = mock_update.call_args + self.assertEqual(reported_ctx.exit_code, 1) + self.assertEqual(reported_ctx.job_error_message, "Abort requested") + mock_exit.assert_called_once_with(1) + + def test_abort_handler_logs_actual_signal_name(self): + task = _RecordingAbortTask() + task.output_context = OutputContext(0, "", {}, []) + wrapper.dai_task_object = task + + with mock.patch.object(wrapper, "update_output_context"), \ + mock.patch.object(wrapper.os, "_exit"), \ + mock.patch.object(wrapper.dai_logger, "info") as mock_log_info: + wrapper.abort_handler(signal.SIGABRT, None) + + logged_messages = [call.args[0] for call in mock_log_info.call_args_list] + self.assertTrue(any("SIGABRT" in msg for msg in logged_messages)) + + +class _CooperativeTask(BaseTask): + """Task whose execute() polls is_abort_requested() and stops promptly, reporting + a distinct, custom result so tests can tell it stopped cooperatively rather + than being torn down by the abort handler.""" + + def execute(self) -> None: + while not self.is_abort_requested(): + time.sleep(0.02) + self.set_output_property("stopped_cooperatively", True) + self.set_error_message("Stopped cooperatively") + + +class WrapperExecuteTaskThreadingTests(unittest.TestCase): + """In-process tests for wrapper.execute_task()'s worker-thread + guard behavior.""" + + def setUp(self): + wrapper._reset_report_guard() + wrapper.dai_task_object = None + + def tearDown(self): + wrapper._reset_report_guard() + wrapper.dai_task_object = None + + def test_execute_task_normal_completion_reports_once(self): + class _FastTask(BaseTask): + def execute(self) -> None: + self.set_output_property("greeting", "Hello World") + + task = _FastTask() + + with mock.patch.object(wrapper, "update_output_context") as mock_update: + wrapper.execute_task(task) + + mock_update.assert_called_once() + (reported_ctx,), _ = mock_update.call_args + self.assertEqual(reported_ctx.exit_code, 0) + self.assertEqual(reported_ctx.output_properties, {"greeting": "Hello World"}) + + def test_cooperative_task_stops_promptly_when_abort_requested(self): + task = _CooperativeTask() + + def _request_abort_shortly(): + time.sleep(0.1) + task._request_abort() + + threading.Thread(target=_request_abort_shortly, daemon=True).start() + + with mock.patch.object(wrapper, "update_output_context") as mock_update: + start = time.monotonic() + wrapper.execute_task(task) + elapsed = time.monotonic() - start + + # The worker should notice is_abort_requested() and return well within + # a couple of poll intervals, without needing a real signal. + self.assertLess(elapsed, 2.0) + mock_update.assert_called_once() + (reported_ctx,), _ = mock_update.call_args + self.assertEqual(reported_ctx.job_error_message, "Stopped cooperatively") + self.assertTrue(reported_ctx.output_properties.get("stopped_cooperatively")) + + def test_execute_task_resets_guard_across_daemon_iterations(self): + class _FastTask(BaseTask): + def execute(self) -> None: + pass + + task1 = _FastTask() + task2 = _FastTask() + + with mock.patch.object(wrapper, "update_output_context") as mock_update: + wrapper.execute_task(task1) + wrapper.execute_task(task2) + + # Each iteration must flush its own task's result; the guard from the + # first iteration must not suppress the second. + self.assertEqual(mock_update.call_count, 2) + self.assertIs(wrapper.dai_task_object, task2) + + +class WrapperRealSigabrtEndToEndTest(unittest.TestCase): + """ + Full end-to-end test: runs the real wrapper module as a subprocess against a + task that blocks in execute() without polling is_abort_requested(), sends it + a genuine SIGABRT partway through, and asserts it reports an aborted result + promptly instead of waiting for the 30s sleep to finish. + """ + + def setUp(self): + self.input_path = os.path.join(THIS_DIR, "blocking_input.json") + self.output_path = os.path.join(THIS_DIR, "blocking_output.json") + if os.path.exists(self.output_path): + os.remove(self.output_path) + + def tearDown(self): + for path in (self.input_path, self.output_path): + if os.path.exists(path): + os.remove(path) + + def test_real_sigabrt_interrupts_blocking_task(self): + input_context = { + "release": { + "id": "Applications/Folder1f65c7220b394afbb941154342fd9fc6/Release31de09e95c8e4ebb95aaed29a8082d0b", + "automatedTaskAsUser": {"username": "admin", "password": "admin"}, + }, + "task": { + "id": "Applications/Folder1f65c7220b394afbb941154342fd9fc6/Release31de09e95c8e4ebb95aaed29a8082d0b/Phase723a601c78804f7dbcaa8b05b83708f5/Task3a35b67b42b6428b854857fba470b39a", + "type": "containerExamples.BlockingTask", + "properties": [ + {"name": "capabilities", "value": ["remote"], "kind": "SET_OF_STRING", "category": "input", "password": False}, + ], + }, + } + with open(self.input_path, "w") as f: + json.dump(input_context, f) + + env = dict(os.environ) + env["INPUT_LOCATION"] = "blocking_input.json" + env["OUTPUT_LOCATION"] = "blocking_output.json" + env["RELEASE_URL"] = "http://localhost:5516" + + proc = subprocess.Popen( + [sys.executable, "-m", "digitalai.release.integration.wrapper"], + cwd=THIS_DIR, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + # Give the task time to enter its blocking sleep(30) call. + time.sleep(1.0) + start = time.monotonic() + proc.send_signal(signal.SIGABRT) + stdout, stderr = proc.communicate(timeout=10) + elapsed = time.monotonic() - start + except subprocess.TimeoutExpired: + proc.kill() + stdout, stderr = proc.communicate() + self.fail( + f"wrapper did not exit promptly after SIGABRT\nstdout:\n{stdout}\nstderr:\n{stderr}" + ) + + self.assertEqual( + proc.returncode, 1, + msg=f"wrapper exited with {proc.returncode}\nstdout:\n{stdout}\nstderr:\n{stderr}", + ) + # Must be well under the task's 30s sleep -- proves the abort was + # timely rather than relying on the runner's Phase-2 force-kill. + self.assertLess(elapsed, 10.0) + + self.assertTrue(os.path.exists(self.output_path), "wrapper did not produce blocking_output.json") + with open(self.output_path) as f: + actual_output = json.load(f) + + self.assertEqual(actual_output["exitCode"], 1) + self.assertEqual(actual_output["jobErrorMessage"], "Abort requested") + + +if __name__ == '__main__': + unittest.main()