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