From 1343f59c274bd5c4cea3ed5ca3dea092000f2b13 Mon Sep 17 00:00:00 2001 From: Nikolaus Rath Date: Thu, 4 Jul 2019 21:20:41 +0100 Subject: [PATCH] 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. --- test/conftest.py | 146 ++++++++++++++++++++++-------------------- test/test_ctests.py | 14 ++-- test/test_examples.py | 49 ++++++++------ 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'), -- 2.30.2