meson
ninja
# Test packages:
-looseversion
pytest
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
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
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
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.
@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")
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()
@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',
# 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:
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,
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...
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')
projects / qemu-gpiodev / libfuse.git / commitdiff