11766741741

Committed 10 Nov 2024 04:26PM CUT coverage: 94.813% (-0.3%) from 95.089%

Build # 11766741741

Build Type

Pull #491

github

Committed by

web-flow

Commit Message

Merge e34b5de8a into 78062fd9b

Pull Request Pull Request #491: Empty cache

Run Details

9 of 12 new or added lines in 2 files covered. (75.0%)

914 of 964 relevant lines covered (94.81%)

0.95 hits per line

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

90.0

/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 = {}
    if "cwd" in resource_dict:
        cwd = resource_dict["cwd"]
    else:
        cwd = cache_directory
    return subprocess.Popen(command, universal_newlines=True, cwd=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✔
43	resource_dict = {}	×
44	if "cwd" in resource_dict:	1✔
45	cwd = resource_dict["cwd"]	1✔
46	else:
47	cwd = cache_directory	×
48	return subprocess.Popen(command, universal_newlines=True, cwd=cwd)	1✔
49
50
51	def terminate_subprocess(task):	1✔
52	"""
53	Terminate a subprocess and wait for it to complete.
54
55	Args:
56	task (subprocess.Popen): The subprocess.Popen instance to terminate
57	"""
58	task.terminate()	1✔
59	while task.poll() is None:	1✔
60	time.sleep(0.1)	1✔

pyiron / executorlib / 11766741741

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