Coverage for /home/jenkins/workspace/NDS/Zserio/NDS_ZSERIO-linux-build/compiler/extensions/python/runtime/tests/test_walker.py: 100%

319 statements  

« prev     ^ index     » next       coverage.py v6.5.0, created at 2023-12-13 15:12 +0000

1import unittest 

2 

3from test_object.api import WalkerBitmask, WalkerNested, WalkerObject, WalkerUnion 

4 

5from zserio import (WalkObserver, WalkFilter, Walker, DefaultWalkObserver, PythonRuntimeException, 

6 ArrayLengthWalkFilter, DepthWalkFilter, RegexWalkFilter, AndWalkFilter) 

7from zserio.typeinfo import TypeInfo, TypeAttribute, MemberInfo, MemberAttribute 

8 

9class TestWalkObserver(WalkObserver): 

10 

11 def __init__(self): 

12 self._captures = { 

13 "begin_root": None, 

14 "end_root": None, 

15 "begin_array": [], 

16 "end_array": [], 

17 "begin_compound": [], 

18 "end_compound": [], 

19 "visit_value": [] 

20 } 

21 

22 @property 

23 def captures(self): 

24 return self._captures 

25 

26 def begin_root(self, compound): 

27 self._captures["begin_root"] = compound 

28 

29 def end_root(self, compound): 

30 self._captures["end_root"] = compound 

31 

32 def begin_array(self, array, member_info): 

33 self._captures["begin_array"].append(array) 

34 

35 def end_array(self, array, member_info): 

36 self._captures["end_array"].append(array) 

37 

38 def begin_compound(self, compound, member_info, element_index = None): 

39 self._captures["begin_compound"].append(compound) 

40 

41 def end_compound(self, compound, member_info, element_index = None): 

42 self._captures["end_compound"].append(compound) 

43 

44 def visit_value(self, value, member_info, element_index = None): 

45 self._captures["visit_value"].append(value) 

46 

47 

48class TestWalkFilter(WalkFilter): 

49 

50 def __init__(self, *, before_array=True, after_array=True, only_first_element=False, 

51 before_compound=True, after_compound=True, before_value=True, after_value=True): 

52 self._config = { 

53 "before_array": before_array, 

54 "after_array": after_array, 

55 "only_first_element": only_first_element, 

56 "before_compound": before_compound, 

57 "after_compound": after_compound, 

58 "before_value": before_value, 

59 "after_value": after_value 

60 } 

61 self._is_first_element = False 

62 

63 def before_array(self, array, member_info): 

64 self._is_first_element = True 

65 return self._config["before_array"] 

66 

67 def after_array(self, array, member_info): 

68 self._is_first_element = False 

69 return self._config["after_array"] 

70 

71 def before_compound(self, compound, member_info, element_index = None): 

72 return self._config["before_compound"] 

73 

74 def after_compound(self, compound, member_info, element_index = None): 

75 go_to_next = not (self._config["only_first_element"] and self._is_first_element) 

76 self._is_first_element = False 

77 return go_to_next and self._config["after_compound"] 

78 

79 def before_value(self, value, member_info, element_index = None): 

80 return self._config["before_value"] 

81 

82 def after_value(self, value, member_info, element_index = None): 

83 return self._config["after_value"] 

84 

85 

86class WalkerTest(unittest.TestCase): 

87 

88 def test_observer(self): 

89 observer = WalkObserver() 

90 

91 walker_object = object() 

92 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

93 

94 with self.assertRaises(NotImplementedError): 

95 observer.begin_root(walker_object) 

96 with self.assertRaises(NotImplementedError): 

97 observer.end_root(walker_object) 

98 with self.assertRaises(NotImplementedError): 

99 observer.begin_array([], walker_member_info) 

100 with self.assertRaises(NotImplementedError): 

101 observer.end_array([], walker_member_info) 

102 with self.assertRaises(NotImplementedError): 

103 observer.begin_compound(walker_object, walker_member_info) 

104 with self.assertRaises(NotImplementedError): 

105 observer.end_compound(walker_object, walker_member_info) 

