Merge branch '782-the-executor-doesnt-work-when-reconnect-is-used' of github.com:simvue-io/python-api into 782-the-executor-doesnt-work-when-reconnect-is-used

Author: kzscisoft

simvue/executor.py

@@ -16,6 +16,7 @@
 import multiprocessing
 import threading
 import os
+import shutil
 import psutil
 import subprocess
 import contextlib
@@ -205,6 +206,19 @@ def callback_function(status_code: int, std_out: str, std_err: str) -> None:
                 "due to function pickling restrictions"
             )
 
+        # To check the executable provided by the user exists combine with environment
+        # PATH variable if exists, if not provided use the current environment
+        _session_path: str | None = (os.environ.copy() | (env or {})).get("PATH", None)
+
+        if (
+            executable
+            and not pathlib.Path(executable).exists()
+            and not shutil.which(executable, path=_session_path)
+        ):
+            raise FileNotFoundError(
+                f"Executable '{executable}' does not exist, please check the path/environment."
+            )
+
         if script:
             self._runner.save_file(file_path=script, category="code")
 
@@ -377,7 +391,7 @@ def _update_alerts(self) -> None:
             # This is so that if a process incorrectly reports its return code,
             # the user can manually set the correct status depending on logs etc.
             _alert = UserAlert(identifier=self._alert_ids[proc_id])
-            _is_set = _alert.get_status(run_id=self._runner._id)
+            _is_set = _alert.get_status(run_id=self._runner.id)
 
             if process.returncode != 0:
                 # If the process fails then purge the dispatcher event queue
@@ -404,6 +418,10 @@ def _update_alerts(self) -> None:
 
     def _save_output(self) -> None:
         """Save the output to Simvue"""
+        if self._runner.status != "running":
+            logger.debug("Run is not active, skipping output save.")
+            return
+
         for proc_id in self._processes.keys():
             # Only save the file if the contents are not empty
             if self.std_err(proc_id):

simvue/run.py

