minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[router]Replace requests lib with openai in e2e_response_api (#13293)

Browse Source
This commit is contained in:
Xinyue Zhang
2025-11-14 16:56:40 -08:00
committed by GitHub
parent b223669136
commit 14dc052382
8 changed files with 305 additions and 479 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sgl-router/py_test/e2e_response_api/backends/test_grpc_backend.py
+19 -18
View File
@@ -11,6 +11,8 @@ import sys
import unittest import unittest
from pathlib import Path from pathlib import Path
import openai
# Add e2e_response_api directory for imports # Add e2e_response_api directory for imports
_TEST_DIR = Path(__file__).parent.parent _TEST_DIR = Path(__file__).parent.parent
sys.path.insert(0, str(_TEST_DIR)) sys.path.insert(0, str(_TEST_DIR))
@@ -51,6 +53,7 @@ class TestGrpcBackend(StateManagementTests, MCPTests, StructuredOutputBaseTest):
) )
cls.base_url = cls.cluster["base_url"] cls.base_url = cls.cluster["base_url"]
cls.client = openai.Client(api_key=cls.api_key, base_url=cls.base_url + "/v1")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
@@ -64,8 +67,7 @@ class TestGrpcBackend(StateManagementTests, MCPTests, StructuredOutputBaseTest):
def test_structured_output_json_schema(self): def test_structured_output_json_schema(self):
"""Override with simpler schema for Llama model (complex schemas not well supported).""" """Override with simpler schema for Llama model (complex schemas not well supported)."""
data = { params = {
"model": self.model,
"input": [ "input": [
{ {
"role": "system", "role": "system",
@@ -89,28 +91,26 @@ class TestGrpcBackend(StateManagementTests, MCPTests, StructuredOutputBaseTest):
}, },
} }
create_resp = self.make_request("/v1/responses", "POST", data) create_resp = self.create_response(**params)
self.assertEqual(create_resp.status_code, 200) self.assertIsNone(create_resp.error)
self.assertIsNotNone(create_resp.id)
create_data = create_resp.json() self.assertIsNotNone(create_resp.output)
self.assertIn("id", create_data) self.assertIsNotNone(create_resp.text)
self.assertIn("output", create_data)
self.assertIn("text", create_data)
# Verify text format was echoed back correctly # Verify text format was echoed back correctly
self.assertIn("format", create_data["text"]) self.assertIsNotNone(create_resp.text.format)
self.assertEqual(create_data["text"]["format"]["type"], "json_schema") self.assertEqual(create_resp.text.format.type, "json_schema")
self.assertEqual(create_data["text"]["format"]["name"], "math_answer") self.assertEqual(create_resp.text.format.name, "math_answer")
self.assertIn("schema", create_data["text"]["format"]) self.assertIsNotNone(create_resp.text.format.schema_)
# Find the message output # Find the message output
output_text = next( output_text = next(
( (
content.get("text", "") content.text
for item in create_data.get("output", []) for item in create_resp.output
if item.get("type") == "message" if item.type == "message"
for content in item.get("content", []) for content in item.content
if content.get("type") == "output_text" if content.type == "output_text"
), ),
None, None,
) )
@@ -154,6 +154,7 @@ class TestGrpcHarmonyBackend(
) )
cls.base_url = cls.cluster["base_url"] cls.base_url = cls.cluster["base_url"]
cls.client = openai.Client(api_key=cls.api_key, base_url=cls.base_url + "/v1")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
sgl-router/py_test/e2e_response_api/backends/test_http_backend.py
+4
View File
@@ -13,6 +13,8 @@ import sys
import unittest import unittest
from pathlib import Path from pathlib import Path
import openai
# Add e2e_response_api directory for imports # Add e2e_response_api directory for imports
_TEST_DIR = Path(__file__).parent.parent _TEST_DIR = Path(__file__).parent.parent
sys.path.insert(0, str(_TEST_DIR)) sys.path.insert(0, str(_TEST_DIR))
@@ -52,6 +54,7 @@ class TestOpenaiBackend(
) )
cls.base_url = cls.cluster["base_url"] cls.base_url = cls.cluster["base_url"]
cls.client = openai.Client(api_key=cls.api_key, base_url=cls.base_url + "/v1")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
@@ -93,6 +96,7 @@ class TestXaiBackend(StateManagementTests):
) )
cls.base_url = cls.cluster["base_url"] cls.base_url = cls.cluster["base_url"]
cls.client = openai.Client(api_key=cls.api_key, base_url=cls.base_url + "/v1")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
sgl-router/py_test/e2e_response_api/mixins/basic_crud.py
+143 -213
View File
@@ -5,14 +5,16 @@ This module provides base test classes that can be reused across different backe
(OpenAI, XAI, gRPC) with common test logic. (OpenAI, XAI, gRPC) with common test logic.
""" """
import json from __future__ import annotations
import sys import sys
import time import time
import unittest import unittest
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional, Union
import requests import openai
from openai.types import conversations, responses
# Add current directory for local imports # Add current directory for local imports
_TEST_DIR = Path(__file__).parent _TEST_DIR = Path(__file__).parent
@@ -28,44 +30,11 @@ class ResponseAPIBaseTest(CustomTestCase):
base_url: str = None base_url: str = None
api_key: str = None api_key: str = None
model: str = None model: str = None
client: openai.OpenAI = None
def make_request(
self,
endpoint: str,
method: str = "POST",
json_data: Optional[dict] = None,
params: Optional[dict] = None,
) -> requests.Response:
"""
Make HTTP request to router.
Args:
endpoint: Endpoint path (e.g., "/v1/responses")
method: HTTP method (GET, POST, DELETE)
json_data: JSON body for POST requests
params: Query parameters
Returns:
requests.Response object
"""
url = f"{self.base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
if method == "POST":
resp = requests.post(url, json=json_data, headers=headers, params=params)
elif method == "GET":
resp = requests.get(url, headers=headers, params=params)
elif method == "DELETE":
resp = requests.delete(url, headers=headers, params=params)
else:
raise ValueError(f"Unsupported method: {method}")
return resp
def create_response( def create_response(
self, self,
input_text: str, input: Union[str, responses.ResponseInputParam],
instructions: Optional[str] = None, instructions: Optional[str] = None,
stream: bool = False, stream: bool = False,
max_output_tokens: Optional[int] = None, max_output_tokens: Optional[int] = None,
@@ -75,12 +44,12 @@ class ResponseAPIBaseTest(CustomTestCase):
tools: Optional[list] = None, tools: Optional[list] = None,
background: bool = False, background: bool = False,
**kwargs, **kwargs,
) -> requests.Response: ) -> responses.Response | openai.Stream[responses.ResponseStreamEvent]:
""" """
Create a response via POST /v1/responses. Create a response via POST /v1/responses.
Args: Args:
input_text: User input input: User input
instructions: Optional system instructions instructions: Optional system instructions
stream: Whether to stream response stream: Whether to stream response
max_output_tokens: Optional max tokens to generate max_output_tokens: Optional max tokens to generate
@@ -92,178 +61,128 @@ class ResponseAPIBaseTest(CustomTestCase):
**kwargs: Additional request parameters **kwargs: Additional request parameters
Returns: Returns:
requests.Response object Response object for non-stream request
ResponseStreamEvent for stream request
""" """
data = { params = {
"model": self.model, "model": self.model,
"input": input_text, "input": input,
"stream": stream, "stream": stream,
**kwargs, **kwargs,
} }
if instructions: if instructions:
data["instructions"] = instructions params["instructions"] = instructions
if max_output_tokens is not None: if max_output_tokens is not None:
data["max_output_tokens"] = max_output_tokens params["max_output_tokens"] = max_output_tokens
if temperature is not None: if temperature is not None:
data["temperature"] = temperature params["temperature"] = temperature
if previous_response_id: if previous_response_id:
data["previous_response_id"] = previous_response_id params["previous_response_id"] = previous_response_id
if conversation: if conversation:
data["conversation"] = conversation params["conversation"] = conversation
if tools: if tools:
data["tools"] = tools params["tools"] = tools
if background: if background:
data["background"] = background params["background"] = background
if stream: return self.client.responses.create(**params)
# For streaming, we need to handle SSE
return self._create_streaming_response(data)
else:
return self.make_request("/v1/responses", "POST", data)
def _create_streaming_response(self, data: dict) -> requests.Response: def get_response(
"""Handle streaming response creation.""" self, response_id: str
url = f"{self.base_url}/v1/responses" ) -> responses.Response | openai.Stream[responses.ResponseStreamEvent]:
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
# Return response object with stream=True
return requests.post(url, json=data, headers=headers, stream=True)
def get_response(self, response_id: str) -> requests.Response:
"""Get response by ID via GET /v1/responses/{response_id}.""" """Get response by ID via GET /v1/responses/{response_id}."""
return self.make_request(f"/v1/responses/{response_id}", "GET") return self.client.responses.retrieve(response_id=response_id)
def delete_response(self, response_id: str) -> requests.Response: def delete_response(self, response_id: str) -> None:
"""Delete response by ID via DELETE /v1/responses/{response_id}.""" """Delete response by ID via DELETE /v1/responses/{response_id}."""
return self.make_request(f"/v1/responses/{response_id}", "DELETE") return self.client.responses.delete(response_id=response_id)
def cancel_response(self, response_id: str) -> requests.Response: def cancel_response(self, response_id: str) -> responses.Response:
"""Cancel response by ID via POST /v1/responses/{response_id}/cancel.""" """Cancel response by ID via POST /v1/responses/{response_id}/cancel."""
return self.make_request(f"/v1/responses/{response_id}/cancel", "POST", {}) return self.client.responses.cancel(response_id=response_id)
def get_response_input_items(self, response_id: str) -> requests.Response: def get_response_input_items(
self, response_id: str
) -> openai.pagination.SyncCursorPage[responses.ResponseItem]:
"""Get response input items via GET /v1/responses/{response_id}/input_items.""" """Get response input items via GET /v1/responses/{response_id}/input_items."""
return self.make_request(f"/v1/responses/{response_id}/input_items", "GET") return self.client.responses.input_items.list(response_id=response_id)
def create_conversation(self, metadata: Optional[dict] = None) -> requests.Response: def create_conversation(
self, metadata: Optional[dict] = None
) -> conversations.Conversation:
"""Create conversation via POST /v1/conversations.""" """Create conversation via POST /v1/conversations."""
data = {} params = {}
if metadata: if metadata:
data["metadata"] = metadata params["metadata"] = metadata
return self.make_request("/v1/conversations", "POST", data) return self.client.conversations.create(**params)
def get_conversation(self, conversation_id: str) -> requests.Response: def get_conversation(self, conversation_id: str) -> conversations.Conversation:
"""Get conversation by ID via GET /v1/conversations/{conversation_id}.""" """Get conversation by ID via GET /v1/conversations/{conversation_id}."""
return self.make_request(f"/v1/conversations/{conversation_id}", "GET") return self.client.conversations.retrieve(conversation_id=conversation_id)
def update_conversation( def update_conversation(
self, conversation_id: str, metadata: dict self, conversation_id: str, metadata: dict
) -> requests.Response: ) -> conversations.Conversation:
"""Update conversation via POST /v1/conversations/{conversation_id}.""" """Update conversation via POST /v1/conversations/{conversation_id}."""
return self.make_request( return self.client.conversations.update(
f"/v1/conversations/{conversation_id}", "POST", {"metadata": metadata} conversation_id=conversation_id, metadata=metadata
) )
def delete_conversation(self, conversation_id: str) -> requests.Response: def delete_conversation(
self, conversation_id: str
) -> conversations.ConversationDeletedResource:
"""Delete conversation via DELETE /v1/conversations/{conversation_id}.""" """Delete conversation via DELETE /v1/conversations/{conversation_id}."""
return self.make_request(f"/v1/conversations/{conversation_id}", "DELETE") return self.client.conversations.delete(conversation_id=conversation_id)
def list_conversation_items( def list_conversation_items(
self, self,
conversation_id: str, conversation_id: str,
limit: Optional[int] = None, limit: Optional[int] = None,
after: Optional[str] = None, after: Optional[str] = None,
before: Optional[str] = None,
order: str = "asc", order: str = "asc",
) -> requests.Response: ) -> openai.pagination.SyncConversationCursorPage[conversations.ConversationItem]:
"""List conversation items via GET /v1/conversations/{conversation_id}/items.""" """List conversation items via GET /v1/conversations/{conversation_id}/items."""
params = {"order": order} params = {"conversation_id": conversation_id, "order": order}
if limit: if limit:
params["limit"] = limit params["limit"] = limit
if after: if after:
params["after"] = after params["after"] = after
if before: return self.client.conversations.items.list(**params)
params["before"] = before
return self.make_request(
f"/v1/conversations/{conversation_id}/items", "GET", params=params
)
def create_conversation_items( def create_conversation_items(
self, conversation_id: str, items: list self, conversation_id: str, items: list
) -> requests.Response: ) -> conversations.ConversationItemList:
"""Create conversation items via POST /v1/conversations/{conversation_id}/items.""" """Create conversation items via POST /v1/conversations/{conversation_id}/items."""
return self.make_request( return self.client.conversations.items.create(
f"/v1/conversations/{conversation_id}/items", "POST", {"items": items} conversation_id=conversation_id, items=items
) )
def get_conversation_item( def get_conversation_item(
self, conversation_id: str, item_id: str self, conversation_id: str, item_id: str
) -> requests.Response: ) -> conversations.ConversationItem:
"""Get conversation item via GET /v1/conversations/{conversation_id}/items/{item_id}.""" """Get conversation item via GET /v1/conversations/{conversation_id}/items/{item_id}."""
return self.make_request( return self.client.conversations.items.retrieve(
f"/v1/conversations/{conversation_id}/items/{item_id}", "GET" conversation_id=conversation_id, item_id=item_id
) )
def delete_conversation_item( def delete_conversation_item(
self, conversation_id: str, item_id: str self, conversation_id: str, item_id: str
) -> requests.Response: ) -> conversations.Conversation:
"""Delete conversation item via DELETE /v1/conversations/{conversation_id}/items/{item_id}.""" """Delete conversation item via DELETE /v1/conversations/{conversation_id}/items/{item_id}."""
return self.make_request( return self.client.conversations.items.delete(
f"/v1/conversations/{conversation_id}/items/{item_id}", "DELETE" conversation_id=conversation_id, item_id=item_id
) )
def parse_sse_events(self, response: requests.Response) -> list:
"""
Parse Server-Sent Events from streaming response.
Args:
response: requests.Response with stream=True
Returns:
List of event dictionaries with 'event' and 'data' keys
"""
events = []
current_event = None
for line in response.iter_lines():
if not line:
# Empty line signals end of event
if current_event and current_event.get("data"):
events.append(current_event)
current_event = None
continue
line = line.decode("utf-8")
if line.startswith("event:"):
current_event = {"event": line[6:].strip()}
elif line.startswith("data:"):
if current_event is None:
current_event = {}
data_str = line[5:].strip()
try:
current_event["data"] = json.loads(data_str)
except json.JSONDecodeError:
current_event["data"] = data_str
# Don't forget the last event if stream ends without empty line
if current_event and current_event.get("data"):
events.append(current_event)
return events
def wait_for_background_task( def wait_for_background_task(
self, response_id: str, timeout: int = 30, poll_interval: float = 0.5 self, response_id: str, timeout: int = 30, poll_interval: float = 0.5
) -> dict: ) -> responses.Response:
""" """
Wait for background task to complete. Wait for background task to complete.
@@ -283,17 +202,15 @@ class ResponseAPIBaseTest(CustomTestCase):
while time.time() - start_time < timeout: while time.time() - start_time < timeout:
resp = self.get_response(response_id) resp = self.get_response(response_id)
self.assertEqual(resp.status_code, 200) self.assertIsNone(resp.error)
self.assertEqual(resp.id, response_id)
data = resp.json() status = resp.status
status = data.get("status")
if status == "completed": if status == "completed":
return data return resp
elif status == "failed": elif status == "failed":
raise AssertionError( raise AssertionError(f"Background task failed: {resp.error}")
f"Background task failed: {data.get('error', 'Unknown error')}"
)
elif status == "cancelled": elif status == "cancelled":
raise AssertionError("Background task was cancelled") raise AssertionError("Background task was cancelled")
@@ -310,31 +227,29 @@ class StateManagementBaseTest(ResponseAPIBaseTest):
def test_basic_response_creation(self): def test_basic_response_creation(self):
"""Test basic response creation without state.""" """Test basic response creation without state."""
resp = self.create_response("What is 2+2?", max_output_tokens=50) resp = self.create_response("What is 2+2?", max_output_tokens=50)
self.assertEqual(resp.status_code, 200)
data = resp.json() self.assertIsNotNone(resp.id)
self.assertIn("id", data) self.assertIsNone(resp.error)
self.assertIn("output", data) self.assertEqual(resp.status, "completed")
self.assertEqual(data["status"], "completed") self.assertGreater(len(resp.output_text), 0)
self.assertIn("usage", data) self.assertGreater(resp.usage.input_tokens, 0)
self.assertGreater(resp.usage.output_tokens, 0)
self.assertGreater(resp.usage.total_tokens, 0)
def test_streaming_response(self): def test_streaming_response(self):
"""Test streaming response.""" """Test streaming response."""
resp = self.create_response("Count to 5", stream=True, max_output_tokens=50) resp = self.create_response("Count to 5", stream=True, max_output_tokens=50)
self.assertEqual(resp.status_code, 200)
events = self.parse_sse_events(resp)
self.assertGreater(len(events), 0)
# Check for response.created event # Check for response.created event
created_events = [e for e in events if e.get("event") == "response.created"] events = [event for event in resp]
created_events = [event for event in events if event.type == "response.created"]
self.assertGreater(len(created_events), 0) self.assertGreater(len(created_events), 0)
# Check for final completed event or in_progress events # Check for final completed event or in_progress events
self.assertTrue( self.assertTrue(
any( any(
e.get("event") in ["response.completed", "response.in_progress"] event.type in ["response.completed", "response.in_progress"]
for e in events for event in events
) )
) )
@@ -346,41 +261,40 @@ class ResponseCRUDBaseTest(ResponseAPIBaseTest):
"""Test creating response and retrieving it.""" """Test creating response and retrieving it."""
# Create response # Create response
create_resp = self.create_response("Hello, world!") create_resp = self.create_response("Hello, world!")
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
self.assertIsNone(create_resp.error)
create_data = create_resp.json() self.assertEqual(create_resp.status, "completed")
response_id = create_data["id"] self.assertGreater(len(create_resp.output_text), 0)
response_id = create_resp.id
# Get response # Get response
get_resp = self.get_response(response_id) get_resp = self.get_response(response_id)
self.assertEqual(get_resp.status_code, 200) self.assertIsNone(get_resp.error)
self.assertEqual(get_resp.id, response_id)
self.assertEqual(get_resp.status, "completed")
get_data = get_resp.json() input_resp = self.get_response_input_items(get_resp.id)
self.assertEqual(get_data["id"], response_id) self.assertIsNotNone(input_resp.data)
self.assertEqual(get_data["status"], "completed") self.assertGreater(len(input_resp.data), 0)
input_resp = self.get_response_input_items(get_data["id"])
self.assertEqual(input_resp.status_code, 200)
input_data = input_resp.json()
self.assertIn("data", input_data)
self.assertGreater(len(input_data["data"]), 0)
@unittest.skip("TODO: Add delete response feature") @unittest.skip("TODO: Add delete response feature")
def test_delete_response(self): def test_delete_response(self):
"""Test deleting response.""" """Test deleting response."""
# Create response # Create response
create_resp = self.create_response("Test deletion", max_output_tokens=50) create_resp = self.create_response("Test deletion")
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
self.assertIsNone(create_resp.error)
self.assertEqual(create_resp.status, "completed")
self.assertGreater(len(create_resp.output_text), 0)
response_id = create_resp.json()["id"] response_id = create_resp.id
# Delete response # Delete response
delete_resp = self.delete_response(response_id) self.delete_response(response_id)
self.assertEqual(delete_resp.status_code, 200)
# Verify it's deleted (should return 404) # Verify it's deleted (should return 404)
get_resp = self.get_response(response_id) with self.assertRaises(openai.NotFoundError):
self.assertEqual(get_resp.status_code, 404) self.get_response(response_id)
@unittest.skip("TODO: Add background response feature") @unittest.skip("TODO: Add background response feature")
def test_background_response(self): def test_background_response(self):
@@ -389,15 +303,15 @@ class ResponseCRUDBaseTest(ResponseAPIBaseTest):
create_resp = self.create_response( create_resp = self.create_response(
"Write a short story", background=True, max_output_tokens=100 "Write a short story", background=True, max_output_tokens=100
) )
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
self.assertIsNone(create_resp.error)
self.assertIn(create_resp.status, ["in_progress", "queued"])
create_data = create_resp.json() response_id = create_resp.id
response_id = create_data["id"]
self.assertEqual(create_data["status"], "in_progress")
# Wait for completion # Wait for completion
final_data = self.wait_for_background_task(response_id, timeout=60) final_data = self.wait_for_background_task(response_id, timeout=60)
self.assertEqual(final_data["status"], "completed") self.assertEqual(final_data.status, "completed")
class ConversationCRUDBaseTest(ResponseAPIBaseTest): class ConversationCRUDBaseTest(ResponseAPIBaseTest):
@@ -407,72 +321,88 @@ class ConversationCRUDBaseTest(ResponseAPIBaseTest):
"""Test creating and retrieving conversation.""" """Test creating and retrieving conversation."""
# Create conversation # Create conversation
create_resp = self.create_conversation(metadata={"user": "test_user"}) create_resp = self.create_conversation(metadata={"user": "test_user"})
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
self.assertIsNotNone(create_resp.created_at)
create_data = create_resp.json() create_data = create_resp.metadata
conversation_id = create_data["id"] self.assertEqual(create_data["user"], "test_user")
self.assertEqual(create_data["metadata"]["user"], "test_user") conversation_id = create_resp.id
# Get conversation # Get conversation
get_resp = self.get_conversation(conversation_id) get_resp = self.get_conversation(conversation_id)
self.assertEqual(get_resp.status_code, 200) self.assertIsNotNone(get_resp.id)
self.assertIsNotNone(get_resp.created_at)
get_data = get_resp.json() get_data = get_resp.metadata
self.assertEqual(get_data["id"], conversation_id) self.assertEqual(get_resp.id, conversation_id)
self.assertEqual(get_data["metadata"]["user"], "test_user") self.assertEqual(get_data["user"], "test_user")
def test_update_conversation(self): def test_update_conversation(self):
"""Test updating conversation metadata.""" """Test updating conversation metadata."""
# Create conversation # Create conversation
create_resp = self.create_conversation(metadata={"key1": "value1"}) create_resp = self.create_conversation(metadata={"key1": "value1"})
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
conversation_id = create_resp.json()["id"] self.assertIsNotNone(create_resp.created_at)
create_data = create_resp.metadata
self.assertEqual(create_data["key1"], "value1")
self.assertNotIn("key2", create_data)
conversation_id = create_resp.id
# Update conversation # Update conversation
update_resp = self.update_conversation( update_resp = self.update_conversation(
conversation_id, metadata={"key1": "value1", "key2": "value2"} conversation_id, metadata={"key1": "value1", "key2": "value2"}
) )
self.assertEqual(update_resp.status_code, 200) self.assertEqual(update_resp.id, conversation_id)
update_data = update_resp.metadata
self.assertEqual(update_data["key1"], "value1")
self.assertEqual(update_data["key2"], "value2")
# Verify update # Verify update
get_resp = self.get_conversation(conversation_id) get_resp = self.get_conversation(conversation_id)
get_data = get_resp.json() get_data = get_resp.metadata
self.assertEqual(get_data["metadata"]["key2"], "value2") self.assertEqual(get_data["key1"], "value1")
self.assertEqual(get_data["key2"], "value2")
def test_delete_conversation(self): def test_delete_conversation(self):
"""Test deleting conversation.""" """Test deleting conversation."""
# Create conversation # Create conversation
create_resp = self.create_conversation() create_resp = self.create_conversation()
self.assertEqual(create_resp.status_code, 200) self.assertIsNotNone(create_resp.id)
conversation_id = create_resp.json()["id"] self.assertIsNotNone(create_resp.created_at)
conversation_id = create_resp.id
# Delete conversation # Delete conversation
delete_resp = self.delete_conversation(conversation_id) delete_resp = self.delete_conversation(conversation_id)
self.assertEqual(delete_resp.status_code, 200) self.assertIsNotNone(delete_resp.id)
self.assertTrue(delete_resp.deleted)
# Verify deletion # Verify deletion
get_resp = self.get_conversation(conversation_id) with self.assertRaises(openai.NotFoundError):
self.assertEqual(get_resp.status_code, 404) self.get_conversation(conversation_id)
def test_list_conversation_items(self): def test_list_conversation_items(self):
"""Test listing conversation items.""" """Test listing conversation items."""
# Create conversation # Create conversation
conv_resp = self.create_conversation() conv_resp = self.create_conversation()
conversation_id = conv_resp.json()["id"] self.assertIsNotNone(conv_resp.id)
conversation_id = conv_resp.id
# Create response with conversation # Create response with conversation
self.create_response( resp1 = self.create_response(
"First message", conversation=conversation_id, max_output_tokens=50 "First message", conversation=conversation_id, max_output_tokens=50
) )
self.create_response( self.assertIsNone(resp1.error)
resp2 = self.create_response(
"Second message", conversation=conversation_id, max_output_tokens=50 "Second message", conversation=conversation_id, max_output_tokens=50
) )
self.assertIsNone(resp2.error)
# List items # List items
list_resp = self.list_conversation_items(conversation_id) list_resp = self.list_conversation_items(conversation_id)
self.assertEqual(list_resp.status_code, 200) self.assertIsNotNone(list_resp)
self.assertIsNotNone(list_resp.data)
list_data = list_resp.json() list_data = list_resp.data
self.assertIn("data", list_data)
# Should have at least 4 items (2 inputs + 2 outputs) # Should have at least 4 items (2 inputs + 2 outputs)
self.assertGreaterEqual(len(list_data["data"]), 4) self.assertGreaterEqual(len(list_data), 4)
sgl-router/py_test/e2e_response_api/mixins/function_call.py
+26 -87
View File
@@ -13,49 +13,10 @@ from pathlib import Path
_TEST_DIR = Path(__file__).parent _TEST_DIR = Path(__file__).parent
sys.path.insert(0, str(_TEST_DIR)) sys.path.insert(0, str(_TEST_DIR))
from util import CustomTestCase from basic_crud import ResponseAPIBaseTest
class ResponseAPIBaseTest(CustomTestCase):
"""Base class for Response API tests with common utilities."""
# To be set by subclasses
base_url: str = None
api_key: str = None
model: str = None
def make_request(
self,
endpoint: str,
method: str = "POST",
json_data: dict = None,
params: dict = None,
):
"""
Make HTTP request to router.
This is a minimal implementation - subclasses should import from basic_crud.
"""
import requests
url = f"{self.base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
if method == "POST":
resp = requests.post(url, json=json_data, headers=headers, params=params)
elif method == "GET":
resp = requests.get(url, headers=headers, params=params)
elif method == "DELETE":
resp = requests.delete(url, headers=headers, params=params)
else:
raise ValueError(f"Unsupported method: {method}")
return resp
class FunctionCallingBaseTest(ResponseAPIBaseTest): class FunctionCallingBaseTest(ResponseAPIBaseTest):
"""Base class for function calling tests."""
def test_basic_function_call(self): def test_basic_function_call(self):
""" """
@@ -99,54 +60,41 @@ class FunctionCallingBaseTest(ResponseAPIBaseTest):
] ]
# 2. Prompt the model with tools defined # 2. Prompt the model with tools defined
resp = self.make_request( resp = self.create_response(input=input_list, tools=tools)
"/v1/responses",
"POST",
{
"model": self.model,
"tools": tools,
"input": input_list,
},
)
# Should successfully make the request # Should successfully make the request
self.assertEqual(resp.status_code, 200) self.assertIsNone(resp.error)
data = resp.json()
# Basic response structure # Basic response structure
self.assertIn("id", data) self.assertIsNotNone(resp.id)
self.assertIn("status", data) self.assertEqual(resp.status, "completed")
self.assertEqual(data["status"], "completed") self.assertIsNotNone(resp.output)
self.assertIn("output", data)
# Verify output array is not empty # Verify output array is not empty
output = data["output"] output = resp.output
self.assertIsInstance(output, list) self.assertIsInstance(output, list)
self.assertGreater(len(output), 0) self.assertGreater(len(output), 0)
# Check for function_call in output # Check for function_call in output
function_calls = [ function_calls = [item for item in output if item.type == "function_call"]
item for item in output if item.get("type") == "function_call"
]
self.assertGreater( self.assertGreater(
len(function_calls), 0, "Response should contain at least one function_call" len(function_calls), 0, "Response should contain at least one function_call"
) )
# Verify function_call structure # Verify function_call structure
function_call = function_calls[0] function_call = function_calls[0]
self.assertIn("call_id", function_call) self.assertIsNotNone(function_call.call_id)
self.assertIn("name", function_call) self.assertIsNotNone(function_call.name)
self.assertEqual(function_call["name"], "get_horoscope") self.assertEqual(function_call.name, "get_horoscope")
self.assertIn("arguments", function_call) self.assertIsNotNone(function_call.arguments)
# Parse arguments # Parse arguments
args = json.loads(function_call["arguments"]) args = json.loads(function_call.arguments)
self.assertIn("sign", args) self.assertIn("sign", args)
self.assertEqual(args["sign"].lower(), "aquarius") self.assertEqual(args["sign"].lower(), "aquarius")
# 3. Save function call outputs for subsequent requests # 3. Save function call outputs for subsequent requests
input_list += output input_list.append(function_call)
# 4. Execute the function logic for get_horoscope # 4. Execute the function logic for get_horoscope
horoscope = f"{args['sign']}: Next Tuesday you will befriend a baby otter." horoscope = f"{args['sign']}: Next Tuesday you will befriend a baby otter."
@@ -155,47 +103,38 @@ class FunctionCallingBaseTest(ResponseAPIBaseTest):
input_list.append( input_list.append(
{ {
"type": "function_call_output", "type": "function_call_output",
"call_id": function_call["call_id"], "call_id": function_call.call_id,
"output": json.dumps({"horoscope": horoscope}), "output": json.dumps({"horoscope": horoscope}),
} }
) )
# 6. Make second request with function output # 6. Make second request with function output
resp2 = self.make_request( resp2 = self.create_response(
"/v1/responses", input=input_list,
"POST", instructions="Respond only with a horoscope generated by a tool.",
{ tools=tools,
"model": self.model,
"instructions": "Respond only with a horoscope generated by a tool.",
"tools": tools,
"input": input_list,
},
) )
data2 = resp2.json() self.assertIsNone(resp2.error)
self.assertEqual(data2["status"], "completed") self.assertEqual(resp2.status, "completed")
# The model should be able to give a response using the function output # The model should be able to give a response using the function output
output2 = data2["output"] output2 = resp2.output
self.assertGreater(len(output2), 0) self.assertGreater(len(output2), 0)
# Find message output # Find message output
messages = [item for item in output2 if item.get("type") == "message"] messages = [item for item in output2 if item.type == "message"]
self.assertGreater( self.assertGreater(
len(messages), 0, "Response should contain at least one message" len(messages), 0, "Response should contain at least one message"
) )
# Verify message contains the horoscope # Verify message contains the horoscope
message = messages[0] message = messages[0]
self.assertIn("content", message) self.assertIsNotNone(message.content)
content_parts = message["content"] content_parts = message.content
self.assertGreater(len(content_parts), 0) self.assertGreater(len(content_parts), 0)
# Get text from content # Get text from content
text_parts = [ text_parts = [part.text for part in content_parts if part.type == "output_text"]
part.get("text", "")
for part in content_parts
if part.get("type") == "output_text"
]
full_text = " ".join(text_parts).lower() full_text = " ".join(text_parts).lower()
# Should mention the horoscope or baby otter # Should mention the horoscope or baby otter
sgl-router/py_test/e2e_response_api/mixins/mcp.py
+57 -79
View File
@@ -56,24 +56,19 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Should successfully make the request # Should successfully make the request
self.assertEqual(resp.status_code, 200) self.assertIsNone(resp.error)
data = resp.json()
# Basic response structure # Basic response structure
self.assertIn("id", data) self.assertIsNotNone(resp.id)
self.assertIn("status", data) self.assertEqual(resp.status, "completed")
self.assertEqual(data["status"], "completed") self.assertIsNotNone(resp.model)
self.assertIn("output", data) self.assertIsNotNone(resp.output)
self.assertIn("model", data)
# Verify output array is not empty # Verify output array is not empty
output = data["output"] self.assertGreater(len(resp.output_text), 0)
self.assertIsInstance(output, list)
self.assertGreater(len(output), 0)
# Check for MCP-specific output types # Check for MCP-specific output types
output_types = [item.get("type") for item in output] output_types = [item.type for item in resp.output]
# Should have mcp_list_tools - tools are listed before calling # Should have mcp_list_tools - tools are listed before calling
self.assertIn( self.assertIn(
@@ -81,40 +76,38 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Should have at least one mcp_call # Should have at least one mcp_call
mcp_calls = [item for item in output if item.get("type") == "mcp_call"] mcp_calls = [item for item in resp.output if item.type == "mcp_call"]
self.assertGreater( self.assertGreater(
len(mcp_calls), 0, "Response should contain at least one mcp_call" len(mcp_calls), 0, "Response should contain at least one mcp_call"
) )
# Verify mcp_call structure # Verify mcp_call structure
for mcp_call in mcp_calls: for mcp_call in mcp_calls:
self.assertIn("id", mcp_call) self.assertIsNotNone(mcp_call.id)
self.assertIn("status", mcp_call) self.assertEqual(mcp_call.status, "completed")
self.assertEqual(mcp_call["status"], "completed") self.assertEqual(mcp_call.server_label, "brave")
self.assertIn("server_label", mcp_call) self.assertIsNotNone(mcp_call.name)
self.assertEqual(mcp_call["server_label"], "brave") self.assertIsNotNone(mcp_call.arguments)
self.assertIn("name", mcp_call) self.assertIsNotNone(mcp_call.output)
self.assertIn("arguments", mcp_call)
self.assertIn("output", mcp_call)
# Strict mode: additional validation for HTTP backends # Strict mode: additional validation for HTTP backends
if self.mcp_validation_mode == "strict": if self.mcp_validation_mode == "strict":
# Should have final message output # Should have final message output
messages = [item for item in output if item.get("type") == "message"] messages = [item for item in resp.output if item.type == "message"]
self.assertGreater( self.assertGreater(
len(messages), 0, "Response should contain at least one message" len(messages), 0, "Response should contain at least one message"
) )
# Verify message structure # Verify message structure
for msg in messages: for msg in messages:
self.assertIn("content", msg) self.assertIsNotNone(msg.content)
self.assertIsInstance(msg["content"], list) self.assertIsInstance(msg.content, list)
# Check content has text # Check content has text
for content_item in msg["content"]: for content_item in msg.content:
if content_item.get("type") == "output_text": if content_item.type == "output_text":
self.assertIn("text", content_item) self.assertIsNotNone(content_item.text)
self.assertIsInstance(content_item["text"], str) self.assertIsInstance(content_item.text, str)
self.assertGreater(len(content_item["text"]), 0) self.assertGreater(len(content_item.text), 0)
def test_mcp_basic_tool_call_streaming(self): def test_mcp_basic_tool_call_streaming(self):
"""Test basic MCP tool call (streaming). """Test basic MCP tool call (streaming).
@@ -130,13 +123,10 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Should successfully make the request # Should successfully make the request
self.assertEqual(resp.status_code, 200) events = [event for event in resp]
events = self.parse_sse_events(resp)
self.assertGreater(len(events), 0) self.assertGreater(len(events), 0)
event_types = [e.get("event") for e in events] event_types = [event.type for event in events]
# Check for lifecycle events # Check for lifecycle events
self.assertIn( self.assertIn(
"response.created", event_types, "Should have response.created event" "response.created", event_types, "Should have response.created event"
@@ -185,31 +175,31 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Verify final completed event has full response # Verify final completed event has full response
completed_events = [e for e in events if e.get("event") == "response.completed"] completed_events = [e for e in events if e.type == "response.completed"]
self.assertEqual(len(completed_events), 1) self.assertEqual(len(completed_events), 1)
final_response = completed_events[0].get("data", {}).get("response", {}) final_response = completed_events[0].response
self.assertIn("id", final_response) self.assertIsNotNone(final_response.id)
self.assertEqual(final_response.get("status"), "completed") self.assertEqual(final_response.status, "completed")
self.assertIn("output", final_response) self.assertIsNotNone(final_response.output)
# Verify final output contains expected items # Verify final output contains expected items
final_output = final_response.get("output", []) final_output = final_response.output
final_output_types = [item.get("type") for item in final_output] final_output_types = [item.type for item in final_output]
self.assertIn("mcp_list_tools", final_output_types) self.assertIn("mcp_list_tools", final_output_types)
self.assertIn("mcp_call", final_output_types) self.assertIn("mcp_call", final_output_types)
# Verify mcp_call items in final output # Verify mcp_call items in final output
mcp_calls = [item for item in final_output if item.get("type") == "mcp_call"] mcp_calls = [item for item in final_output if item.type == "mcp_call"]
self.assertGreater(len(mcp_calls), 0) self.assertGreater(len(mcp_calls), 0)
for mcp_call in mcp_calls: for mcp_call in mcp_calls:
self.assertEqual(mcp_call.get("status"), "completed") self.assertEqual(mcp_call.status, "completed")
self.assertEqual(mcp_call.get("server_label"), "brave") self.assertEqual(mcp_call.server_label, "brave")
self.assertIn("name", mcp_call) self.assertIsNotNone(mcp_call.name)
self.assertIn("arguments", mcp_call) self.assertIsNotNone(mcp_call.arguments)
self.assertIn("output", mcp_call) self.assertIsNotNone(mcp_call.output)
# Strict mode: additional validation for HTTP backends # Strict mode: additional validation for HTTP backends
if self.mcp_validation_mode == "strict": if self.mcp_validation_mode == "strict":
@@ -239,19 +229,17 @@ class MCPTests(ResponseAPIBaseTest):
# Verify text deltas combine to final message # Verify text deltas combine to final message
text_deltas = [ text_deltas = [
e.get("data", {}).get("delta", "") e.delta for e in events if e.type == "response.output_text.delta"
for e in events
if e.get("event") == "response.output_text.delta"
] ]
self.assertGreater(len(text_deltas), 0, "Should have text deltas") self.assertGreater(len(text_deltas), 0, "Should have text deltas")
# Get final text from output_text.done event # Get final text from output_text.done event
text_done_events = [ text_done_events = [
e for e in events if e.get("event") == "response.output_text.done" e for e in events if e.type == "response.output_text.done"
] ]
self.assertGreater(len(text_done_events), 0) self.assertGreater(len(text_done_events), 0)
final_text = text_done_events[0].get("data", {}).get("text", "") final_text = text_done_events[0].text
self.assertGreater(len(final_text), 0, "Final text should not be empty") self.assertGreater(len(final_text), 0, "Final text should not be empty")
def test_mixed_mcp_and_function_tools(self): def test_mixed_mcp_and_function_tools(self):
@@ -264,38 +252,33 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Should successfully make the request # Should successfully make the request
self.assertEqual(resp.status_code, 200) self.assertIsNone(resp.error)
data = resp.json()
# Basic response structure # Basic response structure
self.assertIn("id", data) self.assertIsNotNone(resp.id)
self.assertIn("status", data) self.assertIsNotNone(resp.status)
self.assertIn("output", data) self.assertIsNotNone(resp.output)
# Verify output array is not empty # Verify output array is not empty
output = data["output"] output = resp.output
self.assertIsInstance(output, list) self.assertIsInstance(output, list)
self.assertGreater(len(output), 0) self.assertGreater(len(output), 0)
# Check for function_call (not mcp_call for get_weather) # Check for function_call (not mcp_call for get_weather)
function_calls = [ function_calls = [item for item in output if item.type == "function_call"]
item for item in output if item.get("type") == "function_call"
]
self.assertGreater( self.assertGreater(
len(function_calls), 0, "Response should contain at least one function_call" len(function_calls), 0, "Response should contain at least one function_call"
) )
# Verify function_call structure for get_weather # Verify function_call structure for get_weather
weather_call = function_calls[0] weather_call = function_calls[0]
self.assertIn("name", weather_call) self.assertEqual(weather_call.name, "get_weather")
self.assertEqual(weather_call["name"], "get_weather") self.assertIsNotNone(weather_call.call_id)
self.assertIn("call_id", weather_call) self.assertIsNotNone(weather_call.arguments)
self.assertIn("arguments", weather_call) self.assertIsNotNone(weather_call.status)
self.assertIn("status", weather_call)
# Parse and verify arguments # Parse and verify arguments
args = json.loads(weather_call["arguments"]) args = json.loads(weather_call.arguments)
self.assertIn("location", args) self.assertIn("location", args)
self.assertIn("seattle", args["location"].lower()) self.assertIn("seattle", args["location"].lower())
@@ -309,12 +292,10 @@ class MCPTests(ResponseAPIBaseTest):
) )
# Should successfully make the request # Should successfully make the request
self.assertEqual(resp.status_code, 200) events = [event for event in resp]
events = self.parse_sse_events(resp)
self.assertGreater(len(events), 0) self.assertGreater(len(events), 0)
event_types = [e.get("event") for e in events] event_types = [e.type for e in events]
# Check for lifecycle events # Check for lifecycle events
self.assertIn( self.assertIn(
@@ -345,8 +326,8 @@ class MCPTests(ResponseAPIBaseTest):
mcp_call_arg_events = [ mcp_call_arg_events = [
e e
for e in events for e in events
if e.get("event") == "response.mcp_call_arguments.delta" if e.type == "response.mcp_call_arguments.delta"
and "get_weather" in str(e.get("data", {})) and "get_weather" in str(e.delta)
] ]
self.assertEqual( self.assertEqual(
len(mcp_call_arg_events), len(mcp_call_arg_events),
@@ -356,9 +337,7 @@ class MCPTests(ResponseAPIBaseTest):
# Verify function_call_arguments.delta event structure # Verify function_call_arguments.delta event structure
func_arg_deltas = [ func_arg_deltas = [
e e for e in events if e.type == "response.function_call_arguments.delta"
for e in events
if e.get("event") == "response.function_call_arguments.delta"
] ]
self.assertGreater( self.assertGreater(
len(func_arg_deltas), 0, "Should have function_call_arguments.delta events" len(func_arg_deltas), 0, "Should have function_call_arguments.delta events"
@@ -367,8 +346,7 @@ class MCPTests(ResponseAPIBaseTest):
# Check that at least one delta event contains location arguments # Check that at least one delta event contains location arguments
has_location = False has_location = False
for event in func_arg_deltas: for event in func_arg_deltas:
data = event.get("data", {}) delta = event.delta
delta = data.get("delta", "")
if "location" in delta.lower() or "seattle" in delta.lower(): if "location" in delta.lower() or "seattle" in delta.lower():
has_location = True has_location = True
break break
sgl-router/py_test/e2e_response_api/mixins/state_management.py
+35 -36
View File
@@ -7,6 +7,7 @@ These tests should work across all backends (OpenAI, XAI, gRPC).
import unittest import unittest
import openai
from basic_crud import ResponseAPIBaseTest from basic_crud import ResponseAPIBaseTest
@@ -19,57 +20,59 @@ class StateManagementTests(ResponseAPIBaseTest):
resp1 = self.create_response( resp1 = self.create_response(
"My name is Alice and my friend is Bob. Remember it." "My name is Alice and my friend is Bob. Remember it."
) )
self.assertEqual(resp1.status_code, 200) self.assertIsNone(resp1.error)
response1_id = resp1.json()["id"] self.assertEqual(resp1.status, "completed")
response1_id = resp1.id
# Second response referencing first # Second response referencing first
resp2 = self.create_response( resp2 = self.create_response(
"What is my name", previous_response_id=response1_id "What is my name", previous_response_id=response1_id
) )
self.assertEqual(resp2.status_code, 200) self.assertIsNone(resp2.error)
response2_data = resp2.json() self.assertEqual(resp2.status, "completed")
# The model should remember the name from previous response # The model should remember the name from previous response
output_text = self._extract_output_text(response2_data) self.assertIn("Alice", resp2.output_text)
self.assertIn("Alice", output_text)
# Third response referencing second # Third response referencing second
resp3 = self.create_response( resp3 = self.create_response(
"What is my friend name?", "What is my friend name?",
previous_response_id=response2_data["id"], previous_response_id=resp2.id,
) )
response3_data = resp3.json() self.assertIsNone(resp3.error)
output_text = self._extract_output_text(response3_data) self.assertEqual(resp3.status, "completed")
self.assertEqual(resp3.status_code, 200) self.assertIn("Bob", resp3.output_text)
self.assertIn("Bob", output_text)
@unittest.skip("TODO: Add the invalid previous_response_id check") @unittest.skip("TODO: Add the invalid previous_response_id check")
def test_previous_response_id_invalid(self): def test_previous_response_id_invalid(self):
"""Test using invalid previous_response_id.""" """Test using invalid previous_response_id."""
resp = self.create_response( with self.assertRaises(openai.BadRequestError):
"Test", previous_response_id="resp_invalid123", max_output_tokens=50 self.create_response(
) "Test", previous_response_id="resp_invalid123", max_output_tokens=50
self.assertIn(resp.status_code, [400, 404]) )
def test_conversation_with_multiple_turns(self): def test_conversation_with_multiple_turns(self):
"""Test state management using conversation ID.""" """Test state management using conversation ID."""
# Create conversation # Create conversation
conv_resp = self.create_conversation(metadata={"topic": "math"}) conv_resp = self.create_conversation(metadata={"topic": "math"})
self.assertEqual(conv_resp.status_code, 200) self.assertIsNotNone(conv_resp.id)
self.assertIsNotNone(conv_resp.created_at)
conversation_id = conv_resp.json()["id"] conversation_id = conv_resp.id
# First response in conversation # First response in conversation
resp1 = self.create_response("I have 5 apples.", conversation=conversation_id) resp1 = self.create_response("I have 5 apples.", conversation=conversation_id)
self.assertEqual(resp1.status_code, 200) self.assertIsNone(resp1.error)
self.assertEqual(resp1.status, "completed")
# Second response in same conversation # Second response in same conversation
resp2 = self.create_response( resp2 = self.create_response(
"How many apples do I have?", "How many apples do I have?",
conversation=conversation_id, conversation=conversation_id,
) )
self.assertEqual(resp2.status_code, 200) self.assertIsNone(resp2.error)
output_text = self._extract_output_text(resp2.json()) self.assertEqual(resp2.status, "completed")
output_text = resp2.output_text
# Should remember "5 apples" # Should remember "5 apples"
self.assertTrue("5" in output_text or "five" in output_text.lower()) self.assertTrue("5" in output_text or "five" in output_text.lower())
@@ -79,14 +82,15 @@ class StateManagementTests(ResponseAPIBaseTest):
"If I get 3 more, how many total?", "If I get 3 more, how many total?",
conversation=conversation_id, conversation=conversation_id,
) )
self.assertEqual(resp3.status_code, 200) self.assertIsNone(resp3.error)
output_text = self._extract_output_text(resp3.json()) self.assertEqual(resp3.status, "completed")
output_text = resp3.output_text
# Should calculate 5 + 3 = 8 # Should calculate 5 + 3 = 8
self.assertTrue("8" in output_text or "eight" in output_text.lower()) self.assertTrue("8" in output_text or "eight" in output_text.lower())
list_resp = self.list_conversation_items(conversation_id) list_resp = self.list_conversation_items(conversation_id)
self.assertEqual(list_resp.status_code, 200) self.assertIsNotNone(list_resp.data)
items = list_resp.json()["data"] items = list_resp.data
# Should have at least 6 items (3 inputs + 3 outputs) # Should have at least 6 items (3 inputs + 3 outputs)
self.assertGreaterEqual(len(items), 6) self.assertGreaterEqual(len(items), 6)
@@ -96,20 +100,15 @@ class StateManagementTests(ResponseAPIBaseTest):
conversation_id = "conv_123" conversation_id = "conv_123"
resp1 = self.create_response("Test") resp1 = self.create_response("Test")
response1_id = resp1.json()["id"] response1_id = resp1.id
# Try to use both parameters # Try to use both parameters
resp = self.create_response( with self.assertRaises(openai.BadRequestError):
"This should fail", self.create_response(
previous_response_id=response1_id, "This should fail",
conversation=conversation_id, previous_response_id=response1_id,
) conversation=conversation_id,
)
# Should return 400 Bad Request
self.assertEqual(resp.status_code, 400)
error_data = resp.json()
self.assertIn("error", error_data)
self.assertIn("mutually exclusive", error_data["error"]["message"].lower())
# Helper methods # Helper methods
sgl-router/py_test/e2e_response_api/mixins/structured_output.py
+18 -46
View File
@@ -12,42 +12,16 @@ from pathlib import Path
_TEST_DIR = Path(__file__).parent _TEST_DIR = Path(__file__).parent
sys.path.insert(0, str(_TEST_DIR)) sys.path.insert(0, str(_TEST_DIR))
from util import CustomTestCase from basic_crud import ResponseAPIBaseTest
class StructuredOutputBaseTest(CustomTestCase): class StructuredOutputBaseTest(ResponseAPIBaseTest):
"""Base class for structured output tests with common utilities."""
# To be set by subclasses
base_url: str = None
api_key: str = None
model: str = None
def make_request(self, endpoint, method="GET", data=None):
"""Make HTTP request to the API."""
url = f"{self.base_url}{endpoint}"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}",
}
if method == "GET":
response = self.session.get(url, headers=headers)
elif method == "POST":
response = self.session.post(url, headers=headers, json=data)
elif method == "DELETE":
response = self.session.delete(url, headers=headers)
else:
raise ValueError(f"Unsupported method: {method}")
return response
def test_structured_output_json_schema(self): def test_structured_output_json_schema(self):
"""Test structured output with json_schema format.""" """Test structured output with json_schema format."""
# Create response with structured output # Create response with structured output
data = { params = {
"model": self.model,
"input": [ "input": [
{ {
"role": "system", "role": "system",
@@ -84,29 +58,27 @@ class StructuredOutputBaseTest(CustomTestCase):
}, },
} }
create_resp = self.make_request("/v1/responses", "POST", data) create_resp = self.create_response(**params)
self.assertEqual(create_resp.status_code, 200) self.assertIsNone(create_resp.error)
self.assertIsNotNone(create_resp.id)
create_data = create_resp.json() self.assertIsNotNone(create_resp.output)
self.assertIn("id", create_data) self.assertIsNotNone(create_resp.text)
self.assertIn("output", create_data)
self.assertIn("text", create_data)
# Verify text format was echoed back correctly # Verify text format was echoed back correctly
self.assertIn("format", create_data["text"]) self.assertIsNotNone(create_resp.text.format)
self.assertEqual(create_data["text"]["format"]["type"], "json_schema") self.assertEqual(create_resp.text.format.type, "json_schema")
self.assertEqual(create_data["text"]["format"]["name"], "math_reasoning") self.assertEqual(create_resp.text.format.name, "math_reasoning")
self.assertIn("schema", create_data["text"]["format"]) self.assertIsNotNone(create_resp.text.format.schema_)
self.assertEqual(create_data["text"]["format"]["strict"], True) self.assertEqual(create_resp.text.format.strict, True)
# Find the message output (output[0] may be reasoning, output[1] is message) # Find the message output (output[0] may be reasoning, output[1] is message)
output_text = next( output_text = next(
( (
content.get("text", "") content.text
for item in create_data.get("output", []) for item in create_resp.output
if item.get("type") == "message" if item.type == "message"
for content in item.get("content", []) for content in item.content
if content.get("type") == "output_text" if content.type == "output_text"
), ),
None, None,
) )
sgl-router/py_test/e2e_response_api/persistence/test_oracle_store.py
+3
View File
@@ -12,6 +12,8 @@ import sys
import unittest import unittest
from pathlib import Path from pathlib import Path
import openai
# Add e2e_response_api directory for imports # Add e2e_response_api directory for imports
_TEST_DIR = Path(__file__).parent.parent _TEST_DIR = Path(__file__).parent.parent
sys.path.insert(0, str(_TEST_DIR)) sys.path.insert(0, str(_TEST_DIR))
@@ -39,6 +41,7 @@ class TestOracleStore(ResponseCRUDBaseTest, ConversationCRUDBaseTest):
) )
cls.base_url = cls.cluster["base_url"] cls.base_url = cls.cluster["base_url"]
cls.client = openai.Client(api_key=cls.api_key, base_url=cls.base_url + "/v1")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):