tests/lib/conds/test.py

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