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 # TODO: test received_uuid, stransid, rtransid, stime, and rtime
134 for arg in self.path_or_fd(self.mountpoint):
135 with self.subTest(type=type(arg)):
136 with self.assertRaises(btrfsutil.BtrfsUtilError) as e:
137 # BTRFS_EXTENT_TREE_OBJECTID
138 btrfsutil.subvolume_info(arg, 2)
140 def test_read_only(self):
141 for arg in self.path_or_fd(self.mountpoint):
142 with self.subTest(type=type(arg)):
143 btrfsutil.set_subvolume_read_only(arg)
144 self.assertTrue(btrfsutil.get_subvolume_read_only(arg))
145 self.assertTrue(btrfsutil.subvolume_info(arg).flags & 1)
147 btrfsutil.set_subvolume_read_only(arg, False)
148 self.assertFalse(btrfsutil.get_subvolume_read_only(arg))
149 self.assertFalse(btrfsutil.subvolume_info(arg).flags & 1)
151 btrfsutil.set_subvolume_read_only(arg, True)
152 self.assertTrue(btrfsutil.get_subvolume_read_only(arg))
153 self.assertTrue(btrfsutil.subvolume_info(arg).flags & 1)
155 btrfsutil.set_subvolume_read_only(arg, False)
157 def test_create_subvolume(self):
158 subvol = os.path.join(self.mountpoint, 'subvol')
160 btrfsutil.create_subvolume(subvol + '1')
161 self.assertTrue(btrfsutil.is_subvolume(subvol + '1'))
162 btrfsutil.create_subvolume((subvol + '2').encode())
163 self.assertTrue(btrfsutil.is_subvolume(subvol + '2'))
165 btrfsutil.create_subvolume(PurePath(subvol + '3'))
166 self.assertTrue(btrfsutil.is_subvolume(subvol + '3'))
170 os.chdir(self.mountpoint)
171 btrfsutil.create_subvolume('subvol4')
172 self.assertTrue(btrfsutil.is_subvolume('subvol4'))
176 btrfsutil.create_subvolume(subvol + '5/')
177 self.assertTrue(btrfsutil.is_subvolume(subvol + '5'))
179 btrfsutil.create_subvolume(subvol + '6//')
180 self.assertTrue(btrfsutil.is_subvolume(subvol + '6'))
182 transid = btrfsutil.create_subvolume(subvol + '7', async=True)
183 self.assertTrue(btrfsutil.is_subvolume(subvol + '7'))
184 self.assertGreater(transid, 0)
186 # Test creating subvolumes under '/' in a chroot.
190 os.chroot(self.mountpoint)
192 btrfsutil.create_subvolume('/subvol8')
193 self.assertTrue(btrfsutil.is_subvolume('/subvol8'))
194 with self.assertRaises(btrfsutil.BtrfsUtilError):
195 btrfsutil.create_subvolume('/')
198 traceback.print_exc()
200 wstatus = os.waitpid(pid, 0)[1]
201 self.assertTrue(os.WIFEXITED(wstatus))
202 self.assertEqual(os.WEXITSTATUS(wstatus), 0)