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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
|
From: Debian Science Maintainers
<debian-science-maintainers@lists.alioth.debian.org>
Date: Sun, 17 Mar 2024 13:44:39 +0100
Subject: Python 3.12 fixes
---
diagnostic_aggregator/test/add_analyzers_test.py | 12 ++++++------
diagnostic_aggregator/test/aggregator_test.py | 22 +++++++++++-----------
.../test/discard_stale_not_published_test.py | 12 ++++++------
diagnostic_aggregator/test/expected_stale_test.py | 8 ++++----
diagnostic_aggregator/test/multiple_match_test.py | 10 +++++-----
diagnostic_analysis/test/bag_csv_test.py | 22 +++++++++++-----------
self_test/scripts/test_selftest.py | 18 +++++++++---------
test_diagnostic_aggregator/test/fail_init_test.py | 8 ++++----
.../test/match_analyze_test.py | 8 ++++----
9 files changed, 60 insertions(+), 60 deletions(-)
diff --git a/diagnostic_aggregator/test/add_analyzers_test.py b/diagnostic_aggregator/test/add_analyzers_test.py
index 0cfd771..f2198ed 100755
--- a/diagnostic_aggregator/test/add_analyzers_test.py
+++ b/diagnostic_aggregator/test/add_analyzers_test.py
@@ -78,7 +78,7 @@ class TestAddAnalyzer(unittest.TestCase):
add_diagnostics = rospy.ServiceProxy('/diagnostics_agg/add_diagnostics', AddDiagnostics)
print(self.namespace)
resp = add_diagnostics(load_namespace=self.namespace)
- self.assert_(resp.success, 'Service call was unsuccessful: {0}'.format(resp.message))
+ self.assertTrue(resp.success, 'Service call was unsuccessful: {0}'.format(resp.message))
def wait_for_agg(self):
self.agg_msgs = {}
@@ -91,7 +91,7 @@ class TestAddAnalyzer(unittest.TestCase):
# confirm that the things we're going to add aren't there already
with self._mutex:
agg_paths = [msg.name for name, msg in self.agg_msgs.items()]
- self.assert_(not any(expected in agg_paths for expected in self.expected))
+ self.assertTrue(not any(expected in agg_paths for expected in self.expected))
# add the new groups
self.add_analyzer()
@@ -109,7 +109,7 @@ class TestAddAnalyzer(unittest.TestCase):
# hasn't been fully formed
with self._mutex:
agg_paths = [msg.name for name, msg in self.agg_msgs.items()]
- self.assert_(all(expected in agg_paths for expected in self.expected))
+ self.assertTrue(all(expected in agg_paths for expected in self.expected))
rospy.sleep(rospy.Duration(5)) # wait a bit for the new items to move to the right group
arr.header.stamp = rospy.get_rostime()
@@ -118,10 +118,10 @@ class TestAddAnalyzer(unittest.TestCase):
for name, msg in self.agg_msgs.items():
if name in self.expected: # should have just received messages on the analyzer
- self.assert_(msg.message == 'OK')
+ self.assertTrue(msg.message == 'OK')
agg_paths = [msg.name for name, msg in self.agg_msgs.items()]
- self.assert_(all(expected in agg_paths for expected in self.expected))
+ self.assertTrue(all(expected in agg_paths for expected in self.expected))
self.bond.shutdown()
@@ -130,7 +130,7 @@ class TestAddAnalyzer(unittest.TestCase):
# the aggregator data should no longer contain the paths once the bond is shut down
with self._mutex:
agg_paths = [msg.name for name, msg in self.agg_msgs.items()]
- self.assert_(not any(expected in agg_paths for expected in self.expected))
+ self.assertTrue(not any(expected in agg_paths for expected in self.expected))
if __name__ == '__main__':
print('SYS ARGS:', sys.argv)
diff --git a/diagnostic_aggregator/test/aggregator_test.py b/diagnostic_aggregator/test/aggregator_test.py
index 426def9..9622a0d 100755
--- a/diagnostic_aggregator/test/aggregator_test.py
+++ b/diagnostic_aggregator/test/aggregator_test.py
@@ -207,22 +207,22 @@ class TestAggregator(unittest.TestCase):
if rospy.get_time() - start > self.duration:
break
- self.assert_(not rospy.is_shutdown(), "Rospy shutdown")
+ self.assertTrue(not rospy.is_shutdown(), "Rospy shutdown")
with self._mutex:
all_headers = {}
for name, msg in self.agg_msgs.items():
- self.assert_(name.startswith('/'), "Aggregated name %s doesn't start with \"/\"" % name)
+ self.assertTrue(name.startswith('/'), "Aggregated name %s doesn't start with \"/\"" % name)
# Go through all messages and check that we have them in aggregate
for name, msg in self.diag_msgs.items():
agg_name = name_to_agg_name(name, self.params)
- self.assert_(agg_name is not None, 'Aggregated name is None for %s' % name)
- self.assert_(agg_name in self.agg_msgs, 'No matching name found for name: %s, aggregated name: %s' % (name, agg_name))
- self.assert_(msg.level == self.agg_msgs[agg_name].level, 'Status level of original, aggregated messages doesn\'t match. Name: %s, aggregated name: %s.' % (name, agg_name))
- self.assert_(msg.message == self.agg_msgs[agg_name].message, 'Status message of original, aggregated messages doesn\'t match. Name: %s, aggregated name: %s' % (name, agg_name))
+ self.assertTrue(agg_name is not None, 'Aggregated name is None for %s' % name)
+ self.assertTrue(agg_name in self.agg_msgs, 'No matching name found for name: %s, aggregated name: %s' % (name, agg_name))
+ self.assertTrue(msg.level == self.agg_msgs[agg_name].level, 'Status level of original, aggregated messages doesn\'t match. Name: %s, aggregated name: %s.' % (name, agg_name))
+ self.assertTrue(msg.message == self.agg_msgs[agg_name].message, 'Status message of original, aggregated messages doesn\'t match. Name: %s, aggregated name: %s' % (name, agg_name))
# This is because the analyzers only reports stale if
# all messages underneath it are stale
@@ -243,17 +243,17 @@ class TestAggregator(unittest.TestCase):
if lvl == -1:
lvl = 3
- self.assert_(header in self.agg_msgs, "Header %s not found in messages" % header)
- self.assert_(self.agg_msgs[header].level == lvl, "Level of header %s doesn't match expected value." % header)
+ self.assertTrue(header in self.agg_msgs, "Header %s not found in messages" % header)
+ self.assertTrue(self.agg_msgs[header].level == lvl, "Level of header %s doesn't match expected value." % header)
del self.agg_msgs[header]
# Check that we have the main header message
if len(prefix) > 0:
- self.assert_(len(self.agg_msgs) == 1, "Incorrect number of messages remaining: %d. Messages: %s" % (len(self.agg_msgs), str(self.agg_msgs)))
+ self.assertTrue(len(self.agg_msgs) == 1, "Incorrect number of messages remaining: %d. Messages: %s" % (len(self.agg_msgs), str(self.agg_msgs)))
- self.assert_(prefix in self.agg_msgs, "Global prefix not found in messages: %s. Messages: %s" % (prefix, str(self.agg_msgs)))
+ self.assertTrue(prefix in self.agg_msgs, "Global prefix not found in messages: %s. Messages: %s" % (prefix, str(self.agg_msgs)))
else:
- self.assert_(len(self.agg_msgs) == 0, "Incorrect number of messages remaining: %d. Messages: %s. Expected 0." % (len(self.agg_msgs), str(self.agg_msgs)))
+ self.assertTrue(len(self.agg_msgs) == 0, "Incorrect number of messages remaining: %d. Messages: %s. Expected 0." % (len(self.agg_msgs), str(self.agg_msgs)))
diff --git a/diagnostic_aggregator/test/discard_stale_not_published_test.py b/diagnostic_aggregator/test/discard_stale_not_published_test.py
index e84008c..d92a7ac 100755
--- a/diagnostic_aggregator/test/discard_stale_not_published_test.py
+++ b/diagnostic_aggregator/test/discard_stale_not_published_test.py
@@ -84,9 +84,9 @@ class TestDiscardStale(unittest.TestCase):
with self._mutex:
expecteds = self._expecteds
- self.assert_(len(expecteds.keys()) == 1, "The expected diagnostics are not of length 1."
+ self.assertTrue(len(expecteds.keys()) == 1, "The expected diagnostics are not of length 1."
"Received diagnostics: {}".format(expecteds))
- self.assert_(expecteds['nonexistent2'].level == DiagnosticStatus.WARN)
+ self.assertTrue(expecteds['nonexistent2'].level == DiagnosticStatus.WARN)
self._start_time = rospy.get_time()
duration = 8
@@ -97,12 +97,12 @@ class TestDiscardStale(unittest.TestCase):
break
with self._mutex:
- self.assert_(len(self._agg_expecteds) == 1,
+ self.assertTrue(len(self._agg_expecteds) == 1,
"There should only be one expected aggregated item left, {} found instead!".
format(len(self._agg_expecteds)))
- self.assert_(self._agg_expecteds[0].name == "/Nonexistent2",
+ self.assertTrue(self._agg_expecteds[0].name == "/Nonexistent2",
"The name of the first aggregated message should be '/Nonexistent2'!")
- self.assert_(self._agg_expecteds[0].level == DiagnosticStatus.STALE,
+ self.assertTrue(self._agg_expecteds[0].level == DiagnosticStatus.STALE,
"The level of the first aggregated message should be stale!")
self._start_time = rospy.get_time()
@@ -114,7 +114,7 @@ class TestDiscardStale(unittest.TestCase):
break
with self._mutex:
- self.assert_(len(self._agg_expecteds) == 0,
+ self.assertTrue(len(self._agg_expecteds) == 0,
"There should't be any aggregated items left, {} found instead! {}".
format(len(self._agg_expecteds), self._agg_expecteds))
diff --git a/diagnostic_aggregator/test/expected_stale_test.py b/diagnostic_aggregator/test/expected_stale_test.py
index 442a959..1052dc5 100755
--- a/diagnostic_aggregator/test/expected_stale_test.py
+++ b/diagnostic_aggregator/test/expected_stale_test.py
@@ -100,14 +100,14 @@ class TestExpectedItemsStale(unittest.TestCase):
break
with self._mutex:
- self.assert_(len(self._expecteds) > 0, "No expected items found in raw data!")
+ self.assertTrue(len(self._expecteds) > 0, "No expected items found in raw data!")
for name, item in self._expecteds.items():
- self.assert_(name in self._agg_expecteds, "Item %s not found in aggregated diagnostics output" % name)
+ self.assertTrue(name in self._agg_expecteds, "Item %s not found in aggregated diagnostics output" % name)
if item.is_stale():
- self.assert_(self._agg_expecteds[name].level == 3, "Stale item in diagnostics, but aggregated didn't report as stale. Item: %s, state: %d" %(name, self._agg_expecteds[name].level))
+ self.assertTrue(self._agg_expecteds[name].level == 3, "Stale item in diagnostics, but aggregated didn't report as stale. Item: %s, state: %d" %(name, self._agg_expecteds[name].level))
else:
- self.assert_(self._agg_expecteds[name].level == item.level, "Diagnostic level of aggregated, raw item don't match for %s" % name)
+ self.assertTrue(self._agg_expecteds[name].level == item.level, "Diagnostic level of aggregated, raw item don't match for %s" % name)
if __name__ == '__main__':
rostest.run(PKG, sys.argv[0], TestExpectedItemsStale, sys.argv)
diff --git a/diagnostic_aggregator/test/multiple_match_test.py b/diagnostic_aggregator/test/multiple_match_test.py
index 8c58a2c..6d50543 100755
--- a/diagnostic_aggregator/test/multiple_match_test.py
+++ b/diagnostic_aggregator/test/multiple_match_test.py
@@ -99,14 +99,14 @@ class TestMultipleMatch(unittest.TestCase):
if rospy.get_time() - self._starttime > DURATION:
break
- self.assert_(not rospy.is_shutdown(), "Rospy shutdown!")
+ self.assertTrue(not rospy.is_shutdown(), "Rospy shutdown!")
with self._mutex:
- self.assert_(HEADER1 in self._multi_items, "Didn't have item under %s. Items: %s" % (HEADER1, self._multi_items))
- self.assert_(self._multi_items[HEADER1].name == MULTI_NAME, "Item name under %s didn't match %s" % (HEADER1, MULTI_NAME))
+ self.assertTrue(HEADER1 in self._multi_items, "Didn't have item under %s. Items: %s" % (HEADER1, self._multi_items))
+ self.assertTrue(self._multi_items[HEADER1].name == MULTI_NAME, "Item name under %s didn't match %s" % (HEADER1, MULTI_NAME))
- self.assert_(HEADER2 in self._multi_items, "Didn't have item under %s" % HEADER2)
- self.assert_(self._multi_items[HEADER2].name == MULTI_NAME, "Item name under %s didn't match %s" % (HEADER2, MULTI_NAME))
+ self.assertTrue(HEADER2 in self._multi_items, "Didn't have item under %s" % HEADER2)
+ self.assertTrue(self._multi_items[HEADER2].name == MULTI_NAME, "Item name under %s didn't match %s" % (HEADER2, MULTI_NAME))
if __name__ == '__main__':
diff --git a/diagnostic_analysis/test/bag_csv_test.py b/diagnostic_analysis/test/bag_csv_test.py
index 1c66956..2da65d4 100755
--- a/diagnostic_analysis/test/bag_csv_test.py
+++ b/diagnostic_analysis/test/bag_csv_test.py
@@ -91,8 +91,8 @@ class TestBagToCSV(unittest.TestCase):
##\brief Tests that exported file exists and is not None
def test_file_exists(self):
- self.assert_(self.filename is not None, "CSV file is None")
- self.assert_(os.path.isfile(self.filename), "CSV file doesn't exist")
+ self.assertTrue(self.filename is not None, "CSV file is None")
+ self.assertTrue(os.path.isfile(self.filename), "CSV file doesn't exist")
##\brief Test that CSV file has correct data, number of lines
def test_export(self):
@@ -101,26 +101,26 @@ class TestBagToCSV(unittest.TestCase):
count = -1
for row in input_reader:
if count == -1:
- self.assert_(row[2].strip() == 'Message')
- self.assert_(row[3].strip() == 'Hardware ID')
- self.assert_(row[4].strip() == 'Value A')
+ self.assertTrue(row[2].strip() == 'Message')
+ self.assertTrue(row[3].strip() == 'Hardware ID')
+ self.assertTrue(row[4].strip() == 'Value A')
count += 1
continue
- self.assert_(row[2].strip() == 'OK')
- self.assert_(row[3].strip() == 'HW ID')
- self.assert_(row[4].strip() == str(count))
+ self.assertTrue(row[2].strip() == 'OK')
+ self.assertTrue(row[3].strip() == 'HW ID')
+ self.assertTrue(row[4].strip() == str(count))
count += 1
- self.assert_(count == row_count, "Row count doesn't match")
+ self.assertTrue(count == row_count, "Row count doesn't match")
##\brief Tests that sparse CSV made with 'skip' option has correct number of lines
def test_sparse_skip(self):
- self.assert_(len(open(self.skip_10).read().split('\n')) <= int(row_count / 10) + 2, "Length of sparse CSV (skipped) incorrect")
+ self.assertTrue(len(open(self.skip_10).read().split('\n')) <= int(row_count / 10) + 2, "Length of sparse CSV (skipped) incorrect")
##\brief Tests that sparse CSV made with 'length' option has correct number of lines
def test_sparse_length(self):
- self.assert_(len(open(self.length_10).read().split('\n')) == 12, "Length of sparse CSV incorrect")
+ self.assertTrue(len(open(self.length_10).read().split('\n')) == 12, "Length of sparse CSV incorrect")
def tearDown(self):
self.bag.close()
diff --git a/self_test/scripts/test_selftest.py b/self_test/scripts/test_selftest.py
index 3b6ed66..f5fa7f9 100755
--- a/self_test/scripts/test_selftest.py
+++ b/self_test/scripts/test_selftest.py
@@ -79,32 +79,32 @@ class TestSelfTest(unittest.TestCase):
try:
rospy.wait_for_service(SRV_NAME, 15)
except Exception as e:
- self.assert_(False, "Service %s did not respond. Unable to test self_test" % SRV_NAME)
+ self.assertTrue(False, "Service %s did not respond. Unable to test self_test" % SRV_NAME)
try:
res = proxy()
except Exception as e:
import traceback
- self.assert_(False, "Error calling self_test service. Exception: %s" % traceback.format_exc())
+ self.assertTrue(False, "Error calling self_test service. Exception: %s" % traceback.format_exc())
if self.no_id:
- self.assert_(res.id == '', "Result had node ID even though ID was not expected. ID: %s" % res.id)
+ self.assertTrue(res.id == '', "Result had node ID even though ID was not expected. ID: %s" % res.id)
else:
- self.assert_(res.id != '', "Result had no node ID")
+ self.assertTrue(res.id != '', "Result had no node ID")
if self.expect_fail or self.exception:
- self.assert_(res.passed == 0, "Self test passed, but it shouldn't have. Result: %d" % res.passed)
+ self.assertTrue(res.passed == 0, "Self test passed, but it shouldn't have. Result: %d" % res.passed)
max_val = 0
for tst in res.status:
max_val = max(max_val, tst.level)
- self.assert_(max_val > 0, "Self test failed, but no sub tests reported a failure or warning")
+ self.assertTrue(max_val > 0, "Self test failed, but no sub tests reported a failure or warning")
else:
- self.assert_(res.passed, "Self test failed, but we expected a pass")
+ self.assertTrue(res.passed, "Self test failed, but we expected a pass")
for tst in res.status:
- self.assert_(tst.level == 0, "Self test subtest failed, but we marked it as a pass")
+ self.assertTrue(tst.level == 0, "Self test subtest failed, but we marked it as a pass")
if self.exception:
@@ -113,7 +113,7 @@ class TestSelfTest(unittest.TestCase):
if tst.message.find('exception') > -1:
found_ex = True
- self.assert_(found_ex, "Self test threw and exception, but we didn't catch it and report it")
+ self.assertTrue(found_ex, "Self test threw and exception, but we didn't catch it and report it")
if __name__ == '__main__':
diff --git a/test_diagnostic_aggregator/test/fail_init_test.py b/test_diagnostic_aggregator/test/fail_init_test.py
index e743d39..e68aa91 100755
--- a/test_diagnostic_aggregator/test/fail_init_test.py
+++ b/test_diagnostic_aggregator/test/fail_init_test.py
@@ -96,12 +96,12 @@ class TestFailInit(unittest.TestCase):
if rospy.get_time() - self._starttime > DURATION:
break
- self.assert_(not rospy.is_shutdown(), "Rospy shutdown!")
+ self.assertTrue(not rospy.is_shutdown(), "Rospy shutdown!")
with self._mutex:
- self.assert_(self._ns, "Namespace is none. Option --ns not given")
- self.assert_(self._item, "No item with name %s found in diag_agg" % self._ns)
- self.assert_(self._item.level == 3, "Item failed to initialize, but was not stale. Level: %d" % self._item.level)
+ self.assertTrue(self._ns, "Namespace is none. Option --ns not given")
+ self.assertTrue(self._item, "No item with name %s found in diag_agg" % self._ns)
+ self.assertTrue(self._item.level == 3, "Item failed to initialize, but was not stale. Level: %d" % self._item.level)
if __name__ == '__main__':
diff --git a/test_diagnostic_aggregator/test/match_analyze_test.py b/test_diagnostic_aggregator/test/match_analyze_test.py
index 29ae52b..73a48f9 100755
--- a/test_diagnostic_aggregator/test/match_analyze_test.py
+++ b/test_diagnostic_aggregator/test/match_analyze_test.py
@@ -97,12 +97,12 @@ class TestMatchAnalyze(unittest.TestCase):
if rospy.get_time() - self._starttime > DURATION:
break
- self.assert_(not rospy.is_shutdown(), "Rospy shutdown!")
+ self.assertTrue(not rospy.is_shutdown(), "Rospy shutdown!")
with self._mutex:
- self.assert_(self.header, "Header is none. Option --header not given")
- self.assert_(len(self.match_headers) == 1, "Multiple analyzers reported our item! Headers: %s" % self.match_headers)
- self.assert_(self.match_headers.count(self.header) > 0, "Didn't have item under header \"%s\". Header: \"%s\"" % (self.header, self.match_headers[0]))
+ self.assertTrue(self.header, "Header is none. Option --header not given")
+ self.assertTrue(len(self.match_headers) == 1, "Multiple analyzers reported our item! Headers: %s" % self.match_headers)
+ self.assertTrue(self.match_headers.count(self.header) > 0, "Didn't have item under header \"%s\". Header: \"%s\"" % (self.header, self.match_headers[0]))
if __name__ == '__main__':
|
