[elbe-devel] [PATCH 3/5] shellhelper: Add doctests
Olivier Dion
dion at linutronix.de
Mon May 18 16:09:58 CEST 2020
Tests were made in order to execute all code paths and to keep stdout
clean. Thus, nothing is ever printed to the terminal.
The tests for do(), chroot() and get_command_out() require to
manipulate the loggers of the Python's logging module.
Fixed an eror in the second 'Popen' of get_command_out(). For some
reason, stdout was redirected to elbe's logging when stdin is not
None.
Signed-off-by: Olivier Dion <dion at linutronix.de>
---
elbepack/shellhelper.py | 182 +++++++++++++++++++++++++++++++++++++++-
1 file changed, 179 insertions(+), 3 deletions(-)
diff --git a/elbepack/shellhelper.py b/elbepack/shellhelper.py
index 203b5f4c..69481926 100644
--- a/elbepack/shellhelper.py
+++ b/elbepack/shellhelper.py
@@ -17,7 +17,6 @@ from io import TextIOWrapper, BytesIO
log = logging.getLogger("log")
soap = logging.getLogger("soap")
-
class CommandError(Exception):
def __init__(self, cmd, returncode):
self.returncode = returncode
@@ -28,6 +27,22 @@ class CommandError(Exception):
self.returncode, self.cmd)
def system(cmd, allow_fail=False, env_add=None):
+ """system() - Execute cmd in a shell.
+
+ Throws a CommandError if cmd returns none-zero and allow_fail=False
+
+ --
+
+ >>> system("true")
+
+ >>> system("false", allow_fail=True)
+
+ >>> system("$FALSE", env_add={"FALSE":"false"}) # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+
+ """
new_env = os.environ.copy()
if env_add:
new_env.update(env_add)
@@ -40,6 +55,28 @@ def system(cmd, allow_fail=False, env_add=None):
def command_out(cmd, stdin=None, output=PIPE, env_add=None):
+ """command_out() - Execute cmd in a shell.
+
+ Returns a tuple with the exitcode and the output of cmd.
+
+ --
+
+ >>> command_out("true")
+ (0, '')
+
+ >>> command_out("$TRUE && true", env_add={"TRUE":"true"})
+ (0, '')
+
+ >>> command_out("cat -", stdin=b"ELBE")
+ (0, 'ELBE')
+
+ >>> command_out("2>&1 cat -", stdin=b"ELBE")
+ (0, 'ELBE')
+
+ >>> command_out("false")
+ (1, '')
+
+ """
new_env = os.environ.copy()
if env_add:
new_env.update(env_add)
@@ -59,6 +96,22 @@ def command_out(cmd, stdin=None, output=PIPE, env_add=None):
def system_out(cmd, stdin=None, allow_fail=False, env_add=None):
+ """system_out() - Wrapper around command_out().
+
+ On failure, raises an exception if allow_fail=False, on success,
+ returns the output of cmd.
+
+ --
+
+ >>> system_out("false") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+
+ >>> system_out("false", allow_fail=True)
+ ''
+
+ """
code, out = command_out(cmd, stdin=stdin, env_add=env_add)
if code != 0:
@@ -69,6 +122,22 @@ def system_out(cmd, stdin=None, allow_fail=False, env_add=None):
def command_out_stderr(cmd, stdin=None, env_add=None):
+ """command_out_stderr() - Execute cmd in a shell.
+
+ Returns a tuple of the exitcode, stdout and stderr of cmd.
+
+ --
+
+ >>> command_out_stderr("$TRUE && cat -", stdin=b"ELBE", env_add={"TRUE":"true"})
+ (0, 'ELBE', '')
+
+ >>> command_out_stderr("1>&2 cat - && false", stdin=b"ELBE")
+ (1, '', 'ELBE')
+
+ >>> command_out_stderr("true")
+ (0, '', '')
+
+ """
new_env = os.environ.copy()
if env_add:
new_env.update(env_add)
@@ -89,6 +158,24 @@ def command_out_stderr(cmd, stdin=None, env_add=None):
def system_out_stderr(cmd, stdin=None, allow_fail=False, env_add=None):
+ """system_out_stderr() - Wrapper around command_out_stderr()
+
+ Throws CommandError if cmd failed and allow_fail=False. Otherwise,
+ returns the stdout and stderr of cmd.
+
+ --
+
+ >>> system_out_stderr("false") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+
+ >>> system_out_stderr("cat - && false", allow_fail=True, stdin=b"ELBE")
+ ('ELBE', '')
+
+ >>> system_out_stderr("1>&2 cat -", allow_fail=True, stdin=b"ELBE")
+ ('', 'ELBE')
+ """
code, out, err = command_out_stderr(cmd, stdin, env_add)
if code != 0:
@@ -99,6 +186,43 @@ def system_out_stderr(cmd, stdin=None, allow_fail=False, env_add=None):
def do(cmd, allow_fail=False, stdin=None, env_add=None):
+ """do() - Execute cmd in a shell and redirect outputs to logging.
+
+ Throws a CommandError if cmd failed with allow_Fail=False.
+
+ --
+
+ Let's redirect the loggers to current stdout
+ >>> import logging
+ >>> import sys
+ >>> from elbepack.log import context_fmt
+ >>> root = logging.getLogger()
+ >>> for hdlr in root.handlers:
+ ... root.removeHandler(hdlr)
+ >>> hdlr = logging.StreamHandler(sys.stdout)
+ >>> hdlr.setFormatter(context_fmt)
+ >>> root.addHandler(hdlr)
+
+ >>> do("true")
+ [CMD] true
+
+ >>> do("false", allow_fail=True)
+ [CMD] false
+
+ >>> do("cat -", stdin=b"ELBE")
+ [CMD] cat -
+
+ >>> do("cat - && false", stdin=b"ELBE") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+
+ >>> do("false") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+ """
+
new_env = os.environ.copy()
if env_add:
new_env.update(env_add)
@@ -121,6 +245,27 @@ def do(cmd, allow_fail=False, stdin=None, env_add=None):
def chroot(directory, cmd, env_add=None, **kwargs):
+ """chroot() - Wrapper around do().
+
+ --
+
+ Let's redirect the loggers to current stdout
+ >>> import logging
+ >>> import sys
+ >>> from elbepack.log import context_fmt
+ >>> root = logging.getLogger()
+ >>> for hdlr in root.handlers:
+ ... root.removeHandler(hdlr)
+ >>> hdlr = logging.StreamHandler(sys.stdout)
+ >>> hdlr.setFormatter(context_fmt)
+ >>> root.addHandler(hdlr)
+
+ >>> chroot("/", "true") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+ """
+
new_env = {"LANG":"C",
"LANGUAGE":"C",
"LC_ALL":"C"}
@@ -130,6 +275,37 @@ def chroot(directory, cmd, env_add=None, **kwargs):
do(chcmd, env_add=new_env, **kwargs)
def get_command_out(cmd, stdin=None, allow_fail=False, env_add={}):
+ """get_command_out() - Like do() but returns stdout.
+
+ --
+
+ Let's quiet the loggers
+ >>> import logging
+ >>> import sys
+ >>> from elbepack.log import context_fmt
+ >>> log = logging.getLogger("log")
+ >>> for hdlr in log.handlers:
+ ... log.removeHandler(hdlr)
+ >>> hdlr = logging.StreamHandler(sys.stdout)
+ >>> hdlr.setFormatter(context_fmt)
+ >>> log.addHandler(hdlr)
+
+ >>> get_command_out("echo ELBE")
+ b'ELBE\\n'
+
+ >>> get_command_out("false") # doctest: +ELLIPSIS
+ Traceback (most recent call last):
+ ...
+ elbepack.shellhelper.CommandError: ...
+
+ >>> get_command_out("false", allow_fail=True)
+ b''
+
+ >>> get_command_out("cat -", stdin=b"ELBE", env_add={"TRUE":"true"})
+ b'ELBE'
+
+ """
+
new_env = os.environ.copy()
new_env.update(env_add)
@@ -140,10 +316,10 @@ def get_command_out(cmd, stdin=None, allow_fail=False, env_add={}):
if stdin is None:
p = Popen(cmd, shell=True, stdout=PIPE, stderr=w, env=new_env)
else:
- p = Popen(cmd, shell=True, stdin=PIPE, stdout=w, stderr=STDOUT, env=new_env)
+ p = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, stderr=w, env=new_env)
async_logging(r, w, soap, log)
- stdout, stderr = p.communicate(input=stdin)
+ stdout, _ = p.communicate(input=stdin)
if p.returncode and not allow_fail:
raise CommandError(cmd, p.returncode)
--
2.26.2
More information about the elbe-devel
mailing list