From dae4fa8a324afbd28f486809c6289196da0874cb Mon Sep 17 00:00:00 2001 From: Christian Herdtweck Date: Mon, 8 Oct 2018 09:44:02 +0200 Subject: [PATCH] Bring back call helpers Were not used sofar but will be extendend and used again by autotest code --- src/call_helpers.py | 83 +++++++++++++++++++++++++++++++++++++++++++ test/test_call_helpers.py | 86 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 0 deletions(-) create mode 100644 src/call_helpers.py create mode 100644 test/test_call_helpers.py diff --git a/src/call_helpers.py b/src/call_helpers.py new file mode 100644 index 0000000..4e945e1 --- /dev/null +++ b/src/call_helpers.py @@ -0,0 +1,83 @@ +# The software in this package is distributed under the GNU General +# Public License version 2 (with a special exception described below). +# +# A copy of GNU General Public License (GPL) is included in this distribution, +# in the file COPYING.GPL. +# +# As a special exception, if other files instantiate templates or use macros +# or inline functions from this file, or you compile this file and link it +# with other works to produce a work based on this file, this file +# does not by itself cause the resulting work to be covered +# by the GNU General Public License. +# +# However the source code for this file must still be made available +# in accordance with section (3) of the GNU General Public License. +# +# This exception does not invalidate any other reasons why a work based +# on this file might be covered by the GNU General Public License. +# +# Copyright (c) 2016-2018 Intra2net AG + +""" Helpers for calling commands, capture their output, return result code + +Subprocess library just does not provide all the simplicity we would like + +Stay python2 compatible --> no timeouts +""" + +from subprocess import Popen, PIPE + + +def call_and_capture(command, stdin_data=None, split_lines=True, + *args, **kwargs): + """ call command, captures stdout, stderr and return code, return when done + + Use only for commands with little output since all output has to be + buffered! + Quoting :py:mod:`subprocess`: + + ..note:: The data read is buffered in memory, so do not use this method if + the data size is large or unlimited. + + Forwards all args to Popen constructor, except: + stdout=PIPE (forced, ignored if in kwargs) + stderr=PIPE (forced, ignored if in kwargs) + shell=False (except if set in kwargs) + universal_newlines=True (except if set in kwargs) + + :param command: forwarded as first arg to Popen constructor + :param str stdin_data: forwarded to stdin of process through communicate + :param bool split_lines: True (default) to split output line-wise and + return list of strings; False --> return single + string for out and one for err + :param args: forwarded to Popen constructor + :param kwargs: forwarded to Popen constructor + :returns: (return_code, stdout, stderr); stdout and stderr are lists of + text lines (without terminating newlines); if universal_newlines is + True (default), the lines are of type str, otherwise they are non- + unicode text (type str in py2, bytes in py3). If split_lines is False + (not default), then stdout and stderr are single multi-line strings + + :raise: OSError (e.g., if command does not exist), ValueError if args are + invalid; no :py:class:`subprocess.CalledProcessError` nor + :py:class:`subprocess.TimeoutExpired` + """ + + # construct args + enforced_params = ('stdout', 'stderr') + my_kwargs = dict(stdout=PIPE, stderr=PIPE, + shell=False, universal_newlines=True) + for key, value in kwargs.items(): + if key not in enforced_params: + my_kwargs[key] = value + + # run command + proc = Popen(command, *args, **my_kwargs) + stdout_data, stderr_data = proc.communicate(stdin_data) + + # return + if split_lines: + return proc.returncode, stdout_data.splitlines(), \ + stderr_data.splitlines() + else: + return proc.returncode, stdout_data, stderr_data diff --git a/test/test_call_helpers.py b/test/test_call_helpers.py new file mode 100644 index 0000000..83924db --- /dev/null +++ b/test/test_call_helpers.py @@ -0,0 +1,86 @@ +# The software in this package is distributed under the GNU General +# Public License version 2 (with a special exception described below). +# +# A copy of GNU General Public License (GPL) is included in this distribution, +# in the file COPYING.GPL. +# +# As a special exception, if other files instantiate templates or use macros +# or inline functions from this file, or you compile this file and link it +# with other works to produce a work based on this file, this file +# does not by itself cause the resulting work to be covered +# by the GNU General Public License. +# +# However the source code for this file must still be made available +# in accordance with section (3) of the GNU General Public License. +# +# This exception does not invalidate any other reasons why a work based +# on this file might be covered by the GNU General Public License. +# +# Copyright (c) 2016-2018 Intra2net AG + +""" call_helper_unittest.py: unit tests for call_helpers + +Should be run from python2 and python3! +""" + +from __future__ import absolute_import + +import unittest +import os + +from src import call_helpers +from src.type_helpers import is_unicode + + +class CallHelperTester(unittest.TestCase): + + def test_call_and_capture(self): + """ tests call_and_capture with command ls -a / """ + + cmd = ['ls', '-a', '/'] + return_code, out_data, err_data = call_helpers.call_and_capture(cmd) + self.assertEqual(return_code, 0) + self.assertEqual(err_data, []) + + # get contents of root dir ourselves + true_contents = os.listdir('/') + ['.', '..'] + + # compare to out_data + self.assertEqual(sorted(out_data), sorted(true_contents)) + + # check type + self.assertTrue(all(isinstance(entry, str) for entry in out_data), + "Output is not of type str!") + + def test_call_and_capture_bytes(self): + """ tests call_and_capture with command ls -a / """ + + cmd = ['ls', '-a', '/'] + return_code, out_data, err_data = \ + call_helpers.call_and_capture(cmd, universal_newlines=False) + self.assertEqual(return_code, 0) + self.assertEqual(err_data, []) + + # get contents of root dir ourselves + true_contents = os.listdir(b'/') + [b'.', b'..'] + + # compare to out_data (ignoring order of entries) + self.assertEqual(sorted(out_data), sorted(true_contents)) + + # check type + self.assertFalse(is_unicode(out_data[0]), "Output is unicode!") + + def test_call_and_capture_err(self): + """ tests call_and_capture with command ls -a / """ + + cmd = ['ls', '-e'] + return_code, out_data, err_data = call_helpers.call_and_capture(cmd) + self.assertEqual(return_code, 2) + self.assertEqual(out_data, []) + self.assertEqual(len(err_data), 2) + self.assertEqual(err_data[0], "ls: invalid option -- 'e'") + self.assertEqual(err_data[1], "Try 'ls --help' for more information.") + + +if __name__ == '__main__': + unittest.main() -- 1.7.1