tests/lib/conds/test.py

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