(self)
| 624 | await pool.execute('DROP TABLE exmany') |
| 625 | |
| 626 | async def test_pool_max_inactive_time_01(self): |
| 627 | async with self.create_pool( |
| 628 | database='postgres', min_size=1, max_size=1, |
| 629 | max_inactive_connection_lifetime=0.1) as pool: |
| 630 | |
| 631 | # Test that it's OK if a query takes longer time to execute |
| 632 | # than `max_inactive_connection_lifetime`. |
| 633 | |
| 634 | con = pool._holders[0]._con |
| 635 | |
| 636 | for _ in range(3): |
| 637 | await pool.execute('SELECT pg_sleep(0.5)') |
| 638 | self.assertIs(pool._holders[0]._con, con) |
| 639 | |
| 640 | self.assertEqual( |
| 641 | await pool.execute('SELECT 1::int'), |
| 642 | 'SELECT 1') |
| 643 | self.assertIs(pool._holders[0]._con, con) |
| 644 | |
| 645 | async def test_pool_max_inactive_time_02(self): |
| 646 | async with self.create_pool( |
nothing calls this directly
no test coverage detected