Function test_ws_queue_overflow

tests/test_ws_api.py:201–224

WebSocket API should not overflow queue

(clientAsync)

@pytest.mark.skipif(sys.version_info < (3, 8), reason="websockets_proxy Python 3.8+")
@pytest.mark.asyncio
async def test_ws_queue_overflow(clientAsync):
    """WebSocket API should not overflow queue"""
    #
    original_size = clientAsync.ws_api.max_queue_size
    clientAsync.ws_api.max_queue_size = 1

    try:
        # Request multiple order books concurrently
        symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]
        tasks = [clientAsync.ws_get_order_book(symbol=symbol) for symbol in symbols]

        # Execute all requests concurrently and wait for results
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Check that we got valid responses or expected overflow errors
        valid_responses = [r for r in results if not isinstance(r, Exception)]
        assert len(valid_responses) == len(symbols), "Should get at least one valid response"

        for result in valid_responses:
            assert_ob(result)

    finally:
        # Restore original queue size
        clientAsync.ws_api.MAX_QUEUE_SIZE = original_size


@pytest.mark.skipif(sys.version_info < (3, 8), reason="websockets_proxy Python 3.8+")