1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: AGPL-3.0
15 from pathlib import Path
16 from unittest import mock
19 import arvados_fuse as fuse
22 from arvados_fuse import fusedir
24 from . import run_test_server
25 from .integration_test import IntegrationTest
26 from .mount_test_base import MountTestBase
27 from .test_tmp_collection import storage_classes_desired
29 logger = logging.getLogger('arvados.arv-mount')
31 class AssertWithTimeout(object):
32 """Allow some time for an assertion to pass."""
34 def __init__(self, timeout=0):
35 self.timeout = timeout
38 self.deadline = time.time() + self.timeout
47 def attempt(self, fn, *args, **kwargs):
50 except AssertionError:
51 if time.time() > self.deadline:
57 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
58 class FuseMountTest(MountTestBase):
60 super(FuseMountTest, self).setUp()
62 cw = arvados.collection.Collection()
63 with cw.open('thing1.txt', 'w') as f:
65 with cw.open('thing2.txt', 'w') as f:
68 with cw.open('dir1/thing3.txt', 'w') as f:
70 with cw.open('dir1/thing4.txt', 'w') as f:
73 with cw.open('dir2/thing5.txt', 'w') as f:
75 with cw.open('dir2/thing6.txt', 'w') as f:
78 with cw.open('dir2/dir3/thing7.txt', 'w') as f:
80 with cw.open('dir2/dir3/thing8.txt', 'w') as f:
83 for fnm in ":/.../-/*/ ".split("/"):
84 with cw.open('edgecases/'+fnm, 'w') as f:
87 for fnm in ":/.../-/*/ ".split("/"):
88 with cw.open('edgecases/dirs/'+fnm+'/x/x', 'w') as f:
91 self.testcollection = cw.portable_data_hash()
92 self.test_manifest = cw.manifest_text()
93 self.api.collections().create(body={"manifest_text": self.test_manifest}).execute()
96 self.make_mount(fuse.CollectionDirectory, collection_record=self.testcollection)
98 self.assertDirContents(None, ['thing1.txt', 'thing2.txt',
99 'edgecases', 'dir1', 'dir2'])
100 self.assertDirContents('dir1', ['thing3.txt', 'thing4.txt'])
101 self.assertDirContents('dir2', ['thing5.txt', 'thing6.txt', 'dir3'])
102 self.assertDirContents('dir2/dir3', ['thing7.txt', 'thing8.txt'])
103 self.assertDirContents('edgecases',
104 "dirs/:/.../-/*/ ".split("/"))
105 self.assertDirContents('edgecases/dirs',
106 ":/.../-/*/ ".split("/"))
108 files = {'thing1.txt': 'data 1',
109 'thing2.txt': 'data 2',
110 'dir1/thing3.txt': 'data 3',
111 'dir1/thing4.txt': 'data 4',
112 'dir2/thing5.txt': 'data 5',
113 'dir2/thing6.txt': 'data 6',
114 'dir2/dir3/thing7.txt': 'data 7',
115 'dir2/dir3/thing8.txt': 'data 8'}
117 for k, v in files.items():
118 with open(os.path.join(self.mounttmp, k), 'rb') as f:
119 self.assertEqual(v, f.read().decode())
122 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
123 class FuseMagicTest(MountTestBase):
124 def setUp(self, api=None):
125 super(FuseMagicTest, self).setUp(api=api)
127 self.test_project = run_test_server.fixture('groups')['aproject']['uuid']
128 self.non_project_group = run_test_server.fixture('groups')['public_role']['uuid']
129 self.filter_group = run_test_server.fixture('groups')['afiltergroup']['uuid']
130 self.collection_in_test_project = run_test_server.fixture('collections')['foo_collection_in_aproject']['name']
131 self.collection_in_filter_group = run_test_server.fixture('collections')['baz_file']['name']
133 cw = arvados.collection.Collection()
134 with cw.open('thing1.txt', 'w') as f:
137 self.testcollection = cw.portable_data_hash()
138 self.test_manifest = cw.manifest_text()
139 coll = self.api.collections().create(body={"manifest_text":self.test_manifest}).execute()
140 self.test_manifest_pdh = coll['portable_data_hash']
143 self.make_mount(fuse.MagicDirectory)
145 mount_ls = llfuse.listdir(self.mounttmp)
146 self.assertIn('README', mount_ls)
147 self.assertFalse(any(arvados.util.keep_locator_pattern.match(fn) or
148 arvados.util.uuid_pattern.match(fn)
150 "new FUSE MagicDirectory has no collections or projects")
151 self.assertDirContents(self.testcollection, ['thing1.txt'])
152 self.assertDirContents(os.path.join('by_id', self.testcollection),
154 self.assertIn(self.collection_in_test_project,
155 llfuse.listdir(os.path.join(self.mounttmp, self.test_project)))
156 self.assertIn(self.collection_in_test_project,
157 llfuse.listdir(os.path.join(self.mounttmp, 'by_id', self.test_project)))
158 self.assertIn(self.collection_in_filter_group,
159 llfuse.listdir(os.path.join(self.mounttmp, self.filter_group)))
160 self.assertIn(self.collection_in_filter_group,
161 llfuse.listdir(os.path.join(self.mounttmp, 'by_id', self.filter_group)))
164 mount_ls = llfuse.listdir(self.mounttmp)
165 self.assertIn('README', mount_ls)
166 self.assertIn(self.testcollection, mount_ls)
167 self.assertIn(self.testcollection,
168 llfuse.listdir(os.path.join(self.mounttmp, 'by_id')))
169 self.assertIn(self.test_project, mount_ls)
170 self.assertIn(self.test_project,
171 llfuse.listdir(os.path.join(self.mounttmp, 'by_id')))
172 self.assertIn(self.filter_group,
173 llfuse.listdir(os.path.join(self.mounttmp, 'by_id')))
175 with self.assertRaises(OSError):
176 llfuse.listdir(os.path.join(self.mounttmp, 'by_id', self.non_project_group))
179 files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
181 for k, v in files.items():
182 with open(os.path.join(self.mounttmp, k), 'rb') as f:
183 self.assertEqual(v, f.read().decode())
186 class FuseTagsTest(MountTestBase):
188 self.make_mount(fuse.TagsDirectory)
190 d1 = llfuse.listdir(self.mounttmp)
192 self.assertEqual(['foo_tag'], d1)
194 d2 = llfuse.listdir(os.path.join(self.mounttmp, 'foo_tag'))
196 self.assertEqual(['zzzzz-4zz18-fy296fx3hot09f7'], d2)
198 d3 = llfuse.listdir(os.path.join(self.mounttmp, 'foo_tag', 'zzzzz-4zz18-fy296fx3hot09f7'))
200 self.assertEqual(['foo'], d3)
203 class FuseTagsUpdateTest(MountTestBase):
204 def tag_collection(self, coll_uuid, tag_name):
205 return self.api.links().create(
206 body={'link': {'head_uuid': coll_uuid,
212 self.make_mount(fuse.TagsDirectory, poll_time=1)
214 self.assertIn('foo_tag', llfuse.listdir(self.mounttmp))
216 bar_uuid = run_test_server.fixture('collections')['bar_file']['uuid']
217 self.tag_collection(bar_uuid, 'fuse_test_tag')
218 for attempt in AssertWithTimeout(10):
219 attempt(self.assertIn, 'fuse_test_tag', llfuse.listdir(self.mounttmp))
220 self.assertDirContents('fuse_test_tag', [bar_uuid])
222 baz_uuid = run_test_server.fixture('collections')['baz_file']['uuid']
223 l = self.tag_collection(baz_uuid, 'fuse_test_tag')
224 for attempt in AssertWithTimeout(10):
225 attempt(self.assertDirContents, 'fuse_test_tag', [bar_uuid, baz_uuid])
227 self.api.links().delete(uuid=l['uuid']).execute()
228 for attempt in AssertWithTimeout(10):
229 attempt(self.assertDirContents, 'fuse_test_tag', [bar_uuid])
232 def fuseSharedTestHelper(mounttmp):
233 class Test(unittest.TestCase):
235 # Double check that we can open and read objects in this folder as a file,
236 # and that its contents are what we expect.
237 baz_path = os.path.join(
241 'collection in FUSE project',
243 with open(baz_path) as f:
244 self.assertEqual("baz", f.read())
246 # check mtime on collection
247 st = os.stat(baz_path)
249 mtime = st.st_mtime_ns // 1000000000
250 except AttributeError:
252 self.assertEqual(mtime, 1391448174)
254 # shared_dirs is a list of the directories exposed
255 # by fuse.SharedDirectory (i.e. any object visible
256 # to the current user)
257 shared_dirs = llfuse.listdir(mounttmp)
259 self.assertIn('FUSE User', shared_dirs)
261 # fuse_user_objs is a list of the objects owned by the FUSE
262 # test user (which present as files in the 'FUSE User'
264 fuse_user_objs = llfuse.listdir(os.path.join(mounttmp, 'FUSE User'))
265 fuse_user_objs.sort()
266 self.assertEqual(['FUSE Test Project', # project owned by user
267 'collection #1 owned by FUSE', # collection owned by user
268 'collection #2 owned by FUSE' # collection owned by user
271 # test_proj_files is a list of the files in the FUSE Test Project.
272 test_proj_files = llfuse.listdir(os.path.join(mounttmp, 'FUSE User', 'FUSE Test Project'))
273 test_proj_files.sort()
274 self.assertEqual(['collection in FUSE project'
280 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
281 class FuseSharedTest(MountTestBase):
283 self.make_mount(fuse.SharedDirectory,
284 exclude=self.api.users().current().execute()['uuid'])
285 keep = arvados.keep.KeepClient()
286 keep.put("baz".encode())
288 self.pool.apply(fuseSharedTestHelper, (self.mounttmp,))
291 class FuseHomeTest(MountTestBase):
293 self.make_mount(fuse.ProjectDirectory,
294 project_object=self.api.users().current().execute())
296 d1 = llfuse.listdir(self.mounttmp)
297 self.assertIn('Unrestricted public data', d1)
299 d2 = llfuse.listdir(os.path.join(self.mounttmp, 'Unrestricted public data'))
300 public_project = run_test_server.fixture('groups')[
301 'anonymously_accessible_project']
304 for name, item in run_test_server.fixture('collections').items():
305 if 'name' not in item:
307 elif item['owner_uuid'] == public_project['uuid']:
308 self.assertIn(item['name'], d2)
311 # Artificial assumption here: there is no public
312 # collection fixture with the same name as a
313 # non-public collection.
314 self.assertNotIn(item['name'], d2)
316 self.assertNotEqual(0, found_in)
317 self.assertNotEqual(0, found_not_in)
319 d3 = llfuse.listdir(os.path.join(self.mounttmp, 'Unrestricted public data', 'GNU General Public License, version 3'))
320 self.assertEqual(["GNU_General_Public_License,_version_3.pdf"], d3)
323 def fuseModifyFileTestHelperReadStartContents(mounttmp):
324 class Test(unittest.TestCase):
326 d1 = llfuse.listdir(mounttmp)
327 self.assertEqual(["file1.txt"], d1)
328 with open(os.path.join(mounttmp, "file1.txt")) as f:
329 self.assertEqual("blub", f.read())
332 def fuseModifyFileTestHelperReadEndContents(mounttmp):
333 class Test(unittest.TestCase):
335 d1 = llfuse.listdir(mounttmp)
336 self.assertEqual(["file1.txt"], d1)
337 with open(os.path.join(mounttmp, "file1.txt")) as f:
338 self.assertEqual("plnp", f.read())
341 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
342 class FuseModifyFileTest(MountTestBase):
344 collection = arvados.collection.Collection(api_client=self.api)
345 with collection.open("file1.txt", "w") as f:
348 collection.save_new()
350 m = self.make_mount(fuse.CollectionDirectory)
352 m.new_collection(collection.api_response(), collection)
354 self.pool.apply(fuseModifyFileTestHelperReadStartContents, (self.mounttmp,))
356 with collection.open("file1.txt", "w") as f:
359 self.pool.apply(fuseModifyFileTestHelperReadEndContents, (self.mounttmp,))
362 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
363 class FuseAddFileToCollectionTest(MountTestBase):
365 collection = arvados.collection.Collection(api_client=self.api)
366 with collection.open("file1.txt", "w") as f:
369 collection.save_new()
371 m = self.make_mount(fuse.CollectionDirectory)
373 m.new_collection(collection.api_response(), collection)
375 d1 = llfuse.listdir(self.mounttmp)
376 self.assertEqual(["file1.txt"], d1)
378 with collection.open("file2.txt", "w") as f:
381 d1 = llfuse.listdir(self.mounttmp)
382 self.assertEqual(["file1.txt", "file2.txt"], sorted(d1))
385 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
386 class FuseRemoveFileFromCollectionTest(MountTestBase):
388 collection = arvados.collection.Collection(api_client=self.api)
389 with collection.open("file1.txt", "w") as f:
392 with collection.open("file2.txt", "w") as f:
395 collection.save_new()
397 m = self.make_mount(fuse.CollectionDirectory)
399 m.new_collection(collection.api_response(), collection)
401 d1 = llfuse.listdir(self.mounttmp)
402 self.assertEqual(["file1.txt", "file2.txt"], sorted(d1))
404 collection.remove("file2.txt")
406 d1 = llfuse.listdir(self.mounttmp)
407 self.assertEqual(["file1.txt"], d1)
410 def fuseCreateFileTestHelper(mounttmp):
411 class Test(unittest.TestCase):
413 with open(os.path.join(mounttmp, "file1.txt"), "w") as f:
417 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
418 class FuseCreateFileTest(MountTestBase):
420 collection = arvados.collection.Collection(api_client=self.api)
421 collection.save_new()
423 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
424 self.assertEqual(collection2["manifest_text"], "")
426 collection.save_new()
428 m = self.make_mount(fuse.CollectionDirectory)
430 m.new_collection(collection.api_response(), collection)
431 self.assertTrue(m.writable())
433 self.assertNotIn("file1.txt", collection)
435 self.pool.apply(fuseCreateFileTestHelper, (self.mounttmp,))
437 self.assertIn("file1.txt", collection)
439 d1 = llfuse.listdir(self.mounttmp)
440 self.assertEqual(["file1.txt"], d1)
442 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
443 self.assertRegex(collection2["manifest_text"],
444 r'\. d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:file1\.txt$')
447 def fuseWriteFileTestHelperWriteFile(mounttmp):
448 class Test(unittest.TestCase):
450 with open(os.path.join(mounttmp, "file1.txt"), "w") as f:
451 f.write("Hello world!")
454 def fuseWriteFileTestHelperReadFile(mounttmp):
455 class Test(unittest.TestCase):
457 with open(os.path.join(mounttmp, "file1.txt"), "r") as f:
458 self.assertEqual(f.read(), "Hello world!")
461 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
462 class FuseWriteFileTest(MountTestBase):
464 collection = arvados.collection.Collection(api_client=self.api)
465 collection.save_new()
467 m = self.make_mount(fuse.CollectionDirectory)
469 m.new_collection(collection.api_response(), collection)
470 self.assertTrue(m.writable())
472 self.assertNotIn("file1.txt", collection)
474 self.assertEqual(0, self.operations.write_counter.get())
475 self.pool.apply(fuseWriteFileTestHelperWriteFile, (self.mounttmp,))
476 self.assertEqual(12, self.operations.write_counter.get())
478 with collection.open("file1.txt") as f:
479 self.assertEqual(f.read(), "Hello world!")
481 self.assertEqual(0, self.operations.read_counter.get())
482 self.pool.apply(fuseWriteFileTestHelperReadFile, (self.mounttmp,))
483 self.assertEqual(12, self.operations.read_counter.get())
485 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
486 self.assertRegex(collection2["manifest_text"],
487 r'\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
490 def fuseUpdateFileTestHelper(mounttmp):
491 class Test(unittest.TestCase):
493 with open(os.path.join(mounttmp, "file1.txt"), "w") as f:
494 f.write("Hello world!")
496 with open(os.path.join(mounttmp, "file1.txt"), "r+") as f:
498 self.assertEqual(fr, "Hello world!")
500 f.write("Hola mundo!")
503 self.assertEqual(fr, "Hola mundo!!")
505 with open(os.path.join(mounttmp, "file1.txt"), "r") as f:
506 self.assertEqual(f.read(), "Hola mundo!!")
510 @parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
511 class FuseUpdateFileTest(MountTestBase):
513 collection = arvados.collection.Collection(api_client=self.api)
514 collection.save_new()
516 m = self.make_mount(fuse.CollectionDirectory)
518 m.new_collection(collection.api_response(), collection)
519 self.assertTrue(m.writable())
521 # See note in MountTestBase.setUp
522 self.pool.apply(fuseUpdateFileTestHelper, (self.mounttmp,))
524 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
525 self.assertRegex(collection2["manifest_text"],
526 r'\. daaef200ebb921e011e3ae922dd3266b\+11\+A\S+ 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:11:file1\.txt 22:1:file1\.txt$')
529 def fuseMkdirTestHelper(mounttmp):
530 class Test(unittest.TestCase):
532 with self.assertRaises(IOError):
533 with open(os.path.join(mounttmp, "testdir", "file1.txt"), "w") as f:
534 f.write("Hello world!")
536 os.mkdir(os.path.join(mounttmp, "testdir"))
538 with self.assertRaises(OSError):
539 os.mkdir(os.path.join(mounttmp, "testdir"))
541 d1 = llfuse.listdir(mounttmp)
542 self.assertEqual(["testdir"], d1)
544 with open(os.path.join(mounttmp, "testdir", "file1.txt"), "w") as f:
545 f.write("Hello world!")
547 d1 = llfuse.listdir(os.path.join(mounttmp, "testdir"))
548 self.assertEqual(["file1.txt"], d1)
552 class FuseMkdirTest(MountTestBase):
554 collection = arvados.collection.Collection(api_client=self.api)
555 collection.save_new()
557 m = self.make_mount(fuse.CollectionDirectory)
559 m.new_collection(collection.api_response(), collection)
560 self.assertTrue(m.writable())
562 self.pool.apply(fuseMkdirTestHelper, (self.mounttmp,))
564 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
565 self.assertRegex(collection2["manifest_text"],
566 r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
569 def fuseRmTestHelperWriteFile(mounttmp):
570 class Test(unittest.TestCase):
572 os.mkdir(os.path.join(mounttmp, "testdir"))
574 with open(os.path.join(mounttmp, "testdir", "file1.txt"), "w") as f:
575 f.write("Hello world!")
579 def fuseRmTestHelperDeleteFile(mounttmp):
580 class Test(unittest.TestCase):
582 # Can't delete because it's not empty
583 with self.assertRaises(OSError):
584 os.rmdir(os.path.join(mounttmp, "testdir"))
586 d1 = llfuse.listdir(os.path.join(mounttmp, "testdir"))
587 self.assertEqual(["file1.txt"], d1)
590 os.remove(os.path.join(mounttmp, "testdir", "file1.txt"))
592 # Make sure it's empty
593 d1 = llfuse.listdir(os.path.join(mounttmp, "testdir"))
594 self.assertEqual([], d1)
596 # Try to delete it again
597 with self.assertRaises(OSError):
598 os.remove(os.path.join(mounttmp, "testdir", "file1.txt"))
602 def fuseRmTestHelperRmdir(mounttmp):
603 class Test(unittest.TestCase):
605 # Should be able to delete now that it is empty
606 os.rmdir(os.path.join(mounttmp, "testdir"))
608 # Make sure it's empty
609 d1 = llfuse.listdir(os.path.join(mounttmp))
610 self.assertEqual([], d1)
612 # Try to delete it again
613 with self.assertRaises(OSError):
614 os.rmdir(os.path.join(mounttmp, "testdir"))
618 class FuseRmTest(MountTestBase):
620 collection = arvados.collection.Collection(api_client=self.api)
621 collection.save_new()
623 m = self.make_mount(fuse.CollectionDirectory)
625 m.new_collection(collection.api_response(), collection)
626 self.assertTrue(m.writable())
628 self.pool.apply(fuseRmTestHelperWriteFile, (self.mounttmp,))
631 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
632 self.assertRegex(collection2["manifest_text"],
633 r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
634 self.pool.apply(fuseRmTestHelperDeleteFile, (self.mounttmp,))
636 # Empty directories are represented by an empty file named "."
637 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
638 self.assertRegex(collection2["manifest_text"],
639 r'./testdir d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:\\056\n')
641 self.pool.apply(fuseRmTestHelperRmdir, (self.mounttmp,))
643 # manifest should be empty now.
644 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
645 self.assertEqual(collection2["manifest_text"], "")
648 def fuseMvFileTestHelperWriteFile(mounttmp):
649 class Test(unittest.TestCase):
651 os.mkdir(os.path.join(mounttmp, "testdir"))
653 with open(os.path.join(mounttmp, "testdir", "file1.txt"), "w") as f:
654 f.write("Hello world!")
658 def fuseMvFileTestHelperMoveFile(mounttmp):
659 class Test(unittest.TestCase):
661 d1 = llfuse.listdir(os.path.join(mounttmp))
662 self.assertEqual(["testdir"], d1)
663 d1 = llfuse.listdir(os.path.join(mounttmp, "testdir"))
664 self.assertEqual(["file1.txt"], d1)
666 os.rename(os.path.join(mounttmp, "testdir", "file1.txt"), os.path.join(mounttmp, "file1.txt"))
668 d1 = llfuse.listdir(os.path.join(mounttmp))
669 self.assertEqual(["file1.txt", "testdir"], sorted(d1))
670 d1 = llfuse.listdir(os.path.join(mounttmp, "testdir"))
671 self.assertEqual([], d1)
675 class FuseMvFileTest(MountTestBase):
677 collection = arvados.collection.Collection(api_client=self.api)
678 collection.save_new()
680 m = self.make_mount(fuse.CollectionDirectory)
682 m.new_collection(collection.api_response(), collection)
683 self.assertTrue(m.writable())
685 self.pool.apply(fuseMvFileTestHelperWriteFile, (self.mounttmp,))
688 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
689 self.assertRegex(collection2["manifest_text"],
690 r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
692 self.pool.apply(fuseMvFileTestHelperMoveFile, (self.mounttmp,))
694 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
695 self.assertRegex(collection2["manifest_text"],
696 r'\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt\n\./testdir d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:\\056\n')
699 def fuseRenameTestHelper(mounttmp):
700 class Test(unittest.TestCase):
702 os.mkdir(os.path.join(mounttmp, "testdir"))
704 with open(os.path.join(mounttmp, "testdir", "file1.txt"), "w") as f:
705 f.write("Hello world!")
709 class FuseRenameTest(MountTestBase):
711 collection = arvados.collection.Collection(api_client=self.api)
712 collection.save_new()
714 m = self.make_mount(fuse.CollectionDirectory)
716 m.new_collection(collection.api_response(), collection)
717 self.assertTrue(m.writable())
719 self.pool.apply(fuseRenameTestHelper, (self.mounttmp,))
722 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
723 self.assertRegex(collection2["manifest_text"],
724 r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
726 d1 = llfuse.listdir(os.path.join(self.mounttmp))
727 self.assertEqual(["testdir"], d1)
728 d1 = llfuse.listdir(os.path.join(self.mounttmp, "testdir"))
729 self.assertEqual(["file1.txt"], d1)
731 os.rename(os.path.join(self.mounttmp, "testdir"), os.path.join(self.mounttmp, "testdir2"))
733 d1 = llfuse.listdir(os.path.join(self.mounttmp))
734 self.assertEqual(["testdir2"], sorted(d1))
735 d1 = llfuse.listdir(os.path.join(self.mounttmp, "testdir2"))
736 self.assertEqual(["file1.txt"], d1)
738 collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
739 self.assertRegex(collection2["manifest_text"],
740 r'\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
743 class FuseUpdateFromEventTest(MountTestBase):
745 collection = arvados.collection.Collection(api_client=self.api)
746 collection.save_new()
748 m = self.make_mount(fuse.CollectionDirectory)
750 m.new_collection(collection.api_response(), collection)
752 self.operations.listen_for_events()
754 d1 = llfuse.listdir(os.path.join(self.mounttmp))
755 self.assertEqual([], sorted(d1))
757 with arvados.collection.Collection(collection.manifest_locator(), api_client=self.api) as collection2:
758 with collection2.open("file1.txt", "w") as f:
761 for attempt in AssertWithTimeout(10):
762 attempt(self.assertEqual, ["file1.txt"], llfuse.listdir(os.path.join(self.mounttmp)))
765 class FuseDeleteProjectEventTest(MountTestBase):
768 aproject = self.api.groups().create(body={
770 "group_class": "project"
773 bproject = self.api.groups().create(body={
775 "group_class": "project",
776 "owner_uuid": aproject["uuid"]
779 self.make_mount(fuse.ProjectDirectory,
780 project_object=self.api.users().current().execute())
782 self.operations.listen_for_events()
784 d1 = llfuse.listdir(os.path.join(self.mounttmp, "aproject"))
785 self.assertEqual(["bproject"], sorted(d1))
787 self.api.groups().delete(uuid=bproject["uuid"]).execute()
789 for attempt in AssertWithTimeout(10):
790 attempt(self.assertEqual, [], llfuse.listdir(os.path.join(self.mounttmp, "aproject")))
793 def fuseFileConflictTestHelper(mounttmp, uuid, keeptmp, settings):
794 class Test(unittest.TestCase):
796 os.environ['KEEP_LOCAL_STORE'] = keeptmp
798 with open(os.path.join(mounttmp, "file1.txt"), "w") as f:
799 with arvados.collection.Collection(uuid, api_client=arvados.api_from_config('v1', apiconfig=settings)) as collection2:
800 with collection2.open("file1.txt", "w") as f2:
804 d1 = sorted(llfuse.listdir(os.path.join(mounttmp)))
805 self.assertEqual(len(d1), 2)
807 with open(os.path.join(mounttmp, "file1.txt"), "r") as f:
808 self.assertEqual(f.read(), "bar")
810 self.assertRegex(d1[1],
811 r'file1\.txt~\d\d\d\d\d\d\d\d-\d\d\d\d\d\d~conflict~')
813 with open(os.path.join(mounttmp, d1[1]), "r") as f:
814 self.assertEqual(f.read(), "foo")
818 class FuseFileConflictTest(MountTestBase):
820 collection = arvados.collection.Collection(api_client=self.api)
821 collection.save_new()
823 m = self.make_mount(fuse.CollectionDirectory)
825 m.new_collection(collection.api_response(), collection)
827 d1 = llfuse.listdir(os.path.join(self.mounttmp))
828 self.assertEqual([], sorted(d1))
830 # See note in MountTestBase.setUp
831 self.pool.apply(fuseFileConflictTestHelper, (self.mounttmp, collection.manifest_locator(), self.keeptmp, arvados.config.settings()))
834 def fuseUnlinkOpenFileTest(mounttmp):
835 class Test(unittest.TestCase):
837 with open(os.path.join(mounttmp, "file1.txt"), "w+") as f:
840 d1 = llfuse.listdir(os.path.join(mounttmp))
841 self.assertEqual(["file1.txt"], sorted(d1))
843 os.remove(os.path.join(mounttmp, "file1.txt"))
845 d1 = llfuse.listdir(os.path.join(mounttmp))
846 self.assertEqual([], sorted(d1))
849 self.assertEqual(f.read(), "foo")
853 self.assertEqual(f.read(), "foobar")
857 class FuseUnlinkOpenFileTest(MountTestBase):
859 collection = arvados.collection.Collection(api_client=self.api)
860 collection.save_new()
862 m = self.make_mount(fuse.CollectionDirectory)
864 m.new_collection(collection.api_response(), collection)
866 # See note in MountTestBase.setUp
867 self.pool.apply(fuseUnlinkOpenFileTest, (self.mounttmp,))
869 self.assertEqual(collection.manifest_text(), "")
872 def fuseMvFileBetweenCollectionsTest1(mounttmp, uuid1, uuid2):
873 class Test(unittest.TestCase):
875 with open(os.path.join(mounttmp, uuid1, "file1.txt"), "w") as f:
876 f.write("Hello world!")
878 d1 = os.listdir(os.path.join(mounttmp, uuid1))
879 self.assertEqual(["file1.txt"], sorted(d1))
880 d1 = os.listdir(os.path.join(mounttmp, uuid2))
881 self.assertEqual([], sorted(d1))
885 def fuseMvFileBetweenCollectionsTest2(mounttmp, uuid1, uuid2):
886 class Test(unittest.TestCase):
888 os.rename(os.path.join(mounttmp, uuid1, "file1.txt"), os.path.join(mounttmp, uuid2, "file2.txt"))
890 d1 = os.listdir(os.path.join(mounttmp, uuid1))
891 self.assertEqual([], sorted(d1))
892 d1 = os.listdir(os.path.join(mounttmp, uuid2))
893 self.assertEqual(["file2.txt"], sorted(d1))
897 class FuseMvFileBetweenCollectionsTest(MountTestBase):
899 collection1 = arvados.collection.Collection(api_client=self.api)
900 collection1.save_new()
902 collection2 = arvados.collection.Collection(api_client=self.api)
903 collection2.save_new()
905 m = self.make_mount(fuse.MagicDirectory)
907 # See note in MountTestBase.setUp
908 self.pool.apply(fuseMvFileBetweenCollectionsTest1, (self.mounttmp,
909 collection1.manifest_locator(),
910 collection2.manifest_locator()))
915 self.assertRegex(collection1.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
916 self.assertEqual(collection2.manifest_text(), "")
918 self.pool.apply(fuseMvFileBetweenCollectionsTest2, (self.mounttmp,
919 collection1.manifest_locator(),
920 collection2.manifest_locator()))
925 self.assertEqual(collection1.manifest_text(), "")
926 self.assertRegex(collection2.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file2\.txt$")
928 collection1.stop_threads()
929 collection2.stop_threads()
932 def fuseMvDirBetweenCollectionsTest1(mounttmp, uuid1, uuid2):
933 class Test(unittest.TestCase):
935 os.mkdir(os.path.join(mounttmp, uuid1, "testdir"))
936 with open(os.path.join(mounttmp, uuid1, "testdir", "file1.txt"), "w") as f:
937 f.write("Hello world!")
939 d1 = os.listdir(os.path.join(mounttmp, uuid1))
940 self.assertEqual(["testdir"], sorted(d1))
941 d1 = os.listdir(os.path.join(mounttmp, uuid1, "testdir"))
942 self.assertEqual(["file1.txt"], sorted(d1))
944 d1 = os.listdir(os.path.join(mounttmp, uuid2))
945 self.assertEqual([], sorted(d1))
950 def fuseMvDirBetweenCollectionsTest2(mounttmp, uuid1, uuid2):
951 class Test(unittest.TestCase):
953 os.rename(os.path.join(mounttmp, uuid1, "testdir"), os.path.join(mounttmp, uuid2, "testdir2"))
955 d1 = os.listdir(os.path.join(mounttmp, uuid1))
956 self.assertEqual([], sorted(d1))
958 d1 = os.listdir(os.path.join(mounttmp, uuid2))
959 self.assertEqual(["testdir2"], sorted(d1))
960 d1 = os.listdir(os.path.join(mounttmp, uuid2, "testdir2"))
961 self.assertEqual(["file1.txt"], sorted(d1))
963 with open(os.path.join(mounttmp, uuid2, "testdir2", "file1.txt"), "r") as f:
964 self.assertEqual(f.read(), "Hello world!")
968 class FuseMvDirBetweenCollectionsTest(MountTestBase):
970 collection1 = arvados.collection.Collection(api_client=self.api)
971 collection1.save_new()
973 collection2 = arvados.collection.Collection(api_client=self.api)
974 collection2.save_new()
976 m = self.make_mount(fuse.MagicDirectory)
978 # See note in MountTestBase.setUp
979 self.pool.apply(fuseMvDirBetweenCollectionsTest1, (self.mounttmp,
980 collection1.manifest_locator(),
981 collection2.manifest_locator()))
986 self.assertRegex(collection1.manifest_text(), r"\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
987 self.assertEqual(collection2.manifest_text(), "")
989 self.pool.apply(fuseMvDirBetweenCollectionsTest2, (self.mounttmp,
990 collection1.manifest_locator(),
991 collection2.manifest_locator()))
996 self.assertEqual(collection1.manifest_text(), "")
997 self.assertRegex(collection2.manifest_text(), r"\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
999 collection1.stop_threads()
1000 collection2.stop_threads()
1002 def fuseProjectMkdirTestHelper1(mounttmp):
1003 class Test(unittest.TestCase):
1005 os.mkdir(os.path.join(mounttmp, "testcollection"))
1006 with self.assertRaises(OSError):
1007 os.mkdir(os.path.join(mounttmp, "testcollection"))
1010 def fuseProjectMkdirTestHelper2(mounttmp):
1011 class Test(unittest.TestCase):
1013 with open(os.path.join(mounttmp, "testcollection", "file1.txt"), "w") as f:
1014 f.write("Hello world!")
1015 with self.assertRaises(OSError):
1016 os.rmdir(os.path.join(mounttmp, "testcollection"))
1017 os.remove(os.path.join(mounttmp, "testcollection", "file1.txt"))
1018 with self.assertRaises(OSError):
1019 os.remove(os.path.join(mounttmp, "testcollection"))
1020 os.rmdir(os.path.join(mounttmp, "testcollection"))
1023 class FuseProjectMkdirRmdirTest(MountTestBase):
1025 self.make_mount(fuse.ProjectDirectory,
1026 project_object=self.api.users().current().execute())
1028 d1 = llfuse.listdir(self.mounttmp)
1029 self.assertNotIn('testcollection', d1)
1031 self.pool.apply(fuseProjectMkdirTestHelper1, (self.mounttmp,))
1033 d1 = llfuse.listdir(self.mounttmp)
1034 self.assertIn('testcollection', d1)
1036 self.pool.apply(fuseProjectMkdirTestHelper2, (self.mounttmp,))
1038 d1 = llfuse.listdir(self.mounttmp)
1039 self.assertNotIn('testcollection', d1)
1042 def fuseProjectMvTestHelper1(mounttmp):
1043 class Test(unittest.TestCase):
1045 d1 = llfuse.listdir(mounttmp)
1046 self.assertNotIn('testcollection', d1)
1048 os.mkdir(os.path.join(mounttmp, "testcollection"))
1050 d1 = llfuse.listdir(mounttmp)
1051 self.assertIn('testcollection', d1)
1053 with self.assertRaises(OSError):
1054 os.rename(os.path.join(mounttmp, "testcollection"), os.path.join(mounttmp, 'Unrestricted public data'))
1056 os.rename(os.path.join(mounttmp, "testcollection"), os.path.join(mounttmp, 'Unrestricted public data', 'testcollection'))
1058 d1 = llfuse.listdir(mounttmp)
1059 self.assertNotIn('testcollection', d1)
1061 d1 = llfuse.listdir(os.path.join(mounttmp, 'Unrestricted public data'))
1062 self.assertIn('testcollection', d1)
1066 class FuseProjectMvTest(MountTestBase):
1068 self.make_mount(fuse.ProjectDirectory,
1069 project_object=self.api.users().current().execute())
1071 self.pool.apply(fuseProjectMvTestHelper1, (self.mounttmp,))
1074 def fuseFsyncTestHelper(mounttmp, k):
1075 class Test(unittest.TestCase):
1077 fd = os.open(os.path.join(mounttmp, k), os.O_RDONLY)
1083 class FuseFsyncTest(FuseMagicTest):
1085 self.make_mount(fuse.MagicDirectory)
1086 self.pool.apply(fuseFsyncTestHelper, (self.mounttmp, self.testcollection))
1089 class MagicDirApiError(FuseMagicTest):
1091 api = mock.MagicMock()
1092 api.keep.block_cache = mock.MagicMock(cache_max=1)
1093 super(MagicDirApiError, self).setUp(api=api)
1094 api.collections().get().execute.side_effect = iter([
1095 Exception('API fail'),
1097 "manifest_text": self.test_manifest,
1098 "portable_data_hash": self.test_manifest_pdh,
1101 api.keep.get.side_effect = Exception('Keep fail')
1104 with mock.patch('arvados_fuse.fresh.FreshBase._poll_time', new_callable=mock.PropertyMock, return_value=60) as mock_poll_time:
1105 self.make_mount(fuse.MagicDirectory)
1107 self.operations.inodes.inode_cache.cap = 1
1108 self.operations.inodes.inode_cache.min_entries = 2
1110 with self.assertRaises(OSError):
1111 llfuse.listdir(os.path.join(self.mounttmp, self.testcollection))
1113 llfuse.listdir(os.path.join(self.mounttmp, self.testcollection))
1116 class SanitizeFilenameTest(MountTestBase):
1117 def test_sanitize_filename(self):
1118 pdir = fuse.ProjectDirectory(
1119 1, fuse.Inodes(None), self.api, 0, False, None,
1120 project_object=self.api.users().current().execute(),
1140 for f in acceptable:
1141 self.assertEqual(f, pdir.sanitize_filename(f))
1142 for f in unacceptable:
1143 self.assertNotEqual(f, pdir.sanitize_filename(f))
1144 # The sanitized filename should be the same length, though.
1145 self.assertEqual(len(f), len(pdir.sanitize_filename(f)))
1147 self.assertEqual("_", pdir.sanitize_filename(""))
1148 self.assertEqual("_", pdir.sanitize_filename("."))
1149 self.assertEqual("__", pdir.sanitize_filename(".."))
1152 class FuseMagicTestPDHOnly(MountTestBase):
1153 def setUp(self, api=None):
1154 super(FuseMagicTestPDHOnly, self).setUp(api=api)
1156 cw = arvados.collection.Collection()
1157 with cw.open('thing1.txt', 'w') as f:
1160 self.testcollection = cw.portable_data_hash()
1161 self.test_manifest = cw.manifest_text()
1162 created = self.api.collections().create(body={"manifest_text":self.test_manifest}).execute()
1163 self.testcollectionuuid = str(created['uuid'])
1165 def verify_pdh_only(self, pdh_only=False, skip_pdh_only=False):
1166 if skip_pdh_only is True:
1167 self.make_mount(fuse.MagicDirectory) # in this case, the default by_id applies
1169 self.make_mount(fuse.MagicDirectory, pdh_only=pdh_only)
1171 mount_ls = llfuse.listdir(self.mounttmp)
1172 self.assertIn('README', mount_ls)
1173 self.assertFalse(any(arvados.util.keep_locator_pattern.match(fn) or
1174 arvados.util.uuid_pattern.match(fn)
1175 for fn in mount_ls),
1176 "new FUSE MagicDirectory lists Collection")
1178 # look up using pdh should succeed in all cases
1179 self.assertDirContents(self.testcollection, ['thing1.txt'])
1180 self.assertDirContents(os.path.join('by_id', self.testcollection),
1182 mount_ls = llfuse.listdir(self.mounttmp)
1183 self.assertIn('README', mount_ls)
1184 self.assertIn(self.testcollection, mount_ls)
1185 self.assertIn(self.testcollection,
1186 llfuse.listdir(os.path.join(self.mounttmp, 'by_id')))
1189 files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
1191 for k, v in files.items():
1192 with open(os.path.join(self.mounttmp, k), 'rb') as f:
1193 self.assertEqual(v, f.read().decode())
1195 # look up using uuid should fail when pdh_only is set
1196 if pdh_only is True:
1197 with self.assertRaises(OSError):
1198 self.assertDirContents(os.path.join('by_id', self.testcollectionuuid),
1201 self.assertDirContents(os.path.join('by_id', self.testcollectionuuid),
1204 def test_with_pdh_only_true(self):
1205 self.verify_pdh_only(pdh_only=True)
1207 def test_with_pdh_only_false(self):
1208 self.verify_pdh_only(pdh_only=False)
1210 def test_with_default_by_id(self):
1211 self.verify_pdh_only(skip_pdh_only=True)
1214 class SlashSubstitutionTest(IntegrationTest):
1217 '--mount-home', 'zzz',
1222 super(SlashSubstitutionTest, self).setUp()
1224 self.api = arvados.safeapi.ThreadSafeApiCache(
1225 arvados.config.settings(),
1228 self.testcoll = self.api.collections().create(body={"name": "foo/bar/baz"}).execute()
1229 self.testcolleasy = self.api.collections().create(body={"name": "foo-bar-baz"}).execute()
1230 self.fusename = 'foo[SLASH]bar[SLASH]baz'
1232 @IntegrationTest.mount(argv=mnt_args)
1233 def test_slash_substitution_before_listing(self):
1234 self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
1235 self.checkContents()
1237 def _test_slash_substitution_before_listing(self, tmpdir, fusename):
1238 with open(os.path.join(tmpdir, 'foo-bar-baz', 'waz'), 'w') as f:
1240 with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
1243 @IntegrationTest.mount(argv=mnt_args)
1244 @mock.patch('arvados.util.get_config_once')
1245 def test_slash_substitution_after_listing(self, get_config_once):
1246 get_config_once.return_value = {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
1247 self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
1248 self.checkContents()
1250 def _test_slash_substitution_after_listing(self, tmpdir, fusename):
1251 with open(os.path.join(tmpdir, 'foo-bar-baz', 'waz'), 'w') as f:
1254 with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
1257 def checkContents(self):
1258 self.assertRegex(self.api.collections().get(uuid=self.testcoll['uuid']).execute()['manifest_text'], r' acbd18db') # md5(foo)
1259 self.assertRegex(self.api.collections().get(uuid=self.testcolleasy['uuid']).execute()['manifest_text'], r' f561aaf6') # md5(xxx)
1261 @IntegrationTest.mount(argv=mnt_args)
1262 @mock.patch('arvados.util.get_config_once')
1263 def test_slash_substitution_conflict(self, get_config_once):
1264 self.testcollconflict = self.api.collections().create(body={"name": self.fusename}).execute()
1265 get_config_once.return_value = {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
1266 self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
1267 self.assertRegex(self.api.collections().get(uuid=self.testcollconflict['uuid']).execute()['manifest_text'], r' acbd18db') # md5(foo)
1268 # foo/bar/baz collection unchanged, because it is masked by foo[SLASH]bar[SLASH]baz
1269 self.assertEqual(self.api.collections().get(uuid=self.testcoll['uuid']).execute()['manifest_text'], '')
1271 def _test_slash_substitution_conflict(self, tmpdir, fusename):
1272 with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
1275 class StorageClassesTest(IntegrationTest):
1278 '--mount-home', 'homedir',
1282 super(StorageClassesTest, self).setUp()
1283 self.api = arvados.safeapi.ThreadSafeApiCache(
1284 arvados.config.settings(),
1288 @IntegrationTest.mount(argv=mnt_args)
1289 def test_collection_default_storage_classes(self):
1290 coll_path = os.path.join(self.mnt, 'homedir', 'a_collection')
1291 self.api.collections().create(body={'name':'a_collection'}).execute()
1292 self.pool_test(coll_path)
1294 def _test_collection_default_storage_classes(self, coll):
1295 self.assertEqual(storage_classes_desired(coll), ['default'])
1297 @IntegrationTest.mount(argv=mnt_args+['--storage-classes', 'foo'])
1298 def test_collection_custom_storage_classes(self):
1299 coll_path = os.path.join(self.mnt, 'homedir', 'new_coll')
1301 self.pool_test(coll_path)
1303 def _test_collection_custom_storage_classes(self, coll):
1304 self.assertEqual(storage_classes_desired(coll), ['foo'])
1306 def _readonlyCollectionTestHelper(mounttmp):
1307 f = open(os.path.join(mounttmp, 'thing1.txt'), 'rt')
1308 # Testing that close() doesn't raise an error.
1311 class ReadonlyCollectionTest(MountTestBase):
1313 super(ReadonlyCollectionTest, self).setUp()
1314 cw = arvados.collection.Collection()
1315 with cw.open('thing1.txt', 'wt') as f:
1317 cw.save_new(owner_uuid=run_test_server.fixture("groups")["aproject"]["uuid"])
1318 self.testcollection = cw.api_response()
1321 settings = arvados.config.settings().copy()
1322 settings["ARVADOS_API_TOKEN"] = run_test_server.fixture("api_client_authorizations")["project_viewer"]["api_token"]
1323 self.api = arvados.safeapi.ThreadSafeApiCache(settings, version='v1')
1324 self.make_mount(fuse.CollectionDirectory, collection_record=self.testcollection, enable_write=False)
1326 self.pool.apply(_readonlyCollectionTestHelper, (self.mounttmp,))
1329 @parameterized.parameterized_class([
1330 {'root_class': fusedir.ProjectDirectory, 'root_kwargs': {
1331 'project_object': run_test_server.fixture('users')['admin'],
1333 {'root_class': fusedir.ProjectDirectory, 'root_kwargs': {
1334 'project_object': run_test_server.fixture('groups')['public'],
1337 class UnsupportedCreateTest(MountTestBase):
1343 if 'prefs' in self.root_kwargs.get('project_object', ()):
1344 self.root_kwargs['project_object']['prefs'] = {}
1345 self.make_mount(self.root_class, **self.root_kwargs)
1346 # Make sure the directory knows about its top-level ents.
1347 os.listdir(self.mounttmp)
1349 def test_create(self):
1350 test_path = Path(self.mounttmp, 'test_create')
1351 with self.assertRaises(OSError) as exc_check:
1352 with test_path.open('w'):
1354 self.assertEqual(exc_check.exception.errno, errno.ENOTSUP)
1357 # FIXME: IMO, for consistency with the "create inside a project" case,
1358 # these operations should also return ENOTSUP instead of EPERM.
1359 # Right now they're returning EPERM because the clasess' writable() method
1360 # usually returns False, and the Operations class transforms that accordingly.
1361 # However, for cases where the mount will never be writable, I think ENOTSUP
1362 # is a clearer error: it lets the user know they can't fix the problem by
1363 # adding permissions in Arvados, etc.
1364 @parameterized.parameterized_class([
1365 {'root_class': fusedir.MagicDirectory,
1366 'preset_dir': 'by_id',
1367 'preset_file': 'README',
1370 {'root_class': fusedir.SharedDirectory,
1372 'exclude': run_test_server.fixture('users')['admin']['uuid'],
1374 'preset_dir': 'Active User',
1377 {'root_class': fusedir.TagDirectory,
1379 'tag': run_test_server.fixture('links')['foo_collection_tag']['name'],
1381 'preset_dir': run_test_server.fixture('collections')['foo_collection_in_aproject']['uuid'],
1384 {'root_class': fusedir.TagsDirectory,
1385 'preset_dir': run_test_server.fixture('links')['foo_collection_tag']['name'],
1388 class UnsupportedOperationsTest(UnsupportedCreateTest):
1392 def test_create(self):
1393 test_path = Path(self.mounttmp, 'test_create')
1394 with self.assertRaises(OSError) as exc_check:
1395 with test_path.open('w'):
1397 self.assertEqual(exc_check.exception.errno, errno.EPERM)
1399 def test_mkdir(self):
1400 test_path = Path(self.mounttmp, 'test_mkdir')
1401 with self.assertRaises(OSError) as exc_check:
1403 self.assertEqual(exc_check.exception.errno, errno.EPERM)
1405 def test_rename(self):
1406 src_name = self.preset_dir or self.preset_file
1407 if src_name is None:
1409 test_src = Path(self.mounttmp, src_name)
1410 test_dst = test_src.with_name('test_dst')
1411 with self.assertRaises(OSError) as exc_check:
1412 test_src.rename(test_dst)
1413 self.assertEqual(exc_check.exception.errno, errno.EPERM)
1415 def test_rmdir(self):
1416 if self.preset_dir is None:
1418 test_path = Path(self.mounttmp, self.preset_dir)
1419 with self.assertRaises(OSError) as exc_check:
1421 self.assertEqual(exc_check.exception.errno, errno.EPERM)
1423 def test_unlink(self):
1424 if self.preset_file is None:
1426 test_path = Path(self.mounttmp, self.preset_file)
1427 with self.assertRaises(OSError) as exc_check:
1429 self.assertEqual(exc_check.exception.errno, errno.EPERM)