106 with self.assertRaises(NotImplementedError): 

107 observer.visit_value("", walker_member_info) 

108 

109 def test_filter(self): 

110 walk_filter = WalkFilter() 

111 

112 walker_object = object() 

113 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

114 

115 with self.assertRaises(NotImplementedError): 

116 walk_filter.before_array([], walker_member_info) 

117 with self.assertRaises(NotImplementedError): 

118 walk_filter.after_array([], walker_member_info) 

119 with self.assertRaises(NotImplementedError): 

120 walk_filter.before_compound(walker_object, walker_member_info) 

121 with self.assertRaises(NotImplementedError): 

122 walk_filter.after_compound(walker_object, walker_member_info) 

123 with self.assertRaises(NotImplementedError): 

124 walk_filter.before_value("", walker_member_info) 

125 with self.assertRaises(NotImplementedError): 

126 walk_filter.after_value("", walker_member_info) 

127 

128 def test_walk_without_type_info(self): 

129 walker = Walker(TestWalkObserver()) 

130 obj = object() 

131 with self.assertRaises(PythonRuntimeException): 

132 walker.walk(obj) 

133 

134 def test_walk_non_compound(self): 

135 walker = Walker(TestWalkObserver()) 

136 with self.assertRaises(PythonRuntimeException): 

137 walker.walk(WalkerBitmask()) 

138 

139 def test_walk(self): 

140 observer = TestWalkObserver() 

141 walker = Walker(observer) 

142 obj = _create_walker_object() 

143 walker.walk(obj) 

144 self.assertEqual(obj, observer.captures["begin_root"]) 

145 self.assertEqual(obj, observer.captures["end_root"]) 

146 self.assertEqual([obj.union_array, obj.union_array[2].nested_array], observer.captures["begin_array"]) 

147 self.assertEqual([obj.union_array[2].nested_array, obj.union_array], observer.captures["end_array"]) 

148 self.assertEqual( 

149 [ 

150 obj.nested, obj.union_array[0], obj.union_array[1], 

151 obj.union_array[2], obj.union_array[2].nested_array[0] 

152 ], 

153 observer.captures["begin_compound"] 

154 ) 

155 self.assertEqual( 

156 [ 

157 obj.nested, obj.union_array[0], obj.union_array[1], 

158 obj.union_array[2].nested_array[0], obj.union_array[2] 

159 ], 

160 observer.captures["end_compound"] 

161 ) 

162 self.assertEqual([13, "nested", "test", '1', 2, "nestedArray", None], observer.captures["visit_value"]) 

163 

164 def test_walk_skip_compound(self): 

165 observer = TestWalkObserver() 

166 test_filter = TestWalkFilter(before_array=True, after_array=True, 

167 before_compound=False, after_compound=True, 

168 before_value=True, after_value=True) 

169 walker = Walker(observer, AndWalkFilter([test_filter])) 

170 obj = _create_walker_object() 

171 walker.walk(obj) 

172 self.assertEqual(obj, observer.captures["begin_root"]) 

173 self.assertEqual(obj, observer.captures["end_root"]) 

174 self.assertEqual([obj.union_array], observer.captures["begin_array"]) 

175 self.assertEqual([obj.union_array], observer.captures["end_array"]) 

176 self.assertEqual([], observer.captures["begin_compound"]) 

177 self.assertEqual([], observer.captures["end_compound"]) 

178 self.assertEqual([13, "test", None], observer.captures["visit_value"]) 

179 

180 def test_walk_skip_siblings(self): 

181 observer = TestWalkObserver() 

182 test_filter = TestWalkFilter(before_array=True, after_array=True, 

183 before_compound=True, after_compound=True, 

184 before_value=True, after_value=False) 

185 walker = Walker(observer, AndWalkFilter([test_filter])) 

186 obj = _create_walker_object() 

187 walker.walk(obj) 

188 self.assertEqual(obj, observer.captures["begin_root"]) 

189 self.assertEqual(obj, observer.captures["end_root"]) 

190 self.assertEqual([], observer.captures["begin_array"]) 

191 self.assertEqual([], observer.captures["end_array"]) 

