(self, expected, received)
| 366 | Engine._has_events = False |
| 367 | |
| 368 | def _assert_stmts(self, expected, received): |
| 369 | list(received) |
| 370 | for stmt, params, posn in expected: |
| 371 | if not received: |
| 372 | assert False, "Nothing available for stmt: %s" % stmt |
| 373 | while received: |
| 374 | teststmt, testparams, testmultiparams = received.pop(0) |
| 375 | teststmt = ( |
| 376 | re.compile(r"[\n\t ]+", re.M).sub(" ", teststmt).strip() |
| 377 | ) |
| 378 | if teststmt.startswith(stmt) and ( |
| 379 | testparams == params or testparams == posn |
| 380 | ): |
| 381 | break |
| 382 | |
| 383 | def test_engine_connect(self, testing_engine): |
| 384 | e1 = testing_engine(config.db_url) |
nothing calls this directly
no test coverage detected