diff options
| -rw-r--r-- | snekbox/filesystem.py (renamed from snekbox/libmount.py) | 7 | ||||
| -rw-r--r-- | snekbox/memfs.py | 2 | ||||
| -rw-r--r-- | tests/test_filesystem.py (renamed from tests/test_libmount.py) | 34 |
3 files changed, 25 insertions, 18 deletions
diff --git a/snekbox/libmount.py b/snekbox/filesystem.py index 5300c5c..0cdfbf5 100644 --- a/snekbox/libmount.py +++ b/snekbox/filesystem.py @@ -1,3 +1,4 @@ +"""Mounts and unmounts filesystems.""" from __future__ import annotations import ctypes @@ -48,6 +49,9 @@ def mount(source: Path | str, target: Path | str, fs: str, **options: str | int) target: Target directory. fs: Filesystem type. **options: Mount options. + + Raises: + OSError: On any mount error. """ if Path(target).is_mount(): raise OSError(f"{target} is already a mount point") @@ -71,6 +75,9 @@ def unmount(target: Path | str, flags: UnmountFlags | int = UnmountFlags.MNT_DET Args: target: Target directory. flags: Unmount flags. + + Raises: + OSError: On any unmount error. """ if not Path(target).is_mount(): raise OSError(f"{target} is not a mount point") diff --git a/snekbox/memfs.py b/snekbox/memfs.py index 6a3a635..ebbbb4a 100644 --- a/snekbox/memfs.py +++ b/snekbox/memfs.py @@ -9,7 +9,7 @@ from types import TracebackType from typing import Type from uuid import uuid4 -from snekbox.libmount import mount, unmount +from snekbox.filesystem import mount, unmount from snekbox.snekio import FileAttachment log = logging.getLogger(__name__) diff --git a/tests/test_libmount.py b/tests/test_filesystem.py index d60c6da..b33322c 100644 --- a/tests/test_libmount.py +++ b/tests/test_filesystem.py @@ -5,7 +5,7 @@ from shutil import rmtree from unittest import TestCase from uuid import uuid4 -from snekbox import libmount +from snekbox.filesystem import UnmountFlags, mount, unmount class LibMountTests(TestCase): @@ -25,11 +25,11 @@ class LibMountTests(TestCase): path = self.temp_dir / str(uuid4()) path.mkdir() try: - libmount.mount(source="", target=path, fs="tmpfs") + mount(source="", target=path, fs="tmpfs") yield path finally: with suppress(OSError): - libmount.unmount(path) + unmount(path) def test_mount(self): """Test normal mounting.""" @@ -54,7 +54,7 @@ class LibMountTests(TestCase): for case, err, msg in cases: with self.subTest(case=case): with self.assertRaises(err) as cm: - libmount.mount(**case) + mount(**case) self.assertIn(msg, str(cm.exception)) def test_mount_duplicate(self): @@ -62,31 +62,31 @@ class LibMountTests(TestCase): path = self.temp_dir / str(uuid4()) path.mkdir() try: - libmount.mount(source="", target=path, fs="tmpfs") + mount(source="", target=path, fs="tmpfs") with self.assertRaises(OSError) as cm: - libmount.mount(source="", target=path, fs="tmpfs") + mount(source="", target=path, fs="tmpfs") self.assertIn("already a mount point", str(cm.exception)) finally: - libmount.unmount(target=path) + unmount(target=path) def test_unmount_flags(self): """Test unmount flags.""" flags = [ - libmount.UnmountFlags.MNT_FORCE, - libmount.UnmountFlags.MNT_DETACH, - libmount.UnmountFlags.UMOUNT_NOFOLLOW, + UnmountFlags.MNT_FORCE, + UnmountFlags.MNT_DETACH, + UnmountFlags.UMOUNT_NOFOLLOW, ] for flag in flags: with self.subTest(flag=flag), self.get_mount() as path: self.assertTrue(path.is_mount()) - libmount.unmount(path, flag) + unmount(path, flag) self.assertFalse(path.is_mount()) def test_unmount_flags_expire(self): """Test unmount MNT_EXPIRE behavior.""" with self.get_mount() as path: with self.assertRaises(BlockingIOError): - libmount.unmount(path, libmount.UnmountFlags.MNT_EXPIRE) + unmount(path, UnmountFlags.MNT_EXPIRE) def test_unmount_errors(self): """Test invalid unmount errors.""" @@ -97,14 +97,14 @@ class LibMountTests(TestCase): for case, err, msg in cases: with self.subTest(case=case): with self.assertRaises(err) as cm: - libmount.unmount(**case) + unmount(**case) self.assertIn(msg, str(cm.exception)) def test_unmount_invalid_args(self): """Test invalid unmount invalid flag.""" with self.get_mount() as path: with self.assertRaises(OSError) as cm: - libmount.unmount(path, 251) + unmount(path, 251) self.assertIn("Invalid argument", str(cm.exception)) def test_threading(self): @@ -119,7 +119,7 @@ class LibMountTests(TestCase): with ThreadPoolExecutor() as pool: res = list( pool.map( - libmount.mount, + mount, [""] * len(paths), paths, ["tmpfs"] * len(paths), @@ -131,7 +131,7 @@ class LibMountTests(TestCase): with self.subTest(path=path): self.assertTrue(path.is_mount()) - unmounts = list(pool.map(libmount.unmount, paths)) + unmounts = list(pool.map(unmount, paths)) self.assertEqual(len(unmounts), len(paths)) for path in paths: @@ -140,4 +140,4 @@ class LibMountTests(TestCase): finally: with suppress(OSError): for path in paths: - libmount.unmount(path) + unmount(path) |