tests/lib/conds/test.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2020 Philippe Proulx <pproulx@efficios.com>
   4
   5 # pyright: strict, reportTypeCommentUsage=false, reportMissingTypeStubs=false
   6
   7 import os
   8 import re
   9 import signal
  10 import os.path
  11 import unittest
  12 import functools
  13 import subprocess
  14
  15 import tjson
  16
  17 # the `conds-triggers` program's full path
  18 _CONDS_TRIGGERS_PATH = os.environ["BT_TESTS_LIB_CONDS_TRIGGER_BIN"]
  19
  20
  21 # test methods are added by _create_tests()
  22 class LibPrePostCondsTestCase(unittest.TestCase):
  23     pass
  24
  25
  26 # a condition trigger descriptor (base)
  27 class _CondTriggerDescriptor:
  28     def __init__(self, index: int, trigger_name: str, cond_id: str):
  29         self._index = index
  30         self._trigger_name = trigger_name
  31         self._cond_id = cond_id
  32
  33     @property
  34     def index(self):
  35         return self._index
  36
  37     @property
  38     def trigger_name(self):
  39         return self._trigger_name
  40
  41     @property
  42     def cond_id(self):
  43         return self._cond_id
  44
  45
  46 # precondition trigger descriptor
  47 class _PreCondTriggerDescriptor(_CondTriggerDescriptor):
  48     @property
  49     def type_str(self):
  50         return "pre"
  51
  52
  53 # postcondition trigger descriptor
  54 class _PostCondTriggerDescriptor(_CondTriggerDescriptor):
  55     @property
  56     def type_str(self):
  57         return "post"
  58
  59
  60 # test method template for `LibPrePostCondsTestCase`
  61 def _test(self: unittest.TestCase, descriptor: _CondTriggerDescriptor):
  62     # Execute:
  63     #
  64     #     $ conds-triggers run <index>
  65     #
  66     # where `<index>` is the descriptor's index.
  67     with subprocess.Popen(
  68         [_CONDS_TRIGGERS_PATH, "run", str(descriptor.index)],
  69         stderr=subprocess.PIPE,
  70         universal_newlines=True,
  71     ) as proc:
  72         # wait for termination and get standard output/error data
  73         timeout = 5
  74
  75         try:
  76             # wait for program end and get standard error pipe's contents
  77             _, stderr = proc.communicate(timeout=timeout)
  78         except subprocess.TimeoutExpired:
  79             self.fail("Process hanged for {} seconds".format(timeout))
  80             return
  81
  82         # assert that program aborted (only available on POSIX)
  83         if os.name == "posix":
  84             self.assertEqual(proc.returncode, -int(signal.SIGABRT))
  85
  86         # assert that the standard error text contains the condition ID
  87         text = "Condition ID: `{}`.".format(descriptor.cond_id)
  88         self.assertIn(text, stderr)
  89
  90
  91 # Condition trigger descriptors from the JSON array returned by
  92 #
  93 #     $ conds-triggers list
  94 def _cond_trigger_descriptors_from_json(json_descr_array: tjson.ArrayVal):
  95     descriptors = []  # type: list[_CondTriggerDescriptor]
  96     descriptor_names = set()  # type: set[str]
  97
  98     for index, json_descr in enumerate(json_descr_array.iter(tjson.ObjVal)):
  99         # sanity check: check for duplicate
 100         trigger_name = json_descr.at("name", tjson.StrVal).val
 101
 102         if trigger_name in descriptor_names:
 103             raise ValueError(
 104                 "Duplicate condition trigger name `{}`".format(trigger_name)
 105             )
 106
 107         # condition ID
 108         cond_id = json_descr.at("cond-id", tjson.StrVal).val
 109
 110         if cond_id.startswith("pre"):
 111             cond_type = _PreCondTriggerDescriptor
 112         elif cond_id.startswith("post"):
 113             cond_type = _PostCondTriggerDescriptor
 114         else:
 115             raise ValueError("Invalid condition ID `{}`".format(cond_id))
 116
 117         descriptors.append(cond_type(index, trigger_name, cond_id))
 118         descriptor_names.add(trigger_name)
 119
 120     return descriptors
 121
 122
 123 # creates the individual tests of `LibPrePostCondsTestCase`
 124 def _create_tests():
 125     # Execute `conds-triggers list` to get a JSON array of condition
 126     # trigger descriptors.
 127     json_descr_array = tjson.loads(
 128         subprocess.check_output(
 129             [_CONDS_TRIGGERS_PATH, "list"], universal_newlines=True
 130         ),
 131         tjson.ArrayVal,
 132     )
 133
 134     # get condition trigger descriptor objects from JSON
 135     descriptors = _cond_trigger_descriptors_from_json(json_descr_array)
 136
 137     # create test methods
 138     for descriptor in descriptors:
 139         # test method name
 140         test_meth_name = "test_{}".format(
 141             re.sub(r"[^a-zA-Z0-9_]", "_", descriptor.trigger_name)
 142         )
 143
 144         # test method
 145         meth = functools.partialmethod(_test, descriptor)
 146         setattr(LibPrePostCondsTestCase, test_meth_name, meth)
 147
 148
 149 _create_tests()
 150
 151
 152 if __name__ == "__main__":
 153     unittest.main()