created call_helpers with call_and_capture and corresponding unittest

diff --git a/call_helpers.py b/call_helpers.py
new file mode 100644 (file)
index 0000000..986066d
--- /dev/null
+++ b/call_helpers.py
@@ -0,0 +1,56 @@
+""" 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
+
+.. codeauthor:: Christian Herdtweck, christian.herdtweck@intra2net.com
+"""
+
+from subprocess import Popen, PIPE
+
+
+def call_and_capture(command, stdin_data=None, *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 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)
+
+    :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
+    return proc.returncode, stdout_data.splitlines(), stderr_data.splitlines()

diff --git a/test/call_helper_unittest.py b/test/call_helper_unittest.py
new file mode 100644 (file)
index 0000000..c9057fd
--- /dev/null
+++ b/test/call_helper_unittest.py
@@ -0,0 +1,66 @@
+""" call_helper_unittest.py: unit tests for call_helpers
+
+Should be run from python2 and python3!
+
+.. codeauthor:: Christian Herdtweck, christian.herdtweck@intra2net.com
+"""
+
+import unittest
+import os
+
+import call_helpers
+from 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()