@@ -146,7 +146,6 @@ def __init__(
         ```
         """
         self._uuid: str = f"{uuid.uuid4()}"
-        self._name: str | None = None
 
         # monitor duration with respect to retention period
         self._timer: float = 0
@@ -160,7 +159,6 @@ def __init__(
         self._executor = Executor(self)
         self._dispatcher: DispatcherBaseClass | None = None
 
-        self._id: str | None = None
         self._folder: Folder | None = None
         self._term_color: bool = True
         self._suppress_errors: bool = False
@@ -259,7 +257,7 @@ def __exit__(
     ) -> None:
         logger.debug(
             "Automatically closing run '%s' in status %s",
-            self._id if self._user_config.run.mode == "online" else "unregistered",
+            self.id if self._user_config.run.mode == "online" else "unregistered",
             self._status,
         )
 
@@ -364,24 +362,25 @@ def _get_internal_metrics(
         # Set join on fail to false as if an error is thrown
         # join would be called on this thread and a thread cannot
         # join itself!
-        self._add_metrics_to_dispatch(
-            _current_system_measure.to_dict(),
-            join_on_fail=False,
-            step=system_metrics_step,
-        )
+        if self.status == "running":
+            self._add_metrics_to_dispatch(
+                _current_system_measure.to_dict(),
+                join_on_fail=False,
+                step=system_metrics_step,
+            )
 
         # For the first emissions metrics reading, the time interval to use
         # Is the time since the run started, otherwise just use the time between readings
         if self._emissions_monitor:
             _estimated = self._emissions_monitor.estimate_co2_emissions(
-                process_id=f"{self._name}",
+                process_id=f"{self._sv_obj.name}",
                 cpu_percent=_current_system_measure.cpu_percent,
                 measure_interval=(time.time() - self._start_time)
                 if system_metrics_step == 0
                 else self._system_metrics_interval,
                 gpu_percent=_current_system_measure.gpu_percent,
             )
-            if _estimated:
+            if _estimated and self.status == "running":
                 self._add_metrics_to_dispatch(
                     self._emissions_monitor.simvue_metrics(),
                     join_on_fail=False,
@@ -394,7 +393,7 @@ def _create_heartbeat_callback(
         """Defines the callback executed at the heartbeat interval for the Run."""
         if (
             self._user_config.run.mode == "online"
-            and (not self._user_config.server.url or not self._id)
+            and (not self._user_config.server.url or not self.id)
         ) or not self._heartbeat_termination_trigger:
             raise RuntimeError("Could not commence heartbeat, run not initialised")
 
@@ -459,7 +458,7 @@ def _create_dispatch_callback(
         executed on metrics and events objects held in a buffer.
         """
 
-        if self._user_config.run.mode == "online" and not self._id:
+        if self._user_config.run.mode == "online" and not self.id:
             raise RuntimeError("Expected identifier for run")
 
         if (
@@ -590,7 +589,6 @@ def _error(self, message: str, join_threads: bool = True) -> None:
         # Simvue support now terminated as the instance of Run has entered
         # the dormant state due to exception throw so set listing to be 'lost'
         if self._status == "running" and self._sv_obj:
-            self._sv_obj.name = self._name
             self._sv_obj.status = "lost"
             self._sv_obj.commit()
 
@@ -701,8 +699,6 @@ def init(
         elif not name and self._user_config.run.mode == "offline":
             name = randomname.get_name()
 
-        self._name = name
-
         self._status = "running" if running else "created"
 
         # Parse the time to live/retention time if specified
@@ -750,28 +746,20 @@ def init(
         self._data = self._sv_obj._staging
         self._sv_obj.commit()
 
-        if self._user_config.run.mode == "online":
-            name = self._sv_obj.name
-
-        self._id = self._sv_obj.id
-
-        if not name:
+        if not self.name:
             return False
 
-        elif name is not True:
-            self._name = name
-
         if self._status == "running":
             self._start()
 
         if self._user_config.run.mode == "online":
             click.secho(
-                f"[simvue] Run {self._name} created",
+                f"[simvue] Run {self._sv_obj.name} created",
                 bold=self._term_color,
                 fg="green" if self._term_color else None,
             )
             click.secho(
-                f"[simvue] Monitor in the UI at {self._user_config.server.url.rsplit('/api', 1)[0]}/dashboard/runs/run/{self._id}",
+                f"[simvue] Monitor in the UI at {self._user_config.server.url.rsplit('/api', 1)[0]}/dashboard/runs/run/{self.id}",
                 bold=self._term_color,
                 fg="green" if self._term_color else None,
             )
@@ -951,7 +939,23 @@ def executor(self) -> Executor:
     @property
     def name(self) -> str | None:
         """Return the name of the run"""
-        return self._name
+        if not self._sv_obj:
+            raise RuntimeError("Run has not been initialised")
+        return self._sv_obj.name
+
+    @property
+    def status(
+        self,
+    ) -> (
+        typing.Literal[
+            "created", "running", "completed", "failed", "terminated", "lost"
+        ]
+        | None
+    ):
+        """Return the status of the run"""
+        if not self._sv_obj:
+            raise RuntimeError("Run has not been initialised")
+        return self._sv_obj.status
 
     @property
     def uid(self) -> str:
@@ -961,7 +965,9 @@ def uid(self) -> str:
     @property
     def id(self) -> str | None:
         """Return the unique id of the run"""
-        return self._id
+        if not self._sv_obj:
+            raise RuntimeError("Run has not been initialised")
+        return self._sv_obj.id
 
     @skip_if_failed("_aborted", "_suppress_errors", False)
     @pydantic.validate_call
@@ -981,7 +987,7 @@ def reconnect(self, run_id: str) -> bool:
         self._status = "running"
 
         self._id = run_id
-        self._sv_obj = RunObject(identifier=self._id, _read_only=False)
+        self._sv_obj = RunObject(identifier=run_id, _read_only=False)
         self._name = self._sv_obj.name
         self._sv_obj.status = self._status
         self._sv_obj.system = get_system()
@@ -1614,7 +1620,7 @@ def _tidy_run(self) -> None:
             and self._status != "created"
         ):
             self._user_config.offline.cache.joinpath(
-                "runs", f"{self._id}.closed"
+                "runs", f"{self.id}.closed"
             ).touch()
 
         if _non_zero := self.executor.exit_status:
@@ -2088,7 +2094,7 @@ def log_alert(
             )
             return False
         _alert.read_only(False)
-        _alert.set_status(run_id=self._id, status=state)
+        _alert.set_status(run_id=self.id, status=state)
         _alert.commit()
 
         return True

tests/conftest.py

@@ -279,8 +279,8 @@ def setup_test_run(run: sv_run.Run, create_objects: bool, request: pytest.Fixtur
     if create_objects:
         TEST_DATA["metrics"] = ("metric_counter", "metric_val")
 
-    TEST_DATA["run_id"] = run._id
-    TEST_DATA["run_name"] = run._name
+    TEST_DATA["run_id"] = run.id
+    TEST_DATA["run_name"] = run.name
     TEST_DATA["url"] = run._user_config.server.url
     TEST_DATA["headers"] = run._headers
     TEST_DATA["pid"] = run._pid

tests/functional/test_run_class.py

@@ -674,7 +674,7 @@ def test_set_folder_details(request: pytest.FixtureRequest) -> None:
     ids=[f"scenario_{i}" for i in range(1, 6)],
 )
 def test_save_file_online(
-    create_plain_run: typing.Tuple[sv_run.Run, dict],
+    create_plain_run: tuple[sv_run.Run, dict],
     valid_mimetype: bool,
     preserve_path: bool,
     name: str | None,
@@ -743,7 +743,7 @@ def test_save_file_online(
     ids=[f"scenario_{i}" for i in range(1, 6)],
 )
 def test_save_file_offline(
-    create_plain_run_offline: typing.Tuple[sv_run.Run, dict],
+    create_plain_run_offline: tuple[sv_run.Run, dict],
     preserve_path: bool,
     name: str | None,
     allow_pickle: bool,
@@ -798,7 +798,7 @@ def test_save_file_offline(
 
 @pytest.mark.run
 def test_update_tags_running(
-    create_plain_run: typing.Tuple[sv_run.Run, dict],
+    create_plain_run: tuple[sv_run.Run, dict],
     request: pytest.FixtureRequest,
 ) -> None:
     simvue_run, _ = create_plain_run
@@ -824,7 +824,7 @@ def test_update_tags_running(
 
 @pytest.mark.run
 def test_update_tags_created(
-    create_pending_run: typing.Tuple[sv_run.Run, dict],
+    create_pending_run: tuple[sv_run.Run, dict],
     request: pytest.FixtureRequest,
 ) -> None:
     simvue_run, _ = create_pending_run
@@ -851,7 +851,7 @@ def test_update_tags_created(
 @pytest.mark.offline
 @pytest.mark.run
 def test_update_tags_offline(
-    create_plain_run_offline: typing.Tuple[sv_run.Run, dict],
+    create_plain_run_offline: tuple[sv_run.Run, dict],
 ) -> None:
     simvue_run, _ = create_plain_run_offline
     run_name = simvue_run._name
@@ -879,7 +879,7 @@ def test_update_tags_offline(
 @pytest.mark.run
 @pytest.mark.parametrize("object_type", ("DataFrame", "ndarray"))
 def test_save_object(
-    create_plain_run: typing.Tuple[sv_run.Run, dict], object_type: str
+    create_plain_run: tuple[sv_run.Run, dict], object_type: str
 ) -> None:
     simvue_run, _ = create_plain_run
 
@@ -1081,7 +1081,7 @@ def abort_callback(abort_run=trigger) -> None:
 
 @pytest.mark.run
 def test_abort_on_alert_python(
-    speedy_heartbeat, create_plain_run: typing.Tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture
+    speedy_heartbeat, create_plain_run: tuple[sv_run.Run, dict], mocker: pytest_mock.MockerFixture
 ) -> None:
     timeout: int = 20
     interval: int = 0
@@ -1094,7 +1094,7 @@ def test_abort_on_alert_python(
 
 @pytest.mark.run
 def test_abort_on_alert_raise(
-    create_plain_run: typing.Tuple[sv_run.Run, dict]
+    create_plain_run: tuple[sv_run.Run, dict]
 ) -> None:
 
     run, _ = create_plain_run
@@ -1119,7 +1119,7 @@ def test_abort_on_alert_raise(
 
 
 @pytest.mark.run
-def test_kill_all_processes(create_plain_run: typing.Tuple[sv_run.Run, dict]) -> None:
+def test_kill_all_processes(create_plain_run: tuple[sv_run.Run, dict]) -> None:
     run, _ = create_plain_run
     run.config(system_metrics_interval=1)
     run.add_process(identifier="forever_long_1", executable="bash", c="sleep 10000")
@@ -1150,7 +1150,7 @@ def test_run_created_with_no_timeout() -> None:
 
 @pytest.mark.parametrize("mode", ("online", "offline"), ids=("online", "offline"))
 @pytest.mark.run
-def test_reconnect(mode, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_reconnect_functionality(mode, monkeypatch: pytest.MonkeyPatch) -> None:
     temp_d: tempfile.TemporaryDirectory | None = None
 
     if mode == "offline":
@@ -1191,3 +1191,16 @@ def test_reconnect(mode, monkeypatch: pytest.MonkeyPatch) -> None:
     if temp_d:
         temp_d.cleanup()
 
+
+def test_reconnect_with_process(create_plain_run: tuple[sv_run.Run, dict]) -> None:
+    run, _ = create_plain_run
+    run.init(name="test_reconnect_with_process", folder="/simvue_unit_testing", retention_period="2 minutes", running=False)
+    run.close()
+
+    with sv_run.Run() as new_run:
+        new_run.reconnect(run.id)
+        run.add_process(
+            identifier="test_process",
+            executable="bash",
+            c="echo 'Hello World!'",
+        )