Bring back call helpers

Were not used sofar but will be extendend and used again by autotest code

diff --git a/src/call_helpers.py b/src/call_helpers.py
new file mode 100644 (file)
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 <info@intra2net.com>
+
+""" 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 (file)
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 <info@intra2net.com>
+
+""" 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()