aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorGravatar ionite34 <[email protected]>2022-11-28 11:40:23 +0800
committerGravatar ionite34 <[email protected]>2022-11-28 11:40:23 +0800
commite72032fc3090d6e5d5d961a3aafaeec4cce787d7 (patch)
tree79063c704b9c03fda044088a9c50a83d01537a04
parentAdd tests for invalid base64 file content (diff)
Rename `libmount` to `filesystem`
-rw-r--r--snekbox/filesystem.py (renamed from snekbox/libmount.py)7
-rw-r--r--snekbox/memfs.py2
-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)