services/fuse/tests/integration_test.py

   1 import arvados
   2 import arvados_fuse
   3 import arvados_fuse.command
   4 import atexit
   5 import functools
   6 import inspect
   7 import multiprocessing
   8 import os
   9 import run_test_server
  10 import signal
  11 import sys
  12 import tempfile
  13 import unittest
  14
  15 _pool = None
  16
  17
  18 @atexit.register
  19 def _pool_cleanup():
  20     global _pool
  21     if _pool is None:
  22         return
  23     _pool.close()
  24     _pool.join()
  25
  26
  27 def wrap_static_test_method(modName, clsName, funcName, args, kwargs):
  28     class Test(unittest.TestCase):
  29         def runTest(self, *args, **kwargs):
  30             getattr(getattr(sys.modules[modName], clsName), funcName)(self, *args, **kwargs)
  31     Test().runTest(*args, **kwargs)
  32
  33
  34 class IntegrationTest(unittest.TestCase):
  35     def pool_test(self, *args, **kwargs):
  36         """Run a static method as a unit test, in a different process.
  37
  38         If called by method 'foobar', the static method '_foobar' of
  39         the same class will be called in the other process.
  40         """
  41         global _pool
  42         if _pool is None:
  43             _pool = multiprocessing.Pool(1, maxtasksperchild=1)
  44         modName = inspect.getmodule(self).__name__
  45         clsName = self.__class__.__name__
  46         funcName = inspect.currentframe().f_back.f_code.co_name
  47         _pool.apply(
  48             wrap_static_test_method,
  49             (modName, clsName, '_'+funcName, args, kwargs))
  50
  51     @classmethod
  52     def setUpClass(cls):
  53         run_test_server.run()
  54         run_test_server.run_keep(enforce_permissions=True, num_servers=2)
  55
  56     @classmethod
  57     def tearDownClass(cls):
  58         run_test_server.stop_keep(num_servers=2)
  59
  60     def setUp(self):
  61         self.mnt = tempfile.mkdtemp()
  62         run_test_server.authorize_with('active')
  63         self.api = arvados.safeapi.ThreadSafeApiCache(arvados.config.settings())
  64
  65     def tearDown(self):
  66         os.rmdir(self.mnt)
  67         run_test_server.reset()
  68
  69     @staticmethod
  70     def mount(argv):
  71         """Decorator. Sets up a FUSE mount at self.mnt with the given args."""
  72         def decorator(func):
  73             @functools.wraps(func)
  74             def wrapper(self, *args, **kwargs):
  75                 with arvados_fuse.command.Mount(
  76                         arvados_fuse.command.ArgumentParser().parse_args(
  77                             argv + ['--foreground',
  78                                     '--unmount-timeout=0.1',
  79                                     self.mnt])):
  80                     return func(self, *args, **kwargs)
  81             return wrapper
  82         return decorator