From 59e58de40c50ac05e24f5dc2d3267890974cde04 Mon Sep 17 00:00:00 2001 From: Nikolaus Rath Date: Tue, 29 Mar 2016 15:30:57 -0700 Subject: Added basic unit tests. Fixes issue #33. --- test/.gitignore | 1 + test/conftest.py | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ test/pytest.ini | 2 ++ test/test.c | 2 +- test/test_fuse.py | 32 +++++++++++++++++++++ test/util.py | 38 +++++++++++++++++++++++++ 6 files changed, 159 insertions(+), 1 deletion(-) create mode 100644 test/conftest.py create mode 100644 test/pytest.ini create mode 100755 test/test_fuse.py create mode 100644 test/util.py (limited to 'test') diff --git a/test/.gitignore b/test/.gitignore index 9daeafb..b7041be 100644 --- a/test/.gitignore +++ b/test/.gitignore @@ -1 +1,2 @@ test +__pycache__/ diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 0000000..d14350d --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,85 @@ +import sys +import pytest +import time +import re + +# If a test fails, wait a moment before retrieving the captured +# stdout/stderr. When using a server process, this makes sure that we capture +# any potential output of the server that comes *after* a test has failed. For +# example, if a request handler raises an exception, the server first signals an +# error to FUSE (causing the test to fail), and then logs the exception. Without +# the extra delay, the exception will go into nowhere. +@pytest.mark.hookwrapper +def pytest_pyfunc_call(pyfuncitem): + outcome = yield + failed = outcome.excinfo is not None + 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) + + for pattern in ('exception', 'error', 'warning', 'fatal', + 'fault', 'crash(?:ed)?', 'abort(?:ed)'): + cp = re.compile(r'\b{}\b'.format(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. + ''' + + 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) diff --git a/test/pytest.ini b/test/pytest.ini new file mode 100644 index 0000000..bc4af36 --- /dev/null +++ b/test/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +addopts = --verbose --assert=rewrite --tb=native -x diff --git a/test/test.c b/test/test.c index 5d5750d..2b38bb2 100644 --- a/test/test.c +++ b/test/test.c @@ -993,7 +993,7 @@ static int do_test_open_acc(int flags, const char *flags_str, int mode, int err) int res; int fd; - start_test("open_acc(%s) mode: 0%03o error: '%s'", flags_str, mode, + start_test("open_acc(%s) mode: 0%03o message: '%s'", flags_str, mode, strerror(err)); unlink(testfile); res = create_file(testfile, data, datalen); diff --git a/test/test_fuse.py b/test/test_fuse.py new file mode 100755 index 0000000..bbba6e0 --- /dev/null +++ b/test/test_fuse.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +import pytest +import sys + +if __name__ == '__main__': + sys.exit(pytest.main([__file__] + sys.argv[1:])) + +import subprocess +import os +from util import wait_for_mount, umount, cleanup + +basename = os.path.join(os.path.dirname(__file__), '..') + +def test_fuse(tmpdir): + mnt_dir = str(tmpdir.mkdir('mnt')) + src_dir = str(tmpdir.mkdir('src')) + + cmdline = [ os.path.join(basename, 'example', 'fusexmp_fh'), + '-f', '-o' , 'use_ino,readdir_ino,kernel_cache', + mnt_dir ] + mount_process = subprocess.Popen(cmdline) + try: + wait_for_mount(mount_process, mnt_dir) + cmdline = [ os.path.join(basename, 'test', 'test'), + os.path.join(mnt_dir, src_dir), + ':' + src_dir ] + subprocess.check_call(cmdline) + except: + cleanup(mnt_dir) + raise + else: + umount(mount_process, mnt_dir) diff --git a/test/util.py b/test/util.py new file mode 100644 index 0000000..48ec995 --- /dev/null +++ b/test/util.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 +import subprocess +import pytest +import os +import time + +def wait_for_mount(mount_process, mnt_dir): + elapsed = 0 + while elapsed < 30: + if os.path.ismount(mnt_dir): + return True + if mount_process.poll() is not None: + pytest.fail('file system process terminated prematurely') + time.sleep(0.1) + elapsed += 0.1 + pytest.fail("mountpoint failed to come up") + +def cleanup(mnt_dir): + subprocess.call(['fusermount', '-z', '-u', mnt_dir], + stdout=subprocess.DEVNULL, + stderr=subprocess.STDOUT) + +def umount(mount_process, mnt_dir): + subprocess.check_call(['fusermount', '-z', '-u', mnt_dir]) + assert not os.path.ismount(mnt_dir) + + # Give mount process a little while to terminate. Popen.wait(timeout) + # was only added in 3.3... + elapsed = 0 + while elapsed < 30: + code = mount_process.poll() + if code is not None: + if code == 0: + return + pytest.fail('file system process terminated with code %s' % (code,)) + time.sleep(0.1) + elapsed += 0.1 + pytest.fail('mount process did not terminate') -- cgit v1.2.3