+ self.assertEqual("_", pdir.sanitize_filename(""))
+ self.assertEqual("_", pdir.sanitize_filename("."))
+ self.assertEqual("__", pdir.sanitize_filename(".."))
+
+
+class FuseMagicTestPDHOnly(MountTestBase):
+ def setUp(self, api=None):
+ super(FuseMagicTestPDHOnly, self).setUp(api=api)
+
+ cw = arvados.CollectionWriter()
+
+ cw.start_new_file('thing1.txt')
+ cw.write("data 1")
+
+ self.testcollection = cw.finish()
+ self.test_manifest = cw.manifest_text()
+ created = self.api.collections().create(body={"manifest_text":self.test_manifest}).execute()
+ self.testcollectionuuid = str(created['uuid'])
+
+ def verify_pdh_only(self, pdh_only=False, skip_pdh_only=False):
+ if skip_pdh_only is True:
+ self.make_mount(fuse.MagicDirectory) # in this case, the default by_id applies
+ else:
+ self.make_mount(fuse.MagicDirectory, pdh_only=pdh_only)
+
+ mount_ls = llfuse.listdir(self.mounttmp)
+ self.assertIn('README', mount_ls)
+ self.assertFalse(any(arvados.util.keep_locator_pattern.match(fn) or
+ arvados.util.uuid_pattern.match(fn)
+ for fn in mount_ls),
+ "new FUSE MagicDirectory lists Collection")
+
+ # look up using pdh should succeed in all cases
+ self.assertDirContents(self.testcollection, ['thing1.txt'])
+ self.assertDirContents(os.path.join('by_id', self.testcollection),
+ ['thing1.txt'])
+ mount_ls = llfuse.listdir(self.mounttmp)
+ self.assertIn('README', mount_ls)
+ self.assertIn(self.testcollection, mount_ls)
+ self.assertIn(self.testcollection,
+ llfuse.listdir(os.path.join(self.mounttmp, 'by_id')))
+
+ files = {}
+ files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
+
+ for k, v in viewitems(files):
+ with open(os.path.join(self.mounttmp, k), 'rb') as f:
+ self.assertEqual(v, f.read().decode())
+
+ # look up using uuid should fail when pdh_only is set
+ if pdh_only is True:
+ with self.assertRaises(OSError):
+ self.assertDirContents(os.path.join('by_id', self.testcollectionuuid),
+ ['thing1.txt'])
+ else:
+ self.assertDirContents(os.path.join('by_id', self.testcollectionuuid),
+ ['thing1.txt'])
+
+ def test_with_pdh_only_true(self):
+ self.verify_pdh_only(pdh_only=True)
+
+ def test_with_pdh_only_false(self):
+ self.verify_pdh_only(pdh_only=False)
+
+ def test_with_default_by_id(self):
+ self.verify_pdh_only(skip_pdh_only=True)
+
+
+class SlashSubstitutionTest(IntegrationTest):
+ mnt_args = [
+ '--read-write',
+ '--mount-home', 'zzz',
+ ]
+
+ def setUp(self):
+ super(SlashSubstitutionTest, self).setUp()
+ self.api = arvados.safeapi.ThreadSafeApiCache(arvados.config.settings())
+ self.api.config = lambda: {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
+ self.testcoll = self.api.collections().create(body={"name": "foo/bar/baz"}).execute()
+ self.testcolleasy = self.api.collections().create(body={"name": "foo-bar-baz"}).execute()
+ self.fusename = 'foo[SLASH]bar[SLASH]baz'
+
+ @IntegrationTest.mount(argv=mnt_args)
+ @mock.patch('arvados.util.get_config_once')
+ def test_slash_substitution_before_listing(self, get_config_once):
+ get_config_once.return_value = {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
+ self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
+ self.checkContents()
+ @staticmethod
+ def _test_slash_substitution_before_listing(self, tmpdir, fusename):
+ with open(os.path.join(tmpdir, 'foo-bar-baz', 'waz'), 'w') as f:
+ f.write('xxx')
+ with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
+ f.write('foo')
+
+ @IntegrationTest.mount(argv=mnt_args)
+ @mock.patch('arvados.util.get_config_once')
+ def test_slash_substitution_after_listing(self, get_config_once):
+ get_config_once.return_value = {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
+ self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
+ self.checkContents()
+ @staticmethod
+ def _test_slash_substitution_after_listing(self, tmpdir, fusename):
+ with open(os.path.join(tmpdir, 'foo-bar-baz', 'waz'), 'w') as f:
+ f.write('xxx')
+ os.listdir(tmpdir)
+ with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
+ f.write('foo')
+
+ def checkContents(self):
+ self.assertRegexpMatches(self.api.collections().get(uuid=self.testcoll['uuid']).execute()['manifest_text'], ' acbd18db') # md5(foo)
+ self.assertRegexpMatches(self.api.collections().get(uuid=self.testcolleasy['uuid']).execute()['manifest_text'], ' f561aaf6') # md5(xxx)
+
+ @IntegrationTest.mount(argv=mnt_args)
+ @mock.patch('arvados.util.get_config_once')
+ def test_slash_substitution_conflict(self, get_config_once):
+ self.testcollconflict = self.api.collections().create(body={"name": self.fusename}).execute()
+ get_config_once.return_value = {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
+ self.pool_test(os.path.join(self.mnt, 'zzz'), self.fusename)
+ self.assertRegexpMatches(self.api.collections().get(uuid=self.testcollconflict['uuid']).execute()['manifest_text'], ' acbd18db') # md5(foo)
+ # foo/bar/baz collection unchanged, because it is masked by foo[SLASH]bar[SLASH]baz
+ self.assertEqual(self.api.collections().get(uuid=self.testcoll['uuid']).execute()['manifest_text'], '')
+ @staticmethod
+ def _test_slash_substitution_conflict(self, tmpdir, fusename):
+ with open(os.path.join(tmpdir, fusename, 'waz'), 'w') as f:
+ f.write('foo')