self.keeptmp = tempfile.mkdtemp()
os.environ['KEEP_LOCAL_STORE'] = self.keeptmp
self.mounttmp = tempfile.mkdtemp()
+ run_test_server.run(False)
+ run_test_server.authorize_with("admin")
+ self.api = api = fuse.SafeApi(arvados.config)
def tearDown(self):
+ run_test_server.stop()
+
# llfuse.close is buggy, so use fusermount instead.
#llfuse.close(unmount=True)
count = 0
cw.write("data 8")
self.testcollection = cw.finish()
+ self.api.collections().create(body={"manifest_text":cw.manifest_text()}).execute()
def runTest(self):
# Create the request handler
operations = fuse.Operations(os.getuid(), os.getgid())
- e = operations.inodes.add_entry(fuse.CollectionDirectory(llfuse.ROOT_INODE, operations.inodes, self.testcollection))
+ e = operations.inodes.add_entry(fuse.CollectionDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0, self.testcollection))
llfuse.init(operations, self.mounttmp, [])
t = threading.Thread(None, lambda: llfuse.main())
cw.write("data 1")
self.testcollection = cw.finish()
+ self.api.collections().create(body={"manifest_text":cw.manifest_text()}).execute()
def runTest(self):
# Create the request handler
operations = fuse.Operations(os.getuid(), os.getgid())
- e = operations.inodes.add_entry(fuse.MagicDirectory(llfuse.ROOT_INODE, operations.inodes))
+ e = operations.inodes.add_entry(fuse.MagicDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0))
self.mounttmp = tempfile.mkdtemp()
# now check some stuff
d1 = os.listdir(self.mounttmp)
d1.sort()
- self.assertEqual([], d1)
+ self.assertEqual(['README'], d1)
d2 = os.listdir(os.path.join(self.mounttmp, self.testcollection))
d2.sort()
d3 = os.listdir(self.mounttmp)
d3.sort()
- self.assertEqual([self.testcollection], d3)
+ self.assertEqual([self.testcollection, 'README'], d3)
files = {}
files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
class FuseTagsTest(MountTestBase):
- def setUp(self):
- super(FuseTagsTest, self).setUp()
-
- cw = arvados.CollectionWriter()
-
- cw.start_new_file('foo')
- cw.write("foo")
-
- self.testcollection = cw.finish()
-
- run_test_server.run()
-
def runTest(self):
- run_test_server.authorize_with("admin")
- api = arvados.api('v1', cache=False)
-
operations = fuse.Operations(os.getuid(), os.getgid())
- e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, api))
+ e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0))
llfuse.init(operations, self.mounttmp, [])
t = threading.Thread(None, lambda: llfuse.main())
d2 = os.listdir(os.path.join(self.mounttmp, 'foo_tag'))
d2.sort()
- self.assertEqual(['1f4b0bc7583c2a7f9102c395f4ffc5e3+45'], d2)
+ self.assertEqual(['zzzzz-4zz18-fy296fx3hot09f7'], d2)
- d3 = os.listdir(os.path.join(self.mounttmp, 'foo_tag', '1f4b0bc7583c2a7f9102c395f4ffc5e3+45'))
+ d3 = os.listdir(os.path.join(self.mounttmp, 'foo_tag', 'zzzzz-4zz18-fy296fx3hot09f7'))
d3.sort()
self.assertEqual(['foo'], d3)
- files = {}
- files[os.path.join(self.mounttmp, 'foo_tag', '1f4b0bc7583c2a7f9102c395f4ffc5e3+45', 'foo')] = 'foo'
-
- for k, v in files.items():
- with open(os.path.join(self.mounttmp, k)) as f:
- self.assertEqual(v, f.read())
-
-
- def tearDown(self):
- run_test_server.stop()
-
- super(FuseTagsTest, self).tearDown()
-
-class FuseTagsUpdateTestBase(MountTestBase):
+class FuseTagsUpdateTest(MountTestBase):
def runRealTest(self):
- run_test_server.authorize_with("admin")
- api = arvados.api('v1', cache=False)
-
operations = fuse.Operations(os.getuid(), os.getgid())
- e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, api, poll_time=1))
+ e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0, poll_time=1))
llfuse.init(operations, self.mounttmp, [])
t = threading.Thread(None, lambda: llfuse.main())
d1.sort()
self.assertEqual(['foo_tag'], d1)
- api.links().create(body={'link': {
+ self.api.links().create(body={'link': {
'head_uuid': 'fa7aeb5140e2848d39b416daeef4ffc5+45',
'link_class': 'tag',
'name': 'bar_tag'
d3.sort()
self.assertEqual(['fa7aeb5140e2848d39b416daeef4ffc5+45'], d3)
- l = api.links().create(body={'link': {
+ l = self.api.links().create(body={'link': {
'head_uuid': 'ea10d51bcf88862dbcc36eb292017dfd+45',
'link_class': 'tag',
'name': 'bar_tag'
d4.sort()
self.assertEqual(['ea10d51bcf88862dbcc36eb292017dfd+45', 'fa7aeb5140e2848d39b416daeef4ffc5+45'], d4)
- api.links().delete(uuid=l['uuid']).execute()
+ self.api.links().delete(uuid=l['uuid']).execute()
time.sleep(1)
self.assertEqual(['fa7aeb5140e2848d39b416daeef4ffc5+45'], d5)
-class FuseTagsUpdateTestWebsockets(FuseTagsUpdateTestBase):
- def setUp(self):
- super(FuseTagsUpdateTestWebsockets, self).setUp()
- run_test_server.run(True)
-
+class FuseSharedTest(MountTestBase):
def runTest(self):
- self.runRealTest()
-
- def tearDown(self):
- run_test_server.stop()
- super(FuseTagsUpdateTestWebsockets, self).tearDown()
-
-
-class FuseTagsUpdateTestPoll(FuseTagsUpdateTestBase):
- def setUp(self):
- super(FuseTagsUpdateTestPoll, self).setUp()
- run_test_server.run(False)
+ operations = fuse.Operations(os.getuid(), os.getgid())
+ e = operations.inodes.add_entry(fuse.SharedDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0, self.api.users().current().execute()['uuid']))
- def runTest(self):
- self.runRealTest()
+ llfuse.init(operations, self.mounttmp, [])
+ t = threading.Thread(None, lambda: llfuse.main())
+ t.start()
- def tearDown(self):
- run_test_server.stop()
- super(FuseTagsUpdateTestPoll, self).tearDown()
+ # wait until the driver is finished initializing
+ operations.initlock.wait()
+ # shared_dirs is a list of the directories exposed
+ # by fuse.SharedDirectory (i.e. any object visible
+ # to the current user)
+ shared_dirs = os.listdir(self.mounttmp)
+ shared_dirs.sort()
+ self.assertIn('FUSE User', shared_dirs)
+
+ # fuse_user_dirs is a list of the directories owned
+ # by the "FUSE User" test user.
+ fuse_user_dirs = os.listdir(os.path.join(self.mounttmp, 'FUSE User'))
+ fuse_user_dirs.sort()
+ self.assertEqual(['FUSE Test Project',
+ 'collection #1 owned by FUSE',
+ 'collection #2 owned by FUSE'
+ ], fuse_user_dirs)
+
+ # test_proj_files is a list of the files in the FUSE Test Project.
+ test_proj_files = os.listdir(os.path.join(self.mounttmp, 'FUSE User', 'FUSE Test Project'))
+ test_proj_files.sort()
+ self.assertEqual(['Pipeline Template in FUSE Project.pipelineTemplate',
+ 'collection in FUSE project',
+ ], test_proj_files)
+
+ with open(os.path.join(
+ self.mounttmp,
+ 'FUSE User',
+ 'FUSE Test Project',
+ 'Pipeline Template in FUSE Project.pipelineTemplate')) as f:
+ j = json.load(f)
+ self.assertEqual("Pipeline Template in FUSE Project", j['name'])
-class FuseGroupsTest(MountTestBase):
- def setUp(self):
- super(FuseGroupsTest, self).setUp()
- run_test_server.run()
+class FuseHomeTest(MountTestBase):
def runTest(self):
- run_test_server.authorize_with("admin")
- api = arvados.api('v1', cache=False)
-
operations = fuse.Operations(os.getuid(), os.getgid())
- e = operations.inodes.add_entry(fuse.GroupsDirectory(llfuse.ROOT_INODE, operations.inodes, api))
+ e = operations.inodes.add_entry(fuse.ProjectDirectory(llfuse.ROOT_INODE, operations.inodes, self.api, 0, self.api.users().current().execute()))
llfuse.init(operations, self.mounttmp, [])
t = threading.Thread(None, lambda: llfuse.main())
d1 = os.listdir(self.mounttmp)
d1.sort()
- self.assertIn('zzzzz-j7d0g-v955i6s2oi1cbso', d1)
+ self.assertIn('Unrestricted public data', d1)
- d2 = os.listdir(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso'))
+ d2 = os.listdir(os.path.join(self.mounttmp, 'Unrestricted public data'))
d2.sort()
- self.assertEqual(['1f4b0bc7583c2a7f9102c395f4ffc5e3+45 added sometime',
- "I'm a job in a folder",
- "I'm a template in a folder",
- "zzzzz-j58dm-5gid26432uujf79",
- "zzzzz-j58dm-7r18rnd5nzhg5yk",
- "zzzzz-j58dm-ypsjlol9dofwijz",
- "zzzzz-j7d0g-axqo7eu9pwvna1x"
- ], d2)
-
- d3 = os.listdir(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso', 'zzzzz-j7d0g-axqo7eu9pwvna1x'))
- d3.sort()
- self.assertEqual(["I'm in a subfolder, too",
- "zzzzz-j58dm-c40lddwcqqr1ffs"
- ], d3)
-
- with open(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso', "I'm a template in a folder")) as f:
- j = json.load(f)
- self.assertEqual("Two Part Pipeline Template", j['name'])
+ self.assertEqual(['GNU General Public License, version 3'], d2)
- def tearDown(self):
- run_test_server.stop()
- super(FuseGroupsTest, self).tearDown()
+ d3 = os.listdir(os.path.join(self.mounttmp, 'Unrestricted public data', 'GNU General Public License, version 3'))
+ d3.sort()
+ self.assertEqual(["GNU_General_Public_License,_version_3.pdf"], d3)
projects / arvados.git / blobdiff