[test] Test with loopback filesystem

mxmlnkn · mxmlnkn · commit 2fa8bc90d860 · 2025-07-08T23:32:54.000+02:00
diff --git a/examples/ioctl.py b/examples/ioctl.py
@@ -11,10 +11,10 @@
 
 from ioctl_opt import IOWR
 
-import mfusepy
+import mfusepy as fuse
 
 
-class Ioctl(mfusepy.LoggingMixIn, mfusepy.Operations):
+class Ioctl(fuse.Operations):
     '''
     Example filesystem based on memory.py to demonstrate ioctl().
 
@@ -57,7 +57,7 @@ def create(self, path, mode, fi=None):
 
     def getattr(self, path, fh=None):
         if path not in self.files:
-            raise mfusepy.FuseOSError(errno.ENOENT)
+            raise fuse.FuseOSError(errno.ENOENT)
 
         return self.files[path]
 
@@ -71,7 +71,7 @@ def ioctl(self, path, cmd, arg, fh, flags, data):
             outbuf = struct.pack('<I', data_out)
             ctypes.memmove(data, outbuf, 4)
         else:
-            raise mfusepy.FuseOSError(errno.ENOTTY)
+            raise fuse.FuseOSError(errno.ENOTTY)
         return 0
 
     def open(self, path, flags):
@@ -91,7 +91,7 @@ def cli(args=None):
     args = parser.parse_args(args)
 
     logging.basicConfig(level=logging.DEBUG)
-    mfusepy.FUSE(Ioctl(), args.mount, foreground=True)
+    fuse.FUSE(Ioctl(), args.mount, foreground=True)
 
 
 if __name__ == '__main__':
diff --git a/examples/loopback.py b/examples/loopback.py
@@ -7,10 +7,10 @@
 import threading
 from os.path import realpath
 
-import mfusepy
+import mfusepy as fuse
 
 
-class Loopback(mfusepy.LoggingMixIn, mfusepy.Operations):
+class Loopback(fuse.LoggingMixIn, fuse.Operations):
     def __init__(self, root):
         self.root = realpath(root)
         self.rwlock = threading.Lock()
@@ -20,7 +20,7 @@ def __call__(self, op, path, *args):
 
     def access(self, path, amode):
         if not os.access(path, amode):
-            raise mfusepy.FuseOSError(errno.EACCES)
+            raise fuse.FuseOSError(errno.EACCES)
 
     chmod = os.chmod
     chown = os.chown
@@ -112,7 +112,7 @@ def cli(args=None):
     args = parser.parse_args(args)
 
     logging.basicConfig(level=logging.DEBUG)
-    mfusepy.FUSE(Loopback(args.root), args.mount, foreground=True)
+    fuse.FUSE(Loopback(args.root), args.mount, foreground=True)
 
 
 if __name__ == '__main__':
diff --git a/examples/memory.py b/examples/memory.py
@@ -8,10 +8,11 @@
 import time
 from typing import Any, Dict
 
-import mfusepy
+import mfusepy as fuse
 
 
-class Memory(mfusepy.LoggingMixIn, mfusepy.Operations):
+# LoggingMixIn should not be used anymore! This is only here for downwards compatibility tests.
+class Memory(fuse.LoggingMixIn, fuse.Operations):
     'Example memory filesystem. Supports only one level of files.'
 
     def __init__(self):
@@ -49,7 +50,7 @@ def create(self, path, mode, fi=None):
 
     def getattr(self, path, fh=None):
         if path not in self.files:
-            raise mfusepy.FuseOSError(errno.ENOENT)
+            raise fuse.FuseOSError(errno.ENOENT)
         return self.files[path]
 
     def getxattr(self, path, name, position=0):
@@ -97,7 +98,7 @@ 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)
+            raise fuse.FuseOSError(errno.ENOENT)
         self.files[new] = self.files.pop(old)
 
     def rmdir(self, path):
@@ -151,7 +152,7 @@ def cli(args=None):
     args = parser.parse_args(args)
 
     logging.basicConfig(level=logging.DEBUG)
-    mfusepy.FUSE(Memory(), args.mount, foreground=True)
+    fuse.FUSE(Memory(), args.mount, foreground=True)
 
 
 if __name__ == '__main__':
diff --git a/examples/memory_nullpath.py b/examples/memory_nullpath.py
@@ -7,10 +7,10 @@
 import time
 from typing import Any, Dict
 
-import mfusepy
+import mfusepy as fuse
 
 
-class Memory(mfusepy.Operations):
+class Memory(fuse.Operations):
     'Example memory filesystem. Supports only one level of files.'
 
     flag_nullpath_ok = True
@@ -64,7 +64,7 @@ def getattr(self, path, fh=None):
         if fh is not None and fh in self._opened:
             path = self._opened[fh]
         if path not in self.files:
-            raise mfusepy.FuseOSError(errno.ENOENT)
+            raise fuse.FuseOSError(errno.ENOENT)
         return self.files[path]
 
     def getxattr(self, path, name, position=0):
@@ -129,7 +129,7 @@ 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)
+            raise fuse.FuseOSError(errno.ENOENT)
         self.files[new] = self.files.pop(old)
 
     def rmdir(self, path):
@@ -185,7 +185,7 @@ def cli(args=None):
     parser.add_argument('mount')
     args = parser.parse_args(args)
 
-    mfusepy.FUSE(Memory(), args.mount, foreground=True)
+    fuse.FUSE(Memory(), args.mount, foreground=True)
 
 
 if __name__ == '__main__':
diff --git a/tests/.pylintrc b/tests/.pylintrc
@@ -16,16 +16,21 @@ max-line-length=120
 disable=broad-except,
         invalid-name,
         too-many-arguments,
+        too-many-branches,
         too-many-instance-attributes,
         too-many-locals,
         too-many-lines,
         too-many-positional-arguments,
         too-many-public-methods,
+        too-many-statements,
         too-few-public-methods,
         # I don't need the style checker to bother me with missing docstrings and todos.
         missing-class-docstring,
         missing-function-docstring,
         missing-module-docstring,
         fixme,
         unused-argument,
-        R0801
+        comparison-with-callable,
+        R0801,
+        global-statement,
+        chained-comparison,  # Only Since Python 3.8
diff --git a/tests/.ruff.toml b/tests/.ruff.toml
@@ -9,37 +9,22 @@ select = [
 ignore = [
     # Preview. Complaining about spaces (aligned arguments) should be a formatter option, not a linter one!
     # https://github.com/astral-sh/ruff/issues/2402
-    "E201", "E202", "E203", "E211", "E221", "E226", "E251", "E265", "E266", "E271",
+    "E201", "E202", "E203", "E211", "E221", "E226", "E251", "E265", "E266",
     "E501",     # A linter should lint, not check for line lengths!
     "F401",     # Wants to from .version import __version__ as __version__ which clashes with pylint errors!
     "B904",
     "N801",     # Some class names are snake_case to match ctypes
-    "N803",     # Argument names are camelCase instead of snake_case (400+ errors).
     "N806",     # Variable names are camelCase instead of snake_case (1100+ errors).
-    "N815",     # Class-scope variable names are camelCase instead of snake_case (10 errors).
-    "N816",     # Global variable names are camelCase instead of snake_case (73 errors).
-    "N999",     # Module names are CamelCase like the classes they provide instead of snake_case (11 errors).
     "COM812",   # Do not force trailing commas where it makes no sense, e.g., function calls for which I'll
                 # never add more arguments.
-    "PERF203",  # Some code parts HAVE to try-except inside loops because only the element should be skipped,
-                # not the whole loop being broken out of. Furthermore, this is a useless micro-optimization,
-                # which is actually removed for Python 3.11+ which introduces zero-cost exceptions.
     "PLW0603",  # Cannot live without global statements, especially for subprocessing.
-    "PLW2901",  # Only false positives for constructs such as for data in ...: data = data.clean(); process(data)
+    "PT017",
     "RET504",   # https://github.com/astral-sh/ruff/issues/17292#issuecomment-3039232890
-    "RUF001",   # Only false positives from tests with Unicode characters.
-    "RUF039",   # https://github.com/astral-sh/ruff/issues/18795
-    "RUF056",   # "options.get('disableUnionMount', False):" Wants me to change False to None,
-                # but it makes no sense semantically (expecting a bool)
     "RUF100",   # BUG: removes necessary noqa: E402 in tests!
     "RUF012",   # Mutable class attributes should be annotated with `typing.ClassVar`, but _fields_ is not mutable.
 
-    "SIM102",
+    "SIM102",   # Sometimes, nested if-statements are more readable.
     "SIM105",
-    "SIM115",   # Too many locations where I do not know how to use context managers for files,
-                # which are stored as object members for example!
-    "TC006",    # Wants to quote cast("IO[bytes]", ...) I don't agree with this style choice.
-    "S101", "S110", "S105", "S311", "S324", "S603", "S607", "S608"
 
     # Bug: SIM118 removes the keys() from row.keys(), which is an sqlite3.Row not a Dict!
 ]
diff --git a/tests/test_memory.py b/tests/test_memory.py
@@ -1,6 +1,7 @@
 # pylint: disable=wrong-import-position
 # pylint: disable=protected-access
 
+import errno
 import io
 import os
 import subprocess
@@ -13,15 +14,16 @@
 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 loopback import cli as cli_loopback  # noqa: E402
 from memory import cli as cli_memory  # noqa: E402
 from memory_nullpath import cli as cli_memory_nullpath  # noqa: E402
 
 
 class RunCLI:
-    def __init__(self, cli, mount_point):
+    def __init__(self, cli, mount_point, arguments):
         self.timeout = 4
         self.mount_point = str(mount_point)
-        self.args = [self.mount_point]
+        self.args = [*arguments, self.mount_point]
         self.thread = threading.Thread(target=cli, args=(self.args,))
 
         self._stdout = None
@@ -49,8 +51,8 @@ def __exit__(self, exception_type, exception_value, exception_traceback):
             output = stdout.read()
             errors = stderr.read()
 
-            problematicWords = ['[Warning]', '[Error]']
-            if any(word in output or word in errors for word in problematicWords):
+            problematic_words = ['[Warning]', '[Error]']
+            if any(word in output or word in errors for word in problematic_words):
                 print("===== stdout =====\n", output)
                 print("===== stderr =====\n", errors)
                 raise AssertionError("There were warnings or errors!")
@@ -60,20 +62,20 @@ def __exit__(self, exception_type, exception_value, exception_traceback):
             self.thread.join(self.timeout)
 
     def get_stdout(self):
-        oldPosition = sys.stdout.tell()
+        old_position = sys.stdout.tell()
         try:
             sys.stdout.seek(0)
             return sys.stdout.read()
         finally:
-            sys.stdout.seek(oldPosition)
+            sys.stdout.seek(old_position)
 
     def get_stderr(self):
-        oldPosition = sys.stderr.tell()
+        old_position = sys.stderr.tell()
         try:
             sys.stderr.seek(0)
             return sys.stderr.read()
         finally:
-            sys.stderr.seek(oldPosition)
+            sys.stderr.seek(old_position)
 
     def wait_for_mount_point(self):
         t0 = time.time()
@@ -111,43 +113,60 @@ def unmount(self):
             time.sleep(0.1)
 
 
-@pytest.mark.parametrize('cli', [cli_memory, cli_memory_nullpath])
-def test_memory_file_system(cli, tmpdir):
-    mount_point = tmpdir
-    with RunCLI(cli, mount_point):
+@pytest.mark.parametrize('cli', [cli_loopback, cli_memory, cli_memory_nullpath])
+def test_read_write_file_system(cli, tmp_path):
+    if cli == cli_loopback:
+        mount_source = tmp_path / "folder"
+        mount_point = tmp_path / "mounted"
+        mount_source.mkdir()
+        mount_point.mkdir()
+        arguments = [str(mount_source)]
+    else:
+        mount_point = tmp_path
+        arguments = []
+    with RunCLI(cli, mount_point, arguments):
         assert os.path.isdir(mount_point)
         assert not os.path.isdir(mount_point / "foo")
 
         path = mount_point / "foo"
-        with open(path, 'wb') as file:
-            assert file.write(b"bar") == 3
+        assert path.write_bytes(b"bar") == 3
 
         assert os.path.exists(path)
         assert os.path.isfile(path)
         assert not os.path.isdir(path)
 
-        with open(path, 'rb') as file:
-            assert file.read() == b"bar"
+        assert path.read_bytes() == b"bar"
 
         os.truncate(path, 2)
-        with open(path, 'rb') as file:
-            assert file.read() == b"ba"
+        assert path.read_bytes() == b"ba"
 
         os.chmod(path, 0)
         assert os.stat(path).st_mode & 0o777 == 0
         os.chmod(path, 0o777)
         assert os.stat(path).st_mode & 0o777 == 0o777
 
-        os.chown(path, 12345, 23456)
-        assert os.stat(path).st_uid == 12345
-        assert os.stat(path).st_gid == 23456
+        try:
+            # Only works for memory file systems.
+            os.chown(path, 12345, 23456)
+            assert os.stat(path).st_uid == 12345
+            assert os.stat(path).st_gid == 23456
+        except PermissionError:
+            assert cli == cli_loopback
+
+        os.chown(path, os.getuid(), os.getgid())
+        assert os.stat(path).st_uid == os.getuid()
+        assert os.stat(path).st_gid == os.getgid()
 
-        assert not os.listxattr(path)
-        os.setxattr(path, b"user.tag-test", b"FOO-RESULT")
-        assert os.listxattr(path)
-        assert os.getxattr(path, b"user.tag-test") == b"FOO-RESULT"
-        os.removexattr(path, b"user.tag-test")
-        assert not os.listxattr(path)
+        try:
+            assert not os.listxattr(path)
+            os.setxattr(path, b"user.tag-test", b"FOO-RESULT")
+            assert os.listxattr(path)
+            assert os.getxattr(path, b"user.tag-test") == b"FOO-RESULT"
+            os.removexattr(path, b"user.tag-test")
+            assert not os.listxattr(path)
+        except OSError as exception:
+            assert cli == cli_loopback
+            assert exception.errno == errno.ENOTSUP
 
         os.utime(path, (1.5, 12.5))
         assert os.stat(path).st_atime == 1.5
@@ -178,10 +197,11 @@ def test_memory_file_system(cli, tmpdir):
         # assert os.path.isfile(path)  # Does not have a follow_symlink argument but it seems to be True, see below.
         assert os.path.isdir(path)
         assert os.path.islink(path)
-        assert os.readlink(path) == mount_point / "bar"
+        assert os.readlink(path) == str(mount_point / "bar")
 
         os.rmdir(mount_point / "bar")
         assert not os.path.exists(mount_point / "bar")
 
-        assert os.statvfs(mount_point).f_bsize == 512
-        assert os.statvfs(mount_point).f_bavail == 2048
+        if cli != cli_loopback:
+            assert os.statvfs(mount_point).f_bsize == 512
+            assert os.statvfs(mount_point).f_bavail == 2048
