3 import arvados_fuse.command
10 import run_test_server
16 """If argparse or arvados_fuse tries to exit, fail the test instead"""
17 class SystemExitCaught(StandardError):
19 @functools.wraps(func)
20 def wrapper(*args, **kwargs):
22 return func(*args, **kwargs)
24 raise SystemExitCaught
27 @contextlib.contextmanager
29 orig, sys.stderr = sys.stderr, open(os.devnull, 'w')
36 class MountArgsTest(unittest.TestCase):
38 self.mntdir = tempfile.mkdtemp()
39 run_test_server.authorize_with('active')
44 def lookup(self, mnt, *path):
45 ent = mnt.operations.inodes[llfuse.ROOT_INODE]
50 def check_ent_type(self, cls, *path):
51 ent = self.lookup(self.mnt, *path)
52 self.assertEqual(ent.__class__, cls)
56 def test_default_all(self):
57 args = arvados_fuse.command.ArgumentParser().parse_args([
58 '--foreground', self.mntdir])
59 self.assertEqual(args.mode, None)
60 self.mnt = arvados_fuse.command.Mount(args)
61 e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'home')
62 self.assertEqual(e.project_object['uuid'],
63 run_test_server.fixture('users')['active']['uuid'])
64 e = self.check_ent_type(arvados_fuse.MagicDirectory, 'by_id')
66 e = self.check_ent_type(arvados_fuse.StringFile, 'README')
67 readme = e.readfrom(0, -1)
68 self.assertRegexpMatches(readme, r'active-user@arvados\.local')
69 self.assertRegexpMatches(readme, r'\n$')
71 e = self.check_ent_type(arvados_fuse.StringFile, 'by_id', 'README')
72 txt = e.readfrom(0, -1)
73 self.assertRegexpMatches(txt, r'portable data hash')
74 self.assertRegexpMatches(txt, r'\n$')
78 args = arvados_fuse.command.ArgumentParser().parse_args([
80 '--foreground', self.mntdir])
81 self.assertEqual(args.mode, 'by_id')
82 self.mnt = arvados_fuse.command.Mount(args)
83 e = self.check_ent_type(arvados_fuse.MagicDirectory)
84 self.assertEqual(e.pdh_only, False)
87 def test_by_pdh(self):
88 args = arvados_fuse.command.ArgumentParser().parse_args([
90 '--foreground', self.mntdir])
91 self.assertEqual(args.mode, 'by_pdh')
92 self.mnt = arvados_fuse.command.Mount(args)
93 e = self.check_ent_type(arvados_fuse.MagicDirectory)
94 self.assertEqual(e.pdh_only, True)
97 def test_by_tag(self):
98 args = arvados_fuse.command.ArgumentParser().parse_args([
100 '--foreground', self.mntdir])
101 self.assertEqual(args.mode, 'by_tag')
102 self.mnt = arvados_fuse.command.Mount(args)
103 e = self.check_ent_type(arvados_fuse.TagsDirectory)
106 def test_collection(self, id_type='uuid'):
107 c = run_test_server.fixture('collections')['public_text_file']
109 args = arvados_fuse.command.ArgumentParser().parse_args([
111 '--foreground', self.mntdir])
112 self.mnt = arvados_fuse.command.Mount(args)
113 e = self.check_ent_type(arvados_fuse.CollectionDirectory)
114 self.assertEqual(e.collection_locator, cid)
116 def test_collection_pdh(self):
117 self.test_collection('portable_data_hash')
121 args = arvados_fuse.command.ArgumentParser().parse_args([
123 '--foreground', self.mntdir])
124 self.assertEqual(args.mode, 'home')
125 self.mnt = arvados_fuse.command.Mount(args)
126 e = self.check_ent_type(arvados_fuse.ProjectDirectory)
127 self.assertEqual(e.project_object['uuid'],
128 run_test_server.fixture('users')['active']['uuid'])
130 def test_mutually_exclusive_args(self):
131 cid = run_test_server.fixture('collections')['public_text_file']['uuid']
132 gid = run_test_server.fixture('groups')['aproject']['uuid']
134 ['--mount-tmp', 'foo', '--collection', cid],
135 ['--mount-tmp', 'foo', '--project', gid],
136 ['--collection', cid, '--project', gid],
137 ['--by-id', '--project', gid],
138 ['--mount-tmp', 'foo', '--by-id'],
141 with self.assertRaises(SystemExit):
142 args = arvados_fuse.command.ArgumentParser().parse_args(
143 badargs + ['--foreground', self.mntdir])
144 arvados_fuse.command.Mount(args)
146 def test_project(self):
147 uuid = run_test_server.fixture('groups')['aproject']['uuid']
148 args = arvados_fuse.command.ArgumentParser().parse_args([
150 '--foreground', self.mntdir])
151 self.mnt = arvados_fuse.command.Mount(args)
152 e = self.check_ent_type(arvados_fuse.ProjectDirectory)
153 self.assertEqual(e.project_object['uuid'], uuid)
156 def test_shared(self):
157 args = arvados_fuse.command.ArgumentParser().parse_args([
159 '--foreground', self.mntdir])
160 self.assertEqual(args.mode, 'shared')
161 self.mnt = arvados_fuse.command.Mount(args)
162 e = self.check_ent_type(arvados_fuse.SharedDirectory)
163 self.assertEqual(e.current_user['uuid'],
164 run_test_server.fixture('users')['active']['uuid'])
167 def test_custom(self):
168 args = arvados_fuse.command.ArgumentParser().parse_args([
169 '--mount-tmp', 'foo',
170 '--mount-tmp', 'bar',
171 '--mount-home', 'my_home',
172 '--foreground', self.mntdir])
173 self.assertEqual(args.mode, None)
174 self.mnt = arvados_fuse.command.Mount(args)
175 self.check_ent_type(arvados_fuse.Directory)
176 self.check_ent_type(arvados_fuse.TmpCollectionDirectory, 'foo')
177 self.check_ent_type(arvados_fuse.TmpCollectionDirectory, 'bar')
178 e = self.check_ent_type(arvados_fuse.ProjectDirectory, 'my_home')
179 self.assertEqual(e.project_object['uuid'],
180 run_test_server.fixture('users')['active']['uuid'])
182 def test_custom_unsupported_layouts(self):
183 for name in ['.', '..', '', 'foo/bar', '/foo']:
185 with self.assertRaises(SystemExit):
186 args = arvados_fuse.command.ArgumentParser().parse_args([
188 '--foreground', self.mntdir])
189 arvados_fuse.command.Mount(args)
191 class MountErrorTest(unittest.TestCase):
193 self.mntdir = tempfile.mkdtemp()
194 run_test_server.run()
195 run_test_server.authorize_with("active")
196 self.logger = logging.getLogger("null")
197 self.logger.setLevel(logging.CRITICAL+1)
200 if os.path.exists(self.mntdir):
201 # If the directory was not unmounted, this will raise an exception.
202 os.rmdir(self.mntdir)
203 run_test_server.reset()
205 def test_no_token(self):
206 del arvados.config._settings["ARVADOS_API_TOKEN"]
207 arvados.config._settings = {}
208 with self.assertRaises(SystemExit) as ex:
209 args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
210 arvados_fuse.command.Mount(args, logger=self.logger).run()
211 self.assertEqual(1, ex.exception.code)
213 def test_no_host(self):
214 del arvados.config._settings["ARVADOS_API_HOST"]
215 with self.assertRaises(SystemExit) as ex:
216 args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
217 arvados_fuse.command.Mount(args, logger=self.logger).run()
218 self.assertEqual(1, ex.exception.code)
220 def test_bogus_host(self):
221 arvados.config._settings["ARVADOS_API_HOST"] = "example.null"
222 with self.assertRaises(SystemExit) as ex:
223 args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
224 arvados_fuse.command.Mount(args, logger=self.logger).run()
225 self.assertEqual(1, ex.exception.code)
227 def test_bogus_token(self):
228 arvados.config._settings["ARVADOS_API_TOKEN"] = "zzzzzzzzzzzzz"
229 with self.assertRaises(SystemExit) as ex:
230 args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
231 arvados_fuse.command.Mount(args, logger=self.logger).run()
232 self.assertEqual(1, ex.exception.code)
234 def test_bogus_mount_dir(self):
235 # All FUSE errors in llfuse.init() are raised as RuntimeError
236 # An easy error to trigger is to supply a nonexistent mount point,
239 # Other possible errors that also raise RuntimeError (but are much
240 # harder to test automatically because they depend on operating
241 # system configuration):
243 # The user doesn't have permission to use FUSE
244 # The user specified --allow-other but user_allow_other is not set
246 os.rmdir(self.mntdir)
247 with self.assertRaises(SystemExit) as ex:
248 args = arvados_fuse.command.ArgumentParser().parse_args([self.mntdir])
249 arvados_fuse.command.Mount(args, logger=self.logger).run()
250 self.assertEqual(1, ex.exception.code)
252 def test_unreadable_collection(self):
253 with self.assertRaises(SystemExit) as ex:
254 args = arvados_fuse.command.ArgumentParser().parse_args([
255 "--collection", "zzzzz-4zz18-zzzzzzzzzzzzzzz", self.mntdir])
256 arvados_fuse.command.Mount(args, logger=self.logger).run()
257 self.assertEqual(1, ex.exception.code)
259 def test_unreadable_project(self):
260 with self.assertRaises(SystemExit) as ex:
261 args = arvados_fuse.command.ArgumentParser().parse_args([
262 "--project", "zzzzz-j7d0g-zzzzzzzzzzzzzzz", self.mntdir])
263 arvados_fuse.command.Mount(args, logger=self.logger).run()
264 self.assertEqual(1, ex.exception.code)