[smg][ci] delete old responses api ci (#16695)
This commit is contained in:
Simo Lin
9
.github/workflows/pr-test-rust.yml
vendored
9
.github/workflows/pr-test-rust.yml
vendored
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
@@ -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"
|
||||
)
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
@@ -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,
|
||||
}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user