7018029928

Committed 28 Nov 2023 11:21AM UTC coverage: 91.699%. Remained the same

Build # 7018029928

Build Type

push

github

Committed by

uwefladrich

Commit Message

Fix handling of stderr in case command fails

Run Details

15 of 15 new or added lines in 2 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

1900 of 2072 relevant lines covered (91.7%)

0.92 hits per line

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

92.21

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

uwefladrich / scriptengine / 7018029928

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