Fix output checking in test cases

py.test's capture plugin does not work reliably when used by
other fixtures. Therefore, implement our own version.

3 files changed, 114 insertions, 95 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
diff --git a/test/test_ctests.py b/test/test_ctests.py
index d2f8582..e4ce668 100644
--- a/test/test_ctests.py
+++ b/test/test_ctests.py
@@ -21,7 +21,7 @@ pytestmark = fuse_test_marker()
 @pytest.mark.skipif('FUSE_CAP_WRITEBACK_CACHE' not in fuse_caps,
                     reason='not supported by running kernel')
 @pytest.mark.parametrize("writeback", (False, True))
-def test_write_cache(tmpdir, writeback):
+def test_write_cache(tmpdir, writeback, output_checker):
     if writeback and LooseVersion(platform.release()) < '3.14':
         pytest.skip('Requires kernel 3.14 or newer')
     # This test hangs under Valgrind when running close(fd)
@@ -33,7 +33,7 @@ def test_write_cache(tmpdir, writeback):
                 mnt_dir ]
     if writeback:
         cmdline.append('-owriteback_cache')
-    subprocess.check_call(cmdline)
+    subprocess.check_call(cmdline, stdout=output_checker.fd, stderr=output_checker.fd)
 
 
 names = [ 'notify_inval_inode', 'invalidate_path' ]
@@ -43,14 +43,15 @@ if fuse_proto >= (7,15):
                     reason='not supported by running kernel')
 @pytest.mark.parametrize("name", names)
 @pytest.mark.parametrize("notify", (True, False))
-def test_notify1(tmpdir, name, notify):
+def test_notify1(tmpdir, name, notify, output_checker):
     mnt_dir = str(tmpdir)
     cmdline = base_cmdline + \
               [ pjoin(basename, 'example', name),
                 '-f', '--update-interval=1', mnt_dir ]
     if not notify:
         cmdline.append('--no-notify')
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         filename = pjoin(mnt_dir, 'current_time')
@@ -72,14 +73,15 @@ def test_notify1(tmpdir, name, notify):
 @pytest.mark.skipif(fuse_proto < (7,12),
                     reason='not supported by running kernel')
 @pytest.mark.parametrize("notify", (True, False))
-def test_notify_file_size(tmpdir, notify):
+def test_notify_file_size(tmpdir, notify, output_checker):
     mnt_dir = str(tmpdir)
     cmdline = base_cmdline + \
               [ pjoin(basename, 'example', 'invalidate_path'),
                 '-f', '--update-interval=1', mnt_dir ]
     if not notify:
         cmdline.append('--no-notify')
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         filename = pjoin(mnt_dir, 'growing')
diff --git a/test/test_examples.py b/test/test_examples.py
index d0da69d..72c4100 100755
--- a/test/test_examples.py
+++ b/test/test_examples.py
@@ -76,9 +76,11 @@ def short_tmpdir():
                                              invoke_mount_fuse_drop_privileges))
 @pytest.mark.parametrize("options", powerset(options))
 @pytest.mark.parametrize("name", ('hello', 'hello_ll'))
-def test_hello(tmpdir, name, options, cmdline_builder):
+def test_hello(tmpdir, name, options, cmdline_builder, output_checker):
     mnt_dir = str(tmpdir)
