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