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