-    mount_process = subprocess.Popen(cmdline_builder(mnt_dir, name, options))
+    mount_process = subprocess.Popen(
+        cmdline_builder(mnt_dir, name, options),
+        stdout=output_checker.fd, stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         assert os.listdir(mnt_dir) == [ 'hello' ]
@@ -100,15 +102,15 @@ def test_hello(tmpdir, name, options, cmdline_builder):
 @pytest.mark.parametrize("writeback", (False, True))
 @pytest.mark.parametrize("name", ('passthrough', 'passthrough_fh', 'passthrough_ll'))
 @pytest.mark.parametrize("debug", (False, True))
-def test_passthrough(short_tmpdir, name, debug, capfd, writeback):
+def test_passthrough(short_tmpdir, name, debug, output_checker, writeback):
     # Avoid false positives from libfuse debug messages
     if debug:
-        capfd.register_output(r'^   unique: [0-9]+, error: -[0-9]+ .+$',
-                              count=0)
+        output_checker.register_output(r'^   unique: [0-9]+, error: -[0-9]+ .+$',
+                                       count=0)
 
     # test_syscalls prints "No error" under FreeBSD
-    capfd.register_output(r"^ \d\d \[[^\]]+ message: 'No error: 0'\]",
-                          count=0)
+    output_checker.register_output(r"^ \d\d \[[^\]]+ message: 'No error: 0'\]",
+                                   count=0)
 
     mnt_dir = str(short_tmpdir.mkdir('mnt'))
     src_dir = str(short_tmpdir.mkdir('src'))
@@ -125,7 +127,8 @@ def test_passthrough(short_tmpdir, name, debug, capfd, writeback):
         cmdline.append('-o')
         cmdline.append('writeback')
         
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         work_dir = mnt_dir + src_dir
@@ -169,7 +172,7 @@ def test_passthrough(short_tmpdir, name, debug, capfd, writeback):
         umount(mount_process, mnt_dir)
 
 @pytest.mark.parametrize("cache", (False, True))
-def test_passthrough_hp(short_tmpdir, cache):
+def test_passthrough_hp(short_tmpdir, cache, output_checker):
     mnt_dir = str(short_tmpdir.mkdir('mnt'))
     src_dir = str(short_tmpdir.mkdir('src'))
 
@@ -180,7 +183,8 @@ def test_passthrough_hp(short_tmpdir, cache):
     if not cache:
         cmdline.append('--nocache')
         
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
 
@@ -224,7 +228,7 @@ def test_passthrough_hp(short_tmpdir, cache):
         
 @pytest.mark.skipif(fuse_proto < (7,11),
                     reason='not supported by running kernel')
-def test_ioctl(tmpdir):
+def test_ioctl(tmpdir, output_checker):
     progname = pjoin(basename, 'example', 'ioctl')
     if not os.path.exists(progname):
         pytest.skip('%s not built' % os.path.basename(progname))
@@ -232,7 +236,8 @@ def test_ioctl(tmpdir):
     mnt_dir = str(tmpdir)
     testfile = pjoin(mnt_dir, 'fioc')
     cmdline = base_cmdline + [progname, '-f', mnt_dir ]
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
 
@@ -252,11 +257,12 @@ def test_ioctl(tmpdir):
     else:
         umount(mount_process, mnt_dir)
 
-def test_poll(tmpdir):
+def test_poll(tmpdir, output_checker):
     mnt_dir = str(tmpdir)
     cmdline = base_cmdline + [pjoin(basename, 'example', 'poll'),
                '-f', mnt_dir ]
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         cmdline = base_cmdline + \
@@ -268,7 +274,7 @@ def test_poll(tmpdir):
     else:
         umount(mount_process, mnt_dir)
 
-def test_null(tmpdir):
+def test_null(tmpdir, output_checker):
     progname = pjoin(basename, 'example', 'null')
     if not os.path.exists(progname):
         pytest.skip('%s not built' % os.path.basename(progname))
@@ -277,7 +283,8 @@ def test_null(tmpdir):
     with open(mnt_file, 'w') as fh:
         fh.write('dummy')
     cmdline = base_cmdline + [ progname, '-f', mnt_file ]
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     def test_fn(name):
         return os.stat(name).st_size > 4000
     try:
@@ -296,7 +303,7 @@ def test_null(tmpdir):
 @pytest.mark.skipif(fuse_proto < (7,12),
                     reason='not supported by running kernel')
 @pytest.mark.parametrize("notify", (True, False))
-def test_notify_inval_entry(tmpdir, notify):
+def test_notify_inval_entry(tmpdir, notify, output_checker):
     mnt_dir = str(tmpdir)
     cmdline = base_cmdline + \
               [ pjoin(basename, 'example', 'notify_inval_entry'),
@@ -304,7 +311,8 @@ def test_notify_inval_entry(tmpdir, notify):
                 '--timeout=5', mnt_dir ]
     if not notify:
         cmdline.append('--no-notify')
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
     try:
         wait_for_mount(mount_process, mnt_dir)
         fname = pjoin(mnt_dir, os.listdir(mnt_dir)[0])
@@ -330,7 +338,7 @@ def test_notify_inval_entry(tmpdir, notify):
 
 @pytest.mark.skipif(os.getuid() != 0,
                     reason='needs to run as root')
-def test_cuse(capfd):
+def test_cuse(capfd, output_checker):
 
     # Valgrind warns about unknown ioctls, that's ok
     capfd.register_output(r'^==([0-9]+).+unhandled ioctl.+\n'
@@ -342,7 +350,8 @@ def test_cuse(capfd):
     cmdline = base_cmdline + \
               [ pjoin(basename, 'example', 'cuse'),
                 '-f', '--name=%s' % devname ]
-    mount_process = subprocess.Popen(cmdline)
+    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
+                                     stderr=output_checker.fd)
 
     cmdline = base_cmdline + \
               [ pjoin(basename, 'example', 'cuse_client'),