services/fuse/tests/integration_test.py

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