Update typing issues and fix tests for checking scenarios for abort

Author: kzscisoft

simvue/api.py

@@ -44,7 +44,7 @@ def set_json_header(headers: dict[str, str]) -> dict[str, str]:
     reraise=True,
 )
 def post(
-    url: str, headers: dict[str, str], data: dict[str, typing.Any], is_json: bool = True
+    url: str, headers: dict[str, str], data: typing.Any, is_json: bool = True
 ) -> requests.Response:
     """HTTP POST with retries
 

simvue/executor.py

@@ -27,6 +27,10 @@
 logger = logging.getLogger(__name__)
 
 
+class CompletionCallback(typing.Protocol):
+    def __call__(self, *, status_code: int, std_out: str, std_err: str) -> None: ...
+
+
 def _execute_process(
     proc_id: str,
     command: typing.List[str],
@@ -67,11 +71,13 @@ def __init__(self, simvue_runner: "simvue.Run", keep_logs: bool = True) -> None:
         """
         self._runner = simvue_runner
         self._keep_logs = keep_logs
-        self._completion_callbacks = {}
-        self._completion_triggers = {}
-        self._exit_codes = {}
-        self._std_err = {}
-        self._std_out = {}
+        self._completion_callbacks: dict[str, typing.Optional[CompletionCallback]] = {}
+        self._completion_triggers: dict[
+            str, typing.Optional[multiprocessing.synchronize.Event]
+        ] = {}
+        self._exit_codes: dict[str, int] = {}
+        self._std_err: dict[str, str] = {}
+        self._std_out: dict[str, str] = {}
         self._alert_ids: dict[str, str] = {}
         self._command_str: dict[str, str] = {}
         self._processes: dict[str, subprocess.Popen] = {}
@@ -84,9 +90,7 @@ def add_process(
         script: typing.Optional[pathlib.Path] = None,
         input_file: typing.Optional[pathlib.Path] = None,
         env: typing.Optional[typing.Dict[str, str]] = None,
-        completion_callback: typing.Optional[
-            typing.Callable[[int, str, str], None]
-        ] = None,
+        completion_callback: typing.Optional[CompletionCallback] = None,
         completion_trigger: typing.Optional[multiprocessing.synchronize.Event] = None,
         **kwargs,
     ) -> None:
