services/fuse/tests/test_unmount.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 import arvados_fuse.unmount
   6 import os
   7 import subprocess
   8 import shutil
   9 import tempfile
  10 import time
  11 import unittest
  12
  13 from integration_test import IntegrationTest
  14
  15 class UnmountTest(IntegrationTest):
  16     def setUp(self):
  17         super(UnmountTest, self).setUp()
  18         self.tmp = self.mnt
  19         self.to_delete = []
  20
  21     def tearDown(self):
  22         for d in self.to_delete:
  23             os.rmdir(d)
  24         super(UnmountTest, self).tearDown()
  25
  26     def test_replace(self):
  27         subprocess.check_call(
  28             ['./bin/arv-mount', '--subtype', 'test', '--replace',
  29              self.mnt])
  30         subprocess.check_call(
  31             ['./bin/arv-mount', '--subtype', 'test', '--replace',
  32              '--unmount-timeout', '10',
  33              self.mnt])
  34         subprocess.check_call(
  35             ['./bin/arv-mount', '--subtype', 'test', '--replace',
  36              '--unmount-timeout', '10',
  37              self.mnt,
  38              '--exec', 'true'])
  39         for m in subprocess.check_output(['mount']).splitlines():
  40             self.assertNotIn(' '+self.mnt+' ', m)
  41
  42     def _mounted(self, mounts):
  43         all_mounts = subprocess.check_output(['mount'])
  44         return [m for m in mounts
  45                 if ' '+m+' ' in all_mounts]
  46
  47     def _wait_for_mounts(self, mounts):
  48         deadline = time.time() + 10
  49         while self._mounted(mounts) != mounts:
  50             time.sleep(0.1)
  51             self.assertLess(time.time(), deadline)
  52
  53     def test_unmount_subtype(self):
  54         mounts = []
  55         for d in ['foo', 'bar']:
  56             mnt = self.tmp+'/'+d
  57             os.mkdir(mnt)
  58             self.to_delete.insert(0, mnt)
  59             mounts.append(mnt)
  60             subprocess.check_call(
  61                 ['./bin/arv-mount', '--subtype', d, mnt])
  62
  63         self._wait_for_mounts(mounts)
  64         self.assertEqual(mounts, self._mounted(mounts))
  65         subprocess.call(['./bin/arv-mount', '--subtype', 'baz', '--unmount-all', self.tmp])
  66         self.assertEqual(mounts, self._mounted(mounts))
  67         subprocess.call(['./bin/arv-mount', '--subtype', 'bar', '--unmount', mounts[0]])
  68         self.assertEqual(mounts, self._mounted(mounts))
  69         subprocess.call(['./bin/arv-mount', '--subtype', '', '--unmount', self.tmp])
  70         self.assertEqual(mounts, self._mounted(mounts))
  71         subprocess.check_call(['./bin/arv-mount', '--subtype', 'foo', '--unmount', mounts[0]])
  72         self.assertEqual(mounts[1:], self._mounted(mounts))
  73         subprocess.check_call(['./bin/arv-mount', '--subtype', '', '--unmount-all', mounts[0]])
  74         self.assertEqual(mounts[1:], self._mounted(mounts))
  75         subprocess.check_call(['./bin/arv-mount', '--subtype', 'bar', '--unmount-all', self.tmp])
  76         self.assertEqual([], self._mounted(mounts))
  77
  78     def test_unmount_children(self):
  79         for d in ['foo', 'foo/bar', 'bar']:
  80             mnt = self.tmp+'/'+d
  81             os.mkdir(mnt)
  82             self.to_delete.insert(0, mnt)
  83         mounts = []
  84         for d in ['bar', 'foo/bar']:
  85             mnt = self.tmp+'/'+d
  86             mounts.append(mnt)
  87             subprocess.check_call(
  88                 ['./bin/arv-mount', '--subtype', 'test', mnt])
  89
  90         self._wait_for_mounts(mounts)
  91         self.assertEqual(mounts, self._mounted(mounts))
  92         subprocess.check_call(['./bin/arv-mount', '--unmount', self.tmp])
  93         self.assertEqual(mounts, self._mounted(mounts))
  94         subprocess.check_call(['./bin/arv-mount', '--unmount-all', self.tmp])
  95         self.assertEqual([], self._mounted(mounts))
  96
  97
  98
  99 class SaferRealpath(unittest.TestCase):
 100     def setUp(self):
 101         self.tmp = tempfile.mkdtemp()
 102
 103     def tearDown(self):
 104         shutil.rmtree(self.tmp)
 105
 106     def test_safer_realpath(self):
 107         os.mkdir(self.tmp+"/dir")
 108         os.mkdir(self.tmp+"/dir/dir2")
 109         os.symlink("missing", self.tmp+"/relative-missing")
 110         os.symlink("dir", self.tmp+"/./relative-dir")
 111         os.symlink("relative-dir", self.tmp+"/relative-indirect")
 112         os.symlink(self.tmp+"/dir", self.tmp+"/absolute-dir")
 113         os.symlink("./dir/../loop", self.tmp+"/loop")
 114         os.symlink(".", self.tmp+"/dir/self")
 115         os.symlink("..", self.tmp+"/dir/dir2/parent")
 116         os.symlink("../dir3", self.tmp+"/dir/dir2/sibling")
 117         os.symlink("../missing/../danger", self.tmp+"/dir/tricky")
 118         os.symlink("/proc/1/fd/12345", self.tmp+"/eperm")
 119         for (inpath, outpath, ok) in [
 120                 ("dir/self", "dir", True),
 121                 ("dir/dir2/parent", "dir", True),
 122                 ("dir/dir2/sibling", "dir/dir3", False),
 123                 ("dir", "dir", True),
 124                 ("relative-dir", "dir", True),
 125                 ("relative-missing", "missing", False),
 126                 ("relative-indirect", "dir", True),
 127                 ("absolute-dir", "dir", True),
 128                 ("loop", "loop", False),
 129                 # "missing" doesn't exist, so "missing/.." isn't our
 130                 # tmpdir; it's important not to contract this to just
 131                 # "danger".
 132                 ("dir/tricky", "missing/../danger", False),
 133                 ("eperm", "/proc/1/fd/12345", False),
 134         ]:
 135             if not outpath.startswith('/'):
 136                 outpath = self.tmp + '/' + outpath
 137             self.assertEqual((outpath, ok), arvados_fuse.unmount.safer_realpath(self.tmp+"/"+inpath))