(self)
| 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( |
| 647 | database='postgres', min_size=1, max_size=1, |
| 648 | max_inactive_connection_lifetime=0.5) as pool: |
| 649 | |
| 650 | # Test that we have a new connection after pool not |
| 651 | # being used longer than `max_inactive_connection_lifetime`. |
| 652 | |
| 653 | con = pool._holders[0]._con |
| 654 | |
| 655 | self.assertEqual( |
| 656 | await pool.execute('SELECT 1::int'), |
| 657 | 'SELECT 1') |
| 658 | self.assertIs(pool._holders[0]._con, con) |
| 659 | |
| 660 | await asyncio.sleep(1) |
| 661 | self.assertIs(pool._holders[0]._con, None) |
| 662 | |
| 663 | self.assertEqual( |
| 664 | await pool.execute('SELECT 1::int'), |
| 665 | 'SELECT 1') |
| 666 | self.assertIsNot(pool._holders[0]._con, con) |
| 667 | |
| 668 | async def test_pool_max_inactive_time_03(self): |
| 669 | async with self.create_pool( |
nothing calls this directly
no test coverage detected