fixup! [test] Add unit tests

Author: mxmlnkn

examples/memory.py

@@ -16,7 +16,6 @@ class Memory(mfusepy.LoggingMixIn, mfusepy.Operations):
 
     def __init__(self):
         self.data = collections.defaultdict(bytes)
-        self.fd = 0
         now = time.time()
         self.files: Dict[str, Dict[str, Any]] = {
             '/': {
@@ -46,9 +45,7 @@ def create(self, path, mode, flags=None):
             'st_mtime': time.time(),
             'st_atime': time.time(),
         }
-
-        self.fd += 1
-        return self.fd
+        return 0
 
     def getattr(self, path, fh=None):
         if path not in self.files:
@@ -79,10 +76,6 @@ def mkdir(self, path, mode):
 
         self.files['/']['st_nlink'] += 1
 
-    def open(self, path, flags):
-        self.fd += 1
-        return self.fd
-
     def read(self, path, size, offset, fh):
         return self.data[path][offset : offset + size]
 

examples/memory_nullpath.py

@@ -0,0 +1,190 @@
+#!/usr/bin/env python3
+
+import argparse
+import collections
+import errno
+import logging
+import stat
+import time
+from typing import Dict
+
+import mfusepy
+
+
+class Memory(mfusepy.Operations):
+    'Example memory filesystem. Supports only one level of files.'
+
+    flag_nullpath_ok = True
+    flag_nopath = True
+
+    def __init__(self):
+        self.files = {}
+        self.data = collections.defaultdict(bytes)
+        self.fd = 0
+        now = time.time()
+        self.files['/'] = {
+            'st_mode': (stat.S_IFDIR | 0o755),
+            'st_ctime': now,
+            'st_mtime': now,
+            'st_atime': now,
+            'st_nlink': 2,
+        }
+        self._opened: Dict[int, str] = {}
+
+    def init_with_config(self, conn_info, config_3) -> None:
+        # This only works for FUSE 3 while the flag_nullpath_ok and flag_nopath class members work for FUSE 2 and 3!
+        if config_3:
+            config_3.nullpath_ok = True
+
+    def chmod(self, path, mode):
+        self.files[path]['st_mode'] &= 0o770000
+        self.files[path]['st_mode'] |= mode
+        return 0
+
+    def chown(self, path, uid, gid):
+        self.files[path]['st_uid'] = uid
+        self.files[path]['st_gid'] = gid
+
+    def create(self, path, mode):
+        print("CREATE")
+        self.files[path] = {
+            'st_mode': (stat.S_IFREG | mode),
+            'st_nlink': 1,
+            'st_size': 0,
+            'st_ctime': time.time(),
+            'st_mtime': time.time(),
+            'st_atime': time.time(),
+        }
+
+        self.fd += 1
+        self._opened[self.fd] = path
+        return self.fd
+
+    def getattr(self, path, fh=None):
+        if path not in self.files:
+            raise mfusepy.FuseOSError(errno.ENOENT)
+        return self.files[path]
+
+    def getxattr(self, path, name, position=0):
+        attrs = self.files[path].get('attrs', {})
+
+        try:
+            return attrs[name]
+        except KeyError:
+            return b''
+
+    def listxattr(self, path):
+        attrs = self.files[path].get('attrs', {})
+        return attrs.keys()
+
+    def mkdir(self, path, mode):
+        self.files[path] = {
+            'st_mode': (stat.S_IFDIR | mode),
+            'st_nlink': 2,
+            'st_size': 0,
+            'st_ctime': time.time(),
+            'st_mtime': time.time(),
+            'st_atime': time.time(),
+        }
+
+        self.files['/']['st_nlink'] += 1
+
+    def open(self, path, flags):
+        self.fd += 1
+        self._opened[self.fd] = path
+        return self.fd
+
+    def read(self, path, size, offset, fh):
+        return self.data[self._opened[fh]][offset : offset + size]
+
+    def release(self, path, fh):
+        del self._opened[fh]
+
+    def opendir(self, path):
+        self.fd += 1
+        self._opened[self.fd] = path
+        return self.fd
+
+    def readdir(self, path, fh):
+        print("PATH", path)
+        path = self._opened[fh]
+        return ['.', '..'] + [x[1:] for x in self.files if x.startswith(path) and len(x) > len(path)]
+
+    def releasedir(self, path, fh):
+        del self._opened[fh]
+
+    def readlink(self, path):
+        return self.data[path]
+
+    def removexattr(self, path, name):
+        attrs = self.files[path].get('attrs', {})
+
+        try:
+            del attrs[name]
+        except KeyError:
+            pass
+
+    def rename(self, old, new):
+        if old in self.data:  # Directories have no data.
+            self.data[new] = self.data.pop(old)
+        if old not in self.files:
+            raise mfusepy.FuseOSError(errno.ENOENT)
+        self.files[new] = self.files.pop(old)
+
+    def rmdir(self, path):
+        # with multiple level support, need to raise ENOTEMPTY if contains any files
+        self.files.pop(path)
+        self.files['/']['st_nlink'] -= 1
+
+    def setxattr(self, path, name, value, options, position=0):
+        # Ignore options
+        attrs = self.files[path].setdefault('attrs', {})
+        attrs[name] = value
+
+    def statfs(self, path):
+        return {'f_bsize': 512, 'f_blocks': 4096, 'f_bavail': 2048}
+
+    def symlink(self, target, source):
+        self.files[target] = {'st_mode': (stat.S_IFLNK | 0o777), 'st_nlink': 1, 'st_size': len(source)}
+
+        self.data[target] = source
+
+    def truncate(self, path, length, fh=None):
+        # make sure extending the file fills in zero bytes
+        self.data[path] = self.data[path][:length].ljust(length, '\x00'.encode('ascii'))
+        self.files[path]['st_size'] = length
+
+    def unlink(self, path):
+        self.data.pop(path)
+        self.files.pop(path)
+
+    def utimens(self, path, times=None):
+        now = time.time()
+        atime, mtime = times if times else (now, now)
+        self.files[path]['st_atime'] = atime
+        self.files[path]['st_mtime'] = mtime
+
+    def write(self, path, data, offset, fh):
+        print("[write]", path, data, offset, fh)
+        path = self._opened[fh]
+        self.data[path] = (
+            # make sure the data gets inserted at the right offset
+            self.data[path][:offset].ljust(offset, '\x00'.encode('ascii'))
+            + data
+            # and only overwrites the bytes that data is replacing
+            + self.data[path][offset + len(data) :]
+        )
+        self.files[path]['st_size'] = len(self.data[path])
+        return len(data)
+
+
+def cli(args=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('mount')
+    args = parser.parse_args(args)
+
+    mfusepy.FUSE(Memory(), args.mount, foreground=True)
+
+
+if __name__ == '__main__':
+    cli()

tests/test_memory.py

@@ -8,14 +8,17 @@
 import threading
 import time
 
+import pytest
+
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../examples')))
 
-from memory import cli  # noqa: E402
+from memory import cli as cli_memory  # noqa: E402
+from memory import cli as cli_memory_nullpath  # noqa: E402
 
 
 class RunCLI:
-    def __init__(self, mount_point):
+    def __init__(self, cli, mount_point):
         self.timeout = 4
         self.mount_point = str(mount_point)
         self.args = [self.mount_point]
@@ -108,9 +111,10 @@ def unmount(self):
             time.sleep(0.1)
 
 
-def test_memory_file_system(tmpdir):
+@pytest.mark.parametrize('cli', [cli_memory, cli_memory_nullpath])
+def test_memory_file_system(cli, tmpdir):
     mount_point = tmpdir
-    with RunCLI(mount_point):
+    with RunCLI(cli, mount_point):
         assert os.path.isdir(mount_point)
         assert not os.path.isdir(mount_point / "foo")