(self, expected, received)
| 1707 | Engine._has_events = False |
| 1708 | |
| 1709 | def _assert_stmts(self, expected, received): |
| 1710 | list(received) |
| 1711 | |
| 1712 | for stmt, params, posn in expected: |
| 1713 | if not received: |
| 1714 | assert False, "Nothing available for stmt: %s" % stmt |
| 1715 | while received: |
| 1716 | teststmt, testparams, testmultiparams = received.pop(0) |
| 1717 | teststmt = ( |
| 1718 | re.compile(r"[\n\t ]+", re.M).sub(" ", teststmt).strip() |
| 1719 | ) |
| 1720 | if teststmt.startswith(stmt) and ( |
| 1721 | testparams == params or testparams == posn |
| 1722 | ): |
| 1723 | break |
| 1724 | |
| 1725 | def test_engine_connect(self, testing_engine): |
| 1726 | e1 = testing_engine(config.db_url) |
no test coverage detected