Skip to content

Extend tests for execute_in_subprocess() #744

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jul 17, 2025
Merged

Extend tests for execute_in_subprocess() #744

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions tests/test_cache_fileexecutor_serial.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import shutil
import unittest
from threading import Thread
from time import sleep

try:
from executorlib.task_scheduler.file.subprocess_spawner import (
Expand Down Expand Up @@ -214,6 +215,14 @@ def test_executor_function_dependence_args(self):
q.put({"shutdown": True, "wait": True})
process.join()

def test_execute_in_subprocess(self):
process = execute_in_subprocess(
command=["sleep", "5"],
file_name="test.h5",
data_dict={"fn": sleep, "args": (5,)},
)
self.assertIsNone(terminate_subprocess(task=process))
Comment on lines +218 to +224
Copy link
Contributor

@coderabbitai coderabbitai bot Jul 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Enhance test robustness and clarity.

The test has several areas for improvement:

  1. Redundant/confusing parameters: The test passes both a shell command ["sleep", "5"] and a data_dict with {"fn": sleep, "args": (5,)}, which seems redundant and potentially confusing.

  2. No verification of subprocess startup: The test immediately terminates without verifying the subprocess actually started.

  3. Potential race condition: Immediate termination might not test the intended behavior if the subprocess hasn't fully started.

Consider this improved version:

 def test_execute_in_subprocess(self):
+    import time
     process = execute_in_subprocess(
-        command=["sleep", "5"],
+        command=["python", "-c", "import time; time.sleep(5)"],
         file_name="test.h5",
-        data_dict={"fn": sleep, "args": (5,)},
+        data_dict={"fn": sleep, "args": (5,)},
     )
+    # Give the subprocess a moment to start
+    time.sleep(0.1)
+    # Verify process is running before termination
+    self.assertIsNotNone(process)
     self.assertIsNone(terminate_subprocess(task=process))
+    # Clean up any created files
+    import os
+    test_file = "test.h5"
+    if os.path.exists(test_file):
+        os.remove(test_file)

Alternatively, if you want to test the data_dict functionality specifically, consider removing the command parameter and letting the function execute the sleep function directly.

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def test_execute_in_subprocess(self):
process = execute_in_subprocess(
command=["sleep", "5"],
file_name="test.h5",
data_dict={"fn": sleep, "args": (5,)},
)
self.assertIsNone(terminate_subprocess(task=process))
def test_execute_in_subprocess(self):
import time
process = execute_in_subprocess(
command=["python", "-c", "import time; time.sleep(5)"],
file_name="test.h5",
data_dict={"fn": sleep, "args": (5,)},
)
# Give the subprocess a moment to start
time.sleep(0.1)
# Verify process is running before termination
self.assertIsNotNone(process)
self.assertIsNone(terminate_subprocess(task=process))
# Clean up any created files
import os
test_file = "test.h5"
if os.path.exists(test_file):
os.remove(test_file)
🤖 Prompt for AI Agents 
In tests/test_cache_fileexecutor_serial.py around lines 218 to 224, the
test_execute_in_subprocess function redundantly passes both a shell command and
a data_dict, which is confusing. To fix this, remove the command parameter if
you want to test the data_dict execution path specifically. Also, add
verification that the subprocess has started before terminating it to avoid race
conditions; for example, wait briefly or check process status before calling
terminate_subprocess. This will make the test more robust and clear.


def test_execute_in_subprocess_errors(self):
file_name = os.path.abspath(os.path.join(__file__, "..", "executorlib_cache", "test.h5"))
os.makedirs(os.path.dirname(file_name))
Expand Down
Loading