+#!/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
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
@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)
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' ]
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')
@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')
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' ]
@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'))
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
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'))
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)
@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))
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)
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 + \
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))
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:
@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'),
'--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])
@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'
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'),
projects / qemu-gpiodev / libfuse.git / commitdiff