(self, query, timeout, **kwargs)
| 139 | class SlowPrepareConnection(pg_connection.Connection): |
| 140 | """Connection class to test timeouts.""" |
| 141 | async def _get_statement(self, query, timeout, **kwargs): |
| 142 | await asyncio.sleep(0.3) |
| 143 | return await super()._get_statement(query, timeout, **kwargs) |
| 144 | |
| 145 | |
| 146 | class TestTimeoutCoversPrepare(tb.ConnectedTestCase): |
nothing calls this directly
no outgoing calls
no test coverage detected