8662991737

Committed 12 Apr 2024 01:38PM UTC coverage: 91.435% (-0.03%) from 91.467%

Build # 8662991737

Build Type

push

github

Committed by

uwefladrich

Commit Message

Update $PWD in environment for base.command processes

Run Details

3 of 4 new or added lines in 1 file covered. (75.0%)

1911 of 2090 relevant lines covered (91.44%)

0.91 hits per line

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

91.36

/src/scriptengine/tasks/base/command.py

"""Command task for ScriptEngine

   Example:
      - command:
            name: ls
            args: [-l, -a]
"""

import os
import subprocess
import threading
from contextlib import contextmanager
from pathlib import Path

from scriptengine.context import Context
from scriptengine.exceptions import (
    ScriptEngineTaskArgumentInvalidError,
    ScriptEngineTaskRunError,
)
from scriptengine.tasks.core import Task, timed_runner


class LogPipe(threading.Thread):
    """Helper class that can be used in the subprocess.run() call as argument
    for stdout and stderr. It implements a pipe that sends whatever it receives
    to a log function (provided at initialisation). This can be used to send
    stdout/stderr of a subprocess to a logger.
    Follows https://codereview.stackexchange.com/questions/6567"""

    def __init__(self, log_function):
        super().__init__()
        self.log_function = log_function
        self.fd_read, self.fd_write = os.pipe()
        self.pipe_reader = os.fdopen(self.fd_read)
        self.start()

    def fileno(self):
        return self.fd_write

    def run(self):
        for line in iter(self.pipe_reader.readline, ""):
            self.log_function(line.strip("\n"))
        self.pipe_reader.close()

    def close(self):
        os.close(self.fd_write)


class Command(Task):
    """Command task, executes a command in a shell"""

    _required_arguments = ("name",)

    def __init__(self, arguments):
        Command.check_arguments(arguments)
        super().__init__(arguments)

    @timed_runner
    def run(self, context):
        self.log_info(
            f"{self.name} "
            f'args={getattr(self, "args", None)} '
            f'cwd={getattr(self, "cwd", None)}'
        )

        command = self.getarg("name", context)

        args = self.getarg("args", context, default=[])
        args = args if isinstance(args, list) else [args]

        cwd = self.getarg("cwd", context, default=None)
        ignore_error = self.getarg("ignore_error", context, default=False)

        self.log_debug(
            f"{command} "
            f'{" ".join(map(str, args))} '
            f"cwd={cwd} "
            f"ignore_error={ignore_error} "
        )

        @contextmanager
        def log_pipe(mode):
            """Returns something that can be used as stdout/stderr argument for
            subprocess.run(). If mode is True, a LogPipe is returned, which
            sends stdout/stderr through the info logger of the tasks. If mode is
            None/False, then None is returned and stdout/stderr are ignored. If
            mode is a string, stdout/stderr is captured by subprocess and later
            stored in the context."""
            if mode is True:
                pipe = LogPipe(self.log_info)
            elif not mode:
                pipe = None
            elif isinstance(mode, str):
                pipe = subprocess.PIPE
            else:
                self.log_error(f"Invalid task argument: {mode}")
                raise ScriptEngineTaskArgumentInvalidError
            try:
                yield pipe
            finally:
                if isinstance(pipe, LogPipe):
                    pipe.close()

        stdout_mode = self.getarg("stdout", context, default=True)
        self.log_debug(
            "stdout mode: "
            f'{stdout_mode if stdout_mode in (True, False) else "context"}'
        )
        stderr_mode = self.getarg("stderr", context, default=True)
        self.log_debug(
            "stderr mode: "
            f'{stderr_mode if stderr_mode in (True, False) else "context"}'
        )

        # Update $PWD in the environment of the command
        # Once support for Python<=3.8 is dropped, this can be done directly in the
        # call to subprocess.run() below:
        # subprocess.run(
        #   ...
        #   env=os.environ | {"PWD": Path(cwd).resolve()} if cwd else {},
        #   ...
        # )
        cmd_env = os.environ.copy()
        if cwd:
            cmd_env["PWD"] = Path(cwd).resolve()

        with log_pipe(stdout_mode) as stdout, log_pipe(stderr_mode) as stderr:
            context_update = Context()
            try:
                cmd_proc = subprocess.run(
                    map(str, (command, *args)),
                    stdout=stdout,
                    stderr=stderr,
                    cwd=cwd,
                    check=True,
                    env=cmd_env,
                    errors="replace",
                )
            except subprocess.CalledProcessError as e:
                if ignore_error:
                    self.log_warning(f"Command returned error code {e.returncode}")
                    stdout = e.stdout
                    stderr = e.stderr
                else:
                    self.log_error(f"Command returned error code {e.returncode}")
                    raise ScriptEngineTaskRunError
            else:
                stdout = cmd_proc.stdout
                stderr = cmd_proc.stderr

            if isinstance(stdout_mode, str):
                self.log_debug(f"Store stdout in context under {stdout_mode}")
                context_update[stdout_mode] = stdout.split("\n")
            if isinstance(stderr_mode, str):
                self.log_debug(f"Store stderr in context under {stderr_mode}")
                context_update[stderr_mode] = stderr.split("\n")

        return context_update or None

