Method test_pool_removes_dead_socket

test/test_pooling.py:226–246

(self)

        self.assertEqual(0, len(cx_pool.conns))

    def test_pool_removes_dead_socket(self):
        # Test that Pool removes dead socket and the socket doesn't return
        # itself PYTHON-344
        cx_pool = self.create_pool(max_pool_size=1, wait_queue_timeout=1)
        cx_pool._check_interval_seconds = 0  # Always check.

        with cx_pool.checkout() as conn:
            # Simulate a closed socket without telling the Connection it's
            # closed.
            conn.conn.close()
            self.assertTrue(conn.conn_closed())

        with cx_pool.checkout() as new_connection:
            self.assertEqual(0, len(cx_pool.conns))
            self.assertNotEqual(conn, new_connection)

        self.assertEqual(1, len(cx_pool.conns))

        # Semaphore was released.
        with cx_pool.checkout():
            pass

    def test_socket_closed(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)