(self)
| 36 | self.assertEqual(await self.con.fetch('select 1'), [(1,)]) |
| 37 | |
| 38 | async def test_timeout_03(self): |
| 39 | task = self.loop.create_task( |
| 40 | self.con.fetch('select pg_sleep(10)', timeout=0.2)) |
| 41 | await asyncio.sleep(0.05) |
| 42 | task.cancel() |
| 43 | with self.assertRaises(asyncio.CancelledError), \ |
| 44 | self.assertRunUnder(MAX_RUNTIME): |
| 45 | await task |
| 46 | self.assertEqual(await self.con.fetch('select 1'), [(1,)]) |
| 47 | |
| 48 | async def test_timeout_04(self): |
| 49 | st = await self.con.prepare('select pg_sleep(10)', timeout=0.1) |
nothing calls this directly
no test coverage detected