1	"""Command task for ScriptEngine	1✔
2
3	Example:
4	- command:
5	name: ls
6	args: [-l, -a]
7	"""
8
9	import os	1✔
10	import subprocess	1✔
11	import threading	1✔
12	from contextlib import contextmanager	1✔
13	from pathlib import Path	1✔
14
15	from scriptengine.context import Context	1✔
16	from scriptengine.exceptions import (	1✔
17	ScriptEngineTaskArgumentInvalidError,
18	ScriptEngineTaskRunError,
19	)
20	from scriptengine.tasks.core import Task, timed_runner	1✔
21
22
23	class LogPipe(threading.Thread):	1✔
24	"""Helper class that can be used in the subprocess.run() call as argument	1✔
25	for stdout and stderr. It implements a pipe that sends whatever it receives
26	to a log function (provided at initialisation). This can be used to send
27	stdout/stderr of a subprocess to a logger.
28	Follows https://codereview.stackexchange.com/questions/6567"""
29
30	def __init__(self, log_function):	1✔
31	super().__init__()	1✔
32	self.log_function = log_function	1✔
33	self.fd_read, self.fd_write = os.pipe()	1✔
34	self.pipe_reader = os.fdopen(self.fd_read)	1✔
35	self.start()	1✔
36
37	def fileno(self):	1✔
38	return self.fd_write	1✔
39
40	def run(self):	1✔
41	for line in iter(self.pipe_reader.readline, ""):	1✔
42	self.log_function(line.strip("\n"))	×
43	self.pipe_reader.close()	1✔
44
45	def close(self):	1✔
46	os.close(self.fd_write)	1✔
47
48
49	class Command(Task):	1✔
50	"""Command task, executes a command in a shell"""	1✔
51
52	_required_arguments = ("name",)	1✔
53
54	def __init__(self, arguments):	1✔
55	Command.check_arguments(arguments)	1✔
56	super().__init__(arguments)	1✔
57
58	@timed_runner	1✔
59	def run(self, context):	1✔
60	self.log_info(	1✔
61	f"{self.name} "
62	f'args={getattr(self, "args", None)} '
63	f'cwd={getattr(self, "cwd", None)}'
64	)
65
66	command = self.getarg("name", context)	1✔
67
68	args = self.getarg("args", context, default=[])	1✔
69	args = args if isinstance(args, list) else [args]	1✔
70
71	cwd = self.getarg("cwd", context, default=None)	1✔
72	ignore_error = self.getarg("ignore_error", context, default=False)	1✔
73
74	self.log_debug(	1✔
75	f"{command} "
76	f'{" ".join(map(str, args))} '
77	f"cwd={cwd} "
78	f"ignore_error={ignore_error} "
79	)
80
81	@contextmanager	1✔
82	def log_pipe(mode):	1✔
83	"""Returns something that can be used as stdout/stderr argument for
84	subprocess.run(). If mode is True, a LogPipe is returned, which
85	sends stdout/stderr through the info logger of the tasks. If mode is
86	None/False, then None is returned and stdout/stderr are ignored. If
87	mode is a string, stdout/stderr is captured by subprocess and later
88	stored in the context."""
89	if mode is True:	1✔
90	pipe = LogPipe(self.log_info)	1✔
91	elif not mode:	1✔
92	pipe = None	×
93	elif isinstance(mode, str):	1✔
94	pipe = subprocess.PIPE	1✔
95	else:
96	self.log_error(f"Invalid task argument: {mode}")	×
97	raise ScriptEngineTaskArgumentInvalidError	×
98	try:	1✔
99	yield pipe	1✔
100	finally:
101	if isinstance(pipe, LogPipe):	1✔
102	pipe.close()	1✔
103
104	stdout_mode = self.getarg("stdout", context, default=True)	1✔
105	self.log_debug(	1✔
106	"stdout mode: "
107	f'{stdout_mode if stdout_mode in (True, False) else "context"}'
108	)
109	stderr_mode = self.getarg("stderr", context, default=True)	1✔
110	self.log_debug(	1✔
111	"stderr mode: "
112	f'{stderr_mode if stderr_mode in (True, False) else "context"}'
113	)
114
115	# Update $PWD in the environment of the command
116	# Once support for Python<=3.8 is dropped, this can be done directly in the
117	# call to subprocess.run() below:
118	# subprocess.run(
119	# ...
120	# env=os.environ \| {"PWD": Path(cwd).resolve()} if cwd else {},
121	# ...
122	# )
123	cmd_env = os.environ.copy()	1✔
124	if cwd:	1✔
NEW 125	cmd_env["PWD"] = Path(cwd).resolve()	×
126
127	with log_pipe(stdout_mode) as stdout, log_pipe(stderr_mode) as stderr:	1✔
128	context_update = Context()	1✔
129	try:	1✔
130	cmd_proc = subprocess.run(	1✔
131	map(str, (command, *args)),
132	stdout=stdout,
133	stderr=stderr,
134	cwd=cwd,
135	check=True,
136	env=cmd_env,
137	errors="replace",
138	)
139	except subprocess.CalledProcessError as e:	1✔
140	if ignore_error:	1✔
141	self.log_warning(f"Command returned error code {e.returncode}")	1✔
142	stdout = e.stdout	1✔
143	stderr = e.stderr	1✔
144	else:
145	self.log_error(f"Command returned error code {e.returncode}")	×
146	raise ScriptEngineTaskRunError	×
147	else:
148	stdout = cmd_proc.stdout	1✔
149	stderr = cmd_proc.stderr	1✔
150
151	if isinstance(stdout_mode, str):	1✔
152	self.log_debug(f"Store stdout in context under {stdout_mode}")	1✔
153	context_update[stdout_mode] = stdout.split("\n")	1✔
154	if isinstance(stderr_mode, str):	1✔
155	self.log_debug(f"Store stderr in context under {stderr_mode}")	1✔
156	context_update[stderr_mode] = stderr.split("\n")	1✔
157
158	return context_update or None	1✔

uwefladrich / scriptengine / 8662991737

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