diff options
Diffstat (limited to 'test/conftest.py')
| -rw-r--r-- | test/conftest.py | 146 |
1 files changed, 77 insertions, 69 deletions
diff --git a/test/conftest.py b/test/conftest.py index 70cd0c6..08b1b56 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,7 +1,12 @@ +#!/usr/bin/env python3 + import sys import pytest import time import re +import os +import threading + # If a test fails, wait a moment before retrieving the captured # stdout/stderr. When using a server process, this makes sure that we capture @@ -16,74 +21,77 @@ def pytest_pyfunc_call(pyfuncitem): if failed: time.sleep(1) -@pytest.fixture() -def pass_capfd(request, capfd): - '''Provide capfd object to UnitTest instances''' - request.instance.capfd = capfd - -def check_test_output(capfd): - (stdout, stderr) = capfd.readouterr() - - # Write back what we've read (so that it will still be printed. - sys.stdout.write(stdout) - sys.stderr.write(stderr) - - # Strip out false positives - for (pattern, flags, count) in capfd.false_positives: - cp = re.compile(pattern, flags) - (stdout, cnt) = cp.subn('', stdout, count=count) - if count == 0 or count - cnt > 0: - stderr = cp.sub('', stderr, count=count - cnt) - - patterns = [ r'\b{}\b'.format(x) for x in - ('exception', 'error', 'warning', 'fatal', 'traceback', - 'fault', 'crash(?:ed)?', 'abort(?:ed)', - 'uninitiali[zs]ed') ] - patterns += ['^==[0-9]+== '] - for pattern in patterns: - cp = re.compile(pattern, re.IGNORECASE | re.MULTILINE) - hit = cp.search(stderr) - if hit: - raise AssertionError('Suspicious output to stderr (matched "%s")' % hit.group(0)) - hit = cp.search(stdout) - if hit: - raise AssertionError('Suspicious output to stdout (matched "%s")' % hit.group(0)) - -def register_output(self, pattern, count=1, flags=re.MULTILINE): - '''Register *pattern* as false positive for output checking - - This prevents the test from failing because the output otherwise - appears suspicious. + +class OutputChecker: + '''Check output data for suspicious patters. + + Everything written to check_output.fd will be scanned for suspicious + messages and then written to sys.stdout. ''' - self.false_positives.append((pattern, flags, count)) - -# This is a terrible hack that allows us to access the fixtures from the -# pytest_runtest_call hook. Among a lot of other hidden assumptions, it probably -# relies on tests running sequential (i.e., don't dare to use e.g. the xdist -# plugin) -current_capfd = None -@pytest.yield_fixture(autouse=True) -def save_cap_fixtures(request, capfd): - global current_capfd - capfd.false_positives = [] - - # Monkeypatch in a function to register false positives - type(capfd).register_output = register_output - - if request.config.getoption('capture') == 'no': - capfd = None - current_capfd = capfd - bak = current_capfd - yield - - # Try to catch problems with this hack (e.g. when running tests - # simultaneously) - assert bak is current_capfd - current_capfd = None - -@pytest.hookimpl(trylast=True) -def pytest_runtest_call(item): - capfd = current_capfd - if capfd is not None: - check_test_output(capfd) + def __init__(self): + (fd_r, fd_w) = os.pipe() + self.fd = fd_w + self._false_positives = [] + self._buf = bytearray() + self._thread = threading.Thread(target=self._loop, daemon=True, args=(fd_r,)) + self._thread.start() + + def register_output(self, pattern, count=1, flags=re.MULTILINE): + '''Register *pattern* as false positive for output checking + + This prevents the test from failing because the output otherwise + appears suspicious. + ''' + + self._false_positives.append((pattern, flags, count)) + + def _loop(self, ifd): + BUFSIZE = 128*1024 + ofd = sys.stdout.fileno() + while True: + buf = os.read(ifd, BUFSIZE) + if not buf: + break + os.write(ofd, buf) + self._buf += buf + + def _check(self): + os.close(self.fd) + self._thread.join() + + buf = self._buf.decode('utf8', errors='replace') + + # Strip out false positives + for (pattern, flags, count) in self._false_positives: + cp = re.compile(pattern, flags) + (buf, cnt) = cp.subn('', buf, count=count) + + patterns = [ r'\b{}\b'.format(x) for x in + ('exception', 'error', 'warning', 'fatal', 'traceback', + 'fault', 'crash(?:ed)?', 'abort(?:ed)', + 'uninitiali[zs]ed') ] + patterns += ['^==[0-9]+== '] + + for pattern in patterns: + cp = re.compile(pattern, re.IGNORECASE | re.MULTILINE) + hit = cp.search(buf) + if hit: + raise AssertionError('Suspicious output to stderr (matched "%s")' + % hit.group(0)) + +@pytest.fixture() +def output_checker(request): + checker = OutputChecker() + yield checker + checker._check() + + +# Make test outcome available to fixtures +# (from https://github.com/pytest-dev/pytest/issues/230) +@pytest.hookimpl(hookwrapper=True, tryfirst=True) +def pytest_runtest_makereport(item, call): + outcome = yield + rep = outcome.get_result() + setattr(item, "rep_" + rep.when, rep) + return rep |