(self)
| 297 | self.assertEqual(pool.operation_count, 0) |
| 298 | |
| 299 | async def test_pool_check(self): |
| 300 | # Test that Pool recovers from two connection failures in a row. |
| 301 | # This exercises code at the end of Pool._check(). |
| 302 | cx_pool = await self.create_pool(max_pool_size=1, connect_timeout=1, wait_queue_timeout=1) |
| 303 | cx_pool._check_interval_seconds = 0 # Always check. |
| 304 | self.addAsyncCleanup(cx_pool.close) |
| 305 | |
| 306 | async with cx_pool.checkout() as conn: |
| 307 | # Simulate a closed socket without telling the Connection it's |
| 308 | # closed. |
| 309 | await conn.conn.close() |
| 310 | |
| 311 | # Swap pool's address with a bad one. |
| 312 | address, cx_pool.address = cx_pool.address, ("foo.com", 1234) |
| 313 | with self.assertRaises(AutoReconnect): |
| 314 | async with cx_pool.checkout(): |
| 315 | pass |
| 316 | |
| 317 | # Back to normal, semaphore was correctly released. |
| 318 | cx_pool.address = address |
| 319 | async with cx_pool.checkout(): |
| 320 | pass |
| 321 | |
| 322 | async def test_wait_queue_timeout(self): |
| 323 | wait_queue_timeout = 2 # Seconds |
nothing calls this directly
no test coverage detected