11543861772

Committed 27 Oct 2024 09:13PM UTC coverage: 94.084% (-0.3%) from 94.335%

Build # 11543861772

Build Type

push

github

Committed by

web-flow

Commit Message

Cache: Add queue test (#452)

* Cache: Add queue test

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix mpich bug

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Run Details

11 of 14 new or added lines in 2 files covered. (78.57%)

827 of 879 relevant lines covered (94.08%)

0.94 hits per line

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

78.95

/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,
) -> 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.

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

pyiron / executorlib / 11543861772

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