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', '60',
  33              self.mnt])
  34         subprocess.check_call(
  35             ['./bin/arv-mount', '--subtype', 'test', '--replace',
  36              '--unmount-timeout', '60',
  37              self.mnt,
  38              '--exec', 'true'])
  39         for m in subprocess.check_output(['mount']).splitlines():
  40             expected = bytes(' ' + self.mnt + ' ', encoding='utf-8')
  41             self.assertNotIn(expected, m)
  42
  43     def _mounted(self, mounts):
  44         all_mounts = subprocess.check_output(['mount'])
  45         return [m for m in mounts
  46                 if bytes(' ' + m + ' ', encoding='utf-8') in all_mounts]
  47
  48     def _wait_for_mounts(self, mounts):
  49         deadline = time.time() + 10
  50         while self._mounted(mounts) != mounts:
  51             time.sleep(0.1)
  52             self.assertLess(time.time(), deadline)
  53
  54     def test_unmount_subtype(self):
  55         mounts = []
  56         for d in ['foo', 'bar']:
  57             mnt = self.tmp+'/'+d
  58             os.mkdir(mnt)
  59             self.to_delete.insert(0, mnt)
  60             mounts.append(mnt)
  61             subprocess.check_call(
  62                 ['./bin/arv-mount', '--subtype', d, mnt])
  63
  64         self._wait_for_mounts(mounts)
  65         self.assertEqual(mounts, self._mounted(mounts))
  66         subprocess.call(['./bin/arv-mount', '--subtype', 'baz', '--unmount-all', self.tmp])
  67         self.assertEqual(mounts, self._mounted(mounts))
  68         subprocess.call(['./bin/arv-mount', '--subtype', 'bar', '--unmount', mounts[0]])
  69         self.assertEqual(mounts, self._mounted(mounts))
  70         subprocess.call(['./bin/arv-mount', '--subtype', '', '--unmount', self.tmp])
  71         self.assertEqual(mounts, self._mounted(mounts))
  72         subprocess.check_call(['./bin/arv-mount', '--subtype', 'foo', '--unmount', mounts[0]])
  73         self.assertEqual(mounts[1:], self._mounted(mounts))
  74         subprocess.check_call(['./bin/arv-mount', '--subtype', '', '--unmount-all', mounts[0]])
  75         self.assertEqual(mounts[1:], self._mounted(mounts))
  76         subprocess.check_call(['./bin/arv-mount', '--subtype', 'bar', '--unmount-all', self.tmp])
  77         self.assertEqual([], self._mounted(mounts))
  78
  79     def test_unmount_children(self):
  80         for d in ['foo', 'foo/bar', 'bar']:
  81             mnt = self.tmp+'/'+d
  82             os.mkdir(mnt)
  83             self.to_delete.insert(0, mnt)
  84         mounts = []
  85         for d in ['bar', 'foo/bar']:
  86             mnt = self.tmp+'/'+d
  87             mounts.append(mnt)
  88             subprocess.check_call(
  89                 ['./bin/arv-mount', '--subtype', 'test', mnt])
  90
  91         self._wait_for_mounts(mounts)
  92         self.assertEqual(mounts, self._mounted(mounts))
  93         subprocess.check_call(['./bin/arv-mount', '--unmount', self.tmp])
  94         self.assertEqual(mounts, self._mounted(mounts))
  95         subprocess.check_call(['./bin/arv-mount', '--unmount-all', self.tmp])
  96         self.assertEqual([], self._mounted(mounts))
  97
  98
  99
 100 class SaferRealpath(unittest.TestCase):
 101     def setUp(self):
 102         self.tmp = tempfile.mkdtemp()
 103
 104     def tearDown(self):
 105         shutil.rmtree(self.tmp)
 106
 107     def test_safer_realpath(self):
 108         os.mkdir(self.tmp+"/dir")
 109         os.mkdir(self.tmp+"/dir/dir2")
 110         os.symlink("missing", self.tmp+"/relative-missing")
 111         os.symlink("dir", self.tmp+"/./relative-dir")
 112         os.symlink("relative-dir", self.tmp+"/relative-indirect")
 113         os.symlink(self.tmp+"/dir", self.tmp+"/absolute-dir")
 114         os.symlink("./dir/../loop", self.tmp+"/loop")
 115         os.symlink(".", self.tmp+"/dir/self")
 116         os.symlink("..", self.tmp+"/dir/dir2/parent")
 117         os.symlink("../dir3", self.tmp+"/dir/dir2/sibling")
 118         os.symlink("../missing/../danger", self.tmp+"/dir/tricky")
 119         os.symlink("/proc/1/fd/12345", self.tmp+"/eperm")
 120         for (inpath, outpath, ok) in [
 121                 ("dir/self", "dir", True),
 122                 ("dir/dir2/parent", "dir", True),
 123                 ("dir/dir2/sibling", "dir/dir3", False),
 124                 ("dir", "dir", True),
 125                 ("relative-dir", "dir", True),
 126                 ("relative-missing", "missing", False),
 127                 ("relative-indirect", "dir", True),
 128                 ("absolute-dir", "dir", True),
 129                 ("loop", "loop", False),
 130                 # "missing" doesn't exist, so "missing/.." isn't our
 131                 # tmpdir; it's important not to contract this to just
 132                 # "danger".
 133                 ("dir/tricky", "missing/../danger", False),
 134                 ("eperm", "/proc/1/fd/12345", False),
 135         ]:
 136             if not outpath.startswith('/'):
 137                 outpath = self.tmp + '/' + outpath
 138             self.assertEqual((outpath, ok), arvados_fuse.unmount.safer_realpath(self.tmp+"/"+inpath))