1  """
  2  Find tests, identify the ones that the user wants, and invoke them.
  3  """
  4  
  5  from __future__ import annotations
  6  
  7  import importlib
  8  import traceback
  9  from types import ModuleType
 10  from typing import Iterator
 11  
 12  from arti_rpc_tests import FatalException
 13  from arti_rpc_tests.context import TestContext
 14  
 15  # Is this the right way to do this?
 16  #
 17  # Should we be instead listing all the files in $(basename __file__)/tests ?
 18  _TEST_MODS = [
 19      "basic",
 20      "connpt",
 21      "connect",
 22      "meta_features",
 23      "release_obj",
 24  ]
 25  
 26  
 27  # Return a list of all the python modules that we should search for tests.
 28  def all_modules() -> list[ModuleType]:
 29      return [
 30          importlib.import_module(f"arti_rpc_tests.tests.{name}") for name in _TEST_MODS
 31      ]
 32  
 33  
 34  def run_tests(
 35      testfilter: TestFilter, modules: list[ModuleType], context: TestContext
 36  ) -> bool:
 37      """
 38      Run every test listed by `testfilter` in the provided `modules`,
 39      using the facilities in `context`.
 40  
 41      Return True if every test passed, and False otherwise.
 42  
 43      May raise FatalException if a test failed completely.
 44      """
 45      to_run: list[TestCase] = []
 46  
 47      for m in modules:
 48          to_run.extend(testfilter.list_tests(m))
 49  
 50      print(f"Found {len(to_run)} tests")
 51  
 52      successes = failures = 0
 53      for test in to_run:
 54          if test.run(context):
 55              successes += 1
 56          else:
 57              failures += 1
 58  
 59      if failures:
 60          print(f"{failures}/{len(to_run)} tests failed!")
 61      else:
 62          print(f"All {successes} tests succeeded")
 63      assert successes + failures == len(to_run)
 64  
 65      return failures == 0
 66  
 67  
 68  class TestFilter:
 69      """
 70      Selects one or more tests that we should run.
 71      """
 72  
 73      def __init__(self):
 74          # No features supported yet
 75          pass
 76  
 77      def list_tests(self, module: ModuleType) -> Iterator[TestCase]:
 78          """
 79          Yield every test in `module` that this filter permits.
 80          """
 81          for name in sorted(dir(module)):
 82              obj = getattr(module, name)
 83  
 84              if callable(obj) and getattr(obj, "arti_rpc_test", False):
 85                  sname = name.removeprefix("test_")
 86                  yield TestCase(f"{module.__name__}.{sname}", obj)
 87  
 88  
 89  class TestCase:
 90      """
 91      A single test case.
 92      """
 93  
 94      def __init__(self, name, function):
 95          self.name = name
 96          self.function = function
 97  
 98      def run(self, context: TestContext) -> bool:
 99          """
100          Try to run this test within `context`.
101  
102          Returns True on success and False on failure.
103  
104          May raise FatalEception if test execution should stop entirely.
105          """
106          try:
107              print(self.name, "...", flush=True, end="")
108              self.run_inner(context)
109              print("OK")
110              return True
111          except FatalException:
112              print("FATAL EXCEPTION")
113              raise
114          except Exception:
115              print("FAILED")
116              traceback.print_exc()
117              return False
118  
119      def run_inner(self, context: TestContext) -> None:
120          """
121          Run this test; raise an exception on failure.
122  
123          Raise a FatalException if all test execution should stop entirely.
124          """
125          if not context.arti_process_is_running():
126              raise FatalException("Arti process not running at start of test!")
127  
128          self.function(context)
129  
130          if not context.arti_process_is_running():
131              raise FatalException("Arti process not running at end of test!")