services/fuse/tests/integration_test.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 import arvados
   6 import arvados_fuse
   7 import arvados_fuse.command
   8 import atexit
   9 import functools
  10 import inspect
  11 import logging
  12 import multiprocessing
  13 import os
  14 import run_test_server
  15 import signal
  16 import sys
  17 import tempfile
  18 import unittest
  19
  20 @atexit.register
  21 def _pool_cleanup():
  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 # To avoid Python's threading+multiprocessing=deadlock problems, we
  36 # use a single global pool with maxtasksperchild=None for the entire
  37 # test suite.
  38 _pool = None
  39 def workerPool():
  40     global _pool
  41     if _pool is None:
  42         _pool = multiprocessing.Pool(processes=1, maxtasksperchild=None)
  43     return _pool
  44
  45
  46 class IntegrationTest(unittest.TestCase):
  47     def pool_test(self, *args, **kwargs):
  48         """Run a static method as a unit test, in a different process.
  49
  50         If called by method 'foobar', the static method '_foobar' of
  51         the same class will be called in the other process.
  52         """
  53         modName = inspect.getmodule(self).__name__
  54         clsName = self.__class__.__name__
  55         funcName = inspect.currentframe().f_back.f_code.co_name
  56         workerPool().apply(
  57             wrap_static_test_method,
  58             (modName, clsName, '_'+funcName, args, kwargs))
  59
  60     @classmethod
  61     def setUpClass(cls):
  62         run_test_server.run()
  63         run_test_server.run_keep(enforce_permissions=True, num_servers=2)
  64
  65     @classmethod
  66     def tearDownClass(cls):
  67         run_test_server.stop_keep(num_servers=2)
  68
  69     def setUp(self):
  70         self.mnt = tempfile.mkdtemp()
  71         run_test_server.authorize_with('active')
  72
  73     def tearDown(self):
  74         os.rmdir(self.mnt)
  75         run_test_server.reset()
  76
  77     @staticmethod
  78     def mount(argv):
  79         """Decorator. Sets up a FUSE mount at self.mnt with the given args."""
  80         def decorator(func):
  81             @functools.wraps(func)
  82             def wrapper(self, *args, **kwargs):
  83                 self.mount = None
  84                 try:
  85                     with arvados_fuse.command.Mount(
  86                             arvados_fuse.command.ArgumentParser().parse_args(
  87                                 argv + ['--foreground',
  88                                         '--unmount-timeout=2',
  89                                         self.mnt])) as self.mount:
  90                         return func(self, *args, **kwargs)
  91                 finally:
  92                     if self.mount and self.mount.llfuse_thread.is_alive():
  93                         logging.warning("IntegrationTest.mount:"
  94                                             " llfuse thread still alive after umount"
  95                                             " -- killing test suite to avoid deadlock")
  96                         os.kill(os.getpid(), signal.SIGKILL)
  97             return wrapper
  98         return decorator