3198: Implement cache management for directory objects.

[arvados.git] / services / fuse / tests / test_mount.py

diff --git a/services/fuse/tests/test_mount.py b/services/fuse/tests/test_mount.py
index 104373b7ad3972af52e2a17d27573e22b759506e..4802d69c313f1a1053ef2b3015e984e856d77a60 100644 (file)
--- a/services/fuse/tests/test_mount.py
+++ b/services/fuse/tests/test_mount.py
@@ -1,16 +1,19 @@
-import unittest
 import arvados
+import arvados.safeapi
 import arvados_fuse as fuse
-import threading
-import time
-import os
+import glob
+import json
 import llfuse
-import tempfile
+import os
 import shutil
 import subprocess
-import glob
+import sys
+import tempfile
+import threading
+import time
+import unittest
+
 import run_test_server
-import json
 
 class MountTestBase(unittest.TestCase):
     def setUp(self):
@@ -19,10 +22,10 @@ class MountTestBase(unittest.TestCase):
         self.mounttmp = tempfile.mkdtemp()
         run_test_server.run()
         run_test_server.authorize_with("admin")
-        self.api = fuse.SafeApi(arvados.config)
+        self.api = arvados.safeapi.ThreadSafeApiCache(arvados.config.settings())
 
     def make_mount(self, root_class, **root_kwargs):
-        operations = fuse.Operations(os.getuid(), os.getgid())
+        operations = fuse.Operations(os.getuid(), os.getgid(), cache_cap=2)
         operations.inodes.add_entry(root_class(
             llfuse.ROOT_INODE, operations.inodes, self.api, 0, **root_kwargs))
         llfuse.init(operations, self.mounttmp, [])
@@ -240,8 +243,7 @@ class FuseSharedTest(MountTestBase):
         # directory)
         fuse_user_objs = os.listdir(os.path.join(self.mounttmp, 'FUSE User'))
         fuse_user_objs.sort()
-        self.assertEqual(['Empty collection.link',                # permission link on collection
-                          'FUSE Test Project',                    # project owned by user
+        self.assertEqual(['FUSE Test Project',                    # project owned by user
                           'collection #1 owned by FUSE',          # collection owned by user
                           'collection #2 owned by FUSE',          # collection owned by user
                           'pipeline instance owned by FUSE.pipelineInstance',  # pipeline instance owned by user
@@ -257,14 +259,26 @@ class FuseSharedTest(MountTestBase):
 
         # Double check that we can open and read objects in this folder as a file,
         # and that its contents are what we expect.
-        with open(os.path.join(
+        pipeline_template_path = os.path.join(
                 self.mounttmp,
                 'FUSE User',
                 'FUSE Test Project',
-                'pipeline template in FUSE project.pipelineTemplate')) as f:
+                'pipeline template in FUSE project.pipelineTemplate')
+        with open(pipeline_template_path) as f:
             j = json.load(f)
             self.assertEqual("pipeline template in FUSE project", j['name'])
 
+        # check mtime on template
+        st = os.stat(pipeline_template_path)
+        self.assertEqual(st.st_mtime, 1397493304)
+
+        # check mtime on collection
+        st = os.stat(os.path.join(
+                self.mounttmp,
+                'FUSE User',
+                'collection #1 owned by FUSE'))
+        self.assertEqual(st.st_mtime, 1391448174)
+
 
 class FuseHomeTest(MountTestBase):
     def runTest(self):
@@ -275,9 +289,24 @@ class FuseHomeTest(MountTestBase):
         self.assertIn('Unrestricted public data', d1)
 
         d2 = os.listdir(os.path.join(self.mounttmp, 'Unrestricted public data'))
-        self.assertEqual(['GNU General Public License, version 3',
-                          'Pipeline in publicly accessible project.pipelineInstance',
-                          'Pipeline template in publicly accessible project.pipelineTemplate'], d2)
+        public_project = run_test_server.fixture('groups')[
+            'anonymously_accessible_project']
+        found_in = 0
+        found_not_in = 0
+        for name, item in run_test_server.fixture('collections').iteritems():
+            if 'name' not in item:
+                pass
+            elif item['owner_uuid'] == public_project['uuid']:
+                self.assertIn(item['name'], d2)
+                found_in += 1
+            else:
+                # Artificial assumption here: there is no public
+                # collection fixture with the same name as a
+                # non-public collection.
+                self.assertNotIn(item['name'], d2)
+                found_not_in += 1
+        self.assertNotEqual(0, found_in)
+        self.assertNotEqual(0, found_not_in)
 
         d3 = os.listdir(os.path.join(self.mounttmp, 'Unrestricted public data', 'GNU General Public License, version 3'))
         self.assertEqual(["GNU_General_Public_License,_version_3.pdf"], d3)