192 self.assertEqual([], observer.captures["begin_compound"]) 

193 self.assertEqual([], observer.captures["end_compound"]) 

194 self.assertEqual([13], observer.captures["visit_value"]) 

195 

196 def test_walk_skip_after_nested(self): 

197 observer = TestWalkObserver() 

198 test_filter = TestWalkFilter(before_array=True, after_array=True, 

199 before_compound=True, after_compound=False, 

200 before_value=True, after_value=True) 

201 walker = Walker(observer, AndWalkFilter([test_filter])) 

202 obj = _create_walker_object() 

203 walker.walk(obj) 

204 self.assertEqual(obj, observer.captures["begin_root"]) 

205 self.assertEqual(obj, observer.captures["end_root"]) 

206 self.assertEqual([], observer.captures["begin_array"]) 

207 self.assertEqual([], observer.captures["end_array"]) 

208 self.assertEqual([obj.nested], observer.captures["begin_compound"]) 

209 self.assertEqual([obj.nested], observer.captures["end_compound"]) 

210 self.assertEqual([13, "nested"], observer.captures["visit_value"]) 

211 

212 def test_walk_only_first_element(self): 

213 observer = TestWalkObserver() 

214 test_filter = TestWalkFilter(before_array=True, after_array=True, only_first_element=True, 

215 before_compound=True, after_compound=True, 

216 before_value=True, after_value=True) 

217 walker = Walker(observer, AndWalkFilter([test_filter])) 

218 obj = _create_walker_object() 

219 walker.walk(obj) 

220 self.assertEqual(obj, observer.captures["begin_root"]) 

221 self.assertEqual(obj, observer.captures["end_root"]) 

222 self.assertEqual([obj.union_array], observer.captures["begin_array"]) 

223 self.assertEqual([obj.union_array], observer.captures["end_array"]) 

224 self.assertEqual([obj.nested, obj.union_array[0]], observer.captures["begin_compound"]) 

225 self.assertEqual([obj.nested, obj.union_array[0]], observer.captures["end_compound"]) 

226 self.assertEqual([13, "nested", "test", '1', None], observer.captures["visit_value"]) 

227 

228 

229class DefaultObserverTest(unittest.TestCase): 

230 

231 @staticmethod 

232 def test_default(): 

233 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

234 default_observer = DefaultWalkObserver() 

235 

236 default_observer.begin_root(object()) 

237 default_observer.end_root(object()) 

238 default_observer.begin_array([], walker_member_info) 

239 default_observer.end_array([], walker_member_info) 

240 default_observer.begin_compound(object(), walker_member_info) 

241 default_observer.end_compound(object(), walker_member_info) 

242 default_observer.visit_value(None, walker_member_info) 

243 

244class DepthWalkFilterTest(unittest.TestCase): 

245 

246 def test_depth_0(self): 

247 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

248 depth_filter = DepthWalkFilter(0) 

249 

250 self.assertFalse(depth_filter.before_array([], walker_member_info)) 

251 self.assertTrue(depth_filter.after_array([], walker_member_info)) 

252 

253 self.assertFalse(depth_filter.before_compound(object(), walker_member_info)) 

254 self.assertTrue(depth_filter.after_compound(object(), walker_member_info)) 

255 

256 self.assertFalse(depth_filter.before_value(None, walker_member_info)) 

257 self.assertTrue(depth_filter.after_value(None, walker_member_info)) 

258 

259 def test_depth_1(self): 

260 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

261 depth_filter = DepthWalkFilter(1) 

262 

263 self.assertTrue(depth_filter.before_array([], walker_member_info)) # 0 

264 self.assertFalse(depth_filter.before_array([], walker_member_info)) # 1 

265 self.assertTrue(depth_filter.after_array([], walker_member_info)) # 1 

266 self.assertFalse(depth_filter.before_compound(object(), walker_member_info, 0)) # 1 

267 self.assertTrue(depth_filter.after_compound(object(), walker_member_info, 0)) # 1 

268 self.assertFalse(depth_filter.before_value(None, walker_member_info, 1)) # 1 

