11749909901

Committed 08 Nov 2024 09:35PM UTC coverage: 95.011% (-0.2%) from 95.203%

Build # 11749909901

Build Type

Pull #481

github

Committed by

web-flow

Commit Message

Merge 9afcb5665 into 213202831

Pull Request Pull Request #481: Fix working directory

Run Details

2 of 7 new or added lines in 2 files covered. (28.57%)

895 of 942 relevant lines covered (95.01%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.47

/executorlib/standalone/cache/spawner.py

import subprocess
import time
from typing import Optional


def execute_in_subprocess(
    command: list,
    task_dependent_lst: list = [],
    resource_dict: Optional[dict] = None,
    config_directory: Optional[str] = None,
    backend: Optional[str] = None,
    cache_directory: Optional[str] = None,
) -> subprocess.Popen:
    """
    Execute a command in a subprocess.

    Args:
        command (list): The command to be executed.
        task_dependent_lst (list): A list of subprocesses that the current subprocess depends on. Defaults to [].
        resource_dict (dict): resource dictionary, which defines the resources used for the execution of the function.
                              Example resource dictionary: {
                                  cwd: None,
                              }
        config_directory (str, optional): path to the config directory.
        backend (str, optional): name of the backend used to spawn tasks.
        cache_directory (str): The directory to store the HDF5 files.

    Returns:
        subprocess.Popen: The subprocess object.

    """
    while len(task_dependent_lst) > 0:
        task_dependent_lst = [
            task for task in task_dependent_lst if task.poll() is None
        ]
    if config_directory is not None:
        raise ValueError(
            "config_directory parameter is not supported for subprocess spawner."
        )
    if backend is not None:
        raise ValueError("backend parameter is not supported for subprocess spawner.")
    if resource_dict is None:
        resource_dict = {"cwd": cache_directory}
    elif len(resource_dict) == 0:
        resource_dict = {"cwd": cache_directory}
    return subprocess.Popen(command, universal_newlines=True, cwd=resource_dict["cwd"])


def terminate_subprocess(task):
    """
    Terminate a subprocess and wait for it to complete.

    Args:
        task (subprocess.Popen): The subprocess.Popen instance to terminate
    """
    task.terminate()
    while task.poll() is None:
        time.sleep(0.1)

1	import subprocess	1✔
2	import time	1✔
3	from typing import Optional	1✔
4
5
6	def execute_in_subprocess(	1✔
7	command: list,
8	task_dependent_lst: list = [],
9	resource_dict: Optional[dict] = None,
10	config_directory: Optional[str] = None,
11	backend: Optional[str] = None,
12	cache_directory: Optional[str] = None,
13	) -> subprocess.Popen:
14	"""
15	Execute a command in a subprocess.
16
17	Args:
18	command (list): The command to be executed.
19	task_dependent_lst (list): A list of subprocesses that the current subprocess depends on. Defaults to [].
20	resource_dict (dict): resource dictionary, which defines the resources used for the execution of the function.
21	Example resource dictionary: {
22	cwd: None,
23	}
24	config_directory (str, optional): path to the config directory.
25	backend (str, optional): name of the backend used to spawn tasks.
26	cache_directory (str): The directory to store the HDF5 files.
27
28	Returns:
29	subprocess.Popen: The subprocess object.
30
31	"""
32	while len(task_dependent_lst) > 0:	1✔
33	task_dependent_lst = [	1✔
34	task for task in task_dependent_lst if task.poll() is None
35	]
36	if config_directory is not None:	1✔
37	raise ValueError(	1✔
38	"config_directory parameter is not supported for subprocess spawner."
39	)
40	if backend is not None:	1✔
41	raise ValueError("backend parameter is not supported for subprocess spawner.")	1✔
42	if resource_dict is None:	1✔
NEW 43	resource_dict = {"cwd": cache_directory}	×
44	elif len(resource_dict) == 0:	1✔
NEW 45	resource_dict = {"cwd": cache_directory}	×
46	return subprocess.Popen(command, universal_newlines=True, cwd=resource_dict["cwd"])	1✔
47
48
49	def terminate_subprocess(task):	1✔
50	"""
51	Terminate a subprocess and wait for it to complete.
52
53	Args:
54	task (subprocess.Popen): The subprocess.Popen instance to terminate
55	"""
56	task.terminate()	1✔
57	while task.poll() is None:	1✔
58	time.sleep(0.1)	1✔

pyiron / executorlib / 11749909901

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous