diff options
Diffstat (limited to 'pbr/tests/base.py')
-rw-r--r-- | pbr/tests/base.py | 47 |
1 files changed, 45 insertions, 2 deletions
diff --git a/pbr/tests/base.py b/pbr/tests/base.py index 09bd642..e017b06 100644 --- a/pbr/tests/base.py +++ b/pbr/tests/base.py @@ -47,6 +47,7 @@ import sys import fixtures import testresources import testtools +from testtools import content from pbr import options @@ -147,6 +148,48 @@ class BaseTestCase(testtools.TestCase, testresources.ResourcedTestCase): return result +class CapturedSubprocess(fixtures.Fixture): + """Run a process and capture its output. + + :attr stdout: The output (a string). + :attr stderr: The standard error (a string). + :attr returncode: The return code of the process. + + Note that stdout and stderr are decoded from the bytestrings subprocess + returns using error=replace. + """ + + def __init__(self, label, *args, **kwargs): + """Create a CapturedSubprocess. + + :param label: A label for the subprocess in the test log. E.g. 'foo'. + :param *args: The *args to pass to Popen. + :param **kwargs: The **kwargs to pass to Popen. + """ + super(CapturedSubprocess, self).__init__() + self.label = label + self.args = args + self.kwargs = kwargs + self.kwargs['stderr'] = subprocess.PIPE + self.kwargs['stdin'] = subprocess.PIPE + self.kwargs['stdout'] = subprocess.PIPE + + def setUp(self): + super(CapturedSubprocess, self).setUp() + proc = subprocess.Popen(*self.args, **self.kwargs) + out, err = proc.communicate() + self.out = out.decode('utf-8', 'replace') + self.err = err.decode('utf-8', 'replace') + self.addDetail(self.label + '-stdout', content.text_content(self.out)) + self.addDetail(self.label + '-stderr', content.text_content(self.err)) + self.returncode = proc.returncode + if proc.returncode: + raise AssertionError('Failed process %s' % proc.returncode) + self.addCleanup(delattr, self, 'out') + self.addCleanup(delattr, self, 'err') + self.addCleanup(delattr, self, 'returncode') + + def _run_cmd(args, cwd): """Run the command args in cwd. @@ -158,8 +201,8 @@ def _run_cmd(args, cwd): args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd) streams = tuple(s.decode('latin1').strip() for s in p.communicate()) - for content in streams: - print(content) + for stream_content in streams: + print(stream_content) return (streams) + (p.returncode,) |