(self)
| 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( |
| 670 | database='postgres', min_size=1, max_size=1, |
| 671 | max_inactive_connection_lifetime=1) as pool: |
| 672 | |
| 673 | # Test that we start counting inactive time *after* |
| 674 | # the connection is being released back to the pool. |
| 675 | |
| 676 | con = pool._holders[0]._con |
| 677 | |
| 678 | await pool.execute('SELECT pg_sleep(0.5)') |
| 679 | await asyncio.sleep(0.6) |
| 680 | |
| 681 | self.assertIs(pool._holders[0]._con, con) |
| 682 | |
| 683 | self.assertEqual( |
| 684 | await pool.execute('SELECT 1::int'), |
| 685 | 'SELECT 1') |
| 686 | self.assertIs(pool._holders[0]._con, con) |
| 687 | |
| 688 | async def test_pool_max_inactive_time_04(self): |
| 689 | # Chaos test for max_inactive_connection_lifetime. |
nothing calls this directly
no test coverage detected