Merge branch '21535-multi-wf-delete'

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

diff --git a/services/fuse/tests/test_unmount.py b/services/fuse/tests/test_unmount.py
index 716a0e00d704d16b5edcb8436b3ca68de157024b..3949fd5de4d85953ec2ce8b6bf200e6bc8d22427 100644 (file)
--- a/services/fuse/tests/test_unmount.py
+++ b/services/fuse/tests/test_unmount.py
@@ -1,8 +1,16 @@
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
+import arvados_fuse.unmount
 import os
 import subprocess
+import shutil
+import tempfile
 import time
+import unittest
 
-from integration_test import IntegrationTest
+from .integration_test import IntegrationTest
 
 class UnmountTest(IntegrationTest):
     def setUp(self):
@@ -21,20 +29,21 @@ class UnmountTest(IntegrationTest):
              self.mnt])
         subprocess.check_call(
             ['./bin/arv-mount', '--subtype', 'test', '--replace',
-             '--unmount-timeout', '10',
+             '--unmount-timeout', '60',
              self.mnt])
         subprocess.check_call(
             ['./bin/arv-mount', '--subtype', 'test', '--replace',
-             '--unmount-timeout', '10',
+             '--unmount-timeout', '60',
              self.mnt,
              '--exec', 'true'])
         for m in subprocess.check_output(['mount']).splitlines():
-            self.assertNotIn(' '+self.mnt+' ', m)
+            expected = bytes(' ' + self.mnt + ' ', encoding='utf-8')
+            self.assertNotIn(expected, m)
 
     def _mounted(self, mounts):
         all_mounts = subprocess.check_output(['mount'])
         return [m for m in mounts
-                if ' '+m+' ' in all_mounts]
+                if bytes(' ' + m + ' ', encoding='utf-8') in all_mounts]
 
     def _wait_for_mounts(self, mounts):
         deadline = time.time() + 10
@@ -85,3 +94,45 @@ class UnmountTest(IntegrationTest):
         self.assertEqual(mounts, self._mounted(mounts))
         subprocess.check_call(['./bin/arv-mount', '--unmount-all', self.tmp])
         self.assertEqual([], self._mounted(mounts))
+
+
+
+class SaferRealpath(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp)
+
+    def test_safer_realpath(self):
+        os.mkdir(self.tmp+"/dir")
+        os.mkdir(self.tmp+"/dir/dir2")
+        os.symlink("missing", self.tmp+"/relative-missing")
+        os.symlink("dir", self.tmp+"/./relative-dir")
+        os.symlink("relative-dir", self.tmp+"/relative-indirect")
+        os.symlink(self.tmp+"/dir", self.tmp+"/absolute-dir")
+        os.symlink("./dir/../loop", self.tmp+"/loop")
+        os.symlink(".", self.tmp+"/dir/self")
+        os.symlink("..", self.tmp+"/dir/dir2/parent")
+        os.symlink("../dir3", self.tmp+"/dir/dir2/sibling")
+        os.symlink("../missing/../danger", self.tmp+"/dir/tricky")
+        os.symlink("/proc/1/fd/12345", self.tmp+"/eperm")
+        for (inpath, outpath, ok) in [
+                ("dir/self", "dir", True),
+                ("dir/dir2/parent", "dir", True),
+                ("dir/dir2/sibling", "dir/dir3", False),
+                ("dir", "dir", True),
+                ("relative-dir", "dir", True),
+                ("relative-missing", "missing", False),
+                ("relative-indirect", "dir", True),
+                ("absolute-dir", "dir", True),
+                ("loop", "loop", False),
+                # "missing" doesn't exist, so "missing/.." isn't our
+                # tmpdir; it's important not to contract this to just
+                # "danger".
+                ("dir/tricky", "missing/../danger", False),
+                ("eperm", "/proc/1/fd/12345", False),
+        ]:
+            if not outpath.startswith('/'):
+                outpath = self.tmp + '/' + outpath
+            self.assertEqual((outpath, ok), arvados_fuse.unmount.safer_realpath(self.tmp+"/"+inpath))