1 # SPDX-License-Identifier: MIT
3 # Copyright (c) 2020 Philippe Proulx <pproulx@efficios.com>
5 # pyright: strict, reportTypeCommentUsage=false, reportMissingTypeStubs=false
17 # the `conds-triggers` program's full path
18 _CONDS_TRIGGERS_PATH
= os
.environ
["BT_TESTS_LIB_CONDS_TRIGGER_BIN"]
21 # test methods are added by _create_tests()
22 class LibPrePostCondsTestCase(unittest
.TestCase
):
26 # a condition trigger descriptor (base)
27 class _CondTriggerDescriptor
:
28 def __init__(self
, index
: int, trigger_name
: str, cond_id
: str):
30 self
._trigger
_name
= trigger_name
31 self
._cond
_id
= cond_id
38 def trigger_name(self
):
39 return self
._trigger
_name
46 # precondition trigger descriptor
47 class _PreCondTriggerDescriptor(_CondTriggerDescriptor
):
53 # postcondition trigger descriptor
54 class _PostCondTriggerDescriptor(_CondTriggerDescriptor
):
60 # test method template for `LibPrePostCondsTestCase`
61 def _test(self
: unittest
.TestCase
, descriptor
: _CondTriggerDescriptor
):
64 # $ conds-triggers run <index>
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,
72 # wait for termination and get standard output/error data
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
))
82 # assert that program aborted (only available on POSIX)
83 if os
.name
== "posix":
84 self
.assertEqual(proc
.returncode
, -int(signal
.SIGABRT
))
86 # assert that the standard error text contains the condition ID
87 text
= "Condition ID: `{}`.".format(descriptor
.cond_id
)
88 self
.assertIn(text
, stderr
)
91 # Condition trigger descriptors from the JSON array returned by
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]
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
102 if trigger_name
in descriptor_names
:
104 "Duplicate condition trigger name `{}`".format(trigger_name
)
108 cond_id
= json_descr
.at("cond-id", tjson
.StrVal
).val
110 if cond_id
.startswith("pre"):
111 cond_type
= _PreCondTriggerDescriptor
112 elif cond_id
.startswith("post"):
113 cond_type
= _PostCondTriggerDescriptor
115 raise ValueError("Invalid condition ID `{}`".format(cond_id
))
117 descriptors
.append(cond_type(index
, trigger_name
, cond_id
))
118 descriptor_names
.add(trigger_name
)
123 # creates the individual tests of `LibPrePostCondsTestCase`
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
134 # get condition trigger descriptor objects from JSON
135 descriptors
= _cond_trigger_descriptors_from_json(json_descr_array
)
137 # create test methods
138 for descriptor
in descriptors
:
140 test_meth_name
= "test_{}".format(
141 re
.sub(r
"[^a-zA-Z0-9_]", "_", descriptor
.trigger_name
)
145 meth
= functools
.partialmethod(_test
, descriptor
)
146 setattr(LibPrePostCondsTestCase
, test_meth_name
, meth
)
152 if __name__
== "__main__":