mirror of
https://github.com/libfuse/libfuse.git
synced 2024-12-04 01:24:18 +08:00
eb88309882
Before:
```
=============================== warnings summary ===============================
test/test_ctests.py:12
/tmp/libfuse-build-mB50ZC/build-gcc-9/test/test_ctests.py:12: DeprecationWarning: The distutils package is deprecated and slated for removal in Python 3.12. Use setuptools or check PEP 632 for potential alternatives
from distutils.version import LooseVersion
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
```
105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
if __name__ == '__main__':
|
|
import pytest
|
|
import sys
|
|
sys.exit(pytest.main([__file__] + sys.argv[1:]))
|
|
|
|
import subprocess
|
|
import pytest
|
|
import platform
|
|
import sys
|
|
from looseversion import LooseVersion
|
|
from util import (wait_for_mount, umount, cleanup, base_cmdline,
|
|
safe_sleep, basename, fuse_test_marker, fuse_caps,
|
|
fuse_proto)
|
|
from os.path import join as pjoin
|
|
import os.path
|
|
|
|
pytestmark = fuse_test_marker()
|
|
|
|
@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, 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)
|
|
# test_write_cache.c:test_fs(). Most likely this is because of an internal
|
|
# deadlock in valgrind, it probably assumes that until close() returns,
|
|
# control does not come to the program.
|
|
mnt_dir = str(tmpdir)
|
|
cmdline = [ pjoin(basename, 'test', 'test_write_cache'),
|
|
mnt_dir ]
|
|
if writeback:
|
|
cmdline.append('-owriteback_cache')
|
|
elif LooseVersion(platform.release()) >= '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.
|
|
cmdline.append('--delay_ms=200')
|
|
subprocess.check_call(cmdline, stdout=output_checker.fd, stderr=output_checker.fd)
|
|
|
|
|
|
names = [ 'notify_inval_inode', 'invalidate_path' ]
|
|
if fuse_proto >= (7,15):
|
|
names.append('notify_store_retrieve')
|
|
@pytest.mark.skipif(fuse_proto < (7,12),
|
|
reason='not supported by running kernel')
|
|
@pytest.mark.parametrize("name", names)
|
|
@pytest.mark.parametrize("notify", (True, False))
|
|
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, stdout=output_checker.fd,
|
|
stderr=output_checker.fd)
|
|
try:
|
|
wait_for_mount(mount_process, mnt_dir)
|
|
filename = pjoin(mnt_dir, 'current_time')
|
|
with open(filename, 'r') as fh:
|
|
read1 = fh.read()
|
|
safe_sleep(2)
|
|
with open(filename, 'r') as fh:
|
|
read2 = fh.read()
|
|
if notify:
|
|
assert read1 != read2
|
|
else:
|
|
assert read1 == read2
|
|
except:
|
|
cleanup(mount_process, mnt_dir)
|
|
raise
|
|
else:
|
|
umount(mount_process, mnt_dir)
|
|
|
|
@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):
|
|
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, stdout=output_checker.fd,
|
|
stderr=output_checker.fd)
|
|
try:
|
|
wait_for_mount(mount_process, mnt_dir)
|
|
filename = pjoin(mnt_dir, 'growing')
|
|
size = os.path.getsize(filename)
|
|
safe_sleep(2)
|
|
new_size = os.path.getsize(filename)
|
|
if notify:
|
|
assert new_size > size
|
|
else:
|
|
assert new_size == size
|
|
except:
|
|
cleanup(mount_process, mnt_dir)
|
|
raise
|
|
else:
|
|
umount(mount_process, mnt_dir)
|