diff options
Diffstat (limited to 'test/test_ctests.py')
| -rw-r--r-- | test/test_ctests.py | 49 |
1 files changed, 42 insertions, 7 deletions
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") |