• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2
3import sys
4import pytest
5import time
6import re
7import os
8import threading
9
10
11# If a test fails, wait a moment before retrieving the captured
12# stdout/stderr. When using a server process, this makes sure that we capture
13# any potential output of the server that comes *after* a test has failed. For
14# example, if a request handler raises an exception, the server first signals an
15# error to FUSE (causing the test to fail), and then logs the exception. Without
16# the extra delay, the exception will go into nowhere.
17@pytest.hookimpl(hookwrapper=True)
18def pytest_pyfunc_call(pyfuncitem):
19    outcome = yield
20    failed = outcome.excinfo is not None
21    if failed:
22        time.sleep(1)
23
24
25class OutputChecker:
26    '''Check output data for suspicious patterns.
27
28    Everything written to check_output.fd will be scanned for suspicious
29    messages and then written to sys.stdout.
30    '''
31
32    def __init__(self):
33        (fd_r, fd_w) = os.pipe()
34        self.fd = fd_w
35        self._false_positives = []
36        self._buf = bytearray()
37        self._thread = threading.Thread(target=self._loop, daemon=True, args=(fd_r,))
38        self._thread.start()
39
40    def register_output(self, pattern, count=1, flags=re.MULTILINE):
41        '''Register *pattern* as false positive for output checking
42
43        This prevents the test from failing because the output otherwise
44        appears suspicious.
45        '''
46
47        self._false_positives.append((pattern, flags, count))
48
49    def _loop(self, ifd):
50        BUFSIZE = 128*1024
51        ofd = sys.stdout.fileno()
52        while True:
53            buf = os.read(ifd, BUFSIZE)
54            if not buf:
55                break
56            os.write(ofd, buf)
57            self._buf += buf
58
59    def _check(self):
60        os.close(self.fd)
61        self._thread.join()
62
63        buf = self._buf.decode('utf8', errors='replace')
64
65        # Strip out false positives
66        for (pattern, flags, count) in self._false_positives:
67            cp = re.compile(pattern, flags)
68            (buf, cnt) = cp.subn('', buf, count=count)
69
70        patterns = [ r'\b{}\b'.format(x) for x in
71                     ('exception', 'error', 'warning', 'fatal', 'traceback',
72                        'fault', 'crash(?:ed)?', 'abort(?:ed)',
73                        'uninitiali[zs]ed') ]
74        patterns += ['^==[0-9]+== ']
75
76        for pattern in patterns:
77            cp = re.compile(pattern, re.IGNORECASE | re.MULTILINE)
78            hit = cp.search(buf)
79            if hit:
80                raise AssertionError('Suspicious output to stderr (matched "%s")'
81                                     % hit.group(0))
82
83@pytest.fixture()
84def output_checker(request):
85    checker = OutputChecker()
86    yield checker
87    checker._check()
88
89
90# Make test outcome available to fixtures
91# (from https://github.com/pytest-dev/pytest/issues/230)
92@pytest.hookimpl(hookwrapper=True, tryfirst=True)
93def pytest_runtest_makereport(item, call):
94    outcome = yield
95    rep = outcome.get_result()
96    setattr(item, "rep_" + rep.when, rep)
97    return rep
98