diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/ci-build.sh | 2 | ||||
| -rw-r--r-- | test/test_ctests.py | 49 | ||||
| -rwxr-xr-x | test/test_examples.py | 25 | ||||
| -rw-r--r-- | test/util.py | 36 |
4 files changed, 97 insertions, 15 deletions
diff --git a/test/ci-build.sh b/test/ci-build.sh index af1e07a..4f7da4c 100755 --- a/test/ci-build.sh +++ b/test/ci-build.sh @@ -2,7 +2,7 @@ set -e -TEST_CMD="pytest -v --maxfail=1 --log-level=DEBUG --log-cli-level=DEBUG test/" +TEST_CMD="pytest -v --maxfail=1 --log-level=INFO --log-cli-level=INFO test/" SAN="-Db_sanitize=address,undefined" # not default diff --git a/test/test_ctests.py b/test/test_ctests.py index 55db156..feefc07 100644 --- a/test/test_ctests.py +++ b/test/test_ctests.py @@ -10,10 +10,11 @@ import pytest import platform import sys import os -from looseversion import LooseVersion +import logging +from packaging import version from util import (wait_for_mount, umount, cleanup, base_cmdline, safe_sleep, basename, fuse_test_marker, fuse_caps, - fuse_proto, create_tmpdir) + fuse_proto, create_tmpdir, parse_kernel_version) from os.path import join as pjoin import os.path @@ -23,7 +24,7 @@ pytestmark = fuse_test_marker() reason='not supported by running kernel') @pytest.mark.parametrize("writeback", (False, True)) def test_write_cache(tmpdir, writeback, output_checker): - if writeback and LooseVersion(platform.release()) < '3.14': + if writeback and parse_kernel_version(platform.release()) < version.parse('3.14'): pytest.skip('Requires kernel 3.14 or newer') # This test hangs under Valgrind when running close(fd) # test_write_cache.c:test_fs(). Most likely this is because of an internal @@ -37,7 +38,7 @@ def test_write_cache(tmpdir, writeback, output_checker): mnt_dir ] if writeback: cmdline.append('-owriteback_cache') - elif LooseVersion(platform.release()) >= '5.16': + elif parse_kernel_version(platform.release()) >= version.parse('5.16'): # Test that close(rofd) does not block waiting for pending writes. # This test requires kernel commit a390ccb316be ("fuse: add FOPEN_NOFLUSH") # so opt-in for this test from kernel 5.16. @@ -53,59 +54,93 @@ if fuse_proto >= (7,15): @pytest.mark.parametrize("name", names) @pytest.mark.parametrize("notify", (True, False)) def test_notify1(tmpdir, name, notify, output_checker): + logger = logging.getLogger(__name__) mnt_dir = str(tmpdir) + logger.debug(f"Mount directory: {mnt_dir}") create_tmpdir(mnt_dir) cmdline = base_cmdline + \ [ pjoin(basename, 'example', name), '-f', '--update-interval=1', mnt_dir ] if not notify: cmdline.append('--no-notify') + logger.debug(f"Command line: {' '.join(cmdline)}") mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd, stderr=output_checker.fd) try: wait_for_mount(mount_process, mnt_dir) + logger.debug("Mount completed") filename = pjoin(mnt_dir, 'current_time') + logger.debug(f"Target filename: {filename}") with open(filename, 'r') as fh: read1 = fh.read() + logger.debug(f"First read: {read1}") + logger.debug("Sleeping for 2 seconds...") safe_sleep(2) + logger.debug("Sleep completed") with open(filename, 'r') as fh: read2 = fh.read() + logger.debug(f"Second read: {read2}") if notify: + logger.debug("Expecting reads to be different") assert read1 != read2 else: + logger.debug("Expecting reads to be the same") assert read1 == read2 + logger.debug("Test completed successfully") except: - print("Failure in notify test: '" + str(cmdline) + "'") + logger.error(f"Failure in notify test: '{' '.join(cmdline)}'") + logger.exception("Exception details:") cleanup(mount_process, mnt_dir) raise else: - umount(mount_process, mnt_dir) + logger.debug("Unmounting...") + try: + umount(mount_process, mnt_dir) + logger.debug("Umount disabled") + except: + logger.error(f"Failure in unmount: '{' '.join(cmdline)}'") + cleanup(mount_process, mnt_dir) + logger.debug("Unmount completed") @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, output_checker): + logger = logging.getLogger(__name__) mnt_dir = str(tmpdir) + logger.debug(f"Mount directory: {mnt_dir}") create_tmpdir(mnt_dir) cmdline = base_cmdline + \ [ pjoin(basename, 'example', 'invalidate_path'), '-f', '--update-interval=1', mnt_dir ] if not notify: cmdline.append('--no-notify') + logger.debug(f"Command line: {' '.join(cmdline)}") mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd, stderr=output_checker.fd) + logger.debug(f"Mount process PID: {mount_process.pid}") try: wait_for_mount(mount_process, mnt_dir) filename = pjoin(mnt_dir, 'growing') size = os.path.getsize(filename) + logger.debug(f"Initial file size: {size}") + logger.debug("Sleeping for 2 seconds...") safe_sleep(2) + logger.debug("Sleep completed") new_size = os.path.getsize(filename) + logger.debug(f"New file size: {new_size}") if notify: assert new_size > size else: assert new_size == size + logger.debug("Test completed successfully") except: cleanup(mount_process, mnt_dir) raise else: - umount(mount_process, mnt_dir) + try: + umount(mount_process, mnt_dir) + except: + logger.error(f"Failure in unmount: '{' '.join(cmdline)}'") + cleanup(mount_process, mnt_dir) + logger.debug("Unmount completed") diff --git a/test/test_examples.py b/test/test_examples.py index 3b1fd1b..e0fb8c4 100755 --- a/test/test_examples.py +++ b/test/test_examples.py @@ -18,13 +18,15 @@ import time import errno import sys import platform -from looseversion import LooseVersion +import re +from packaging import version from tempfile import NamedTemporaryFile from contextlib import contextmanager from util import (wait_for_mount, umount, cleanup, base_cmdline, safe_sleep, basename, fuse_test_marker, test_printcap, - fuse_proto, fuse_caps, powerset) + fuse_proto, fuse_caps, powerset, parse_kernel_version) from os.path import join as pjoin +import logging pytestmark = fuse_test_marker() @@ -88,31 +90,46 @@ def readdir_inode(dir): @pytest.mark.parametrize("options", powerset(options)) @pytest.mark.parametrize("name", ('hello', 'hello_ll')) def test_hello(tmpdir, name, options, cmdline_builder, output_checker): + logger = logging.getLogger(__name__) mnt_dir = str(tmpdir) + logger.debug(f"Mount directory: {mnt_dir}") + cmdline = cmdline_builder(mnt_dir, name, options) + logger.debug(f"Command line: {' '.join(cmdline)}") mount_process = subprocess.Popen( - cmdline_builder(mnt_dir, name, options), + cmdline, stdout=output_checker.fd, stderr=output_checker.fd) + logger.debug(f"Mount process PID: {mount_process.pid}") try: + logger.debug("Waiting for mount...") wait_for_mount(mount_process, mnt_dir) + logger.debug("Mount completed") assert os.listdir(mnt_dir) == [ 'hello' ] + logger.debug("Verified 'hello' file exists in mount directory") filename = pjoin(mnt_dir, 'hello') with open(filename, 'r') as fh: assert fh.read() == 'Hello World!\n' + logger.debug("Verified contents of 'hello' file") with pytest.raises(IOError) as exc_info: open(filename, 'r+') assert exc_info.value.errno == errno.EACCES + logger.debug("Verified EACCES error when trying to open file for writing") with pytest.raises(IOError) as exc_info: open(filename + 'does-not-exist', 'r+') assert exc_info.value.errno == errno.ENOENT + logger.debug("Verified ENOENT error for non-existent file") if name == 'hello_ll': + logger.debug("Testing xattr for hello_ll") tst_xattr(mnt_dir) path = os.path.join(mnt_dir, 'hello') tst_xattr(path) except: + logger.error("Exception occurred during test", exc_info=True) cleanup(mount_process, mnt_dir) raise else: + logger.debug("Unmounting...") umount(mount_process, mnt_dir) + logger.debug("Test completed successfully") @pytest.mark.parametrize("writeback", (False, True)) @pytest.mark.parametrize("name", ('passthrough', 'passthrough_plus', @@ -251,7 +268,7 @@ def test_passthrough_hp(short_tmpdir, cache, output_checker): # unlinked testfiles check fails without kernel fix # "fuse: fix illegal access to inode with reused nodeid" # so opt-in for this test from kernel 5.14 - if LooseVersion(platform.release()) >= '5.14': + if parse_kernel_version(platform.release()) >= version.parse('5.14'): syscall_test_cmd.append('-u') subprocess.check_call(syscall_test_cmd) except: diff --git a/test/util.py b/test/util.py index 623b031..a421e72 100644 --- a/test/util.py +++ b/test/util.py @@ -8,9 +8,18 @@ from os.path import join as pjoin import sys import re import itertools +from packaging import version +import logging basename = pjoin(os.path.dirname(__file__), '..') +def parse_kernel_version(release): + # Extract the first three numbers from the kernel version string + match = re.match(r'^(\d+\.\d+\.\d+)', release) + if match: + return version.parse(match.group(1)) + return version.parse('0') + def get_printcap(): cmdline = base_cmdline + [ pjoin(basename, 'example', 'printcap') ] proc = subprocess.Popen(cmdline, stdout=subprocess.PIPE, @@ -63,21 +72,40 @@ def cleanup(mount_process, mnt_dir): mount_process.kill() def umount(mount_process, mnt_dir): + logger = logging.getLogger(__name__) + logger.debug(f"Unmounting {mnt_dir}") if 'bsd' in sys.platform or 'dragonfly' in sys.platform: cmdline = [ 'umount', mnt_dir ] + logger.debug("Using BSD-style umount command") else: + logger.debug("Using fusermount3 for unmounting") # fusermount3 will be setuid root, so we can only trace it with # valgrind if we're root if os.getuid() == 0: cmdline = base_cmdline + logger.debug("Running as root, using valgrind if configured") else: cmdline = [] + logger.debug("Not running as root, skipping valgrind for fusermount3") cmdline = cmdline + [ pjoin(basename, 'util', 'fusermount3'), '-z', '-u', mnt_dir ] - subprocess.check_call(cmdline) - assert not os.path.ismount(mnt_dir) + logger.debug(f"Unmount command: {' '.join(cmdline)}") + try: + result = subprocess.run(cmdline, capture_output=True, text=True, check=True) + if result.stdout: + logger.debug(f"Unmount command stdout: {result.stdout}") + if result.stderr: + logger.debug(f"Unmount command stderr: {result.stderr}") + except subprocess.CalledProcessError as e: + logger.error(f"Unmount command failed with return code {e.returncode}\nStdout: {e.stdout}\nStderr: {e.stderr}") + raise + + if not os.path.ismount(mnt_dir): + logger.debug(f"{mnt_dir} is no longer a mount point") + else: + logger.warning(f"{mnt_dir} is still a mount point after unmount command") # Give mount process a little while to terminate. Popen.wait(timeout) # was only added in 3.3... @@ -87,9 +115,11 @@ def umount(mount_process, mnt_dir): if code is not None: if code == 0: return - pytest.fail('file system process terminated with code %s' % (code,)) + logger.error(f"File system process terminated with code {code}") + pytest.fail(f'file system process terminated with code {code}') time.sleep(0.1) elapsed += 0.1 + logger.error("Mount process did not terminate within 30 seconds") pytest.fail('mount process did not terminate') |