1 # SPDX-License-Identifier: MIT
3 # Copyright (c) 2020 Philippe Proulx <pproulx@efficios.com>
15 # the `conds-triggers` program's full path
16 _CONDS_TRIGGERS_PATH
= os
.environ
["BT_TESTS_LIB_CONDS_TRIGGER_BIN"]
19 # test methods are added by _create_tests()
20 class LibPrePostCondsTestCase(unittest
.TestCase
):
24 # a condition trigger descriptor (base)
25 class _CondTriggerDescriptor
:
26 def __init__(self
, index
, trigger_name
, cond_id
):
28 self
._trigger
_name
= trigger_name
29 self
._cond
_id
= cond_id
36 def trigger_name(self
):
37 return self
._trigger
_name
44 # precondition trigger descriptor
45 class _PreCondTriggerDescriptor(_CondTriggerDescriptor
):
51 # postcondition trigger descriptor
52 class _PostCondTriggerDescriptor(_CondTriggerDescriptor
):
58 # test method template for `LibPrePostCondsTestCase`
59 def _test(self
, descriptor
):
62 # $ conds-triggers run <index>
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,
70 # wait for termination and get standard output/error data
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
))
80 # assert that program aborted (only available on POSIX)
81 if os
.name
== "posix":
82 self
.assertEqual(proc
.returncode
, -int(signal
.SIGABRT
))
84 # assert that the standard error text contains the condition ID
85 text
= "Condition ID: `{}`.".format(descriptor
.cond_id
)
86 self
.assertIn(text
, stderr
)
89 # Condition trigger descriptors from the JSON array returned by
91 # $ conds-triggers list
92 def _cond_trigger_descriptors_from_json(json_descr_array
):
94 descriptor_names
= set()
96 for index
, json_descr
in enumerate(json_descr_array
):
97 # sanity check: check for duplicate
98 trigger_name
= json_descr
["name"]
100 if trigger_name
in descriptor_names
:
102 "Duplicate condition trigger name `{}`".format(trigger_name
)
106 cond_id
= json_descr
["cond-id"]
108 if cond_id
.startswith("pre"):
109 cond_type
= _PreCondTriggerDescriptor
110 elif cond_id
.startswith("post"):
111 cond_type
= _PostCondTriggerDescriptor
113 raise ValueError("Invalid condition ID `{}`".format(cond_id
))
115 descriptors
.append(cond_type(index
, trigger_name
, cond_id
))
116 descriptor_names
.add(trigger_name
)
121 # creates the individual tests of `LibPrePostCondsTestCase`
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)
129 # get condition trigger descriptor objects from JSON
130 descriptors
= _cond_trigger_descriptors_from_json(json_descr_array
)
132 # create test methods
133 for descriptor
in descriptors
:
135 test_meth_name
= "test_{}".format(
136 re
.sub(r
"[^a-zA-Z0-9_]", "_", descriptor
.trigger_name
)
140 meth
= functools
.partialmethod(_test
, descriptor
)
141 setattr(LibPrePostCondsTestCase
, test_meth_name
, meth
)
147 if __name__
== "__main__":