Import s3ql_1.7.orig.tar.bz2

1 files changed, 35 insertions, 3 deletions

diff --git a/tests/t4_fuse.py b/tests/t4_fuse.py
index dfede15..afd8391 100644
--- a/tests/t4_fuse.py
+++ b/tests/t4_fuse.py
@@ -121,6 +121,31 @@ class TimeoutError(Exception):
 
     pass
 
+def skip_if_no_fusermount():
+    '''Raise SkipTest if fusermount is not available'''
+
+    which = subprocess.Popen(['which', 'fusermount'], stdout=subprocess.PIPE)
+    fusermount_path = which.communicate()[0].strip()
+    
+    if not fusermount_path or which.wait() != 0:
+        raise unittest.SkipTest("Can't find fusermount executable")
+    
+    if not os.path.exists('/dev/fuse'):
+        raise unittest.SkipTest("FUSE kernel module does not seem to be loaded")
+                          
+    if os.getuid() == 0:
+        return
+    
+    mode = os.stat(fusermount_path).st_mode
+    if mode & stat.S_ISUID == 0:
+        raise unittest.SkipTest('fusermount executable not setuid, and we are not root.')
+    
+    try:
+        subprocess.check_call([fusermount_path, '-V'], 
+                              stdout=open('/dev/null', 'wb'))
+    except subprocess.CalledProcessError:
+        raise unittest.SkipTest('Unable to execute fusermount')
+    
 if __name__ == '__main__':
     mypath = sys.argv[0]
 else:
@@ -130,6 +155,8 @@ BASEDIR = os.path.abspath(os.path.join(os.path.dirname(mypath), '..'))
 class fuse_tests(TestCase):
     
     def setUp(self):
+        skip_if_no_fusermount()
+        
         # We need this to test multi block operations
         self.src = __file__
         if os.path.getsize(self.src) < 1048:
@@ -165,7 +192,11 @@ class fuse_tests(TestCase):
                                                   stdin=subprocess.PIPE)
         print(self.passphrase, file=self.mount_process.stdin)
         self.mount_process.stdin.close()
-        retry(30, os.path.ismount, self.mnt_dir)                              
+        def poll():
+            if os.path.ismount(self.mnt_dir):
+                return True
+            self.assertIsNone(self.mount_process.poll())
+        retry(30, poll)                              
 
     def umount(self):
         devnull = open('/dev/null', 'wb')
@@ -174,7 +205,7 @@ class fuse_tests(TestCase):
         
         proc = subprocess.Popen([os.path.join(BASEDIR, 'bin', 'umount.s3ql'), 
                                  '--quiet', self.mnt_dir])
-        retry(30, lambda : proc.poll() is not None)
+        retry(60, lambda : proc.poll() is not None)
         self.assertEquals(proc.wait(), 0)
         
         self.assertEqual(self.mount_process.wait(), 0)
@@ -195,7 +226,8 @@ class fuse_tests(TestCase):
         os.rmdir(self.mnt_dir)
         
         # Give mount process a little while to terminate
-        retry(10, lambda : self.mount_process.poll() is not None)
+        if self.mount_process is not None:
+            retry(10, lambda : self.mount_process.poll() is not None)
               
         shutil.rmtree(self.cache_dir)
         shutil.rmtree(self.bucket_dir)