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
|
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager, suppress
from pathlib import Path
from unittest import TestCase
from uuid import uuid4
from snekbox import libmount
class LibMountTests(TestCase):
def setUp(self):
self.temp_dir = Path("/tmp/snekbox-test")
self.temp_dir.mkdir(exist_ok=True, parents=True)
super().setUp()
@contextmanager
def get_mount(self):
"""Yields a valid mount point, unmounts after context."""
path = self.temp_dir / str(uuid4())
path.mkdir()
try:
libmount.mount(source="", target=path, fs="tmpfs")
yield path
finally:
libmount.unmount(target=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, fs="tmpfs", invalid_opt="?"),
OSError,
"Invalid argument",
),
]
for case, err, msg in cases:
with self.subTest(case=case):
with self.assertRaises(err) as cm:
libmount.mount(**case)
self.assertIn(msg, str(cm.exception))
def test_mount_duplicate(self):
"""Test attempted mount after mounted."""
path = self.temp_dir / str(uuid4())
path.mkdir()
try:
libmount.mount(source="", target=path, fs="tmpfs")
with self.assertRaises(OSError) as cm:
libmount.mount(source="", target=path, fs="tmpfs")
self.assertIn("already a mount point", str(cm.exception))
finally:
libmount.unmount(target=path)
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:
libmount.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)
self.assertIn("Invalid argument", str(cm.exception))
def test_threading(self):
"""Test concurrent mounting works in multi-thread environments."""
paths = [self.temp_dir / 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(
libmount.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(libmount.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:
libmount.unmount(path)
|
