sdk/python/test_mount.py

   1 import unittest
   2 import arvados
   3 import arvados.fuse as fuse
   4 import threading
   5 import time
   6 import os
   7 import llfuse
   8 import tempfile
   9 import shutil
  10 import subprocess
  11 import glob
  12 import run_test_server
  13 import json
  14
  15 class MountTestBase(unittest.TestCase):
  16     def setUp(self):
  17         self.keeptmp = tempfile.mkdtemp()
  18         os.environ['KEEP_LOCAL_STORE'] = self.keeptmp
  19         self.mounttmp = tempfile.mkdtemp()
  20
  21     def tearDown(self):
  22         # llfuse.close is buggy, so use fusermount instead.
  23         #llfuse.close(unmount=True)
  24         subprocess.call(["fusermount", "-u", self.mounttmp])
  25
  26         os.rmdir(self.mounttmp)
  27         shutil.rmtree(self.keeptmp)
  28
  29
  30 class FuseMountTest(MountTestBase):
  31     def setUp(self):
  32         super(FuseMountTest, self).setUp()
  33
  34         cw = arvados.CollectionWriter()
  35
  36         cw.start_new_file('thing1.txt')
  37         cw.write("data 1")
  38         cw.start_new_file('thing2.txt')
  39         cw.write("data 2")
  40         cw.start_new_stream('dir1')
  41
  42         cw.start_new_file('thing3.txt')
  43         cw.write("data 3")
  44         cw.start_new_file('thing4.txt')
  45         cw.write("data 4")
  46
  47         cw.start_new_stream('dir2')
  48         cw.start_new_file('thing5.txt')
  49         cw.write("data 5")
  50         cw.start_new_file('thing6.txt')
  51         cw.write("data 6")
  52
  53         cw.start_new_stream('dir2/dir3')
  54         cw.start_new_file('thing7.txt')
  55         cw.write("data 7")
  56
  57         cw.start_new_file('thing8.txt')
  58         cw.write("data 8")
  59
  60         self.testcollection = cw.finish()
  61
  62     def runTest(self):
  63         # Create the request handler
  64         operations = fuse.Operations(os.getuid(), os.getgid())
  65         e = operations.inodes.add_entry(fuse.CollectionDirectory(llfuse.ROOT_INODE, operations.inodes, self.testcollection))
  66
  67         llfuse.init(operations, self.mounttmp, [])
  68         t = threading.Thread(None, lambda: llfuse.main())
  69         t.start()
  70
  71         # wait until the driver is finished initializing
  72         operations.initlock.wait()
  73
  74         # now check some stuff
  75         d1 = os.listdir(self.mounttmp)
  76         d1.sort()
  77         self.assertEqual(['dir1', 'dir2', 'thing1.txt', 'thing2.txt'], d1)
  78
  79         d2 = os.listdir(os.path.join(self.mounttmp, 'dir1'))
  80         d2.sort()
  81         self.assertEqual(['thing3.txt', 'thing4.txt'], d2)
  82
  83         d3 = os.listdir(os.path.join(self.mounttmp, 'dir2'))
  84         d3.sort()
  85         self.assertEqual(['dir3', 'thing5.txt', 'thing6.txt'], d3)
  86
  87         d4 = os.listdir(os.path.join(self.mounttmp, 'dir2/dir3'))
  88         d4.sort()
  89         self.assertEqual(['thing7.txt', 'thing8.txt'], d4)
  90
  91         files = {'thing1.txt': 'data 1',
  92                  'thing2.txt': 'data 2',
  93                  'dir1/thing3.txt': 'data 3',
  94                  'dir1/thing4.txt': 'data 4',
  95                  'dir2/thing5.txt': 'data 5',
  96                  'dir2/thing6.txt': 'data 6',
  97                  'dir2/dir3/thing7.txt': 'data 7',
  98                  'dir2/dir3/thing8.txt': 'data 8'}
  99
 100         for k, v in files.items():
 101             with open(os.path.join(self.mounttmp, k)) as f:
 102                 self.assertEqual(v, f.read())
 103
 104
 105 class FuseMagicTest(MountTestBase):
 106     def setUp(self):
 107         super(FuseMagicTest, self).setUp()
 108
 109         cw = arvados.CollectionWriter()
 110
 111         cw.start_new_file('thing1.txt')
 112         cw.write("data 1")
 113
 114         self.testcollection = cw.finish()
 115
 116     def runTest(self):
 117         # Create the request handler
 118         operations = fuse.Operations(os.getuid(), os.getgid())
 119         e = operations.inodes.add_entry(fuse.MagicDirectory(llfuse.ROOT_INODE, operations.inodes))
 120
 121         self.mounttmp = tempfile.mkdtemp()
 122
 123         llfuse.init(operations, self.mounttmp, [])
 124         t = threading.Thread(None, lambda: llfuse.main())
 125         t.start()
 126
 127         # wait until the driver is finished initializing
 128         operations.initlock.wait()
 129
 130         # now check some stuff
 131         d1 = os.listdir(self.mounttmp)
 132         d1.sort()
 133         self.assertEqual([], d1)
 134
 135         d2 = os.listdir(os.path.join(self.mounttmp, self.testcollection))
 136         d2.sort()
 137         self.assertEqual(['thing1.txt'], d2)
 138
 139         d3 = os.listdir(self.mounttmp)
 140         d3.sort()
 141         self.assertEqual([self.testcollection], d3)
 142
 143         files = {}
 144         files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
 145
 146         for k, v in files.items():
 147             with open(os.path.join(self.mounttmp, k)) as f:
 148                 self.assertEqual(v, f.read())
 149
 150
 151 class FuseTagsTest(MountTestBase):
 152     def setUp(self):
 153         super(FuseTagsTest, self).setUp()
 154
 155         cw = arvados.CollectionWriter()
 156
 157         cw.start_new_file('foo')
 158         cw.write("foo")
 159
 160         self.testcollection = cw.finish()
 161
 162         run_test_server.run()
 163
 164     def runTest(self):
 165         run_test_server.authorize_with("admin")
 166         api = arvados.api('v1', cache=False)
 167
 168         operations = fuse.Operations(os.getuid(), os.getgid())
 169         e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, api))
 170
 171         llfuse.init(operations, self.mounttmp, [])
 172         t = threading.Thread(None, lambda: llfuse.main())
 173         t.start()
 174
 175         # wait until the driver is finished initializing
 176         operations.initlock.wait()
 177
 178         d1 = os.listdir(self.mounttmp)
 179         d1.sort()
 180         self.assertEqual(['foo_tag'], d1)
 181
 182         d2 = os.listdir(os.path.join(self.mounttmp, 'foo_tag'))
 183         d2.sort()
 184         self.assertEqual(['1f4b0bc7583c2a7f9102c395f4ffc5e3+45'], d2)
 185
 186         d3 = os.listdir(os.path.join(self.mounttmp, 'foo_tag', '1f4b0bc7583c2a7f9102c395f4ffc5e3+45'))
 187         d3.sort()
 188         self.assertEqual(['foo'], d3)
 189
 190         files = {}
 191         files[os.path.join(self.mounttmp, 'foo_tag', '1f4b0bc7583c2a7f9102c395f4ffc5e3+45', 'foo')] = 'foo'
 192
 193         for k, v in files.items():
 194             with open(os.path.join(self.mounttmp, k)) as f:
 195                 self.assertEqual(v, f.read())
 196
 197
 198     def tearDown(self):
 199         run_test_server.stop()
 200
 201         super(FuseTagsTest, self).tearDown()
 202
 203 class FuseTagsUpdateTestBase(MountTestBase):
 204
 205     def runRealTest(self):
 206         run_test_server.authorize_with("admin")
 207         api = arvados.api('v1', cache=False)
 208
 209         operations = fuse.Operations(os.getuid(), os.getgid())
 210         e = operations.inodes.add_entry(fuse.TagsDirectory(llfuse.ROOT_INODE, operations.inodes, api, poll_time=1))
 211
 212         llfuse.init(operations, self.mounttmp, [])
 213         t = threading.Thread(None, lambda: llfuse.main())
 214         t.start()
 215
 216         # wait until the driver is finished initializing
 217         operations.initlock.wait()
 218
 219         d1 = os.listdir(self.mounttmp)
 220         d1.sort()
 221         self.assertEqual(['foo_tag'], d1)
 222
 223         api.links().create(body={'link': {
 224             'head_uuid': 'fa7aeb5140e2848d39b416daeef4ffc5+45',
 225             'link_class': 'tag',
 226             'name': 'bar_tag'
 227         }}).execute()
 228
 229         time.sleep(1)
 230
 231         d2 = os.listdir(self.mounttmp)
 232         d2.sort()
 233         self.assertEqual(['bar_tag', 'foo_tag'], d2)
 234
 235         d3 = os.listdir(os.path.join(self.mounttmp, 'bar_tag'))
 236         d3.sort()
 237         self.assertEqual(['fa7aeb5140e2848d39b416daeef4ffc5+45'], d3)
 238
 239         l = api.links().create(body={'link': {
 240             'head_uuid': 'ea10d51bcf88862dbcc36eb292017dfd+45',
 241             'link_class': 'tag',
 242             'name': 'bar_tag'
 243         }}).execute()
 244
 245         time.sleep(1)
 246
 247         d4 = os.listdir(os.path.join(self.mounttmp, 'bar_tag'))
 248         d4.sort()
 249         self.assertEqual(['ea10d51bcf88862dbcc36eb292017dfd+45', 'fa7aeb5140e2848d39b416daeef4ffc5+45'], d4)
 250
 251         api.links().delete(uuid=l['uuid']).execute()
 252
 253         time.sleep(1)
 254
 255         d5 = os.listdir(os.path.join(self.mounttmp, 'bar_tag'))
 256         d5.sort()
 257         self.assertEqual(['fa7aeb5140e2848d39b416daeef4ffc5+45'], d5)
 258
 259
 260 class FuseTagsUpdateTestWebsockets(FuseTagsUpdateTestBase):
 261     def setUp(self):
 262         super(FuseTagsUpdateTestWebsockets, self).setUp()
 263         run_test_server.run(True)
 264
 265     def runTest(self):
 266         self.runRealTest()
 267
 268     def tearDown(self):
 269         run_test_server.stop()
 270         super(FuseTagsUpdateTestWebsockets, self).tearDown()
 271
 272
 273 class FuseTagsUpdateTestPoll(FuseTagsUpdateTestBase):
 274     def setUp(self):
 275         super(FuseTagsUpdateTestPoll, self).setUp()
 276         run_test_server.run(False)
 277
 278     def runTest(self):
 279         self.runRealTest()
 280
 281     def tearDown(self):
 282         run_test_server.stop()
 283         super(FuseTagsUpdateTestPoll, self).tearDown()
 284
 285
 286 class FuseGroupsTest(MountTestBase):
 287     def setUp(self):
 288         super(FuseGroupsTest, self).setUp()
 289         run_test_server.run()
 290
 291     def runTest(self):
 292         run_test_server.authorize_with("admin")
 293         api = arvados.api('v1', cache=False)
 294
 295         operations = fuse.Operations(os.getuid(), os.getgid())
 296         e = operations.inodes.add_entry(fuse.GroupsDirectory(llfuse.ROOT_INODE, operations.inodes, api))
 297
 298         llfuse.init(operations, self.mounttmp, [])
 299         t = threading.Thread(None, lambda: llfuse.main())
 300         t.start()
 301
 302         # wait until the driver is finished initializing
 303         operations.initlock.wait()
 304
 305         d1 = os.listdir(self.mounttmp)
 306         d1.sort()
 307         self.assertIn('zzzzz-j7d0g-v955i6s2oi1cbso', d1)
 308
 309         d2 = os.listdir(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso'))
 310         d2.sort()
 311         self.assertEqual(['1f4b0bc7583c2a7f9102c395f4ffc5e3+45 added sometime',
 312                           "I'm a job in a folder",
 313                           "I'm a template in a folder",
 314                           "zzzzz-j58dm-5gid26432uujf79",
 315                           "zzzzz-j58dm-7r18rnd5nzhg5yk",
 316                           "zzzzz-j58dm-ypsjlol9dofwijz",
 317                           "zzzzz-j7d0g-axqo7eu9pwvna1x"
 318                       ], d2)
 319
 320         d3 = os.listdir(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso', 'zzzzz-j7d0g-axqo7eu9pwvna1x'))
 321         d3.sort()
 322         self.assertEqual(["I'm in a subfolder, too",
 323                           "zzzzz-j58dm-c40lddwcqqr1ffs"
 324                       ], d3)
 325
 326         with open(os.path.join(self.mounttmp, 'zzzzz-j7d0g-v955i6s2oi1cbso', "I'm a template in a folder")) as f:
 327             j = json.load(f)
 328             self.assertEqual("Two Part Pipeline Template", j['name'])
 329
 330     def tearDown(self):
 331         run_test_server.stop()
 332         super(FuseGroupsTest, self).tearDown()