| 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() |