From 109fe03ad13140c26c53716f7fbff9b2eece5033 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Wed, 7 Jan 2026 17:40:03 -0800 Subject: [PATCH] [smg][ci] Migrate Response API e2e tests to shared infrastructure (#16680) Co-authored-by: Chang Su Co-authored-by: key4ng --- .github/workflows/pr-test-rust.yml | 9 + sgl-model-gateway/e2e_test/backends.py | 49 +- sgl-model-gateway/e2e_test/fixtures/hooks.py | 5 + .../e2e_test/fixtures/setup_backend.py | 45 +- .../e2e_test/infra/model_pool.py | 5 + .../e2e_test/infra/model_specs.py | 10 + .../e2e_test/responses/__init__.py | 9 + .../e2e_test/responses/test_basic_crud.py | 441 +++++++++ .../responses/test_state_management.py | 349 +++++++ .../responses/test_streaming_events.py | 256 +++++ .../responses/test_structured_output.py | 289 ++++++ .../e2e_test/responses/test_tools_call.py | 902 ++++++++++++++++++ 12 files changed, 2314 insertions(+), 55 deletions(-) create mode 100644 sgl-model-gateway/e2e_test/responses/__init__.py create mode 100644 sgl-model-gateway/e2e_test/responses/test_basic_crud.py create mode 100644 sgl-model-gateway/e2e_test/responses/test_state_management.py create mode 100644 sgl-model-gateway/e2e_test/responses/test_streaming_events.py create mode 100644 sgl-model-gateway/e2e_test/responses/test_structured_output.py create mode 100644 sgl-model-gateway/e2e_test/responses/test_tools_call.py diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml index 6bbc75528..5e66f54e6 100644 --- a/.github/workflows/pr-test-rust.yml +++ b/.github/workflows/pr-test-rust.yml @@ -182,6 +182,15 @@ jobs: setup_oracle: true setup_brave: true parallel_opts: "" # Legacy tests, not yet migrated for parallel + - name: responses + timeout: 45 + test_dirs: "e2e_test/responses" + extra_deps: "" + env_vars: "SHOW_WORKER_LOGS=0 SHOW_ROUTER_LOGS=1" + reruns: "--reruns 2 --reruns-delay 5" + setup_oracle: true + setup_brave: true + parallel_opts: "" # Cloud backend tests not compatible with parallel execution - name: grpc timeout: 32 test_dirs: "e2e_test/e2e_grpc" diff --git a/sgl-model-gateway/e2e_test/backends.py b/sgl-model-gateway/e2e_test/backends.py index e3f0319dd..588c5be5e 100644 --- a/sgl-model-gateway/e2e_test/backends.py +++ b/sgl-model-gateway/e2e_test/backends.py @@ -24,7 +24,7 @@ logger = logging.getLogger(__name__) CLOUD_RUNTIMES: dict[str, dict[str, Any]] = { "openai": { "description": "OpenAI API", - "model": "gpt-4o-mini", + "model": "gpt-5-nano", "api_key_env": "OPENAI_API_KEY", }, "xai": { @@ -34,19 +34,8 @@ CLOUD_RUNTIMES: dict[str, dict[str, Any]] = { }, } -# Keep CLOUD_BACKENDS as alias for backward compatibility during migration -# TODO: Remove after e2e_response_api migration -CLOUD_BACKENDS: dict[str, dict[str, Any]] = { - **CLOUD_RUNTIMES, - # Legacy entry for tests that parameterize on history backend - "oracle_store": { - "description": "OpenAI API with Oracle history backend", - "model": "gpt-4o-mini", - "api_key_env": "OPENAI_API_KEY", - "history_backend": "oracle", - "_runtime": "openai", # Actual runtime to use - }, -} +# Backward compatibility alias +CLOUD_BACKENDS = CLOUD_RUNTIMES def get_cloud_runtime_config(runtime: str) -> dict[str, Any]: @@ -92,33 +81,7 @@ def launch_cloud_gateway( return gateway -# Backward compatibility aliases - TODO: Remove after migration -def get_cloud_backend_config(backend: str) -> dict[str, Any]: - """Deprecated: Use get_cloud_runtime_config instead.""" - if backend not in CLOUD_BACKENDS: - raise KeyError( - f"Unknown cloud backend: {backend}. Available: {list(CLOUD_BACKENDS.keys())}" - ) - return CLOUD_BACKENDS[backend] - - -def launch_cloud_backend(backend: str, **kwargs: Any) -> Gateway: - """Deprecated: Use launch_cloud_gateway instead.""" - cfg = get_cloud_backend_config(backend) - - # Handle legacy oracle_store entry - runtime = cfg.get("_runtime", backend) - history_backend = kwargs.pop( - "history_backend", cfg.get("history_backend", "memory") - ) - - return launch_cloud_gateway( - runtime, - history_backend=history_backend, - extra_args=kwargs.pop("router_args", None), - **kwargs, - ) - - -# Keep old function name as alias +# Backward compatibility aliases +get_cloud_backend_config = get_cloud_runtime_config +launch_cloud_backend = launch_cloud_gateway launch_cloud_router = launch_cloud_gateway diff --git a/sgl-model-gateway/e2e_test/fixtures/hooks.py b/sgl-model-gateway/e2e_test/fixtures/hooks.py index e22180305..a827f0c61 100644 --- a/sgl-model-gateway/e2e_test/fixtures/hooks.py +++ b/sgl-model-gateway/e2e_test/fixtures/hooks.py @@ -377,6 +377,11 @@ def pytest_configure(config: pytest.Config) -> None: "markers", "thread_unsafe: mark test as incompatible with parallel thread execution", ) + config.addinivalue_line( + "markers", + "storage(backend): mark test to use a specific history storage backend " + "(memory, oracle). Default is memory.", + ) # --------------------------------------------------------------------------- diff --git a/sgl-model-gateway/e2e_test/fixtures/setup_backend.py b/sgl-model-gateway/e2e_test/fixtures/setup_backend.py index 53bdd4ed0..2f61f1bf0 100644 --- a/sgl-model-gateway/e2e_test/fixtures/setup_backend.py +++ b/sgl-model-gateway/e2e_test/fixtures/setup_backend.py @@ -115,8 +115,11 @@ def setup_backend(request: pytest.FixtureRequest, model_pool: "ModelPool"): ) return + # Get storage backend from marker (default: memory) + storage_backend = get_marker_value(request, "storage", default="memory") + # Cloud backends: launch cloud router - yield from _setup_cloud_backend(backend_name) + yield from _setup_cloud_backend(backend_name, storage_backend, gateway_config) def _setup_pd_backend( @@ -357,34 +360,52 @@ def _setup_local_backend( inst.release() -def _setup_cloud_backend(backend_name: str): - """Setup cloud backend (openai, xai, etc.).""" +def _setup_cloud_backend( + backend_name: str, + storage_backend: str = "memory", + gateway_config: dict | None = None, +): + """Setup cloud backend (openai, xai, etc.). + + Args: + backend_name: Cloud backend name (openai, xai). + storage_backend: History storage backend (memory, oracle). + gateway_config: Gateway configuration from marker. + """ import openai - from backends import CLOUD_BACKENDS, launch_cloud_backend + from backends import CLOUD_RUNTIMES, launch_cloud_gateway - if backend_name not in CLOUD_BACKENDS: - pytest.fail(f"Unknown backend: {backend_name}") + if backend_name not in CLOUD_RUNTIMES: + pytest.fail(f"Unknown cloud runtime: {backend_name}") - cfg = CLOUD_BACKENDS[backend_name] + cfg = CLOUD_RUNTIMES[backend_name] api_key_env = cfg.get("api_key_env") if api_key_env and not os.environ.get(api_key_env): pytest.skip(f"{api_key_env} not set, skipping {backend_name} tests") - logger.info("Launching cloud backend: %s", backend_name) - router = launch_cloud_backend(backend_name) + extra_args = gateway_config.get("extra_args") if gateway_config else None + + logger.info( + "Launching cloud backend: %s with storage=%s", backend_name, storage_backend + ) + gateway = launch_cloud_gateway( + backend_name, + history_backend=storage_backend, + extra_args=extra_args, + ) api_key = os.environ.get(api_key_env) if api_key_env else "not-used" client = openai.OpenAI( - base_url=f"{router.base_url}/v1", + base_url=f"{gateway.base_url}/v1", api_key=api_key, ) try: - yield backend_name, cfg["model"], client, router + yield backend_name, cfg["model"], client, gateway finally: logger.info("Tearing down cloud backend: %s", backend_name) - router.shutdown() + gateway.shutdown() @pytest.fixture diff --git a/sgl-model-gateway/e2e_test/infra/model_pool.py b/sgl-model-gateway/e2e_test/infra/model_pool.py index 2b78c5ed9..92bcb25ed 100644 --- a/sgl-model-gateway/e2e_test/infra/model_pool.py +++ b/sgl-model-gateway/e2e_test/infra/model_pool.py @@ -494,6 +494,11 @@ class ModelPool: if ib_device: cmd.extend(["--disaggregation-ib-device", ib_device]) + # Additional worker args from model spec (e.g., --context-length) + worker_args = spec.get("worker_args", []) + if worker_args: + cmd.extend(worker_args) + # Build key based on worker type (or use custom key) if instance_key: key = instance_key diff --git a/sgl-model-gateway/e2e_test/infra/model_specs.py b/sgl-model-gateway/e2e_test/infra/model_specs.py index 090d4b468..81c31bba5 100644 --- a/sgl-model-gateway/e2e_test/infra/model_specs.py +++ b/sgl-model-gateway/e2e_test/infra/model_specs.py @@ -46,6 +46,16 @@ MODEL_SPECS: dict[str, dict] = { "tp": 1, "features": ["chat", "streaming", "function_calling", "pythonic_tools"], }, + # Function calling specialist (larger, for Response API tests) + "qwen-14b": { + "model": _resolve_model_path("Qwen/Qwen2.5-14B-Instruct"), + "memory_gb": 28, + "tp": 2, + "features": ["chat", "streaming", "function_calling", "pythonic_tools"], + "worker_args": [ + "--context-length=1000" + ], # Faster startup, prevents memory issues + }, # Reasoning model "deepseek-7b": { "model": _resolve_model_path("deepseek-ai/DeepSeek-R1-Distill-Qwen-7B"), diff --git a/sgl-model-gateway/e2e_test/responses/__init__.py b/sgl-model-gateway/e2e_test/responses/__init__.py new file mode 100644 index 000000000..e6858e2cb --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/__init__.py @@ -0,0 +1,9 @@ +"""Response API E2E tests. + +Tests for the Response API endpoints including: +- Basic CRUD operations for responses and conversations +- State management (previous_response_id, conversation-based) +- Streaming events and output validation +- Structured output (json_schema) +- Tool calling (function tools and MCP) +""" diff --git a/sgl-model-gateway/e2e_test/responses/test_basic_crud.py b/sgl-model-gateway/e2e_test/responses/test_basic_crud.py new file mode 100644 index 000000000..e40e002c0 --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/test_basic_crud.py @@ -0,0 +1,441 @@ +"""Basic CRUD tests for Response API. + +Tests for Response and Conversation CRUD operations against cloud backends. + +Source: Migrated from e2e_response_api/features/test_basic_crud.py +""" + +from __future__ import annotations + +import logging +import time + +import openai +import pytest +from openai import OpenAI +from openai.types import responses + +logger = logging.getLogger(__name__) + + +def wait_for_background_task( + client: OpenAI, response_id: str, timeout: int = 30, poll_interval: float = 0.5 +) -> responses.Response: + """Wait for background task to complete. + + Args: + client: OpenAI client + response_id: Response ID to poll + timeout: Max seconds to wait + poll_interval: Seconds between polls + + Returns: + Final response data + + Raises: + TimeoutError: If task doesn't complete in time + AssertionError: If task fails + """ + start_time = time.time() + + while time.time() - start_time < timeout: + resp = client.responses.retrieve(response_id=response_id) + assert resp.error is None + assert resp.id == response_id + + if resp.status == "completed": + return resp + elif resp.status == "failed": + raise AssertionError(f"Background task failed: {resp.error}") + elif resp.status == "cancelled": + raise AssertionError("Background task was cancelled") + + time.sleep(poll_interval) + + raise TimeoutError( + f"Background task {response_id} did not complete within {timeout}s" + ) + + +# ============================================================================= +# Response CRUD Tests (Memory Storage) +# ============================================================================= + + +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestResponseCRUD: + """Tests for Response API CRUD operations.""" + + def test_create_and_get_response(self, setup_backend): + """Test creating response and retrieving it.""" + _, model, client, gateway = setup_backend + + # Create response + create_resp = client.responses.create(model=model, input="Hello, world!") + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status == "completed" + assert len(create_resp.output_text) > 0 + response_id = create_resp.id + + # Get response + get_resp = client.responses.retrieve(response_id=response_id) + assert get_resp.error is None + assert get_resp.id == response_id + assert get_resp.status == "completed" + + input_resp = client.responses.input_items.list(response_id=get_resp.id) + assert input_resp.data is not None + assert len(input_resp.data) > 0 + + @pytest.mark.skip(reason="TODO: Add delete response feature") + def test_delete_response(self, setup_backend): + """Test deleting response.""" + _, model, client, gateway = setup_backend + + # Create response + create_resp = client.responses.create(model=model, input="Test deletion") + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status == "completed" + assert len(create_resp.output_text) > 0 + + response_id = create_resp.id + + # Delete response + client.responses.delete(response_id=response_id) + + # Verify it's deleted (should return 404) + with pytest.raises(openai.NotFoundError): + client.responses.retrieve(response_id=response_id) + + @pytest.mark.skip(reason="TODO: Add background response feature") + def test_background_response(self, setup_backend): + """Test background response execution.""" + _, model, client, gateway = setup_backend + + # Create background response + create_resp = client.responses.create( + model=model, + input="Write a short story", + background=True, + max_output_tokens=100, + ) + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status in ["in_progress", "queued"] + + response_id = create_resp.id + + # Wait for completion + final_data = wait_for_background_task(client, response_id, timeout=60) + assert final_data.status == "completed" + + +# ============================================================================= +# Response CRUD Tests (Oracle Storage) +# ============================================================================= + + +@pytest.mark.storage("oracle") +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestResponseCRUDOracleStorage: + """Tests for Response API CRUD operations with Oracle history backend.""" + + def test_create_and_get_response(self, setup_backend): + """Test creating response and retrieving it.""" + _, model, client, gateway = setup_backend + + # Create response + create_resp = client.responses.create(model=model, input="Hello, world!") + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status == "completed" + assert len(create_resp.output_text) > 0 + response_id = create_resp.id + + # Get response + get_resp = client.responses.retrieve(response_id=response_id) + assert get_resp.error is None + assert get_resp.id == response_id + assert get_resp.status == "completed" + + input_resp = client.responses.input_items.list(response_id=get_resp.id) + assert input_resp.data is not None + assert len(input_resp.data) > 0 + + @pytest.mark.skip(reason="TODO: Add delete response feature") + def test_delete_response(self, setup_backend): + """Test deleting response.""" + _, model, client, gateway = setup_backend + + # Create response + create_resp = client.responses.create(model=model, input="Test deletion") + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status == "completed" + assert len(create_resp.output_text) > 0 + + response_id = create_resp.id + + # Delete response + client.responses.delete(response_id=response_id) + + # Verify it's deleted (should return 404) + with pytest.raises(openai.NotFoundError): + client.responses.retrieve(response_id=response_id) + + @pytest.mark.skip(reason="TODO: Add background response feature") + def test_background_response(self, setup_backend): + """Test background response execution.""" + _, model, client, gateway = setup_backend + + # Create background response + create_resp = client.responses.create( + model=model, + input="Write a short story", + background=True, + max_output_tokens=100, + ) + assert create_resp.id is not None + assert create_resp.error is None + assert create_resp.status in ["in_progress", "queued"] + + response_id = create_resp.id + + # Wait for completion + final_data = wait_for_background_task(client, response_id, timeout=60) + assert final_data.status == "completed" + + +# ============================================================================= +# Conversation CRUD Tests (Memory Storage) +# ============================================================================= + + +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestConversationCRUD: + """Tests for Conversation API CRUD operations.""" + + def test_create_and_get_conversation(self, setup_backend): + """Test creating and retrieving conversation.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create(metadata={"user": "test_user"}) + assert create_resp.id is not None + assert create_resp.created_at is not None + + create_data = create_resp.metadata + assert create_data["user"] == "test_user" + conversation_id = create_resp.id + + # Get conversation + get_resp = client.conversations.retrieve(conversation_id=conversation_id) + assert get_resp.id is not None + assert get_resp.created_at is not None + + get_data = get_resp.metadata + assert get_resp.id == conversation_id + assert get_data["user"] == "test_user" + + def test_update_conversation(self, setup_backend): + """Test updating conversation metadata.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create(metadata={"key1": "value1"}) + assert create_resp.id is not None + assert create_resp.created_at is not None + + create_data = create_resp.metadata + assert create_data["key1"] == "value1" + assert "key2" not in create_data + conversation_id = create_resp.id + + # Update conversation + update_resp = client.conversations.update( + conversation_id=conversation_id, + metadata={"key1": "value1", "key2": "value2"}, + ) + assert update_resp.id == conversation_id + update_data = update_resp.metadata + assert update_data["key1"] == "value1" + assert update_data["key2"] == "value2" + + # Verify update + get_resp = client.conversations.retrieve(conversation_id=conversation_id) + get_data = get_resp.metadata + assert get_data["key1"] == "value1" + assert get_data["key2"] == "value2" + + def test_delete_conversation(self, setup_backend): + """Test deleting conversation.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create() + assert create_resp.id is not None + assert create_resp.created_at is not None + conversation_id = create_resp.id + + # Delete conversation + delete_resp = client.conversations.delete(conversation_id=conversation_id) + assert delete_resp.id is not None + assert delete_resp.deleted + + # Verify deletion + with pytest.raises(openai.NotFoundError): + client.conversations.retrieve(conversation_id=conversation_id) + + def test_list_conversation_items(self, setup_backend): + """Test listing conversation items.""" + _, model, client, gateway = setup_backend + + # Create conversation + conv_resp = client.conversations.create() + assert conv_resp.id is not None + conversation_id = conv_resp.id + + # Create response with conversation + resp1 = client.responses.create( + model=model, + input="First message", + conversation=conversation_id, + max_output_tokens=50, + ) + assert resp1.error is None + + resp2 = client.responses.create( + model=model, + input="Second message", + conversation=conversation_id, + max_output_tokens=50, + ) + assert resp2.error is None + + # List items + list_resp = client.conversations.items.list(conversation_id=conversation_id) + assert list_resp is not None + assert list_resp.data is not None + + list_data = list_resp.data + # Should have at least 4 items (2 inputs + 2 outputs) + assert len(list_data) >= 4 + + +# ============================================================================= +# Conversation CRUD Tests (Oracle Storage) +# ============================================================================= + + +@pytest.mark.storage("oracle") +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestConversationCRUDOracleStorage: + """Tests for Conversation API CRUD operations with Oracle history backend.""" + + def test_create_and_get_conversation(self, setup_backend): + """Test creating and retrieving conversation.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create(metadata={"user": "test_user"}) + assert create_resp.id is not None + assert create_resp.created_at is not None + + create_data = create_resp.metadata + assert create_data["user"] == "test_user" + conversation_id = create_resp.id + + # Get conversation + get_resp = client.conversations.retrieve(conversation_id=conversation_id) + assert get_resp.id is not None + assert get_resp.created_at is not None + + get_data = get_resp.metadata + assert get_resp.id == conversation_id + assert get_data["user"] == "test_user" + + def test_update_conversation(self, setup_backend): + """Test updating conversation metadata.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create(metadata={"key1": "value1"}) + assert create_resp.id is not None + assert create_resp.created_at is not None + + create_data = create_resp.metadata + assert create_data["key1"] == "value1" + assert "key2" not in create_data + conversation_id = create_resp.id + + # Update conversation + update_resp = client.conversations.update( + conversation_id=conversation_id, + metadata={"key1": "value1", "key2": "value2"}, + ) + assert update_resp.id == conversation_id + update_data = update_resp.metadata + assert update_data["key1"] == "value1" + assert update_data["key2"] == "value2" + + # Verify update + get_resp = client.conversations.retrieve(conversation_id=conversation_id) + get_data = get_resp.metadata + assert get_data["key1"] == "value1" + assert get_data["key2"] == "value2" + + def test_delete_conversation(self, setup_backend): + """Test deleting conversation.""" + _, model, client, gateway = setup_backend + + # Create conversation + create_resp = client.conversations.create() + assert create_resp.id is not None + assert create_resp.created_at is not None + conversation_id = create_resp.id + + # Delete conversation + delete_resp = client.conversations.delete(conversation_id=conversation_id) + assert delete_resp.id is not None + assert delete_resp.deleted + + # Verify deletion + with pytest.raises(openai.NotFoundError): + client.conversations.retrieve(conversation_id=conversation_id) + + def test_list_conversation_items(self, setup_backend): + """Test listing conversation items.""" + _, model, client, gateway = setup_backend + + # Create conversation + conv_resp = client.conversations.create() + assert conv_resp.id is not None + conversation_id = conv_resp.id + + # Create response with conversation + resp1 = client.responses.create( + model=model, + input="First message", + conversation=conversation_id, + max_output_tokens=50, + ) + assert resp1.error is None + + resp2 = client.responses.create( + model=model, + input="Second message", + conversation=conversation_id, + max_output_tokens=50, + ) + assert resp2.error is None + + # List items + list_resp = client.conversations.items.list(conversation_id=conversation_id) + assert list_resp is not None + assert list_resp.data is not None + + list_data = list_resp.data + # Should have at least 4 items (2 inputs + 2 outputs) + assert len(list_data) >= 4 diff --git a/sgl-model-gateway/e2e_test/responses/test_state_management.py b/sgl-model-gateway/e2e_test/responses/test_state_management.py new file mode 100644 index 000000000..b21e41ee8 --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/test_state_management.py @@ -0,0 +1,349 @@ +"""State management tests for Response API. + +Tests both previous_response_id and conversation-based state management. +These tests work across local (gRPC) and cloud (OpenAI, xAI) backends. + +Source: Migrated from e2e_response_api/features/test_state_management.py +""" + +from __future__ import annotations + +import logging + +import openai +import pytest + +logger = logging.getLogger(__name__) + + +# ============================================================================= +# Cloud Backend Tests (OpenAI, xAI) +# ============================================================================= + + +@pytest.mark.parametrize("setup_backend", ["openai", "xai"], indirect=True) +class TestStateManagementCloud: + """State management tests against cloud APIs.""" + + def test_basic_response_creation(self, setup_backend): + """Test basic response creation without state.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create(model=model, input="What is 2+2?") + + assert resp.id is not None + assert resp.error is None + assert resp.status == "completed" + assert len(resp.output_text) > 0 + assert resp.usage is not None + + def test_streaming_response(self, setup_backend): + """Test streaming response.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, input="Count to 5", stream=True, max_output_tokens=50 + ) + + events = list(resp) + created_events = [e for e in events if e.type == "response.created"] + assert len(created_events) > 0 + + assert any( + e.type in ["response.completed", "response.in_progress"] for e in events + ) + + def test_previous_response_id_chaining(self, setup_backend): + """Test chaining responses using previous_response_id.""" + _, model, client, gateway = setup_backend + + # First response + resp1 = client.responses.create( + model=model, input="My name is Alice and my friend is Bob. Remember it." + ) + assert resp1.error is None + assert resp1.status == "completed" + + # Second response referencing first + resp2 = client.responses.create( + model=model, input="What is my name", previous_response_id=resp1.id + ) + assert resp2.error is None + assert resp2.status == "completed" + assert "Alice" in resp2.output_text + + # Third response referencing second + resp3 = client.responses.create( + model=model, + input="What is my friend name?", + previous_response_id=resp2.id, + ) + assert resp3.error is None + assert resp3.status == "completed" + assert "Bob" in resp3.output_text + + def test_conversation_with_multiple_turns(self, setup_backend): + """Test state management using conversation ID.""" + _, model, client, gateway = setup_backend + + # Create conversation + conv_resp = client.conversations.create(metadata={"topic": "math"}) + assert conv_resp.id is not None + assert conv_resp.created_at is not None + conversation_id = conv_resp.id + + # First response in conversation + resp1 = client.responses.create( + model=model, input="I have 5 apples.", conversation=conversation_id + ) + assert resp1.error is None + assert resp1.status == "completed" + + # Second response in same conversation + resp2 = client.responses.create( + model=model, + input="How many apples do I have?", + conversation=conversation_id, + ) + assert resp2.error is None + assert resp2.status == "completed" + assert "5" in resp2.output_text or "five" in resp2.output_text.lower() + + # Third response in same conversation + resp3 = client.responses.create( + model=model, + input="If I get 3 more, how many total?", + conversation=conversation_id, + ) + assert resp3.error is None + assert resp3.status == "completed" + assert "8" in resp3.output_text or "eight" in resp3.output_text.lower() + + items = client.conversations.items.list(conversation_id) + assert items.data is not None + assert len(items.data) >= 6 # 3 inputs + 3 outputs + + @pytest.mark.skip(reason="TODO: Add the invalid previous_response_id check") + def test_previous_response_id_invalid(self, setup_backend): + """Test using invalid previous_response_id.""" + _, model, client, gateway = setup_backend + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="Test", + previous_response_id="resp_invalid123", + max_output_tokens=50, + ) + + def test_mutually_exclusive_parameters(self, setup_backend): + """Test that previous_response_id and conversation are mutually exclusive.""" + _, model, client, gateway = setup_backend + + conversation_id = "conv_123" + resp1 = client.responses.create(model=model, input="Test") + + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="This should fail", + previous_response_id=resp1.id, + conversation=conversation_id, + ) + + +# ============================================================================= +# Local Backend Tests (gRPC with Qwen model) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("qwen-14b") +@pytest.mark.gateway( + extra_args=["--tool-call-parser", "qwen", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestStateManagementLocal: + """State management tests against local gRPC backend.""" + + @pytest.mark.skip(reason="TODO: Add the invalid previous_response_id check") + def test_previous_response_id_invalid(self, setup_backend): + """Test using invalid previous_response_id.""" + _, model, client, gateway = setup_backend + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="Test", + previous_response_id="resp_invalid123", + max_output_tokens=50, + ) + + def test_basic_response_creation(self, setup_backend): + """Test basic response creation without state.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create(model=model, input="What is 2+2?") + + assert resp.id is not None + assert resp.error is None + assert resp.status == "completed" + assert len(resp.output_text) > 0 + assert resp.usage is not None + + def test_streaming_response(self, setup_backend): + """Test streaming response.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, input="Count to 5", stream=True, max_output_tokens=50 + ) + + events = list(resp) + created_events = [e for e in events if e.type == "response.created"] + assert len(created_events) > 0 + + assert any( + e.type in ["response.completed", "response.in_progress"] for e in events + ) + + def test_previous_response_id_chaining(self, setup_backend): + """Test chaining responses using previous_response_id.""" + _, model, client, gateway = setup_backend + + # First response + resp1 = client.responses.create( + model=model, input="My name is Alice and my friend is Bob. Remember it." + ) + assert resp1.error is None + assert resp1.status == "completed" + + # Second response referencing first + resp2 = client.responses.create( + model=model, input="What is my name", previous_response_id=resp1.id + ) + assert resp2.error is None + assert resp2.status == "completed" + assert "Alice" in resp2.output_text + + # Third response referencing second + resp3 = client.responses.create( + model=model, + input="What is my friend name?", + previous_response_id=resp2.id, + ) + assert resp3.error is None + assert resp3.status == "completed" + assert "Bob" in resp3.output_text + + def test_mutually_exclusive_parameters(self, setup_backend): + """Test that previous_response_id and conversation are mutually exclusive.""" + _, model, client, gateway = setup_backend + + conversation_id = "conv_123" + resp1 = client.responses.create(model=model, input="Test") + + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="This should fail", + previous_response_id=resp1.id, + conversation=conversation_id, + ) + + +# ============================================================================= +# Local Backend Tests (gRPC with Harmony/Reasoning model) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("gpt-oss") +@pytest.mark.gateway( + extra_args=["--reasoning-parser=gpt-oss", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestStateManagementHarmony: + """State management tests against local gRPC backend with Harmony model.""" + + @pytest.mark.skip(reason="TODO: Add the invalid previous_response_id check") + def test_previous_response_id_invalid(self, setup_backend): + """Test using invalid previous_response_id.""" + _, model, client, gateway = setup_backend + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="Test", + previous_response_id="resp_invalid123", + max_output_tokens=50, + ) + + def test_basic_response_creation(self, setup_backend): + """Test basic response creation without state.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create(model=model, input="What is 2+2?") + + assert resp.id is not None + assert resp.error is None + assert resp.status == "completed" + assert len(resp.output_text) > 0 + assert resp.usage is not None + + def test_streaming_response(self, setup_backend): + """Test streaming response.""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, input="Count to 5", stream=True, max_output_tokens=50 + ) + + events = list(resp) + created_events = [e for e in events if e.type == "response.created"] + assert len(created_events) > 0 + + assert any( + e.type in ["response.completed", "response.in_progress"] for e in events + ) + + def test_previous_response_id_chaining(self, setup_backend): + """Test chaining responses using previous_response_id.""" + _, model, client, gateway = setup_backend + + # First response + resp1 = client.responses.create( + model=model, input="My name is Alice and my friend is Bob. Remember it." + ) + assert resp1.error is None + assert resp1.status == "completed" + + # Second response referencing first + resp2 = client.responses.create( + model=model, input="What is my name", previous_response_id=resp1.id + ) + assert resp2.error is None + assert resp2.status == "completed" + assert "Alice" in resp2.output_text + + # Third response referencing second + resp3 = client.responses.create( + model=model, + input="What is my friend name?", + previous_response_id=resp2.id, + ) + assert resp3.error is None + assert resp3.status == "completed" + assert "Bob" in resp3.output_text + + def test_mutually_exclusive_parameters(self, setup_backend): + """Test that previous_response_id and conversation are mutually exclusive.""" + _, model, client, gateway = setup_backend + + conversation_id = "conv_123" + resp1 = client.responses.create(model=model, input="Test") + + with pytest.raises(openai.BadRequestError): + client.responses.create( + model=model, + input="This should fail", + previous_response_id=resp1.id, + conversation=conversation_id, + ) diff --git a/sgl-model-gateway/e2e_test/responses/test_streaming_events.py b/sgl-model-gateway/e2e_test/responses/test_streaming_events.py new file mode 100644 index 000000000..4aaaf0b0a --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/test_streaming_events.py @@ -0,0 +1,256 @@ +"""Streaming events tests for Response API. + +Tests for streaming event validation including: +- Zero-based output_index for content +- OutputItemDone event emission and output array construction +- Reasoning content with proper output_index + +Source: Migrated from e2e_response_api/features/test_streaming_events.py +""" + +from __future__ import annotations + +import logging + +import pytest + +logger = logging.getLogger(__name__) + + +# ============================================================================= +# Local Backend Tests (gRPC with Qwen model) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("qwen-14b") +@pytest.mark.gateway( + extra_args=["--tool-call-parser", "qwen", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestStreamingEventsLocal: + """Streaming event tests against local gRPC backend.""" + + def test_output_item_event_emitted(self, setup_backend): + """Test that output_index is zero-based in streaming responses. + + Verifies that the first output item has output_index: 0. + """ + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Count from 1 to 3", + stream=True, + max_output_tokens=50, + ) + + events = list(resp) + assert len(events) > 0 + + # Find output_item.added events + output_item_added_events = [ + event for event in events if event.type == "response.output_item.added" + ] + assert len(output_item_added_events) > 0, "Should have output_item.added events" + + # Verify first output item has output_index: 0 + first_item_event = output_item_added_events[0] + assert first_item_event.item is not None + assert first_item_event.output_index is not None + assert ( + first_item_event.output_index == 0 + ), "First output item must have output_index: 0 (zero-based indexing)" + + # Verify subsequent items increment correctly + for i, event in enumerate(output_item_added_events): + assert ( + event.output_index == i + ), f"Output item {i} should have output_index: {i}" + + # Verify output_item.done event exists + output_item_done_events = [ + event for event in events if event.type == "response.output_item.done" + ] + assert len(output_item_done_events) > 0 + + # Verify output_item.done event structure + for event in output_item_done_events: + assert event.item is not None + assert event.output_index is not None + assert event.item.type is not None + + # Find response.completed event + completed_events = [ + event for event in events if event.type == "response.completed" + ] + assert len(completed_events) == 1, "Should have exactly one completed event" + + # Verify output array exists and contains items + completed_event = completed_events[0] + assert completed_event.response.output is not None + output_array = completed_event.response.output + assert isinstance(output_array, list) + assert len(output_array) > 0, "Output array should contain at least one item" + + # Verify each item in output array has proper structure + for item in output_array: + assert item.type is not None + + # Verify output_item.added events match items in final output array + output_item_added_events = [ + event for event in events if event.type == "response.output_item.added" + ] + assert len(output_item_added_events) == len( + output_array + ), "Number of output_item.added events should match output array length" + + +# ============================================================================= +# Local Backend Tests (gRPC with Harmony/Reasoning model) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("gpt-oss") +@pytest.mark.gateway( + extra_args=["--reasoning-parser=gpt-oss", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestStreamingEventsHarmony: + """Streaming event tests against local gRPC backend with Harmony model.""" + + def test_output_item_event_emitted(self, setup_backend): + """Test that output_index is zero-based in streaming responses. + + Verifies that the first output item has output_index: 0. + """ + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Count from 1 to 3", + stream=True, + max_output_tokens=50, + ) + + events = list(resp) + assert len(events) > 0 + + # Find output_item.added events + output_item_added_events = [ + event for event in events if event.type == "response.output_item.added" + ] + assert len(output_item_added_events) > 0, "Should have output_item.added events" + + # Verify first output item has output_index: 0 + first_item_event = output_item_added_events[0] + assert first_item_event.item is not None + assert first_item_event.output_index is not None + assert ( + first_item_event.output_index == 0 + ), "First output item must have output_index: 0 (zero-based indexing)" + + # Verify subsequent items increment correctly + for i, event in enumerate(output_item_added_events): + assert ( + event.output_index == i + ), f"Output item {i} should have output_index: {i}" + + # Verify output_item.done event exists + output_item_done_events = [ + event for event in events if event.type == "response.output_item.done" + ] + assert len(output_item_done_events) > 0 + + # Verify output_item.done event structure + for event in output_item_done_events: + assert event.item is not None + assert event.output_index is not None + assert event.item.type is not None + + # Find response.completed event + completed_events = [ + event for event in events if event.type == "response.completed" + ] + assert len(completed_events) == 1, "Should have exactly one completed event" + + # Verify output array exists and contains items + completed_event = completed_events[0] + assert completed_event.response.output is not None + output_array = completed_event.response.output + assert isinstance(output_array, list) + assert len(output_array) > 0, "Output array should contain at least one item" + + # Verify each item in output array has proper structure + for item in output_array: + assert item.type is not None + + # Verify output_item.added events match items in final output array + output_item_added_events = [ + event for event in events if event.type == "response.output_item.added" + ] + assert len(output_item_added_events) == len( + output_array + ), "Number of output_item.added events should match output array length" + + def test_reasoning_content(self, setup_backend): + """Test that reasoning content has correct zero-based output_index. + + Specifically tests that reasoning item has output_index: 0 + and message item has output_index: 1. + """ + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="What is the capital of France? Think step by step.", + stream=True, + max_output_tokens=200, + ) + + events = list(resp) + assert len(events) > 0 + + # Find output_item.added events + output_item_added_events = [ + event for event in events if event.type == "response.output_item.added" + ] + assert len(output_item_added_events) > 0 + + reasoning_items = [ + item for item in output_item_added_events if item.item.type == "reasoning" + ] + message_items = [ + item for item in output_item_added_events if item.item.type == "message" + ] + + # If reasoning is present, verify it has output_index: 0 + if reasoning_items: + reasoning_item = reasoning_items[0] + assert ( + reasoning_item.output_index == 0 + ), "Reasoning item should have output_index: 0" + + # If message is present after reasoning, verify it has output_index: 1 + if reasoning_items and message_items: + message_item = message_items[0] + assert ( + message_item.output_index == 1 + ), "Message item after reasoning should have output_index: 1" + + # Find response.completed event + completed_events = [ + event for event in events if event.type == "response.completed" + ] + assert len(completed_events) == 1 + + # Get output array from completed event + output_array = completed_events[0].response.output + assert len(output_array) > 0 + + # Check if reasoning items are in output array + reasoning_items_in_output = [ + item for item in output_array if item.type == "reasoning" + ] + assert len(reasoning_items_in_output) > 0 diff --git a/sgl-model-gateway/e2e_test/responses/test_structured_output.py b/sgl-model-gateway/e2e_test/responses/test_structured_output.py new file mode 100644 index 000000000..5cc8e7a14 --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/test_structured_output.py @@ -0,0 +1,289 @@ +"""Structured output tests for Response API. + +Tests for text.format field with json_object and json_schema formats. + +Source: Migrated from e2e_response_api/features/test_structured_output.py +""" + +from __future__ import annotations + +import json +import logging + +import pytest + +logger = logging.getLogger(__name__) + + +# ============================================================================= +# Cloud Backend Tests (OpenAI) +# ============================================================================= + + +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestStructuredOutputCloud: + """Structured output tests against cloud APIs.""" + + def test_structured_output_json_schema(self, setup_backend): + """Test structured output with json_schema format.""" + _, model, client, gateway = setup_backend + + params = { + "model": model, + "input": [ + { + "role": "system", + "content": "You are a helpful math tutor. Guide the user through the solution step by step.", + }, + {"role": "user", "content": "how can I solve 8x + 7 = -23"}, + ], + "text": { + "format": { + "type": "json_schema", + "name": "math_reasoning", + "schema": { + "type": "object", + "properties": { + "steps": { + "type": "array", + "items": { + "type": "object", + "properties": { + "explanation": {"type": "string"}, + "output": {"type": "string"}, + }, + "required": ["explanation", "output"], + "additionalProperties": False, + }, + }, + "final_answer": {"type": "string"}, + }, + "required": ["steps", "final_answer"], + "additionalProperties": False, + }, + "strict": True, + } + }, + } + + create_resp = client.responses.create(**params) + assert create_resp.error is None + assert create_resp.id is not None + assert create_resp.output is not None + assert create_resp.text is not None + + # Verify text format was echoed back correctly + assert create_resp.text.format is not None + assert create_resp.text.format.type == "json_schema" + assert create_resp.text.format.name == "math_reasoning" + assert create_resp.text.format.schema_ is not None + assert create_resp.text.format.strict + + # Find the message output + output_text = next( + ( + content.text + for item in create_resp.output + if item.type == "message" + for content in item.content + if content.type == "output_text" + ), + None, + ) + + assert output_text is not None, "No output_text found in response" + assert output_text.strip(), "output_text is empty" + + # Parse JSON output + output_json = json.loads(output_text) + + # Verify schema structure + assert "steps" in output_json + assert "final_answer" in output_json + assert isinstance(output_json["steps"], list) + assert len(output_json["steps"]) > 0 + + # Verify each step has required fields + for step in output_json["steps"]: + assert "explanation" in step + assert "output" in step + + +# ============================================================================= +# Local Backend Tests (gRPC with Harmony model - complex schema) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("gpt-oss") +@pytest.mark.gateway( + extra_args=["--reasoning-parser=gpt-oss", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestStructuredOutputHarmony: + """Structured output tests against local gRPC backend with Harmony model.""" + + def test_structured_output_json_schema(self, setup_backend): + """Test structured output with json_schema format.""" + _, model, client, gateway = setup_backend + + params = { + "model": model, + "input": [ + { + "role": "system", + "content": "You are a helpful math tutor. Guide the user through the solution step by step.", + }, + {"role": "user", "content": "how can I solve 8x + 7 = -23"}, + ], + "text": { + "format": { + "type": "json_schema", + "name": "math_reasoning", + "schema": { + "type": "object", + "properties": { + "steps": { + "type": "array", + "items": { + "type": "object", + "properties": { + "explanation": {"type": "string"}, + "output": {"type": "string"}, + }, + "required": ["explanation", "output"], + "additionalProperties": False, + }, + }, + "final_answer": {"type": "string"}, + }, + "required": ["steps", "final_answer"], + "additionalProperties": False, + }, + "strict": True, + } + }, + } + + create_resp = client.responses.create(**params) + assert create_resp.error is None + assert create_resp.id is not None + assert create_resp.output is not None + assert create_resp.text is not None + + # Verify text format was echoed back correctly + assert create_resp.text.format is not None + assert create_resp.text.format.type == "json_schema" + assert create_resp.text.format.name == "math_reasoning" + assert create_resp.text.format.schema_ is not None + assert create_resp.text.format.strict + + # Find the message output (output[0] may be reasoning, output[1] is message) + output_text = next( + ( + content.text + for item in create_resp.output + if item.type == "message" + for content in item.content + if content.type == "output_text" + ), + None, + ) + + assert output_text is not None, "No output_text found in response" + assert output_text.strip(), "output_text is empty" + + # Parse JSON output + output_json = json.loads(output_text) + + # Verify schema structure + assert "steps" in output_json + assert "final_answer" in output_json + assert isinstance(output_json["steps"], list) + assert len(output_json["steps"]) > 0 + + # Verify each step has required fields + for step in output_json["steps"]: + assert "explanation" in step + assert "output" in step + + +# ============================================================================= +# Local Backend Tests (gRPC with Qwen model - simple schema) +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("qwen-14b") +@pytest.mark.gateway( + extra_args=["--tool-call-parser", "qwen", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestSimpleSchemaStructuredOutput: + """Structured output tests with simpler schema for models that don't + handle complex schemas well. + """ + + def test_structured_output_json_schema(self, setup_backend): + """Test structured output with simple json_schema format.""" + _, model, client, gateway = setup_backend + + params = { + "model": model, + "input": [ + { + "role": "system", + "content": "You are a math solver. Return ONLY a JSON object that matches the schema-no extra text.", + }, + { + "role": "user", + "content": "What is 1 + 1?", + }, + ], + "text": { + "format": { + "type": "json_schema", + "name": "math_answer", + "schema": { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + }, + } + }, + } + + create_resp = client.responses.create(**params) + assert create_resp.error is None + assert create_resp.id is not None + assert create_resp.output is not None + assert create_resp.text is not None + + # Verify text format was echoed back correctly + assert create_resp.text.format is not None + assert create_resp.text.format.type == "json_schema" + assert create_resp.text.format.name == "math_answer" + assert create_resp.text.format.schema_ is not None + + # Find the message output + output_text = next( + ( + content.text + for item in create_resp.output + if item.type == "message" + for content in item.content + if content.type == "output_text" + ), + None, + ) + + assert output_text is not None, "No output_text found in response" + assert output_text.strip(), "output_text is empty" + + # Parse JSON output + output_json = json.loads(output_text) + + # Verify simple schema structure (just answer field) + assert "answer" in output_json + assert isinstance(output_json["answer"], str) + assert output_json["answer"], "Answer is empty" diff --git a/sgl-model-gateway/e2e_test/responses/test_tools_call.py b/sgl-model-gateway/e2e_test/responses/test_tools_call.py new file mode 100644 index 000000000..d025fb52a --- /dev/null +++ b/sgl-model-gateway/e2e_test/responses/test_tools_call.py @@ -0,0 +1,902 @@ +"""Tool calling tests for Response API. + +Tests for function calling functionality, tool choices and MCP calling +functionality across different backends. + +Source: Migrated from e2e_response_api/features/test_tools_call.py +""" + +from __future__ import annotations + +import json +import logging +import time + +import pytest + +logger = logging.getLogger(__name__) + + +# ============================================================================= +# Shared Tool Definitions +# ============================================================================= + + +SYSTEM_DIAGNOSTICS_FUNCTION = { + "type": "function", + "name": "get_system_diagnostics", + "description": "Retrieve real-time diagnostics for a spacecraft system.", + "parameters": { + "type": "object", + "properties": { + "system_name": { + "type": "string", + "description": "Name of the spacecraft system to query. " + "Example: 'Astra-7 Core Reactor'.", + } + }, + "required": ["system_name"], + }, +} + +GET_WEATHER_FUNCTION = { + "type": "function", + "name": "get_weather", + "description": "Get the current weather in a given location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city name, e.g., San Francisco", + } + }, + "required": ["location"], + }, +} + +CALCULATE_FUNCTION = { + "type": "function", + "name": "calculate", + "description": "Perform a mathematical calculation", + "parameters": { + "type": "object", + "properties": { + "expression": { + "type": "string", + "description": "The mathematical expression to evaluate", + } + }, + "required": ["expression"], + }, +} + +SEARCH_WEB_FUNCTION = { + "type": "function", + "name": "search_web", + "description": "Search the web for information", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, +} + +LOCAL_SEARCH_FUNCTION = { + "type": "function", + "name": "local_search", + "description": "Search local database", + "parameters": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, +} + +GET_HOROSCOPE_FUNCTION = { + "type": "function", + "name": "get_horoscope", + "description": "Get today's horoscope for an astrological sign.", + "parameters": { + "type": "object", + "properties": { + "sign": { + "type": "string", + "description": "An astrological sign like Taurus or Aquarius", + }, + }, + "required": ["sign"], + }, +} + +BRAVE_MCP_TOOL = { + "type": "mcp", + "server_label": "brave", + "server_description": "A Tool to do web search", + "server_url": "http://localhost:8001/sse", + "require_approval": "never", +} + +DEEPWIKI_MCP_TOOL = { + "type": "mcp", + "server_label": "deepwiki", + "server_url": "https://mcp.deepwiki.com/mcp", + "require_approval": "never", +} + +MCP_TEST_PROMPT = ( + "show me some news about sglang router, use the tool to just search " + "one result and return one sentence response" +) + + +# ============================================================================= +# Cloud Backend Tests (OpenAI) - Basic Function Calling +# ============================================================================= + + +@pytest.mark.parametrize("setup_backend", ["openai"], indirect=True) +class TestToolCallingCloud: + """Tool calling tests against cloud APIs.""" + + def test_basic_function_call(self, setup_backend): + """Test basic function calling workflow.""" + _, model, client, gateway = setup_backend + + tools = [GET_HOROSCOPE_FUNCTION] + system_prompt = ( + "You are a helpful assistant that can call functions. " + "When a user asks for horoscope information, call the function. " + "IMPORTANT: Don't reply directly to the user, only call the function. " + ) + + input_list = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": "What is my horoscope? I am an Aquarius."}, + ] + + resp = client.responses.create(model=model, input=input_list, tools=tools) + + assert resp.error is None + assert resp.id is not None + assert resp.status == "completed" + assert resp.output is not None + + output = resp.output + assert isinstance(output, list) + assert len(output) > 0 + + # Check for function_call in output + function_calls = [item for item in output if item.type == "function_call"] + assert ( + len(function_calls) > 0 + ), "Response should contain at least one function_call" + + # Verify function_call structure + function_call = function_calls[0] + assert function_call.call_id is not None + assert function_call.name == "get_horoscope" + assert function_call.arguments is not None + + # Parse arguments + args = json.loads(function_call.arguments) + assert "sign" in args + assert args["sign"].lower() == "aquarius" + + # Provide function call output + input_list.append(function_call) + horoscope = f"{args['sign']}: Next Tuesday you will befriend a baby otter." + input_list.append( + { + "type": "function_call_output", + "call_id": function_call.call_id, + "output": json.dumps({"horoscope": horoscope}), + } + ) + + # Second request with function output + resp2 = client.responses.create( + model=model, + input=input_list, + instructions="Respond only with a horoscope generated by a tool.", + tools=tools, + ) + assert resp2.error is None + assert resp2.status == "completed" + + output2 = resp2.output + assert len(output2) > 0 + + messages = [item for item in output2 if item.type == "message"] + assert len(messages) > 0 + + message = messages[0] + assert message.content is not None + text_parts = [ + part.text for part in message.content if part.type == "output_text" + ] + full_text = " ".join(text_parts).lower() + assert "baby otter" in full_text or "aquarius" in full_text + + def test_mcp_basic_tool_call(self, setup_backend): + """Test basic MCP tool call (non-streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) # Avoid rate limiting + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=False, + reasoning={"effort": "low"}, + ) + + assert resp.error is None + assert resp.id is not None + assert resp.status == "completed" + assert resp.model is not None + assert resp.output is not None + assert len(resp.output_text) > 0 + + output_types = [item.type for item in resp.output] + assert "mcp_list_tools" in output_types + + mcp_calls = [item for item in resp.output if item.type == "mcp_call"] + assert len(mcp_calls) > 0 + + for mcp_call in mcp_calls: + assert mcp_call.id is not None + assert mcp_call.error is None + assert mcp_call.status == "completed" + assert mcp_call.server_label == "brave" + assert mcp_call.name is not None + assert mcp_call.arguments is not None + assert mcp_call.output is not None + + # Strict validation for cloud backends + messages = [item for item in resp.output if item.type == "message"] + assert len(messages) > 0, "Response should contain at least one message" + for msg in messages: + assert msg.content is not None + assert isinstance(msg.content, list) + for content_item in msg.content: + if content_item.type == "output_text": + assert content_item.text is not None + assert isinstance(content_item.text, str) + assert len(content_item.text) > 0 + + def test_mcp_basic_tool_call_streaming(self, setup_backend): + """Test basic MCP tool call (streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) # Avoid rate limiting + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=True, + reasoning={"effort": "low"}, + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [event.type for event in events] + assert "response.created" in event_types, "Should have response.created event" + assert ( + "response.completed" in event_types + ), "Should have response.completed event" + assert ( + "response.output_item.added" in event_types + ), "Should have output_item.added events" + assert ( + "response.mcp_list_tools.in_progress" in event_types + ), "Should have mcp_list_tools.in_progress event" + assert ( + "response.mcp_list_tools.completed" in event_types + ), "Should have mcp_list_tools.completed event" + assert ( + "response.mcp_call.in_progress" in event_types + ), "Should have mcp_call.in_progress event" + assert ( + "response.mcp_call_arguments.delta" in event_types + ), "Should have mcp_call_arguments.delta event" + assert ( + "response.mcp_call_arguments.done" in event_types + ), "Should have mcp_call_arguments.done event" + assert ( + "response.mcp_call.completed" in event_types + ), "Should have mcp_call.completed event" + + completed_events = [e for e in events if e.type == "response.completed"] + assert len(completed_events) == 1 + + final_response = completed_events[0].response + assert final_response.id is not None + assert final_response.status == "completed" + assert final_response.output is not None + + final_output = final_response.output + final_output_types = [item.type for item in final_output] + assert "mcp_list_tools" in final_output_types + assert "mcp_call" in final_output_types + + # Verify mcp_call items in final output + mcp_calls = [item for item in final_output if item.type == "mcp_call"] + assert len(mcp_calls) > 0 + + for mcp_call in mcp_calls: + assert mcp_call.error is None + assert mcp_call.status == "completed" + assert mcp_call.server_label == "brave" + assert mcp_call.name is not None + assert mcp_call.arguments is not None + assert mcp_call.output is not None + + # Strict validation for cloud backends - check for text output events + assert ( + "response.content_part.added" in event_types + ), "Should have content_part.added event" + assert ( + "response.output_text.delta" in event_types + ), "Should have output_text.delta events" + assert ( + "response.output_text.done" in event_types + ), "Should have output_text.done event" + assert ( + "response.content_part.done" in event_types + ), "Should have content_part.done event" + + assert "message" in final_output_types + + # Verify text deltas combine to final message + text_deltas = [ + e.delta for e in events if e.type == "response.output_text.delta" + ] + assert len(text_deltas) > 0, "Should have text deltas" + + # Get final text from output_text.done event + text_done_events = [e for e in events if e.type == "response.output_text.done"] + assert len(text_done_events) > 0 + + final_text = text_done_events[0].text + assert len(final_text) > 0, "Final text should not be empty" + + +# ============================================================================= +# Local Backend Tests (gRPC with Harmony model) - Tool Choice +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("gpt-oss") +@pytest.mark.gateway( + extra_args=["--reasoning-parser=gpt-oss", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestToolChoiceHarmony: + """Tool choice tests against local gRPC backend with Harmony model.""" + + def test_tool_choice_auto(self, setup_backend): + """Test tool_choice="auto" allows model to decide whether to use tools.""" + _, model, client, gateway = setup_backend + + tools = [GET_WEATHER_FUNCTION] + + resp = client.responses.create( + model=model, + input="What is the weather in Seattle?", + tools=tools, + tool_choice="auto", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + assert len(output) > 0 + + function_calls = [item for item in output if item.type == "function_call"] + assert ( + len(function_calls) > 0 + ), "Model should choose to call function with tool_choice='auto'" + + def test_tool_choice_required(self, setup_backend): + """Test tool_choice="required" forces the model to call at least one tool.""" + _, model, client, gateway = setup_backend + + tools = [CALCULATE_FUNCTION] + + resp = client.responses.create( + model=model, + input="What is 15 * 23?", + tools=tools, + tool_choice="required", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert ( + len(function_calls) > 0 + ), "tool_choice='required' must force at least one function call" + + def test_tool_choice_specific_function(self, setup_backend): + """Test tool_choice with specific function name forces that function to be called.""" + _, model, client, gateway = setup_backend + + tools = [SEARCH_WEB_FUNCTION, GET_WEATHER_FUNCTION] + + resp = client.responses.create( + model=model, + input="What's happening in the news today?", + tools=tools, + tool_choice={"type": "function", "function": {"name": "search_web"}}, + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0, "Must call the specified function" + assert ( + function_calls[0].name == "search_web" + ), "Must call the function specified in tool_choice" + + def test_tool_choice_streaming(self, setup_backend): + """Test tool_choice parameter works correctly with streaming.""" + _, model, client, gateway = setup_backend + + tools = [CALCULATE_FUNCTION] + + resp = client.responses.create( + model=model, + input="Calculate 42 * 17", + tools=tools, + tool_choice="required", + stream=True, + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [e.type for e in events] + assert "response.function_call_arguments.delta" in event_types + + completed_events = [e for e in events if e.type == "response.completed"] + assert len(completed_events) == 1 + + output = completed_events[0].response.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + + def test_tool_choice_with_mcp_tools(self, setup_backend): + """Test tool_choice parameter works with MCP tools.""" + _, model, client, gateway = setup_backend + + tools = [DEEPWIKI_MCP_TOOL] + + resp = client.responses.create( + model=model, + input="What transport protocols does the 2025-03-26 version of the MCP spec (modelcontextprotocol/modelcontextprotocol) support?", + tools=tools, + tool_choice="auto", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + mcp_calls = [item for item in output if item.type == "mcp_call"] + assert len(mcp_calls) > 0, "tool_choice='auto' should allow MCP tool calls" + + def test_tool_choice_mixed_function_and_mcp(self, setup_backend): + """Test tool_choice with mixed function and MCP tools.""" + _, model, client, gateway = setup_backend + + tools = [DEEPWIKI_MCP_TOOL, LOCAL_SEARCH_FUNCTION] + + resp = client.responses.create( + model=model, + input="Search for information about Python", + tools=tools, + tool_choice={"type": "function", "function": {"name": "local_search"}}, + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + assert function_calls[0].name == "local_search" + + mcp_calls = [item for item in output if item.type == "mcp_call"] + assert len(mcp_calls) == 0, "Should only call specified function, not MCP tools" + + def test_basic_function_call(self, setup_backend): + """Test basic function calling workflow.""" + _, model, client, gateway = setup_backend + + tools = [GET_HOROSCOPE_FUNCTION] + system_prompt = ( + "You are a helpful assistant that can call functions. " + "When a user asks for horoscope information, call the function. " + "IMPORTANT: Don't reply directly to the user, only call the function. " + ) + + input_list = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": "What is my horoscope? I am an Aquarius."}, + ] + + resp = client.responses.create(model=model, input=input_list, tools=tools) + + assert resp.error is None + assert resp.id is not None + assert resp.status == "completed" + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + + function_call = function_calls[0] + assert function_call.name == "get_horoscope" + + args = json.loads(function_call.arguments) + assert "sign" in args + assert args["sign"].lower() == "aquarius" + + def test_mcp_basic_tool_call(self, setup_backend): + """Test basic MCP tool call (non-streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=False, + reasoning={"effort": "low"}, + ) + + assert resp.error is None + assert resp.id is not None + assert resp.status == "completed" + assert len(resp.output_text) > 0 + + output_types = [item.type for item in resp.output] + assert "mcp_list_tools" in output_types + + mcp_calls = [item for item in resp.output if item.type == "mcp_call"] + assert len(mcp_calls) > 0 + + for mcp_call in mcp_calls: + assert mcp_call.id is not None + assert mcp_call.error is None + assert mcp_call.status == "completed" + assert mcp_call.server_label == "brave" + + def test_mcp_basic_tool_call_streaming(self, setup_backend): + """Test basic MCP tool call (streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=True, + reasoning={"effort": "low"}, + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [event.type for event in events] + assert "response.created" in event_types + assert "response.completed" in event_types + assert "response.mcp_list_tools.completed" in event_types + assert "response.mcp_call.completed" in event_types + + def test_mixed_mcp_and_function_tools(self, setup_backend): + """Test mixed MCP and function tools (non-streaming).""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Give me diagnostics for the Astra-7 Core Reactor.", + tools=[BRAVE_MCP_TOOL, SYSTEM_DIAGNOSTICS_FUNCTION], + stream=False, + tool_choice="auto", + ) + + assert resp.error is None + assert resp.id is not None + assert resp.output is not None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + + system_diagnostics_call = function_calls[0] + assert system_diagnostics_call.name == "get_system_diagnostics" + assert system_diagnostics_call.call_id is not None + + args = json.loads(system_diagnostics_call.arguments) + assert "system_name" in args + assert "astra-7" in args["system_name"].lower() + + def test_mixed_mcp_and_function_tools_streaming(self, setup_backend): + """Test mixed MCP and function tools (streaming).""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Give me diagnostics for the Astra-7 Core Reactor.", + tools=[BRAVE_MCP_TOOL, SYSTEM_DIAGNOSTICS_FUNCTION], + stream=True, + tool_choice="auto", + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [e.type for e in events] + assert "response.created" in event_types + assert "response.mcp_list_tools.completed" in event_types + assert "response.function_call_arguments.delta" in event_types + assert "response.function_call_arguments.done" in event_types + + func_arg_deltas = [ + e for e in events if e.type == "response.function_call_arguments.delta" + ] + assert len(func_arg_deltas) > 0 + + full_delta_event = "".join(e.delta for e in func_arg_deltas) + assert ( + "system_name" in full_delta_event.lower() + and "astra-7" in full_delta_event.lower() + ) + + +# ============================================================================= +# Local Backend Tests (gRPC with Qwen model) - Tool Choice +# ============================================================================= + + +@pytest.mark.e2e +@pytest.mark.model("qwen-14b") +@pytest.mark.gateway( + extra_args=["--tool-call-parser", "qwen", "--history-backend", "memory"] +) +@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) +class TestToolChoiceLocal: + """Tool choice tests against local gRPC backend with Qwen model.""" + + def test_tool_choice_auto(self, setup_backend): + """Test tool_choice="auto" allows model to decide whether to use tools.""" + _, model, client, gateway = setup_backend + + tools = [GET_WEATHER_FUNCTION] + + resp = client.responses.create( + model=model, + input="What is the weather in Seattle?", + tools=tools, + tool_choice="auto", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + assert len(output) > 0 + + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + + def test_tool_choice_required(self, setup_backend): + """Test tool_choice="required" forces the model to call at least one tool.""" + _, model, client, gateway = setup_backend + + tools = [CALCULATE_FUNCTION] + + resp = client.responses.create( + model=model, + input="What is 15 * 23?", + tools=tools, + tool_choice="required", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + function_calls = [item for item in resp.output if item.type == "function_call"] + assert len(function_calls) > 0 + + def test_tool_choice_specific_function(self, setup_backend): + """Test tool_choice with specific function name forces that function to be called.""" + _, model, client, gateway = setup_backend + + tools = [SEARCH_WEB_FUNCTION, GET_WEATHER_FUNCTION] + + resp = client.responses.create( + model=model, + input="What's happening in the news today?", + tools=tools, + tool_choice={"type": "function", "function": {"name": "search_web"}}, + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + function_calls = [item for item in resp.output if item.type == "function_call"] + assert len(function_calls) > 0 + assert function_calls[0].name == "search_web" + + def test_mcp_basic_tool_call(self, setup_backend): + """Test basic MCP tool call (non-streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=False, + reasoning={"effort": "low"}, + ) + + assert resp.error is None + assert resp.id is not None + assert resp.status == "completed" + + output_types = [item.type for item in resp.output] + assert "mcp_list_tools" in output_types + + mcp_calls = [item for item in resp.output if item.type == "mcp_call"] + assert len(mcp_calls) > 0 + + def test_mcp_basic_tool_call_streaming(self, setup_backend): + """Test basic MCP tool call (streaming).""" + _, model, client, gateway = setup_backend + + time.sleep(2) + + resp = client.responses.create( + model=model, + input=MCP_TEST_PROMPT, + tools=[BRAVE_MCP_TOOL], + stream=True, + reasoning={"effort": "low"}, + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [event.type for event in events] + assert "response.created" in event_types + assert "response.completed" in event_types + + def test_tool_choice_with_mcp_tools(self, setup_backend): + """Test tool_choice parameter works with MCP tools.""" + _, model, client, gateway = setup_backend + + tools = [DEEPWIKI_MCP_TOOL] + + resp = client.responses.create( + model=model, + input="What transport protocols does the 2025-03-26 version of the MCP spec (modelcontextprotocol/modelcontextprotocol) support?", + tools=tools, + tool_choice="auto", + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + mcp_calls = [item for item in output if item.type == "mcp_call"] + assert len(mcp_calls) > 0, "tool_choice='auto' should allow MCP tool calls" + + def test_tool_choice_mixed_function_and_mcp(self, setup_backend): + """Test tool_choice with mixed function and MCP tools.""" + _, model, client, gateway = setup_backend + + tools = [DEEPWIKI_MCP_TOOL, LOCAL_SEARCH_FUNCTION] + + resp = client.responses.create( + model=model, + input="Search for information about Python", + tools=tools, + tool_choice={"type": "function", "function": {"name": "local_search"}}, + stream=False, + ) + + assert resp.id is not None + assert resp.error is None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + assert function_calls[0].name == "local_search" + + mcp_calls = [item for item in output if item.type == "mcp_call"] + assert len(mcp_calls) == 0, "Should only call specified function, not MCP tools" + + def test_mixed_mcp_and_function_tools(self, setup_backend): + """Test mixed MCP and function tools (non-streaming).""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Give me diagnostics for the Astra-7 Core Reactor.", + tools=[BRAVE_MCP_TOOL, SYSTEM_DIAGNOSTICS_FUNCTION], + stream=False, + tool_choice="auto", + ) + + assert resp.error is None + assert resp.id is not None + assert resp.output is not None + + output = resp.output + function_calls = [item for item in output if item.type == "function_call"] + assert len(function_calls) > 0 + + system_diagnostics_call = function_calls[0] + assert system_diagnostics_call.name == "get_system_diagnostics" + assert system_diagnostics_call.call_id is not None + + args = json.loads(system_diagnostics_call.arguments) + assert "system_name" in args + assert "astra-7" in args["system_name"].lower() + + def test_mixed_mcp_and_function_tools_streaming(self, setup_backend): + """Test mixed MCP and function tools (streaming).""" + _, model, client, gateway = setup_backend + + resp = client.responses.create( + model=model, + input="Give me diagnostics for the Astra-7 Core Reactor.", + tools=[BRAVE_MCP_TOOL, SYSTEM_DIAGNOSTICS_FUNCTION], + stream=True, + tool_choice="auto", + ) + + events = list(resp) + assert len(events) > 0 + + event_types = [e.type for e in events] + assert "response.created" in event_types + assert "response.mcp_list_tools.completed" in event_types + assert "response.function_call_arguments.delta" in event_types + assert "response.function_call_arguments.done" in event_types + + func_arg_deltas = [ + e for e in events if e.type == "response.function_call_arguments.delta" + ] + assert len(func_arg_deltas) > 0 + + full_delta_event = "".join(e.delta for e in func_arg_deltas) + assert ( + "system_name" in full_delta_event.lower() + and "astra-7" in full_delta_event.lower() + )