diff options
Diffstat (limited to 'test/test_examples.py')
| -rwxr-xr-x | test/test_examples.py | 106 |
1 files changed, 73 insertions, 33 deletions
diff --git a/test/test_examples.py b/test/test_examples.py index 31d2fae..5ddc860 100755 --- a/test/test_examples.py +++ b/test/test_examples.py @@ -14,9 +14,10 @@ import shutil import filecmp import errno from tempfile import NamedTemporaryFile -from util import wait_for_mount, umount, cleanup +from util import wait_for_mount, umount, cleanup, base_cmdline +from os.path import join as pjoin -basename = os.path.join(os.path.dirname(__file__), '..') +basename = pjoin(os.path.dirname(__file__), '..') TEST_FILE = __file__ with open(TEST_FILE, 'rb') as fh: @@ -29,13 +30,17 @@ def name_generator(__ctr=[0]): @pytest.mark.parametrize("name", ('hello', 'hello_ll')) def test_hello(tmpdir, name): mnt_dir = str(tmpdir) - cmdline = [os.path.join(basename, 'example', name), - '-f', mnt_dir ] + cmdline = base_cmdline + \ + [ pjoin(basename, 'example', name), + '-f', mnt_dir ] + if name == 'hello_ll': + # supports single-threading only + cmdline.append('-s') mount_process = subprocess.Popen(cmdline) try: wait_for_mount(mount_process, mnt_dir) assert os.listdir(mnt_dir) == [ 'hello' ] - filename = os.path.join(mnt_dir, 'hello') + filename = pjoin(mnt_dir, 'hello') with open(filename, 'r') as fh: assert fh.read() == 'Hello World!\n' with pytest.raises(IOError) as exc_info: @@ -50,18 +55,51 @@ def test_hello(tmpdir, name): else: umount(mount_process, mnt_dir) +def test_fuse_lo_plus(tmpdir): + mnt_dir = str(tmpdir.mkdir('mnt')) + src_dir = str(tmpdir.mkdir('src')) + + cmdline = base_cmdline + \ + [ pjoin(basename, 'example', 'fuse_lo-plus'), + '-f', '-s', mnt_dir ] + mount_process = subprocess.Popen(cmdline) + try: + wait_for_mount(mount_process, mnt_dir) + work_dir = pjoin(mnt_dir, src_dir) + tst_write(work_dir) + tst_mkdir(work_dir) + tst_symlink(work_dir) + tst_mknod(work_dir) + if os.getuid() == 0: + tst_chown(work_dir) + # Underlying fs may not have full nanosecond resolution + tst_utimens(work_dir, ns_tol=1000) + tst_link(work_dir) + tst_readdir(work_dir) + tst_statvfs(work_dir) + tst_truncate_path(work_dir) + tst_truncate_fd(work_dir) + tst_unlink(work_dir) + tst_passthrough(src_dir, work_dir) + except: + cleanup(mnt_dir) + raise + else: + umount(mount_process, mnt_dir) + @pytest.mark.parametrize("name", ('fusexmp', 'fusexmp_fh')) def test_fusexmp_fh(tmpdir, name): mnt_dir = str(tmpdir.mkdir('mnt')) src_dir = str(tmpdir.mkdir('src')) - cmdline = [os.path.join(basename, 'example', name), - '-f', '-o' , 'use_ino,readdir_ino,kernel_cache', + cmdline = base_cmdline + \ + [ pjoin(basename, 'example', name), + '-f', '-o', 'use_ino,readdir_ino,kernel_cache', mnt_dir ] mount_process = subprocess.Popen(cmdline) try: wait_for_mount(mount_process, mnt_dir) - work_dir = os.path.join(mnt_dir, src_dir) + work_dir = pjoin(mnt_dir, src_dir) tst_write(work_dir) tst_mkdir(work_dir) tst_symlink(work_dir) @@ -85,20 +123,21 @@ def test_fusexmp_fh(tmpdir, name): def test_fioc(tmpdir): mnt_dir = str(tmpdir) - testfile = os.path.join(mnt_dir, 'fioc') - cmdline = [os.path.join(basename, 'example', 'fioc'), - '-f', mnt_dir ] + testfile = pjoin(mnt_dir, 'fioc') + cmdline = base_cmdline + \ + [pjoin(basename, 'example', 'fioc'), '-f', mnt_dir ] mount_process = subprocess.Popen(cmdline) try: wait_for_mount(mount_process, mnt_dir) - base_cmd = [ os.path.join(basename, 'example', 'fioclient'), - testfile ] - assert subprocess.check_output(base_cmd) == b'0\n' + cmdline = base_cmdline + \ + [ pjoin(basename, 'example', 'fioclient'), + testfile ] + assert subprocess.check_output(cmdline) == b'0\n' with open(testfile, 'wb') as fh: fh.write(b'foobar') - assert subprocess.check_output(base_cmd) == b'6\n' - subprocess.check_call(base_cmd + [ '3' ]) + assert subprocess.check_output(cmdline) == b'6\n' + subprocess.check_call(cmdline + [ '3' ]) with open(testfile, 'rb') as fh: assert fh.read()== b'foo' except: @@ -109,12 +148,13 @@ def test_fioc(tmpdir): def test_fsel(tmpdir): mnt_dir = str(tmpdir) - cmdline = [os.path.join(basename, 'example', 'fsel'), + cmdline = base_cmdline + [pjoin(basename, 'example', 'fsel'), '-f', mnt_dir ] mount_process = subprocess.Popen(cmdline) try: wait_for_mount(mount_process, mnt_dir) - cmdline = [ os.path.join(basename, 'example', 'fselclient') ] + cmdline = base_cmdline + \ + [ pjoin(basename, 'example', 'fselclient') ] subprocess.check_call(cmdline, cwd=mnt_dir) except: cleanup(mnt_dir) @@ -123,7 +163,7 @@ def test_fsel(tmpdir): umount(mount_process, mnt_dir) def checked_unlink(filename, path, isdir=False): - fullname = os.path.join(path, filename) + fullname = pjoin(path, filename) if isdir: os.rmdir(fullname) else: @@ -156,7 +196,7 @@ def tst_symlink(mnt_dir): checked_unlink(linkname, mnt_dir) def tst_mknod(mnt_dir): - filename = os.path.join(mnt_dir, name_generator()) + filename = pjoin(mnt_dir, name_generator()) shutil.copyfile(TEST_FILE, filename) fstat = os.lstat(filename) assert stat.S_ISREG(fstat.st_mode) @@ -166,7 +206,7 @@ def tst_mknod(mnt_dir): checked_unlink(filename, mnt_dir) def tst_chown(mnt_dir): - filename = os.path.join(mnt_dir, name_generator()) + filename = pjoin(mnt_dir, name_generator()) os.mkdir(filename) fstat = os.lstat(filename) uid = fstat.st_uid @@ -187,17 +227,17 @@ def tst_chown(mnt_dir): checked_unlink(filename, mnt_dir, isdir=True) def tst_write(mnt_dir): - name = os.path.join(mnt_dir, name_generator()) + name = pjoin(mnt_dir, name_generator()) shutil.copyfile(TEST_FILE, name) assert filecmp.cmp(name, TEST_FILE, False) checked_unlink(name, mnt_dir) def tst_unlink(mnt_dir): - name = os.path.join(mnt_dir, name_generator()) + name = pjoin(mnt_dir, name_generator()) data1 = b'foo' data2 = b'bar' - with open(os.path.join(mnt_dir, name), 'wb+', buffering=0) as fh: + with open(pjoin(mnt_dir, name), 'wb+', buffering=0) as fh: fh.write(data1) checked_unlink(name, mnt_dir) fh.write(data2) @@ -208,8 +248,8 @@ def tst_statvfs(mnt_dir): os.statvfs(mnt_dir) def tst_link(mnt_dir): - name1 = os.path.join(mnt_dir, name_generator()) - name2 = os.path.join(mnt_dir, name_generator()) + name1 = pjoin(mnt_dir, name_generator()) + name2 = pjoin(mnt_dir, name_generator()) shutil.copyfile(TEST_FILE, name1) assert filecmp.cmp(name1, TEST_FILE, False) os.link(name1, name2) @@ -228,7 +268,7 @@ def tst_link(mnt_dir): os.unlink(name1) def tst_readdir(mnt_dir): - dir_ = os.path.join(mnt_dir, name_generator()) + dir_ = pjoin(mnt_dir, name_generator()) file_ = dir_ + "/" + name_generator() subdir = dir_ + "/" + name_generator() subfile = subdir + "/" + name_generator() @@ -252,7 +292,7 @@ def tst_readdir(mnt_dir): def tst_truncate_path(mnt_dir): assert len(TEST_DATA) > 1024 - filename = os.path.join(mnt_dir, name_generator()) + filename = pjoin(mnt_dir, name_generator()) with open(filename, 'wb') as fh: fh.write(TEST_DATA) @@ -298,7 +338,7 @@ def tst_truncate_fd(mnt_dir): assert fh.read(size) == TEST_DATA[:size-1024] def tst_utimens(mnt_dir, ns_tol=0): - filename = os.path.join(mnt_dir, name_generator()) + filename = pjoin(mnt_dir, name_generator()) os.mkdir(filename) fstat = os.lstat(filename) @@ -323,8 +363,8 @@ def tst_utimens(mnt_dir, ns_tol=0): def tst_passthrough(src_dir, mnt_dir): name = name_generator() - src_name = os.path.join(src_dir, name) - mnt_name = os.path.join(src_dir, name) + src_name = pjoin(src_dir, name) + mnt_name = pjoin(src_dir, name) assert name not in os.listdir(src_dir) assert name not in os.listdir(mnt_dir) with open(src_name, 'w') as fh: @@ -334,8 +374,8 @@ def tst_passthrough(src_dir, mnt_dir): assert os.stat(src_name) == os.stat(mnt_name) name = name_generator() - src_name = os.path.join(src_dir, name) - mnt_name = os.path.join(src_dir, name) + src_name = pjoin(src_dir, name) + mnt_name = pjoin(src_dir, name) assert name not in os.listdir(src_dir) assert name not in os.listdir(mnt_dir) with open(mnt_name, 'w') as fh: |