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