🧪 Fix wrong string formatting for additional python version in tests

Author: kzscisoft

tests/conftest.py

@@ -44,8 +44,8 @@ def emit(self, record):
 
 
 def clear_out_files() -> None:
-    out_files = list(pathlib.Path.cwd().glob(f"test_*_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.out"))
-    out_files += list(pathlib.Path.cwd().glob(f"test_*_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.err"))
+    out_files = list(pathlib.Path.cwd().glob(f"test_*_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.out"))
+    out_files += list(pathlib.Path.cwd().glob(f"test_*_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.err"))
 
     for file_obj in out_files:
         file_obj.unlink()

tests/functional/test_executor.py

@@ -39,7 +39,7 @@ def completion_callback(*_, trigger=trigger, **__):
         retention_period="2 mins"
     )
     run.add_process(
-        identifier=f"test_add_process_{'success' if successful else 'fail'}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+        identifier=f"test_add_process_{'success' if successful else 'fail'}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
         c=f"exit {0 if successful else 1}",
         executable="bash" if sys.platform != "win32" else "powershell",
         completion_callback=completion_callback
@@ -83,7 +83,7 @@ def callback(*_, evts=events, ident=i, **__):
                 callbacks[i] = callback
                 out_file = pathlib.Path(tempd).joinpath(f"out_file_{i}.dat")
                 run.add_process(
-                    f"cmd_{i}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+                    f"cmd_{i}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
                     executable="bash",
                     c="for i in {0..10}; do sleep 0.5; echo $i >> "+ f"{out_file}; done",
                     completion_trigger=triggers[i],
@@ -96,8 +96,8 @@ def callback(*_, evts=events, ident=i, **__):
             assert out_file.exists()
             assert triggers[i].is_set()
     for i in range(10):
-        os.remove(f"test_executor_multiprocess_cmd_{i}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.err")
-        os.remove(f"test_executor_multiprocess_cmd_{i}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.out")
+        os.remove(f"test_executor_multiprocess_cmd_{i}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.err")
+        os.remove(f"test_executor_multiprocess_cmd_{i}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.out")
     with contextlib.suppress(ObjectNotFoundError):
         folder.delete(recursive=True, delete_runs=True)
 
@@ -140,7 +140,7 @@ def test_add_process_command_assembly(request: pytest.FixtureRequest) -> None:
                 tags=[platform.system(), "simvue_client_tests", request.node.name]
             )
             run.add_process(
-                identifier=(exe_id := f"advanced_run_{os.environ.get("PYTEST_XDIST_WORKER", 0)}"),
+                identifier=(exe_id := f"advanced_run_{os.environ.get('PYTEST_XDIST_WORKER', 0)}"),
                 executable="python",
                 script=f"{code_file}",
                 input_file=f"{in_file}",
@@ -166,7 +166,7 @@ def completion_callback(*_, success: dict[str, bool]=success, **__):
             tags=[platform.system(), "simvue_client_tests", request.node.name]
         )
         run.add_process(
-            identifier=f"test_completion_callbacks_var_change_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+            identifier=f"test_completion_callbacks_var_change_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
             executable="bash",
             c="exit 0",
             completion_callback=completion_callback
@@ -194,7 +194,7 @@ def test_completion_trigger_set(request: pytest.FixtureRequest) -> None:
             tags=[platform.system(), "simvue_client_tests", request.node.name]
         )
         run.add_process(
-            identifier=f"test_completion_trigger_set_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+            identifier=f"test_completion_trigger_set_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
             executable="bash",
             c="exit 0",
             completion_trigger=trigger
@@ -225,7 +225,7 @@ def completion_callback(*_, trigger=trigger, **__):
             tags=[platform.system(), "simvue_client_tests", request.node.name]
         )
         run.add_process(
-            identifier=f"test_completion_callbacks_trigger_set_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+            identifier=f"test_completion_callbacks_trigger_set_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
             executable="bash",
             c="exit 0",
             completion_callback=completion_callback

tests/functional/test_run_class.py

@@ -1138,7 +1138,7 @@ def abort_callback(abort_run=trigger) -> None:
     run._heartbeat_interval = 1
     run._testing = True
     run.add_process(
-        identifier=f"forever_long_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+        identifier=f"forever_long_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
         executable="bash",
         c="&".join(["sleep 10"] * N_PROCESSES),
     )
@@ -1195,7 +1195,7 @@ def test_abort_on_alert_raise(
     run._heartbeat_interval = 1
     run._testing = True
     alert_id = run.create_user_alert("abort_test", trigger_abort=True)
-    run.add_process(identifier=f"forever_long_other_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", executable="bash", c="sleep 10")
+    run.add_process(identifier=f"forever_long_other_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", executable="bash", c="sleep 10")
     run.log_alert(identifier=alert_id, state="critical")
     _alert = Alert(identifier=alert_id)
     assert _alert.get_status(run.id) == "critical"
@@ -1213,8 +1213,8 @@ def test_abort_on_alert_raise(
 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=f"forever_long_a_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", executable="bash", c="sleep 10000")
-    run.add_process(identifier=f"forever_long_b_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", executable="bash", c="sleep 10000")
+    run.add_process(identifier=f"forever_long_a_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", executable="bash", c="sleep 10000")
+    run.add_process(identifier=f"forever_long_b_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", executable="bash", c="sleep 10000")
     processes = [
         psutil.Process(process.pid) for process in run._executor._processes.values()
     ]
@@ -1321,7 +1321,7 @@ def test_reconnect_with_process() -> None:
     with sv_run.Run() as new_run:
         new_run.reconnect(run.id)
         run.add_process(
-            identifier=f"test_process_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+            identifier=f"test_process_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
             executable="bash",
             c="echo 'Hello World!'",
         )

tests/functional/test_run_execute_process.py

@@ -15,9 +15,9 @@
 def test_monitor_processes(create_plain_run_offline: tuple[Run, dict]):
     _run: Run
     _run, _ = create_plain_run_offline
-    _run.add_process(f"process_1_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", "Hello world!", executable="echo", n=True)
-    _run.add_process(f"process_2_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", "bash" if sys.platform != "win32" else "powershell", debug=True, c="exit 0")
-    _run.add_process(f"process_3_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", "ls", "-ltr")
+    _run.add_process(f"process_1_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", "Hello world!", executable="echo", n=True)
+    _run.add_process(f"process_2_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", "bash" if sys.platform != "win32" else "powershell", debug=True, c="exit 0")
+    _run.add_process(f"process_3_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", "ls", "-ltr")
     sender(_run._sv_obj._local_staging_file.parents[1], 1, 10, ["folders", "runs", "alerts"])
 
 
@@ -34,8 +34,8 @@ def test_abort_all_processes(create_plain_run: tuple[Run, dict]) -> None:
             ])
 
         for i in range(1, 3):
-            _run.add_process(f"process_{i}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}", executable="bash", script=temp_f.name)
-            assert _run.executor.get_command(f"process_{i}_{os.environ.get("PYTEST_XDIST_WORKER", 0)}") == f"bash {temp_f.name}"
+            _run.add_process(f"process_{i}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}", executable="bash", script=temp_f.name)
+            assert _run.executor.get_command(f"process_{i}_{os.environ.get('PYTEST_XDIST_WORKER', 0)}") == f"bash {temp_f.name}"
 
 
         time.sleep(3)
@@ -44,7 +44,7 @@ def test_abort_all_processes(create_plain_run: tuple[Run, dict]) -> None:
 
         # Check that for when one of the processes has stopped
         _attempts: int = 0
-        _first_out = next(pathlib.Path.cwd().glob(f"*process_*_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.out"))
+        _first_out = next(pathlib.Path.cwd().glob(f"*process_*_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.out"))
 
         while _first_out.stat().st_size == 0 and _attempts < 10:
             time.sleep(1)
@@ -54,14 +54,14 @@ def test_abort_all_processes(create_plain_run: tuple[Run, dict]) -> None:
             raise AssertionError("Failed to terminate processes")
 
         # Check the Python process did not error
-        _out_err = pathlib.Path.cwd().glob(f"*process_*_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.err")
+        _out_err = pathlib.Path.cwd().glob(f"*process_*_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.err")
         for file in _out_err:
             with file.open() as in_f:
                 assert not in_f.readlines()
 
         # Now check the counter in the process was terminated
         # just beyond the sleep time
-        _out_files = pathlib.Path.cwd().glob(f"*process_*_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.out")
+        _out_files = pathlib.Path.cwd().glob(f"*process_*_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.out")
         for file in _out_files:
             with file.open() as in_f:
                 assert (lines := in_f.readlines())
@@ -88,20 +88,20 @@ def test_processes_cwd(create_plain_run: dict[Run, dict]) -> None:
             with open(temp_file.name, "w") as out_f:
                 out_f.writelines([
                     "import os\n",
-                    f"f = open('new_file_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.txt', 'w')\n",
+                    f"f = open('new_file_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.txt', 'w')\n",
                     "f.write('Test Line')\n",
                     "f.close()"
                 ])
 
             run_id = run.id
             run.add_process(
-                identifier=f"sleep_10_process_{os.environ.get("PYTEST_XDIST_WORKER", 0)}",
+                identifier=f"sleep_10_process_{os.environ.get('PYTEST_XDIST_WORKER', 0)}",
                 executable="python",
                 script=temp_file.name,
                 cwd=temp_dir
             )
             time.sleep(1)
-            run.save_file(os.path.join(temp_dir, f"new_file_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.txt"), 'output')
+            run.save_file(os.path.join(temp_dir, f"new_file_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.txt"), 'output')
 
             client = Client()
 
@@ -110,8 +110,8 @@ def test_processes_cwd(create_plain_run: dict[Run, dict]) -> None:
             client.get_artifact_as_file(run_id, os.path.basename(temp_file.name), output_dir=os.path.join(temp_dir, "downloaded"))
             assert filecmp.cmp(os.path.join(temp_dir, "downloaded", os.path.basename(temp_file.name)), temp_file.name)
 
-            client.get_artifact_as_file(run_id, f"new_file_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.txt", output_dir=os.path.join(temp_dir, "downloaded"))
-            with open(os.path.join(temp_dir, "downloaded", f"new_file_{os.environ.get("PYTEST_XDIST_WORKER", 0)}.txt"), "r") as new_file:
+            client.get_artifact_as_file(run_id, f"new_file_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.txt", output_dir=os.path.join(temp_dir, "downloaded"))
+            with open(os.path.join(temp_dir, "downloaded", f"new_file_{os.environ.get('PYTEST_XDIST_WORKER', 0)}.txt"), "r") as new_file:
                 assert new_file.read() == "Test Line"
     with contextlib.suppress(FileNotFoundError):
         os.unlink(temp_file.name)