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 signal
  15 import sys
  16 import tempfile
  17 import unittest
  18
  19 import pytest
  20
  21 from . import run_test_server
  22
  23 @atexit.register
  24 def _pool_cleanup():
  25     if _pool is None:
  26         return
  27     _pool.close()
  28     _pool.join()
  29
  30
  31 def wrap_static_test_method(modName, clsName, funcName, args, kwargs):
  32     class Test(unittest.TestCase):
  33         def runTest(self, *args, **kwargs):
  34             getattr(getattr(sys.modules[modName], clsName), funcName)(self, *args, **kwargs)
  35     Test().runTest(*args, **kwargs)
  36
  37
  38 # To avoid Python's threading+multiprocessing=deadlock problems, we
  39 # use a single global pool with maxtasksperchild=None for the entire
  40 # test suite.
  41 _pool = None
  42 def workerPool():
  43     global _pool
  44     if _pool is None:
  45         _pool = multiprocessing.Pool(processes=1, maxtasksperchild=None)
  46     return _pool
  47
  48
  49 class IntegrationTest(unittest.TestCase):
  50     def pool_test(self, *args, **kwargs):
  51         """Run a static method as a unit test, in a different process.
  52
  53         If called by method 'foobar', the static method '_foobar' of
  54         the same class will be called in the other process.
  55         """
  56         modName = inspect.getmodule(self).__name__
  57         clsName = self.__class__.__name__
  58         funcName = inspect.currentframe().f_back.f_code.co_name
  59         workerPool().apply(
  60             wrap_static_test_method,
  61             (modName, clsName, '_'+funcName, args, kwargs))
  62
  63     @classmethod
  64     def setUpClass(cls):
  65         run_test_server.run()
  66         run_test_server.run_keep(blob_signing=True, num_servers=2)
  67
  68     @classmethod
  69     def tearDownClass(cls):
  70         run_test_server.stop_keep(num_servers=2)
  71
  72     def setUp(self):
  73         self.mnt = tempfile.mkdtemp()
  74         run_test_server.authorize_with('active')
  75
  76     def tearDown(self):
  77         os.rmdir(self.mnt)
  78         run_test_server.reset()
  79
  80     @staticmethod
  81     def mount(argv):
  82         """Decorator. Sets up a FUSE mount at self.mnt with the given args."""
  83         def decorator(func):
  84             @functools.wraps(func)
  85             def wrapper(self, *args, **kwargs):
  86                 self.mount = None
  87                 try:
  88                     with arvados_fuse.command.Mount(
  89                             arvados_fuse.command.ArgumentParser().parse_args(
  90                                 argv + ['--foreground',
  91                                         '--unmount-timeout=60',
  92                                         self.mnt])) as self.mount:
  93                         return func(self, *args, **kwargs)
  94                 finally:
  95                     if self.mount and self.mount.llfuse_thread.is_alive():
  96                         logging.warning("IntegrationTest.mount:"
  97                                             " llfuse thread still alive after umount"
  98                                             " -- ending test suite to avoid deadlock")
  99                         # pytest uses exit status 2 when test collection failed.
 100                         # A UnitTest failing in setup/teardown counts as a
 101                         # collection failure, so pytest will exit with status 2
 102                         # no matter what status you specify here. run-tests.sh
 103                         # looks for this status, so specify 2 just to keep
 104                         # everything as consistent as possible.
 105                         # TODO: If we refactor these tests so they're not built
 106                         # on unittest, consider using a dedicated, non-pytest
 107                         # exit code like TEMPFAIL.
 108                         pytest.exit("llfuse thread outlived test", 2)
 109             return wrapper
 110         return decorator