269 self.assertTrue(depth_filter.after_value(None, walker_member_info, 1)) # 1 

270 self.assertTrue(depth_filter.after_array([], walker_member_info)) # 0 

271 

272 self.assertTrue(depth_filter.before_compound(object(), walker_member_info)) # 0 

273 self.assertFalse(depth_filter.before_array([], walker_member_info)) # 1 

274 self.assertTrue(depth_filter.after_array([], walker_member_info)) # 1 

275 self.assertFalse(depth_filter.before_compound(object(), walker_member_info)) # 1 

276 self.assertTrue(depth_filter.after_compound(object(), walker_member_info)) # 1 

277 self.assertFalse(depth_filter.before_value(None, walker_member_info)) # 1 

278 self.assertTrue(depth_filter.after_value(None, walker_member_info)) # 1 

279 self.assertTrue(depth_filter.after_compound(object(), walker_member_info)) # 0 

280 

281 self.assertTrue(depth_filter.before_value(None, walker_member_info)) # 0 

282 self.assertTrue(depth_filter.after_value(None, walker_member_info)) # 0 

283 

284 

285class RegexWalkFilterTest(unittest.TestCase): 

286 

287 def test_regex_all_match(self): 

288 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

289 walker_array_member_info = MemberInfo("walker", TypeInfo("Walker", None), attributes={ 

290 MemberAttribute.ARRAY_LENGTH: None 

291 }) 

292 regex_filter = RegexWalkFilter(".*") 

293 

294 self.assertTrue(regex_filter.before_array([], walker_array_member_info)) 

295 self.assertTrue(regex_filter.after_array([], walker_array_member_info)) 

296 self.assertTrue(regex_filter.before_compound(object(), walker_member_info)) 

297 self.assertTrue(regex_filter.before_compound(object(), walker_member_info)) 

298 self.assertTrue(regex_filter.before_value(None, walker_member_info)) 

299 self.assertTrue(regex_filter.after_value(None, walker_member_info)) 

300 

301 def test_regex_prefix_match(self): 

302 walker_object = _create_walker_object() 

303 regex_filter = RegexWalkFilter("nested\\..*") 

304 

305 identifier_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][0] 

306 self.assertFalse(regex_filter.before_value(walker_object.identifier, identifier_member_info)) 

307 self.assertTrue(regex_filter.after_value(walker_object.identifier, identifier_member_info)) 

308 

309 nested_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][1] 

310 self.assertTrue(regex_filter.before_compound(walker_object.nested, nested_member_info)) 

311 text_member_info = nested_member_info.type_info.attributes[TypeAttribute.FIELDS][0] 

312 self.assertTrue(regex_filter.before_value(walker_object.nested.text, text_member_info)) 

313 self.assertTrue(regex_filter.after_value(walker_object.nested.text, text_member_info)) 

314 self.assertTrue(regex_filter.after_compound(walker_object.nested, nested_member_info)) 

315 

316 # ignore text 

317 

318 union_array_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][3] 

319 self.assertFalse(regex_filter.before_array(walker_object.union_array, union_array_member_info)) 

320 self.assertTrue(regex_filter.after_array(walker_object.union_array, union_array_member_info)) 

321 

322 def test_regex_array_match(self): 

323 walker_object = _create_walker_object() 

324 regex_filter = RegexWalkFilter("unionArray\\[\\d+\\]\\.nes.*") 

325 

326 union_array_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][3] 

327 self.assertTrue(regex_filter.before_array(walker_object.union_array, union_array_member_info)) 

328 

329 self.assertFalse(regex_filter.before_compound(walker_object.union_array[0], union_array_member_info, 0)) 

330 self.assertTrue(regex_filter.after_compound(walker_object.union_array[0], union_array_member_info, 0)) 

331 

332 self.assertFalse(regex_filter.before_compound(walker_object.union_array[1], union_array_member_info, 1)) 

333 self.assertTrue(regex_filter.after_compound(walker_object.union_array[1], union_array_member_info, 1)) 

334 

335 self.assertTrue(regex_filter.before_compound(walker_object.union_array[2], union_array_member_info, 2)) 

