1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager, suppress
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import TestCase
from uuid import uuid4
from snekbox.filesystem import UnmountFlags, mount, unmount
class LibMountTests(TestCase):
temp_dir: TemporaryDirectory
@classmethod
def setUpClass(cls):
cls.temp_dir = TemporaryDirectory(prefix="snekbox_tests")
@classmethod
def tearDownClass(cls):
cls.temp_dir.cleanup()
@contextmanager
def get_mount(self):
"""Yield a valid mount point and unmount after context."""
path = Path(self.temp_dir.name, str(uuid4()))
path.mkdir()
try:
mount(source="", target=path, fs="tmpfs")
yield path
finally:
with suppress(OSError):
unmount(path)
def test_mount(self):
"""Test normal mounting."""
with self.get_mount() as path:
self.assertTrue(path.is_mount())
self.assertTrue(path.exists())
self.assertFalse(path.is_mount())
# Unmounting should not remove the original folder
self.assertTrue(path.exists())
def test_mount_errors(self):
"""Test invalid mount errors."""
cases = [
(dict(source="", target=str(uuid4()), fs="tmpfs"), OSError, "No such file"),
(dict(source=str(uuid4()), target="some/dir", fs="tmpfs"), OSError, "No such file"),
(
dict(source="", target=self.temp_dir.name, fs="tmpfs", invalid_opt="?"),
OSError,
"Invalid argument",
),
]
for case, err, msg in cases:
with self.subTest(case=case):
with self.assertRaises(err) as cm:
mount(**case)
self.assertIn(msg, str(cm.exception))
def test_mount_duplicate(self):
"""Test attempted mount after mounted."""
path = Path(self.temp_dir.name, str(uuid4()))
path.mkdir()
try:
mount(source="", target=path, fs="tmpfs")
with self.assertRaises(OSError) as cm:
mount(source="", target=path, fs="tmpfs")
self.assertIn("already a mount point", str(cm.exception))
finally:
unmount(target=path)
def test_unmount_flags(self):
"""Test unmount flags."""
flags = [
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())
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):
unmount(path, UnmountFlags.MNT_EXPIRE)
def test_unmount_errors(self):
"""Test invalid unmount errors."""
cases = [
(dict(target="not/exist"), OSError, "is not a mount point"),
(dict(target=Path("not/exist")), OSError, "is not a mount point"),
]
for case, err, msg in cases:
with self.subTest(case=case):
with self.assertRaises(err) as cm:
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:
unmount(path, 251)
self.assertIn("Invalid argument", str(cm.exception))
def test_threading(self):
"""Test concurrent mounting works in multi-thread environments."""
paths = [Path(self.temp_dir.name, str(uuid4())) for _ in range(16)]
for path in paths:
path.mkdir()
self.assertFalse(path.is_mount())
try:
with ThreadPoolExecutor() as pool:
res = list(
pool.map(
mount,
[""] * len(paths),
paths,
["tmpfs"] * len(paths),
)
)
self.assertEqual(len(res), len(paths))
for path in paths:
with self.subTest(path=path):
self.assertTrue(path.is_mount())
unmounts = list(pool.map(unmount, paths))
self.assertEqual(len(unmounts), len(paths))
for path in paths:
with self.subTest(path=path):
self.assertFalse(path.is_mount())
finally:
with suppress(OSError):
for path in paths:
unmount(path)
|