@@ -141,6 +145,9 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
         """
         _pos_args = list(args)
 
+        if not self._runner.name:
+            raise RuntimeError("Cannot add process, expected Run instance to have name")
+
         if sys.platform == "win32" and completion_callback:
             logger.warning(
                 "Completion callback for 'add_process' may fail on Windows due to "
@@ -289,20 +296,20 @@ def _save_output(self) -> None:
 
     def kill_process(self, process_id: str) -> None:
         """Kill a running process by ID"""
-        if not (_process := self._processes.get(process_id)):
+        if not (process := self._processes.get(process_id)):
             logger.error(
                 f"Failed to terminate process '{process_id}', no such identifier."
             )
             return
 
-        _parent = psutil.Process(_process.pid)
+        parent = psutil.Process(process.pid)
 
-        for child in _parent.children(recursive=True):
+        for child in parent.children(recursive=True):
             logger.debug(f"Terminating child process {child.pid}: {child.name()}")
             child.kill()
 
-        logger.debug(f"Terminating child process {_process.pid}: {_process.args}")
-        _process.terminate()
+        logger.debug(f"Terminating child process {process.pid}: {process.args}")
+        process.terminate()
 
         self._execute_callback(process_id)
 
@@ -325,14 +332,14 @@ def _execute_callback(self, identifier: str) -> None:
         with open(f"{self._runner.name}_{identifier}.out") as out:
             std_out = out.read()
 
-        if self._completion_callbacks[identifier]:
-            self._completion_callbacks[identifier](
+        if callback := self._completion_callbacks.get(identifier):
+            callback(
                 status_code=self._processes[identifier].returncode,
                 std_out=std_out,
                 std_err=std_err,
             )
-        if self._completion_triggers[identifier]:
-            self._completion_triggers[identifier].set()
+        if completion_trigger := self._completion_triggers.get(identifier):
+            completion_trigger.set()
 
     def wait_for_completion(self) -> None:
         """Wait for all processes to finish then perform tidy up and upload"""

simvue/run.py

@@ -10,6 +10,7 @@
 import datetime
 import json
 import logging
+import pathlib
 import mimetypes
 import multiprocessing.synchronize
 import threading
@@ -50,6 +51,7 @@
     validate_timestamp,
 )
 
+
 if typing.TYPE_CHECKING:
     from .factory.proxy import SimvueBaseClass
     from .factory.dispatch import DispatcherBaseClass
@@ -101,6 +103,7 @@ def __init__(
         self._uuid: str = f"{uuid.uuid4()}"
         self._mode: typing.Literal["online", "offline", "disabled"] = mode
         self._name: typing.Optional[str] = None
+        self._testing: bool = False
         self._abort_on_alert: bool = True
         self._dispatch_mode: typing.Literal["direct", "queued"] = "queued"
         self._executor = Executor(self)
@@ -246,7 +249,7 @@ def _get_sysinfo(self) -> dict[str, typing.Any]:
 
     def _create_heartbeat_callback(
         self,
-    ) -> typing.Callable[[str, dict, str, bool], None]:
+    ) -> typing.Callable[[threading.Event], None]:
         if (
             self._mode == "online" and (not self._url or not self._id)
         ) or not self._heartbeat_termination_trigger:
@@ -354,7 +357,7 @@ def _online_dispatch_callback(
             buffer: list[typing.Any],
             category: str,
             url: str = self._url,
-            run_id: str = self._id,
+            run_id: typing.Optional[str] = self._id,
             headers: dict[str, str] = self._headers,
         ) -> None:
             if not buffer:
@@ -636,7 +639,7 @@ def add_process(
         self,
         identifier: str,
         *cmd_args,
-        executable: typing.Optional[typing.Union[str]] = None,
+        executable: typing.Optional[typing.Union[str, pathlib.Path]] = None,
         script: typing.Optional[pydantic.FilePath] = None,
         input_file: typing.Optional[pydantic.FilePath] = None,
         completion_callback: typing.Optional[
@@ -708,12 +711,20 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
                 "due to function pickling restrictions for multiprocessing"
             )
 
+        if isinstance(executable, pathlib.Path):
+            if not executable.is_file():
+                raise FileNotFoundError(
+                    f"Executable '{executable}' is not a valid file"
+                )
+
+        executable_str = f"{executable}"
+
         _cmd_list: typing.List[str] = []
         _pos_args = list(cmd_args)
 
         # Assemble the command for saving to metadata as string
         if executable:
-            _cmd_list += [executable]
+            _cmd_list += [executable_str]
         else:
             _cmd_list += [_pos_args[0]]
             executable = _pos_args[0]
@@ -742,10 +753,10 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
         self._executor.add_process(
             identifier,
             *_pos_args,
-            executable=executable,
+            executable=executable_str,
             script=script,
             input_file=input_file,
-            completion_callback=completion_callback,
+            completion_callback=completion_callback,  # type: ignore
             completion_trigger=completion_trigger,
             env=env,
             **cmd_kwargs,
@@ -1368,14 +1379,14 @@ def set_status(
         if self._mode == "disabled":
             return True
 
-        if not self._active:
+        if not self._active or not self._name:
             self._error("Run is not active")
             return False
 
         data: dict[str, str] = {"name": self._name, "status": status}
         self._status = status
 
-        if self._simvue.update(data):
+        if self._simvue and self._simvue.update(data):
             return True
 
         return False

simvue/types.py

@@ -5,6 +5,7 @@
 except ImportError:
     from typing_extensions import TypeAlias
 
+
 if typing.TYPE_CHECKING:
     from numpy import ndarray
     from pandas import DataFrame

tests/refactor/test_run_class.py

@@ -1,10 +1,13 @@
 import pytest
+import pytest_mock
 import time
 import typing
 import contextlib
 import inspect
 import tempfile
+import threading
 import uuid
+import psutil
 import pathlib
 import concurrent.futures
 import random
@@ -461,22 +464,37 @@ def test_save_object(
 
 
 @pytest.mark.run
-def test_abort_on_alert_process(create_plain_run: typing.Tuple[sv_run.Run, dict]) -> None:
+def test_abort_on_alert_process(create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture) -> None:
+    def testing_exit(status: int) -> None:
+        raise SystemExit(status)
+    mocker.patch("os._exit", testing_exit)
+    N_PROCESSES: int = 3
     run, _ = create_plain_run
     run.config(resources_metrics_interval=1)
     run._heartbeat_interval = 1
-    run.add_process(identifier="forever_long", executable="bash", c="sleep 10000")
+    run._testing = True
+    run.add_process(identifier="forever_long", executable="bash", c="&".join(["sleep 10"] * N_PROCESSES))
+    process_id = list(run._executor._processes.values())[0].pid
+    process = psutil.Process(process_id)
+    assert len(child_processes := process.children(recursive=True)) == 3
     time.sleep(2)
     client = sv_cl.Client()
     client.abort_run(run._id, reason="testing abort")
     time.sleep(4)
+    for child in child_processes:
+        assert not child.is_running()
     if not run._status == "terminated":
         run.kill_all_processes()
         raise AssertionError("Run was not terminated")
 
 
 @pytest.mark.run
-def test_abort_on_alert_python(create_plain_run: typing.Tuple[sv_run.Run, dict]) -> None:
+def test_abort_on_alert_python(create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture) -> None:
+    abort_set = threading.Event()
+    def testing_exit(status: int) -> None:
+        abort_set.set()
+        raise SystemExit(status)
+    mocker.patch("os._exit", testing_exit)
     run, _ = create_plain_run
     run.config(resources_metrics_interval=1)
     run._heartbeat_interval = 1
@@ -488,10 +506,10 @@ def test_abort_on_alert_python(create_plain_run: typing.Tuple[sv_run.Run, dict])
         if i == 4:
             client.abort_run(run._id, reason="testing abort")
         i += 1
-        if i == 10:
+        if abort_set.is_set() or i > 9:
             break
 
-    assert i == 4
+    assert i < 7
     assert run._status == "terminated"