336 self.assertTrue(regex_filter.after_compound(walker_object.union_array[2], union_array_member_info, 2)) 

337 

338 self.assertTrue(regex_filter.after_array(walker_object.union_array, union_array_member_info)) 

339 

340 def test_regex_array_no_match(self): 

341 walker_object = WalkerObject(13, WalkerNested("nested"), "test", 

342 [WalkerUnion(nested_array_=[WalkerNested("nestedArray")])]) 

343 

344 regex_filter = RegexWalkFilter("^unionArray\\[\\d*\\]\\.te.*") 

345 

346 union_array_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][3] 

347 self.assertFalse(regex_filter.before_array(walker_object.union_array, union_array_member_info)) 

348 self.assertTrue(regex_filter.after_array(walker_object.union_array, union_array_member_info)) 

349 

350 def test_regex_none_compound_match(self): 

351 walker_object = WalkerObject(13, None, "test", [WalkerUnion(text_="1"), WalkerUnion(value_=2)]) 

352 

353 regex_filter = RegexWalkFilter("nested") 

354 

355 nested_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][1] 

356 # note that the None compounds are processed as values! 

357 self.assertTrue(regex_filter.before_value(walker_object.nested, nested_member_info)) 

358 self.assertTrue(regex_filter.after_value(walker_object.nested, nested_member_info)) 

359 

360 def test_regex_none_compound_no_match(self): 

361 walker_object = WalkerObject(13, None, "test", [WalkerUnion(text_="1"), WalkerUnion(value_=2)]) 

362 

363 regex_filter = RegexWalkFilter("^nested\\.text$") 

364 

365 nested_member_info = walker_object.type_info().attributes[TypeAttribute.FIELDS][1] 

366 # note that the None compounds are processed as values! 

367 self.assertFalse(regex_filter.before_value(walker_object.nested, nested_member_info)) 

368 self.assertTrue(regex_filter.after_value(walker_object.nested, nested_member_info)) 

369 

370 def test_regex_none_array_match(self): 

371 regex_filter = RegexWalkFilter("optionalUnionArray") 

372 

373 optional_union_array_member_info = WalkerObject.type_info().attributes[TypeAttribute.FIELDS][4] 

374 # note that the None arrays are processed as values! 

375 self.assertTrue(regex_filter.before_value(None, optional_union_array_member_info)) 

376 self.assertTrue(regex_filter.after_value(None, optional_union_array_member_info)) 

377 

378 def test_regex_none_array_no_match(self): 

379 regex_filter = RegexWalkFilter("^optionalUnionArray\\[\\d+\\]\\.nestedArray.*") 

380 

381 union_array_member_info = WalkerObject.type_info().attributes[TypeAttribute.FIELDS][3] 

382 # note that the None arrays are processed as values! 

383 self.assertFalse(regex_filter.before_value(None, union_array_member_info)) 

384 self.assertTrue(regex_filter.after_value(None, union_array_member_info)) 

385 

386 

387class ArrayLengthWalkFilterTest(unittest.TestCase): 

388 

389 def test_array_length_0(self): 

390 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

391 walker_array_member_info = MemberInfo("walker_array", TypeInfo("Walker", None), attributes={ 

392 MemberAttribute.ARRAY_LENGTH : None 

393 }) 

394 array_length_filter = ArrayLengthWalkFilter(0) 

395 

396 self.assertTrue(array_length_filter.before_array([], walker_array_member_info)) 

397 self.assertFalse(array_length_filter.before_compound(object(), walker_array_member_info, 0)) 

398 self.assertFalse(array_length_filter.after_compound(object(), walker_array_member_info, 0)) 

399 self.assertFalse(array_length_filter.before_value(None, walker_array_member_info, 1)) 

400 self.assertFalse(array_length_filter.after_value(None, walker_array_member_info, 1)) 

401 self.assertTrue(array_length_filter.after_array([], walker_array_member_info)) 

402 

403 self.assertTrue(array_length_filter.before_compound(object(), walker_member_info)) 

