Function test_get_queue_status

tests/api_utils/routers/test_queue.py:104–148

Test get_queue_status endpoint.

()

@pytest.mark.asyncio
async def test_get_queue_status():
    """Test get_queue_status endpoint."""
    queue = asyncio.Queue()
    lock = asyncio.Lock()

    # Add items
    mock_request = MagicMock()
    mock_request.stream = True

    item1 = {
        "req_id": "req_1",
        "enqueue_time": time.time() - 10,
        "request_data": mock_request,
        "cancelled": False,
    }
    item2 = {
        "req_id": "req_2",
        "enqueue_time": time.time() - 5,
        "request_data": mock_request,
        "cancelled": True,
    }

    await queue.put(item1)
    await queue.put(item2)

    # Lock the lock
    await lock.acquire()

    response = await get_queue_status(queue, lock)

    assert isinstance(response, JSONResponse)
    assert response.status_code == 200

    import json

    data = json.loads(bytes(response.body))

    assert data["queue_length"] == 2
    assert data["is_processing_locked"] is True
    assert len(data["items"]) == 2
    assert data["items"][0]["req_id"] == "req_1"
    assert data["items"][1]["req_id"] == "req_2"
    assert data["items"][0]["wait_time_seconds"] >= 10

    lock.release()


@pytest.mark.asyncio