1 # Copyright (C) 2018 Facebook
3 # This file is part of libbtrfsutil.
5 # libbtrfsutil is free software: you can redistribute it and/or modify
6 # it under the terms of the GNU Lesser General Public License as published by
7 # the Free Software Foundation, either version 3 of the License, or
8 # (at your option) any later version.
10 # libbtrfsutil is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU Lesser General Public License for more details.
15 # You should have received a copy of the GNU Lesser General Public License
16 # along with libbtrfsutil. If not, see <http://www.gnu.org/licenses/>.
22 from pathlib import PurePath
26 from tests import BtrfsTestCase, HAVE_PATH_LIKE
29 class TestSubvolume(BtrfsTestCase):
30 def test_is_subvolume(self):
31 dir = os.path.join(self.mountpoint, 'foo')
34 for arg in self.path_or_fd(self.mountpoint):
35 with self.subTest(type=type(arg)):
36 self.assertTrue(btrfsutil.is_subvolume(arg))
37 for arg in self.path_or_fd(dir):
38 with self.subTest(type=type(arg)):
39 self.assertFalse(btrfsutil.is_subvolume(arg))
41 with self.assertRaises(btrfsutil.BtrfsUtilError) as e:
42 btrfsutil.is_subvolume(os.path.join(self.mountpoint, 'bar'))
43 # This is a bit of an implementation detail, but really this is testing
44 # that the exception is initialized correctly.
45 self.assertEqual(e.exception.btrfsutilerror, btrfsutil.ERROR_STATFS_FAILED)
46 self.assertEqual(e.exception.errno, errno.ENOENT)
48 def test_subvolume_id(self):
49 dir = os.path.join(self.mountpoint, 'foo')
52 for arg in self.path_or_fd(self.mountpoint):
53 with self.subTest(type=type(arg)):
54 self.assertEqual(btrfsutil.subvolume_id(arg), 5)
55 for arg in self.path_or_fd(dir):
56 with self.subTest(type=type(arg)):
57 self.assertEqual(btrfsutil.subvolume_id(arg), 5)
59 def test_subvolume_path(self):
60 btrfsutil.create_subvolume(os.path.join(self.mountpoint, 'subvol1'))
61 os.mkdir(os.path.join(self.mountpoint, 'dir1'))
62 os.mkdir(os.path.join(self.mountpoint, 'dir1/dir2'))
63 btrfsutil.create_subvolume(os.path.join(self.mountpoint, 'dir1/dir2/subvol2'))
64 btrfsutil.create_subvolume(os.path.join(self.mountpoint, 'dir1/dir2/subvol2/subvol3'))
65 os.mkdir(os.path.join(self.mountpoint, 'subvol1/dir3'))
66 btrfsutil.create_subvolume(os.path.join(self.mountpoint, 'subvol1/dir3/subvol4'))
68 for arg in self.path_or_fd(self.mountpoint):
69 with self.subTest(type=type(arg)):
70 self.assertEqual(btrfsutil.subvolume_path(arg), '')
71 self.assertEqual(btrfsutil.subvolume_path(arg, 5), '')
72 self.assertEqual(btrfsutil.subvolume_path(arg, 256), 'subvol1')
73 self.assertEqual(btrfsutil.subvolume_path(arg, 257), 'dir1/dir2/subvol2')
74 self.assertEqual(btrfsutil.subvolume_path(arg, 258), 'dir1/dir2/subvol2/subvol3')
75 self.assertEqual(btrfsutil.subvolume_path(arg, 259), 'subvol1/dir3/subvol4')
79 os.chdir(self.mountpoint)
82 name = chr(ord('a') + i) * 255
83 path = os.path.join(path, name)
84 btrfsutil.create_subvolume(name)
86 self.assertEqual(btrfsutil.subvolume_path('.'), path)
90 def test_subvolume_info(self):
91 for arg in self.path_or_fd(self.mountpoint):
92 with self.subTest(type=type(arg)):
93 info = btrfsutil.subvolume_info(arg)
94 self.assertEqual(info.id, 5)
95 self.assertEqual(info.parent_id, 0)
96 self.assertEqual(info.dir_id, 0)
97 self.assertEqual(info.flags, 0)
98 self.assertEqual(info.uuid, bytes(16))
99 self.assertEqual(info.parent_uuid, bytes(16))
100 self.assertEqual(info.received_uuid, bytes(16))
101 self.assertNotEqual(info.generation, 0)
102 self.assertEqual(info.ctransid, 0)
103 self.assertEqual(info.otransid, 0)
104 self.assertEqual(info.stransid, 0)
105 self.assertEqual(info.rtransid, 0)
106 self.assertEqual(info.ctime, 0)
107 self.assertEqual(info.otime, 0)
108 self.assertEqual(info.stime, 0)
109 self.assertEqual(info.rtime, 0)
111 subvol = os.path.join(self.mountpoint, 'subvol')
112 btrfsutil.create_subvolume(subvol)
114 info = btrfsutil.subvolume_info(subvol)
115 self.assertEqual(info.id, 256)
116 self.assertEqual(info.parent_id, 5)
117 self.assertEqual(info.dir_id, 256)
118 self.assertEqual(info.flags, 0)
119 self.assertIsInstance(info.uuid, bytes)
120 self.assertEqual(info.parent_uuid, bytes(16))
121 self.assertEqual(info.received_uuid, bytes(16))
122 self.assertNotEqual(info.generation, 0)
123 self.assertNotEqual(info.ctransid, 0)
124 self.assertNotEqual(info.otransid, 0)
125 self.assertEqual(info.stransid, 0)
126 self.assertEqual(info.rtransid, 0)
127 self.assertNotEqual(info.ctime, 0)
128 self.assertNotEqual(info.otime, 0)
129 self.assertEqual(info.stime, 0)
130 self.assertEqual(info.rtime, 0)
132 subvol_uuid = info.uuid
133 snapshot = os.path.join(self.mountpoint, 'snapshot')
134 btrfsutil.create_snapshot(subvol, snapshot)
136 info = btrfsutil.subvolume_info(snapshot)
137 self.assertEqual(info.parent_uuid, subvol_uuid)
139 # TODO: test received_uuid, stransid, rtransid, stime, and rtime
141 for arg in self.path_or_fd(self.mountpoint):
142 with self.subTest(type=type(arg)):
143 with self.assertRaises(btrfsutil.BtrfsUtilError) as e:
144 # BTRFS_EXTENT_TREE_OBJECTID
145 btrfsutil.subvolume_info(arg, 2)
147 def test_read_only(self):
148 for arg in self.path_or_fd(self.mountpoint):
149 with self.subTest(type=type(arg)):
150 btrfsutil.set_subvolume_read_only(arg)
151 self.assertTrue(btrfsutil.get_subvolume_read_only(arg))
152 self.assertTrue(btrfsutil.subvolume_info(arg).flags & 1)
154 btrfsutil.set_subvolume_read_only(arg, False)
155 self.assertFalse(btrfsutil.get_subvolume_read_only(arg))
156 self.assertFalse(btrfsutil.subvolume_info(arg).flags & 1)
158 btrfsutil.set_subvolume_read_only(arg, True)
159 self.assertTrue(btrfsutil.get_subvolume_read_only(arg))
160 self.assertTrue(btrfsutil.subvolume_info(arg).flags & 1)
162 btrfsutil.set_subvolume_read_only(arg, False)
164 def test_default_subvolume(self):
165 for arg in self.path_or_fd(self.mountpoint):
166 with self.subTest(type=type(arg)):
167 self.assertEqual(btrfsutil.get_default_subvolume(arg), 5)
169 subvol = os.path.join(self.mountpoint, 'subvol')
170 btrfsutil.create_subvolume(subvol)
171 for arg in self.path_or_fd(subvol):
172 with self.subTest(type=type(arg)):
173 btrfsutil.set_default_subvolume(arg)
174 self.assertEqual(btrfsutil.get_default_subvolume(arg), 256)
175 btrfsutil.set_default_subvolume(arg, 5)
176 self.assertEqual(btrfsutil.get_default_subvolume(arg), 5)
178 def test_create_subvolume(self):
179 subvol = os.path.join(self.mountpoint, 'subvol')
181 btrfsutil.create_subvolume(subvol + '1')
182 self.assertTrue(btrfsutil.is_subvolume(subvol + '1'))
183 btrfsutil.create_subvolume((subvol + '2').encode())
184 self.assertTrue(btrfsutil.is_subvolume(subvol + '2'))
186 btrfsutil.create_subvolume(PurePath(subvol + '3'))
187 self.assertTrue(btrfsutil.is_subvolume(subvol + '3'))
191 os.chdir(self.mountpoint)
192 btrfsutil.create_subvolume('subvol4')
193 self.assertTrue(btrfsutil.is_subvolume('subvol4'))
197 btrfsutil.create_subvolume(subvol + '5/')
198 self.assertTrue(btrfsutil.is_subvolume(subvol + '5'))
200 btrfsutil.create_subvolume(subvol + '6//')
201 self.assertTrue(btrfsutil.is_subvolume(subvol + '6'))
203 transid = btrfsutil.create_subvolume(subvol + '7', async=True)
204 self.assertTrue(btrfsutil.is_subvolume(subvol + '7'))
205 self.assertGreater(transid, 0)
207 # Test creating subvolumes under '/' in a chroot.
211 os.chroot(self.mountpoint)
213 btrfsutil.create_subvolume('/subvol8')
214 self.assertTrue(btrfsutil.is_subvolume('/subvol8'))
215 with self.assertRaises(btrfsutil.BtrfsUtilError):
216 btrfsutil.create_subvolume('/')
219 traceback.print_exc()
221 wstatus = os.waitpid(pid, 0)[1]
222 self.assertTrue(os.WIFEXITED(wstatus))
223 self.assertEqual(os.WEXITSTATUS(wstatus), 0)
225 def test_create_snapshot(self):
226 subvol = os.path.join(self.mountpoint, 'subvol')
228 btrfsutil.create_subvolume(subvol)
229 os.mkdir(os.path.join(subvol, 'dir'))
231 for i, arg in enumerate(self.path_or_fd(subvol)):
232 with self.subTest(type=type(arg)):
233 snapshots_dir = os.path.join(self.mountpoint, 'snapshots{}'.format(i))
234 os.mkdir(snapshots_dir)
235 snapshot = os.path.join(snapshots_dir, 'snapshot')
237 btrfsutil.create_snapshot(subvol, snapshot + '1')
238 self.assertTrue(btrfsutil.is_subvolume(snapshot + '1'))
239 self.assertTrue(os.path.exists(os.path.join(snapshot + '1', 'dir')))
241 btrfsutil.create_snapshot(subvol, (snapshot + '2').encode())
242 self.assertTrue(btrfsutil.is_subvolume(snapshot + '2'))
243 self.assertTrue(os.path.exists(os.path.join(snapshot + '2', 'dir')))
246 btrfsutil.create_snapshot(subvol, PurePath(snapshot + '3'))
247 self.assertTrue(btrfsutil.is_subvolume(snapshot + '3'))
248 self.assertTrue(os.path.exists(os.path.join(snapshot + '3', 'dir')))
250 nested_subvol = os.path.join(subvol, 'nested')
251 more_nested_subvol = os.path.join(nested_subvol, 'more_nested')
252 btrfsutil.create_subvolume(nested_subvol)
253 btrfsutil.create_subvolume(more_nested_subvol)
254 os.mkdir(os.path.join(more_nested_subvol, 'nested_dir'))
256 snapshot = os.path.join(self.mountpoint, 'snapshot')
258 btrfsutil.create_snapshot(subvol, snapshot + '1')
260 self.assertEqual(os.stat(os.path.join(snapshot + '1', 'nested')).st_ino, 2)
261 self.assertFalse(os.path.exists(os.path.join(snapshot + '1', 'nested', 'more_nested')))
263 btrfsutil.create_snapshot(subvol, snapshot + '2', recursive=True)
264 self.assertTrue(os.path.exists(os.path.join(snapshot + '2', 'nested/more_nested/nested_dir')))
266 transid = btrfsutil.create_snapshot(subvol, snapshot + '3', recursive=True, async=True)
267 self.assertTrue(os.path.exists(os.path.join(snapshot + '3', 'nested/more_nested/nested_dir')))
268 self.assertGreater(transid, 0)
270 btrfsutil.create_snapshot(subvol, snapshot + '4', read_only=True)
271 self.assertTrue(btrfsutil.get_subvolume_read_only(snapshot + '4'))
273 def test_delete_subvolume(self):
274 subvol = os.path.join(self.mountpoint, 'subvol')
275 btrfsutil.create_subvolume(subvol + '1')
276 self.assertTrue(os.path.exists(subvol + '1'))
277 btrfsutil.create_subvolume(subvol + '2')
278 self.assertTrue(os.path.exists(subvol + '2'))
279 btrfsutil.create_subvolume(subvol + '3')
280 self.assertTrue(os.path.exists(subvol + '3'))
282 btrfsutil.delete_subvolume(subvol + '1')
283 self.assertFalse(os.path.exists(subvol + '1'))
284 btrfsutil.delete_subvolume((subvol + '2').encode())
285 self.assertFalse(os.path.exists(subvol + '2'))
287 btrfsutil.delete_subvolume(PurePath(subvol + '3'))
288 self.assertFalse(os.path.exists(subvol + '3'))
290 # Test deleting subvolumes under '/' in a chroot.
294 os.chroot(self.mountpoint)
296 btrfsutil.create_subvolume('/subvol4')
297 self.assertTrue(os.path.exists('/subvol4'))
298 btrfsutil.delete_subvolume('/subvol4')
299 self.assertFalse(os.path.exists('/subvol4'))
300 with self.assertRaises(btrfsutil.BtrfsUtilError):
301 btrfsutil.delete_subvolume('/')
304 traceback.print_exc()
306 wstatus = os.waitpid(pid, 0)[1]
307 self.assertTrue(os.WIFEXITED(wstatus))
308 self.assertEqual(os.WEXITSTATUS(wstatus), 0)
310 btrfsutil.create_subvolume(subvol + '5')
311 btrfsutil.create_subvolume(subvol + '5/foo')
312 btrfsutil.create_subvolume(subvol + '5/bar')
313 btrfsutil.create_subvolume(subvol + '5/bar/baz')
314 btrfsutil.create_subvolume(subvol + '5/bar/qux')
315 btrfsutil.create_subvolume(subvol + '5/quux')
316 with self.assertRaises(btrfsutil.BtrfsUtilError):
317 btrfsutil.delete_subvolume(subvol + '5')
318 btrfsutil.delete_subvolume(subvol + '5', recursive=True)
319 self.assertFalse(os.path.exists(subvol + '5'))
321 def test_deleted_subvolumes(self):
322 subvol = os.path.join(self.mountpoint, 'subvol')
323 btrfsutil.create_subvolume(subvol + '1')
324 btrfsutil.delete_subvolume(subvol + '1')
325 for arg in self.path_or_fd(self.mountpoint):
326 with self.subTest(type=type(arg)):
327 self.assertEqual(btrfsutil.deleted_subvolumes(arg), [256])
329 def test_subvolume_iterator(self):
332 os.chdir(self.mountpoint)
333 btrfsutil.create_subvolume('foo')
335 path, subvol = next(btrfsutil.SubvolumeIterator('.', info=True))
336 self.assertEqual(path, 'foo')
337 self.assertIsInstance(subvol, btrfsutil.SubvolumeInfo)
338 self.assertEqual(subvol.id, 256)
339 self.assertEqual(subvol.parent_id, 5)
341 btrfsutil.create_subvolume('foo/bar')
342 btrfsutil.create_subvolume('foo/bar/baz')
347 ('foo/bar/baz', 258),
350 for arg in self.path_or_fd('.'):
351 with self.subTest(type=type(arg)):
352 self.assertEqual(list(btrfsutil.SubvolumeIterator(arg)), subvols)
353 self.assertEqual(list(btrfsutil.SubvolumeIterator('.', top=0)), subvols)
355 self.assertEqual(list(btrfsutil.SubvolumeIterator('.', post_order=True)),
356 [('foo/bar/baz', 258),
365 self.assertEqual(list(btrfsutil.SubvolumeIterator('.', top=256)), subvols)
366 self.assertEqual(list(btrfsutil.SubvolumeIterator('foo', top=0)), subvols)
368 os.rename('foo/bar/baz', 'baz')
369 self.assertEqual(sorted(btrfsutil.SubvolumeIterator('.')),
374 with btrfsutil.SubvolumeIterator('.') as it:
375 self.assertGreaterEqual(it.fileno(), 0)
377 with self.assertRaises(ValueError):
379 with self.assertRaises(ValueError):