(self)
| 107 | pool.terminate() |
| 108 | |
| 109 | async def test_pool_05(self): |
| 110 | for n in {1, 3, 5, 10, 20, 100}: |
| 111 | with self.subTest(tasksnum=n): |
| 112 | pool = await self.create_pool(database='postgres', |
| 113 | min_size=5, max_size=10) |
| 114 | |
| 115 | async def worker(): |
| 116 | async with pool.acquire() as con: |
| 117 | self.assertEqual(await con.fetchval('SELECT 1'), 1) |
| 118 | |
| 119 | tasks = [worker() for _ in range(n)] |
| 120 | await asyncio.gather(*tasks) |
| 121 | await pool.close() |
| 122 | |
| 123 | async def test_pool_06(self): |
| 124 | fut = asyncio.Future() |
nothing calls this directly
no test coverage detected