404 self.assertTrue(array_length_filter.before_value(None, walker_member_info)) 

405 self.assertTrue(array_length_filter.after_value(None, walker_member_info)) 

406 self.assertTrue(array_length_filter.before_array([], walker_array_member_info)) 

407 self.assertFalse(array_length_filter.before_value(None, walker_array_member_info, 0)) 

408 self.assertFalse(array_length_filter.after_value(None, walker_array_member_info, 0)) 

409 self.assertTrue(array_length_filter.after_array([], walker_array_member_info)) 

410 self.assertTrue(array_length_filter.after_compound(object(), walker_member_info)) 

411 

412 

413class AndWalkFilterTest(unittest.TestCase): 

414 

415 def test_empty(self): 

416 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

417 walk_filter = AndWalkFilter([]) 

418 

419 self.assertTrue(walk_filter.before_array([], walker_member_info)) 

420 self.assertTrue(walk_filter.after_array([], walker_member_info)) 

421 self.assertTrue(walk_filter.before_compound(object(), walker_member_info)) 

422 self.assertTrue(walk_filter.after_compound(object(), walker_member_info)) 

423 self.assertTrue(walk_filter.before_value(None, walker_member_info)) 

424 self.assertTrue(walk_filter.after_value(None, walker_member_info)) 

425 

426 def test_true_true(self): 

427 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

428 # two filters returning True 

429 walk_filter = AndWalkFilter([TestWalkFilter(), TestWalkFilter()]) 

430 

431 self.assertTrue(walk_filter.before_array([], walker_member_info)) 

432 self.assertTrue(walk_filter.after_array([], walker_member_info)) 

433 self.assertTrue(walk_filter.before_compound(object(), walker_member_info)) 

434 self.assertTrue(walk_filter.after_compound(object(), walker_member_info)) 

435 self.assertTrue(walk_filter.before_value(None, walker_member_info)) 

436 self.assertTrue(walk_filter.after_value(None, walker_member_info)) 

437 

438 def test_false_false(self): 

439 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

440 # two filters returning False 

441 walk_filter = AndWalkFilter([ 

442 TestWalkFilter(before_array=False, after_array=False, 

443 before_compound=False, after_compound=False, 

444 before_value=False, after_value=False), 

445 TestWalkFilter(before_array=False, after_array=False, 

446 before_compound=False, after_compound=False, 

447 before_value=False, after_value=False) 

448 ]) 

449 

450 self.assertFalse(walk_filter.before_array([], walker_member_info)) 

451 self.assertFalse(walk_filter.after_array([], walker_member_info)) 

452 self.assertFalse(walk_filter.before_compound(object(), walker_member_info)) 

453 self.assertFalse(walk_filter.after_compound(object(), walker_member_info)) 

454 self.assertFalse(walk_filter.before_value(None, walker_member_info)) 

455 self.assertFalse(walk_filter.after_value(None, walker_member_info)) 

456 

457 def test_true_false(self): 

458 walker_member_info = MemberInfo("walker", TypeInfo("Walker", None)) 

459 walk_filter = AndWalkFilter([ 

460 TestWalkFilter(), # returning true 

461 TestWalkFilter(before_array=False, after_array=False, 

462 before_compound=False, after_compound=False, 

463 before_value=False, after_value=False) # returning false 

464 ]) 

465 

466 self.assertFalse(walk_filter.before_array([], walker_member_info)) 

467 self.assertFalse(walk_filter.after_array([], walker_member_info)) 

468 self.assertFalse(walk_filter.before_compound(object(), walker_member_info)) 

469 self.assertFalse(walk_filter.after_compound(object(), walker_member_info)) 

470 self.assertFalse(walk_filter.before_value(None, walker_member_info)) 

471 self.assertFalse(walk_filter.after_value(None, walker_member_info)) 

472 

473 

474def _create_walker_object(): 

475 return WalkerObject(13, WalkerNested("nested"), "test", [ 

476 WalkerUnion(text_="1"), 

477 WalkerUnion(value_=2), 

478 WalkerUnion(nested_array_=[WalkerNested("nestedArray")]) 

479 ])