From 4c46ecde8022b786a810b28c260773c9c789e9ef Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Wed, 7 Jan 2026 18:24:12 -0800 Subject: [PATCH] [smg][ci] delete old responses api ci (#16695) --- .github/workflows/pr-test-rust.yml | 9 - .../e2e_test/e2e_response_api/conftest.py | 132 --- .../features/test_basic_crud.py | 241 ------ .../features/test_state_management.py | 161 ---- .../features/test_streaming_events.py | 152 ---- .../features/test_structured_output.py | 172 ---- .../features/test_tools_call.py | 765 ------------------ .../e2e_response_api/router_fixtures.py | 565 ------------- .../e2e_test/e2e_response_api/util.py | 81 -- 9 files changed, 2278 deletions(-) delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/conftest.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/features/test_basic_crud.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/features/test_state_management.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/features/test_streaming_events.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/features/test_structured_output.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/features/test_tools_call.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/router_fixtures.py delete mode 100644 sgl-model-gateway/e2e_test/e2e_response_api/util.py diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml index 5e66f54e6..70b476aca 100644 --- a/.github/workflows/pr-test-rust.yml +++ b/.github/workflows/pr-test-rust.yml @@ -173,15 +173,6 @@ jobs: reruns: "" upload_benchmarks: true parallel_opts: "" # No parallel for benchmarks (performance measurement) - - name: response-api - timeout: 32 - test_dirs: "e2e_test/e2e_response_api" - extra_deps: "" - env_vars: "SHOW_ROUTER_LOGS=1" - reruns: "--reruns 3 --reruns-delay 2" - setup_oracle: true - setup_brave: true - parallel_opts: "" # Legacy tests, not yet migrated for parallel - name: responses timeout: 45 test_dirs: "e2e_test/responses" diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/conftest.py b/sgl-model-gateway/e2e_test/e2e_response_api/conftest.py deleted file mode 100644 index 786c4a912..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/conftest.py +++ /dev/null @@ -1,132 +0,0 @@ -""" -pytest configuration for e2e_response_api tests. - -This configures pytest to not collect base test classes that are meant to be inherited. -""" - -import os - -import openai -import pytest # noqa: F401 -from router_fixtures import ( - popen_launch_openai_xai_router, - popen_launch_workers_and_router, -) -from util import kill_process_tree - -# ------------------------------ -# Backend Configuration Map -# ------------------------------ -BACKENDS = { - "openai": { - "model": "gpt-5-nano", - "base_url_port": "http://127.0.0.1:30010", - "launcher": popen_launch_openai_xai_router, - "launcher_kwargs": { - "backend": "openai", - "history_backend": "memory", - }, - "api_key_env": "OPENAI_API_KEY", - "needs_workers": False, - }, - "xai": { - "model": "grok-4-fast", - "base_url_port": "http://127.0.0.1:30023", - "launcher": popen_launch_openai_xai_router, - "launcher_kwargs": { - "backend": "xai", - "history_backend": "memory", - }, - "api_key_env": "XAI_API_KEY", - "needs_workers": False, - }, - "grpc": { - "model": "/home/ubuntu/models/Qwen/Qwen2.5-14B-Instruct", - "base_url_port": "http://127.0.0.1:30030", - "launcher": popen_launch_workers_and_router, - "launcher_kwargs": { - "timeout": 90, - "num_workers": 1, - "tp_size": 2, - "policy": "round_robin", - "worker_args": ["--context-length=1000"], - "router_args": [ - "--history-backend", - "memory", - "--tool-call-parser", - "qwen", - ], - }, - "api_key_env": None, # grpc does not use API keys - "needs_workers": True, - }, - "grpc_harmony": { - "model": "/home/ubuntu/models/openai/gpt-oss-20b", - "base_url_port": "http://127.0.0.1:30030", - "launcher": popen_launch_workers_and_router, - "launcher_kwargs": { - "timeout": 90, - "num_workers": 1, - "tp_size": 2, - "policy": "round_robin", - "worker_args": ["--reasoning-parser=gpt-oss"], - "router_args": ["--history-backend", "memory"], - }, - "api_key_env": None, - "needs_workers": True, - }, - "oracle_store": { - "model": "gpt-5-nano", - "base_url_port": "http://127.0.0.1:30040", - "launcher": popen_launch_openai_xai_router, - "launcher_kwargs": { - "backend": "openai", - "history_backend": "oracle", - }, - "api_key_env": "OPENAI_API_KEY", - "needs_workers": False, - }, -} - - -@pytest.fixture(scope="class") -def setup_backend(request): - backend = request.param - if backend not in BACKENDS: - raise RuntimeError(f"Unknown backend {backend}") - - cfg = BACKENDS[backend] - - # Launch cluster - cluster = ( - cfg["launcher"]( - cfg["model"], - cfg["base_url_port"], - **cfg["launcher_kwargs"], - ) - if cfg["launcher"] is popen_launch_workers_and_router - else cfg["launcher"]( - backend=cfg["launcher_kwargs"]["backend"], - base_url=cfg["base_url_port"], - history_backend=cfg["launcher_kwargs"]["history_backend"], - ) - ) - - # Build client - api_key = os.environ.get(cfg["api_key_env"]) if cfg["api_key_env"] else None - client = openai.Client( - api_key=api_key, - base_url=cluster["base_url"] + "/v1", - ) - - # Yield data to test - try: - yield backend, cfg["model"], client - finally: - # Always kill router - kill_process_tree(cluster["router"].pid) - - # If workers exist, kill them as well - if cfg["needs_workers"]: - for w in cluster.get("workers", []): - kill_process_tree(w.pid) diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_basic_crud.py b/sgl-model-gateway/e2e_test/e2e_response_api/features/test_basic_crud.py deleted file mode 100644 index 0695feaa2..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_basic_crud.py +++ /dev/null @@ -1,241 +0,0 @@ -""" -Base test class for Response API e2e tests. - -This module provides base test classes that can be reused across different backends -(OpenAI, XAI, gRPC) with common test logic. -""" - -import sys -import time -from pathlib import Path - -import openai -import pytest -from openai import OpenAI -from openai.types import responses - -# Add current directory for local imports -_TEST_DIR = Path(__file__).parent -sys.path.insert(0, str(_TEST_DIR)) - - -@pytest.mark.parametrize("setup_backend", ["openai", "oracle_store"], indirect=True) -class TestResponseCRUD: - """Base class for Response API CRUD tests.""" - - def test_create_and_get_response(self, setup_backend): - """Test creating response and retrieving it.""" - _, model, client = 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 = 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 = 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" - - -@pytest.mark.parametrize("setup_backend", ["openai", "oracle_store"], indirect=True) -class TestConversationCRUD: - """Base class for Conversation API CRUD tests.""" - - def test_create_and_get_conversation(self, setup_backend): - """Test creating and retrieving conversation.""" - _, model, client = 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 = 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 update_data["key1"] == "value1" - assert update_data["key2"] == "value2" - - def test_delete_conversation(self, setup_backend): - """Test deleting conversation.""" - _, model, client = 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 = 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 - - -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 - - status = resp.status - - if status == "completed": - return resp - elif status == "failed": - raise AssertionError(f"Background task failed: {resp.error}") - elif 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" - ) diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_state_management.py b/sgl-model-gateway/e2e_test/e2e_response_api/features/test_state_management.py deleted file mode 100644 index 88f5346a9..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_state_management.py +++ /dev/null @@ -1,161 +0,0 @@ -""" -State management tests for Response API. - -Tests both previous_response_id and conversation-based state management. -These tests should work across all backends (OpenAI, XAI, gRPC). -""" - -import openai -import pytest - - -@pytest.mark.parametrize( - "setup_backend", ["openai", "xai", "grpc", "grpc_harmony"], indirect=True -) -class TestStateManagement: - """Tests for state management using previous_response_id and conversation.""" - - def test_basic_response_creation(self, setup_backend): - """Test basic response creation without state.""" - _, model, client = 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 = setup_backend - - resp = client.responses.create( - model=model, input="Count to 5", stream=True, max_output_tokens=50 - ) - - # Check for response.created event - events = [event for event in resp] - created_events = [event for event in events if event.type == "response.created"] - assert len(created_events) > 0 - - # Check for final completed event or in_progress events - assert any( - event.type in ["response.completed", "response.in_progress"] - for event in events - ) - - def test_previous_response_id_chaining(self, setup_backend): - """Test chaining responses using previous_response_id.""" - _, model, client = 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" - response1_id = resp1.id - - # Second response referencing first - resp2 = client.responses.create( - model=model, input="What is my name", previous_response_id=response1_id - ) - assert resp2.error is None - assert resp2.status == "completed" - - # The model should remember the name from previous response - 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 - - @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 = 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_conversation_with_multiple_turns(self, setup_backend): - """Test state management using conversation ID.""" - backend, model, client = setup_backend - - if backend in ["grpc", "grpc_harmony"]: - pytest.skip("TODO: 501 Not Implemented") - - # 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" - output_text = resp2.output_text - - # Should remember "5 apples" - assert "5" in output_text or "five" in 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" - output_text = resp3.output_text - - # Should calculate 5 + 3 = 8 - assert "8" in output_text or "eight" in output_text.lower() - list_resp = client.conversations.items.list(conversation_id) - assert list_resp.data is not None - items = list_resp.data - # Should have at least 6 items (3 inputs + 3 outputs) - assert len(items) >= 6 - - def test_mutually_exclusive_parameters(self, setup_backend): - """Test that previous_response_id and conversation are mutually exclusive.""" - _, model, client = setup_backend - - # TODO: Remove this once the conversation API is implemented for GRPC backend - conversation_id = "conv_123" - - resp1 = client.responses.create(model=model, input="Test") - response1_id = resp1.id - - # Try to use both parameters - with pytest.raises(openai.BadRequestError): - client.responses.create( - model=model, - input="This should fail", - previous_response_id=response1_id, - conversation=conversation_id, - ) diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_streaming_events.py b/sgl-model-gateway/e2e_test/e2e_response_api/features/test_streaming_events.py deleted file mode 100644 index 7b23d8b1c..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_streaming_events.py +++ /dev/null @@ -1,152 +0,0 @@ -""" -Streaming events tests for Response API. -Tests for streaming event validation including: -- Zero-based output_index for reasoning content -- OutputItemDone event emission and output array construction -""" - -import pytest - - -@pytest.mark.parametrize("setup_backend", ["grpc", "grpc_harmony"], indirect=True) -class TestStreamingEvents: - """Tests for streaming event validation.""" - - 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 = setup_backend - - resp = client.responses.create( - model=model, - input="Count from 1 to 3", - stream=True, - max_output_tokens=50, - ) - - events = [event for event in 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 i, item in enumerate(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. - """ - backend, model, client = setup_backend - if backend in ["grpc"]: - pytest.skip("skip test_reasoning_content for grpc") - - resp = client.responses.create( - model=model, - input="What is the capital of France? Think step by step.", - stream=True, - max_output_tokens=200, - ) - - events = [event for event in 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/e2e_response_api/features/test_structured_output.py b/sgl-model-gateway/e2e_test/e2e_response_api/features/test_structured_output.py deleted file mode 100644 index adfce1e03..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_structured_output.py +++ /dev/null @@ -1,172 +0,0 @@ -""" -Structured output tests for Response API. - -Tests for text.format field with json_object and json_schema formats. -""" - -import json -import sys -from pathlib import Path - -import pytest - -# Add current directory for local imports -_TEST_DIR = Path(__file__).parent -sys.path.insert(0, str(_TEST_DIR)) - - -@pytest.mark.parametrize("setup_backend", ["openai", "grpc_harmony"], indirect=True) -class TestStructuredOutput: - - def test_structured_output_json_schema(self, setup_backend): - """Test structured output with json_schema format.""" - _, model, client = setup_backend - - # Create response with structured output - 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 - - -@pytest.mark.parametrize("setup_backend", ["grpc"], indirect=True) -class TestSimpleSchemaStructuredOutput: - - def test_structured_output_json_schema(self, setup_backend): - """Override with simpler schema for Llama model (complex schemas not well supported).""" - _, model, client = 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/e2e_response_api/features/test_tools_call.py b/sgl-model-gateway/e2e_test/e2e_response_api/features/test_tools_call.py deleted file mode 100644 index 29d58f184..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/features/test_tools_call.py +++ /dev/null @@ -1,765 +0,0 @@ -""" -Test class for tool calling tests. - -This module provides test cases for function calling functionality, tool choices -and mcp calling functionality across different backends. -""" - -import json -import sys -import time -from pathlib import Path - -import pytest - -# Add current directory for local imports -_TEST_DIR = Path(__file__).parent -sys.path.insert(0, str(_TEST_DIR)) - - -@pytest.mark.parametrize( - "setup_backend", ["openai", "grpc", "grpc_harmony"], indirect=True -) -class TestToolCalling: - - # Shared function 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"], - }, - } - - # Shared constants for MCP tests - 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" - ) - - # Test cases for basic function calling functionality - - def test_basic_function_call(self, setup_backend): - """ - Test basic function calling workflow. - - This test follows the pattern from function_call_test.py: - 1. Define a function tool (get_horoscope) - 2. Send user message asking for horoscope - 3. Model should return function_call - 4. Execute function locally and provide output - 5. Model should generate final response using the function output - """ - backend, model, client = setup_backend - - if backend in ["grpc"]: - pytest.skip("skip for grpc") - - # 1. Define a list of callable tools for the model - tools = [ - { - "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"], - }, - }, - ] - 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. " - ) - - # Create a running input list we will add to over time - input_list = [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": "What is my horoscope? I am an Aquarius."}, - ] - - # 2. Prompt the model with tools defined - resp = client.responses.create(model=model, input=input_list, tools=tools) - - # Should successfully make the request - assert resp.error is None - - # Basic response structure - assert resp.id is not None - assert resp.status == "completed" - assert resp.output is not None - - # Verify output array is not empty - 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 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" - - # 3. Save function call outputs for subsequent requests - input_list.append(function_call) - - # 4. Execute the function logic for get_horoscope - horoscope = f"{args['sign']}: Next Tuesday you will befriend a baby otter." - - # 5. Provide function call results to the model - input_list.append( - { - "type": "function_call_output", - "call_id": function_call.call_id, - "output": json.dumps({"horoscope": horoscope}), - } - ) - - # 6. Make 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" - - # The model should be able to give a response using the function output - output2 = resp2.output - assert len(output2) > 0 - - # Find message output - messages = [item for item in output2 if item.type == "message"] - assert len(messages) > 0, "Response should contain at least one message" - - # Verify message contains the horoscope - message = messages[0] - assert message.content is not None - content_parts = message.content - assert len(content_parts) > 0 - - # Get text from content - text_parts = [part.text for part in content_parts if part.type == "output_text"] - full_text = " ".join(text_parts).lower() - - # Should mention the horoscope or baby otter - assert ( - "baby otter" in full_text or "aquarius" in full_text - ), "Response should reference the horoscope content" - - # Test cases for tool_choice parameter support, these tests require --reasoning-parser - - def test_tool_choice_auto(self, setup_backend): - """ - Test tool_choice="auto" allows model to decide whether to use tools. - - The model should be able to choose to call a tool or not. - """ - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip("skip for openai") - - tools = [self.GET_WEATHER_FUNCTION] - - # Query that should trigger tool use - 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 - - # With auto, model should choose to call get_weather for this query - 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. - - The model must make at least one function call. - """ - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip("skip for openai") - - tools = [self.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 - - # Must have at least one function call - 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. - - The model must call the specified function. - """ - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip("skip for openai") - - tools = [self.SEARCH_WEB_FUNCTION, self.GET_WEATHER_FUNCTION] - - # Force specific function call - 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 - - # Must have function call - function_calls = [item for item in output if item.type == "function_call"] - assert len(function_calls) > 0, "Must call the specified function" - - # Must be the specified function - called_function = function_calls[0] - assert ( - called_function.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. - - Verifies that tool_choice constraints are applied in streaming mode. - """ - backend, model, client = setup_backend - - if backend in ["openai", "grpc"]: - pytest.skip("skip for openai") - - tools = [self.CALCULATE_FUNCTION] - - resp = client.responses.create( - model=model, - input="Calculate 42 * 17", - tools=tools, - tool_choice="required", - stream=True, - ) - - events = [event for event in resp] - assert len(events) > 0 - - event_types = [e.type for e in events] - - # Should have function call events - assert ( - "response.function_call_arguments.delta" in event_types - ), "Should have function_call_arguments.delta events" - - # Verify completed event has function call - 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 - ), "Streaming with tool_choice='required' must produce function call" - - def test_tool_choice_with_mcp_tools(self, setup_backend): - """ - Test tool_choice parameter works with MCP tools. - - Verifies that tool_choice can control MCP tool usage. - """ - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip("skip for openai") - - tools = [self.DEEPWIKI_MCP_TOOL] - - # With tool_choice="auto", should allow MCP tool calls - 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 - - # Should have mcp_call with auto - 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. - - Verifies tool_choice can select specific tools when both function and MCP tools are available. - """ - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip("skip for openai") - - tools = [self.DEEPWIKI_MCP_TOOL, self.LOCAL_SEARCH_FUNCTION] - - # Force specific function call - 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 - - # Must call local_search, not MCP - function_calls = [item for item in output if item.type == "function_call"] - assert len(function_calls) > 0 - assert function_calls[0].name == "local_search" - - # Should not have mcp_call - 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" - - # Tests for MCP tool calling in both streaming and non-streaming modes. - - def test_mcp_basic_tool_call(self, setup_backend): - """ - Test basic MCP tool call (non-streaming). - Validation strictness is controlled by parameter `backend` from setup_backend fixture. - Set to "strict" if backend is http. - """ - backend, model, client = setup_backend - - # To avoid being rate-limited by brave search server - time.sleep(2) - - resp = client.responses.create( - model=model, - input=self.MCP_TEST_PROMPT, - tools=[self.BRAVE_MCP_TOOL], - stream=False, - reasoning={"effort": "low"}, - ) - - # Should successfully make the request - assert resp.error is None - - # Basic response structure - assert resp.id is not None - assert resp.status == "completed" - assert resp.model is not None - assert resp.output is not None - - # Verify output array is not empty - assert len(resp.output_text) > 0 - - # Check for MCP-specific output types - output_types = [item.type for item in resp.output] - - # Should have mcp_list_tools - tools are listed before calling - assert ( - "mcp_list_tools" in output_types - ), "Response should contain mcp_list_tools" - - # Should have at least one mcp_call - mcp_calls = [item for item in resp.output if item.type == "mcp_call"] - assert len(mcp_calls) > 0, "Response should contain at least one mcp_call" - - # Verify mcp_call structure - 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 mode: additional validation for HTTP backends - if backend == "openai": - # Should have final message output - messages = [item for item in resp.output if item.type == "message"] - assert len(messages) > 0, "Response should contain at least one message" - # Verify message structure - for msg in messages: - assert msg.content is not None - assert isinstance(msg.content, list) - - # Check content has text - 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). - - Validation strictness is controlled by the class attribute `mcp_validation_mode`. - Set to "strict" in subclasses for additional HTTP-specific validation. - """ - backend, model, client = setup_backend - - # To avoid being rate-limited by brave search server - time.sleep(2) - - resp = client.responses.create( - model=model, - input=self.MCP_TEST_PROMPT, - tools=[self.BRAVE_MCP_TOOL], - stream=True, - reasoning={"effort": "low"}, - ) - - # Should successfully make the request - events = [event for event in resp] - assert len(events) > 0 - - event_types = [event.type for event in events] - # Check for lifecycle events - assert "response.created" in event_types, "Should have response.created event" - assert ( - "response.completed" in event_types - ), "Should have response.completed event" - - # Check for MCP list tools events - 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" - - # Check for MCP call events - 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" - - # Verify final completed event has full response - 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 - - # Verify final output contains expected items - 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 mode: additional validation for HTTP backends - if backend == "openai": - # 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" - - def test_mixed_mcp_and_function_tools(self, setup_backend): - """Test mixed MCP and function tools (non-streaming).""" - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip( - "Requires external MCP server (deepwiki) - may not be accessible in CI" - ) - - resp = client.responses.create( - model=model, - input="Give me diagnostics for the Astra-7 Core Reactor.", - tools=[self.BRAVE_MCP_TOOL, self.SYSTEM_DIAGNOSTICS_FUNCTION], - stream=False, - tool_choice="auto", - ) - - # Should successfully make the request - assert resp.error is None - - # Basic response structure - assert resp.id is not None - assert resp.status is not None - assert resp.output is not None - - # Verify output array is not empty - output = resp.output - assert isinstance(output, list) - assert len(output) > 0 - - # Check for function_call (not mcp_call for get_system_diagnostics) - 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 for get_system_diagnostics - system_diagnostics_call = function_calls[0] - assert system_diagnostics_call.name == "get_system_diagnostics" - assert system_diagnostics_call.call_id is not None - assert system_diagnostics_call.arguments is not None - assert system_diagnostics_call.status is not None - - # Parse and verify arguments - 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).""" - backend, model, client = setup_backend - - if backend in ["openai"]: - pytest.skip( - "Requires external MCP server (deepwiki) - may not be accessible in CI" - ) - - resp = client.responses.create( - model=model, - input="Give me diagnostics for the Astra-7 Core Reactor.", - tools=[self.BRAVE_MCP_TOOL, self.SYSTEM_DIAGNOSTICS_FUNCTION], - stream=True, - tool_choice="auto", # Encourage tool usage - ) - - # Should successfully make the request - events = [event for event in resp] - assert len(events) > 0 - - event_types = [e.type for e in events] - - # Check for lifecycle events - assert "response.created" in event_types, "Should have response.created event" - - # Should have mcp_list_tools events - assert ( - "response.mcp_list_tools.completed" in event_types - ), "Should have mcp_list_tools.completed event" - - # Should have function_call_arguments events (not mcp_call_arguments) - assert ( - "response.function_call_arguments.delta" in event_types - ), "Should have function_call_arguments.delta event for function tools" - assert ( - "response.function_call_arguments.done" in event_types - ), "Should have function_call_arguments.done event for function tools" - - # Should NOT have mcp_call_arguments events for function tools - # (get_system_diagnostics should use function_call_arguments, not mcp_call_arguments) - mcp_call_arg_events = [ - e - for e in events - if e.type == "response.mcp_call_arguments.delta" - and "get_system_diagnostics" in str(e.delta) - ] - assert ( - len(mcp_call_arg_events) == 0 - ), "Should NOT emit mcp_call_arguments.delta for function tools (get_system_diagnostics)" - - # Verify function_call_arguments.delta event structure - func_arg_deltas = [ - e for e in events if e.type == "response.function_call_arguments.delta" - ] - assert ( - len(func_arg_deltas) > 0 - ), "Should have function_call_arguments.delta events" - - # Check that delta event contains system_name arguments - full_delta_event = "" - for event in func_arg_deltas: - full_delta_event += event.delta - - assert ( - "system_name" in full_delta_event.lower() - and "astra-7" in full_delta_event.lower() - ), "function_call_arguments.delta should contain system_name and astra-7" diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/router_fixtures.py b/sgl-model-gateway/e2e_test/e2e_response_api/router_fixtures.py deleted file mode 100644 index 74b6b1b01..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/router_fixtures.py +++ /dev/null @@ -1,565 +0,0 @@ -""" -Fixtures for launching OpenAI/XAI router for response API e2e testing. - -This module provides fixtures for launching SGLang router with OpenAI or XAI backends: - 1. Launch router with --backend openai pointing to OpenAI or XAI API - 2. Configure history backend (memory or oracle) - -This supports testing the Response API against real cloud providers. -""" - -import logging -import os -import socket -import subprocess -import time -from typing import Optional - -import requests - -logger = logging.getLogger(__name__) - - -def wait_for_workers_ready( - router_url: str, - expected_workers: int, - timeout: int = 300, - api_key: Optional[str] = None, -) -> None: - """ - Wait for router to have all workers connected. - - Polls the /workers endpoint until the 'total' field matches expected_workers. - - Example response from /workers endpoint: - {"workers":[],"total":0,"stats":{"prefill_count":0,"decode_count":0,"regular_count":0}} - - Args: - router_url: Base URL of router (e.g., "http://127.0.0.1:30000") - expected_workers: Number of workers expected to be connected - timeout: Max seconds to wait - api_key: Optional API key for authentication - """ - start_time = time.time() - last_error = None - attempt = 0 - - headers = {} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - - with requests.Session() as session: - while time.time() - start_time < timeout: - attempt += 1 - elapsed = int(time.time() - start_time) - - # Log progress every 10 seconds - if elapsed > 0 and elapsed % 10 == 0 and attempt % 10 == 0: - logger.info( - f" Still waiting for workers... ({elapsed}/{timeout}s elapsed)" - ) - - try: - response = session.get( - f"{router_url}/workers", headers=headers, timeout=5 - ) - if response.status_code == 200: - data = response.json() - total_workers = data.get("total", 0) - - if total_workers == expected_workers: - logger.info( - f" All {expected_workers} workers connected after {elapsed}s" - ) - return - else: - last_error = f"Workers: {total_workers}/{expected_workers}" - else: - last_error = f"HTTP {response.status_code}" - except requests.ConnectionError: - last_error = "Connection refused (router not ready yet)" - except requests.Timeout: - last_error = "Timeout" - except requests.RequestException as e: - last_error = str(e) - except (ValueError, KeyError) as e: - last_error = f"Invalid response: {e}" - - time.sleep(1) - - raise TimeoutError( - f"Router at {router_url} did not get {expected_workers} workers within {timeout}s.\n" - f"Last status: {last_error}\n" - f"Hint: Run with SHOW_ROUTER_LOGS=1 to see startup logs" - ) - - -def find_free_port() -> int: - """Find an available port on localhost.""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] - - -def wait_for_router_ready( - router_url: str, - timeout: int = 60, - api_key: Optional[str] = None, -) -> None: - """ - Wait for router to be ready. - - Polls the /health endpoint until it returns 200. - - Args: - router_url: Base URL of router (e.g., "http://127.0.0.1:30000") - timeout: Max seconds to wait - api_key: Optional API key for authentication - """ - start_time = time.time() - last_error = None - attempt = 0 - - headers = {} - if api_key: - headers["Authorization"] = f"Bearer {api_key}" - - with requests.Session() as session: - while time.time() - start_time < timeout: - attempt += 1 - elapsed = int(time.time() - start_time) - - # Log progress every 10 seconds - if elapsed > 0 and elapsed % 10 == 0 and attempt % 10 == 0: - logger.info( - f" Still waiting for router... ({elapsed}/{timeout}s elapsed)" - ) - - try: - response = session.get( - f"{router_url}/health", headers=headers, timeout=5 - ) - if response.status_code == 200: - logger.info(f" Router ready after {elapsed}s") - return - else: - last_error = f"HTTP {response.status_code}" - except requests.ConnectionError: - last_error = "Connection refused (router not ready yet)" - except requests.Timeout: - last_error = "Timeout" - except requests.RequestException as e: - last_error = str(e) - - time.sleep(1) - - raise TimeoutError( - f"Router at {router_url} did not become ready within {timeout}s.\n" - f"Last status: {last_error}\n" - f"Hint: Run with SHOW_ROUTER_LOGS=1 to see startup logs" - ) - - -def popen_launch_openai_xai_router( - backend: str, # "openai" or "xai" - base_url: str, - timeout: int = 60, - history_backend: str = "memory", - api_key: Optional[str] = None, - router_args: Optional[list] = None, - stdout=None, - stderr=None, - prometheus_port: Optional[int] = None, -) -> dict: - """ - Launch SGLang router with OpenAI or XAI backend. - - This approach: - 1. Starts router with --backend openai - 2. Points to OpenAI or XAI API via --worker-urls - 3. Configures history backend (memory or oracle) - 4. Waits for router health check to pass - - Args: - backend: "openai" or "xai" - base_url: Base URL for router (e.g., "http://127.0.0.1:30000") - timeout: Timeout for router startup (default: 60s) - history_backend: "memory" or "oracle" (default: memory) - api_key: Optional API key for router authentication - router_args: Additional arguments for router - stdout: Optional file handle for router stdout - stderr: Optional file handle for router stderr - - Returns: - dict with: - - router: router process object - - base_url: router URL (HTTP endpoint) - - Example: - >>> cluster = popen_launch_openai_xai_router( - ... "openai", "http://127.0.0.1:30000" - ... ) - >>> # Use cluster['base_url'] for HTTP requests - >>> # Cleanup: - >>> kill_process_tree(cluster['router'].pid) - """ - show_output = os.environ.get("SHOW_ROUTER_LOGS", "0") == "1" - - # Parse router port from base_url - if ":" in base_url.split("//")[-1]: - router_port = int(base_url.split(":")[-1]) - else: - router_port = find_free_port() - - logger.info(f"\n{'='*70}") - logger.info(f"Launching {backend.upper()} router") - logger.info(f"{'='*70}") - logger.info(f" Backend: {backend}") - logger.info(f" Router port: {router_port}") - logger.info(f" History backend: {history_backend}") - - # Determine worker URL based on backend - if backend == "openai": - worker_url = "https://api.openai.com" - # Get API key from environment - backend_api_key = os.environ.get("OPENAI_API_KEY") - if not backend_api_key: - raise ValueError( - "OPENAI_API_KEY environment variable must be set for OpenAI backend" - ) - elif backend == "xai": - worker_url = "https://api.x.ai" - # Get API key from environment - backend_api_key = os.environ.get("XAI_API_KEY") - if not backend_api_key: - raise ValueError( - "XAI_API_KEY environment variable must be set for XAI backend" - ) - else: - raise ValueError(f"Unsupported backend: {backend}") - - logger.info(f" Worker URL: {worker_url}") - - # Build router command - router_cmd = [ - "python3", - "-m", - "sglang_router.launch_router", - "--host", - "127.0.0.1", - "--port", - str(router_port), - "--backend", - "openai", - "--worker-urls", - worker_url, - "--history-backend", - history_backend, - "--log-level", - "warn", - ] - - # Note: Not adding --api-key to router command for local testing - # The router will not require authentication - - # Add Prometheus port to avoid conflicts (use unique port or disable) - if prometheus_port is None: - # Auto-assign a unique prometheus port based on router port - prometheus_port = router_port + 1000 - router_cmd.extend(["--prometheus-port", str(prometheus_port)]) - - # Add router-specific args - if router_args: - router_cmd.extend(router_args) - - if show_output: - logger.info(f" Command: {' '.join(router_cmd)}") - - # Set up environment with backend API key - env = os.environ.copy() - if backend == "openai": - env["OPENAI_API_KEY"] = backend_api_key - else: - env["XAI_API_KEY"] = backend_api_key - - # Launch router - if show_output: - router_proc = subprocess.Popen( - router_cmd, - env=env, - stdout=stdout, - stderr=stderr, - ) - else: - router_proc = subprocess.Popen( - router_cmd, - stdout=stdout if stdout is not None else subprocess.PIPE, - stderr=stderr if stderr is not None else subprocess.PIPE, - env=env, - ) - - print(f" PID: {router_proc.pid}") - - # Wait for router to be ready - router_url = f"http://127.0.0.1:{router_port}" - print(f"\nWaiting for router to start at {router_url}...") - - try: - wait_for_router_ready(router_url, timeout=timeout, api_key=None) - logger.info(f"✓ Router ready at {router_url}") - except TimeoutError: - logger.error(f"✗ Router failed to start") - # Cleanup: kill router - try: - router_proc.kill() - except: - pass - raise - - logger.info(f"\n{'='*70}") - logger.info(f"✓ {backend.upper()} router ready!") - logger.info(f" Router: {router_url}") - logger.info(f"{'='*70}\n") - - return { - "router": router_proc, - "base_url": router_url, - } - - -def popen_launch_workers_and_router( - model: str, - base_url: str, - timeout: int = 300, - num_workers: int = 2, - policy: str = "round_robin", - api_key: Optional[str] = None, - worker_args: Optional[list] = None, - router_args: Optional[list] = None, - tp_size: int = 1, - env: Optional[dict] = None, - stdout=None, - stderr=None, -) -> dict: - """ - Launch SGLang workers and gRPC router separately. - - This approach: - 1. Starts N SGLang workers with --grpc-mode flag - 2. Waits for workers to initialize (process startup) - 3. Starts a gRPC router pointing to those workers - 4. Waits for router health check to pass (router validates worker connectivity) - - This matches production deployment patterns better than the integrated approach. - - Args: - model: Model path (e.g., /home/ubuntu/models/llama-3.1-8b-instruct) - base_url: Base URL for router (e.g., "http://127.0.0.1:8080") - timeout: Timeout for server startup (default: 300s) - num_workers: Number of workers to launch - policy: Routing policy (round_robin, random, power_of_two, cache_aware) - api_key: Optional API key for router - worker_args: Additional arguments for workers (e.g., ["--context-len", "8192"]) - router_args: Additional arguments for router (e.g., ["--max-total-token", "1536"]) - tp_size: Tensor parallelism size for workers (default: 1) - env: Optional environment variables for workers (e.g., {"SGLANG_CLIP_MAX_NEW_TOKENS_ESTIMATION": "256"}) - stdout: Optional file handle for worker stdout (default: subprocess.PIPE) - stderr: Optional file handle for worker stderr (default: subprocess.PIPE) - - Returns: - dict with: - - workers: list of worker process objects - - worker_urls: list of gRPC worker URLs - - router: router process object - - base_url: router URL (HTTP endpoint) - - Example: - >>> cluster = popen_launch_workers_and_router(model, base_url, num_workers=2) - >>> # Use cluster['base_url'] for HTTP requests - >>> # Cleanup: - >>> for worker in cluster['workers']: - >>> kill_process_tree(worker.pid) - >>> kill_process_tree(cluster['router'].pid) - """ - show_output = os.environ.get("SHOW_ROUTER_LOGS", "0") == "1" - - # Parse router port from base_url - if ":" in base_url.split("//")[-1]: - router_port = int(base_url.split(":")[-1]) - else: - router_port = find_free_port() - - logger.info(f"\n{'='*70}") - logger.info(f"Launching gRPC cluster (separate workers + router)") - logger.info(f"{'='*70}") - logger.info(f" Model: {model}") - logger.info(f" Router port: {router_port}") - logger.info(f" Workers: {num_workers}") - logger.info(f" TP size: {tp_size}") - logger.info(f" Policy: {policy}") - - # Step 1: Launch workers with gRPC enabled - workers = [] - worker_urls = [] - - for i in range(num_workers): - worker_port = find_free_port() - worker_url = f"grpc://127.0.0.1:{worker_port}" - worker_urls.append(worker_url) - - logger.info(f"\n[Worker {i+1}/{num_workers}]") - logger.info(f" Port: {worker_port}") - logger.info(f" URL: {worker_url}") - - # Build worker command - worker_cmd = [ - "python3", - "-m", - "sglang.launch_server", - "--model-path", - model, - "--host", - "127.0.0.1", - "--port", - str(worker_port), - "--grpc-mode", # Enable gRPC for this worker - "--mem-fraction-static", - "0.8", - ] - - # Add TP size - if tp_size > 1: - worker_cmd.extend(["--tp-size", str(tp_size)]) - - # Add worker-specific args - if worker_args: - worker_cmd.extend(worker_args) - - # Launch worker with optional environment variables - if show_output: - worker_proc = subprocess.Popen( - worker_cmd, - env=env, - stdout=stdout, - stderr=stderr, - ) - else: - worker_proc = subprocess.Popen( - worker_cmd, - stdout=stdout if stdout is not None else subprocess.PIPE, - stderr=stderr if stderr is not None else subprocess.PIPE, - env=env, - ) - - workers.append(worker_proc) - logger.info(f" PID: {worker_proc.pid}") - - # Give workers a moment to start binding to ports - # The router will check worker health when it starts - logger.info(f"\nWaiting for {num_workers} workers to initialize (20s)...") - time.sleep(20) - - # Quick check: make sure worker processes are still alive - for i, worker in enumerate(workers): - if worker.poll() is not None: - logger.error( - f" ✗ Worker {i+1} died during startup (exit code: {worker.poll()})" - ) - # Cleanup: kill all workers - for w in workers: - try: - w.kill() - except: - pass - raise RuntimeError(f"Worker {i+1} failed to start") - - logger.info( - f"✓ All {num_workers} workers started (router will verify connectivity)" - ) - - # Step 2: Launch router pointing to workers - logger.info(f"\n[Router]") - logger.info(f" Port: {router_port}") - logger.info(f" Worker URLs: {', '.join(worker_urls)}") - - # Build router command - router_cmd = [ - "python3", - "-m", - "sglang_router.launch_router", - "--host", - "127.0.0.1", - "--port", - str(router_port), - "--prometheus-port", - "9321", - "--policy", - policy, - "--model-path", - model, - "--log-level", - "warn", - ] - - # Add worker URLs - router_cmd.append("--worker-urls") - router_cmd.extend(worker_urls) - - # Add API key - if api_key: - router_cmd.extend(["--api-key", api_key]) - - # Add router-specific args - if router_args: - router_cmd.extend(router_args) - - if show_output: - logger.info(f" Command: {' '.join(router_cmd)}") - - # Launch router - if show_output: - router_proc = subprocess.Popen(router_cmd) - else: - router_proc = subprocess.Popen( - router_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - - logger.info(f" PID: {router_proc.pid}") - - # Wait for router to be ready - router_url = f"http://127.0.0.1:{router_port}" - logger.info(f"\nWaiting for router to start at {router_url}...") - - try: - wait_for_workers_ready( - router_url, expected_workers=num_workers, timeout=180, api_key=api_key - ) - logger.info(f"✓ Router ready at {router_url}") - except TimeoutError: - logger.error(f"✗ Router failed to start") - # Cleanup: kill router and all workers - try: - router_proc.kill() - except: - pass - for worker in workers: - try: - worker.kill() - except: - pass - raise - - logger.info(f"\n{'='*70}") - logger.info(f"✓ gRPC cluster ready!") - logger.info(f" Router: {router_url}") - logger.info(f" Workers: {len(workers)}") - logger.info(f"{'='*70}\n") - - return { - "workers": workers, - "worker_urls": worker_urls, - "router": router_proc, - "base_url": router_url, - } diff --git a/sgl-model-gateway/e2e_test/e2e_response_api/util.py b/sgl-model-gateway/e2e_test/e2e_response_api/util.py deleted file mode 100644 index 222b648b6..000000000 --- a/sgl-model-gateway/e2e_test/e2e_response_api/util.py +++ /dev/null @@ -1,81 +0,0 @@ -""" -Utility functions for Response API e2e tests. -""" - -import logging -import os -import signal -import threading -import unittest - -import psutil - -logger = logging.getLogger(__name__) - - -def kill_process_tree(parent_pid, include_parent: bool = True, skip_pid: int = None): - """ - Kill the process and all its child processes. - - Args: - parent_pid: PID of the parent process - include_parent: Whether to kill the parent process itself - skip_pid: Optional PID to skip during cleanup - """ - # Remove sigchld handler to avoid spammy logs - if threading.current_thread() is threading.main_thread(): - signal.signal(signal.SIGCHLD, signal.SIG_DFL) - - if parent_pid is None: - parent_pid = os.getpid() - include_parent = False - - try: - itself = psutil.Process(parent_pid) - except psutil.NoSuchProcess: - return - - children = itself.children(recursive=True) - for child in children: - if child.pid == skip_pid: - continue - try: - child.kill() - except psutil.NoSuchProcess: - pass - - if include_parent: - try: - itself.kill() - except psutil.NoSuchProcess: - pass - - -class CustomTestCase(unittest.TestCase): - """ - Custom test case base class with retry support. - - This provides automatic test retry functionality based on environment variables. - """ - - def _callTestMethod(self, method): - """Override to add retry logic.""" - max_retry = int(os.environ.get("SGLANG_TEST_MAX_RETRY", "0")) - - if max_retry == 0: - # No retry, just run once - return super(CustomTestCase, self)._callTestMethod(method) - - # Retry logic - for attempt in range(max_retry + 1): - try: - return super(CustomTestCase, self)._callTestMethod(method) - except Exception as e: - if attempt < max_retry: - logger.info( - f"Test failed on attempt {attempt + 1}/{max_retry + 1}, retrying..." - ) - continue - else: - # Last attempt, re-raise the exception - raise