• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2
3if __name__ == '__main__':
4    import pytest
5    import sys
6    sys.exit(pytest.main([__file__] + sys.argv[1:]))
7
8import subprocess
9import pytest
10import platform
11import sys
12from distutils.version import LooseVersion
13from util import (wait_for_mount, umount, cleanup, base_cmdline,
14                  safe_sleep, basename, fuse_test_marker, fuse_caps,
15                  fuse_proto)
16from os.path import join as pjoin
17import os.path
18
19pytestmark = fuse_test_marker()
20
21@pytest.mark.skipif('FUSE_CAP_WRITEBACK_CACHE' not in fuse_caps,
22                    reason='not supported by running kernel')
23@pytest.mark.parametrize("writeback", (False, True))
24def test_write_cache(tmpdir, writeback, output_checker):
25    if writeback and LooseVersion(platform.release()) < '3.14':
26        pytest.skip('Requires kernel 3.14 or newer')
27    # This test hangs under Valgrind when running close(fd)
28    # test_write_cache.c:test_fs(). Most likely this is because of an internal
29    # deadlock in valgrind, it probably assumes that until close() returns,
30    # control does not come to the program.
31    mnt_dir = str(tmpdir)
32    cmdline = [ pjoin(basename, 'test', 'test_write_cache'),
33                mnt_dir ]
34    if writeback:
35        cmdline.append('-owriteback_cache')
36    subprocess.check_call(cmdline, stdout=output_checker.fd, stderr=output_checker.fd)
37
38
39names = [ 'notify_inval_inode', 'invalidate_path' ]
40if fuse_proto >= (7,15):
41    names.append('notify_store_retrieve')
42@pytest.mark.skipif(fuse_proto < (7,12),
43                    reason='not supported by running kernel')
44@pytest.mark.parametrize("name", names)
45@pytest.mark.parametrize("notify", (True, False))
46def test_notify1(tmpdir, name, notify, output_checker):
47    mnt_dir = str(tmpdir)
48    cmdline = base_cmdline + \
49              [ pjoin(basename, 'example', name),
50                '-f', '--update-interval=1', mnt_dir ]
51    if not notify:
52        cmdline.append('--no-notify')
53    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
54                                     stderr=output_checker.fd)
55    try:
56        wait_for_mount(mount_process, mnt_dir)
57        filename = pjoin(mnt_dir, 'current_time')
58        with open(filename, 'r') as fh:
59            read1 = fh.read()
60        safe_sleep(2)
61        with open(filename, 'r') as fh:
62            read2 = fh.read()
63        if notify:
64            assert read1 != read2
65        else:
66            assert read1 == read2
67    except:
68        cleanup(mount_process, mnt_dir)
69        raise
70    else:
71        umount(mount_process, mnt_dir)
72
73@pytest.mark.skipif(fuse_proto < (7,12),
74                    reason='not supported by running kernel')
75@pytest.mark.parametrize("notify", (True, False))
76def test_notify_file_size(tmpdir, notify, output_checker):
77    mnt_dir = str(tmpdir)
78    cmdline = base_cmdline + \
79              [ pjoin(basename, 'example', 'invalidate_path'),
80                '-f', '--update-interval=1', mnt_dir ]
81    if not notify:
82        cmdline.append('--no-notify')
83    mount_process = subprocess.Popen(cmdline, stdout=output_checker.fd,
84                                     stderr=output_checker.fd)
85    try:
86        wait_for_mount(mount_process, mnt_dir)
87        filename = pjoin(mnt_dir, 'growing')
88        size = os.path.getsize(filename)
89        safe_sleep(2)
90        new_size = os.path.getsize(filename)
91        if notify:
92            assert new_size > size
93        else:
94            assert new_size == size
95    except:
96        cleanup(mount_process, mnt_dir)
97        raise
98    else:
99        umount(mount_process, mnt_dir)
100