jpmschweitzerandClaude
5b67f5b66c
The 21 the automatic pass could not make on its own. `ruff check` and `ruff format --check` are both clean now; typecheck is still red and is next. `in_reasoning` in chat/service.py was a complete state machine that nothing read: initialised False, set True when a reasoning delta arrived, set False when the summary ended — three assignments, zero reads. Ruff reported one at a time, and removing each revealed the next, so what looked like a single stray variable took three passes to bottom out. The branches themselves do real work and are untouched; only the flag is gone. Four `raise HTTPException` inside `except` blocks now chain with `from e`. Until now a failure while handling an error was indistinguishable from the error, which matters most in exactly the situation where the traceback is all you have. In biographer/tools.py the binding was unused but the call is not: MemoryType() is called for the ValueError it raises on an invalid name. The binding is gone and the call and its comment stay, because dropping the line would have removed the validation. The rest are unused bindings in tests where the assertions are on something else (call_args, mostly), plus three unused loop variables and an isinstance tuple. One correction to my own work: removing a dead comprehension in test_error_handling.py left an `if` block with nothing but comments in it, which is a SyntaxError. Ruff caught it immediately. The block now says what the test actually pins — that the stream parses without crashing, which reaching that line demonstrates — rather than computing a list nobody asserts on. `make test` is intermittent here, and it is not this change. test_tatlock_tool_call_logging_calculator failed in two of five full runs across both HEAD and this branch, and passes in the other three; it also fails in isolation at HEAD while passing in isolation here. Order- or timing-dependent. Recorded rather than chased, since tests are not gated in this repo yet. Co-Authored-By: Claude <noreply@anthropic.com>
256 lines
7.9 KiB
Python
256 lines
7.9 KiB
Python
"""
|
|
Tests for error handling in Responses API.
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from httpx import AsyncClient
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_model_not_found_error(client: TestClient) -> None:
|
|
"""Test response when model doesn't exist."""
|
|
request_data = {
|
|
"model": "nonexistent-model-12345",
|
|
"input": [{"role": "user", "content": "Hello"}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "detail" in data
|
|
assert "not found" in data["detail"].lower()
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_validation_error_missing_model(client: TestClient) -> None:
|
|
"""Test validation error when model field is missing."""
|
|
request_data = {"input": [{"role": "user", "content": "Hello"}]}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_validation_error_invalid_temperature(client: TestClient) -> None:
|
|
"""Test validation error for out-of-range temperature."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "Hello"}],
|
|
"temperature": 3.0, # Max is 2.0
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_rate_limit_error(client: TestClient) -> None:
|
|
"""Test rate limit error trigger."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "trigger_rate_limit"}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
# Should get an error response
|
|
assert response.status_code in [429, 500] # Rate limit or internal error
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_context_overflow_error(client: TestClient) -> None:
|
|
"""Test context length overflow error trigger."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "trigger_context_overflow"}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
# Should get an error response
|
|
assert response.status_code in [400, 500] # Bad request or internal error
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_streaming_model_not_found(async_client: AsyncClient) -> None:
|
|
"""Test streaming response with nonexistent model."""
|
|
request_data = {
|
|
"model": "nonexistent-streaming-model",
|
|
"input": [{"role": "user", "content": "Hello"}],
|
|
"stream": True,
|
|
}
|
|
|
|
async with async_client.stream(
|
|
"POST",
|
|
"/v1/responses",
|
|
json=request_data,
|
|
timeout=20.0,
|
|
) as response:
|
|
# Streaming always returns 200 OK, errors are sent as events
|
|
assert response.status_code == 200
|
|
|
|
# Collect events and look for error event
|
|
error_found = False
|
|
async for line in response.aiter_lines():
|
|
if not line.strip():
|
|
continue
|
|
|
|
if line.startswith("event: "):
|
|
event_type = line[7:].strip()
|
|
if event_type == "error":
|
|
error_found = True
|
|
break
|
|
|
|
# Should have received an error event
|
|
assert error_found, "Expected error event in stream"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_streaming_rate_limit_error(async_client: AsyncClient) -> None:
|
|
"""Test streaming with rate limit error."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "trigger_rate_limit"}],
|
|
"stream": True,
|
|
}
|
|
|
|
async with async_client.stream(
|
|
"POST",
|
|
"/v1/responses",
|
|
json=request_data,
|
|
timeout=20.0,
|
|
) as response:
|
|
# May get error status or error event
|
|
# Collect all events
|
|
events = []
|
|
async for line in response.aiter_lines():
|
|
if not line.strip():
|
|
continue
|
|
|
|
if line.startswith("event: "):
|
|
event_type = line[7:].strip()
|
|
elif line.startswith("data: "):
|
|
data_str = line[6:].strip()
|
|
try:
|
|
data = json.loads(data_str)
|
|
if event_type:
|
|
events.append({"event": event_type, "data": data})
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Should either have error status or error event
|
|
if response.status_code == 200:
|
|
# An error event may or may not appear, depending on where the failure
|
|
# occurs. What this test pins is that the stream parses and does not
|
|
# crash; arriving here is that assertion.
|
|
pass
|
|
# May or may not have error event depending on where error occurs
|
|
# At minimum, should not crash
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_streaming_context_overflow_error(async_client: AsyncClient) -> None:
|
|
"""Test streaming with context overflow error."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "trigger_context_overflow"}],
|
|
"stream": True,
|
|
}
|
|
|
|
async with async_client.stream(
|
|
"POST",
|
|
"/v1/responses",
|
|
json=request_data,
|
|
timeout=20.0,
|
|
) as response:
|
|
# May get error status or error event
|
|
events = []
|
|
async for line in response.aiter_lines():
|
|
if not line.strip():
|
|
continue
|
|
|
|
if line.startswith("event: "):
|
|
event_type = line[7:].strip()
|
|
elif line.startswith("data: "):
|
|
data_str = line[6:].strip()
|
|
try:
|
|
data = json.loads(data_str)
|
|
if event_type:
|
|
events.append({"event": event_type, "data": data})
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Should not crash
|
|
assert response.status_code in [200, 400, 500]
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_function_call_output_item(client: TestClient) -> None:
|
|
"""Test response with function call items."""
|
|
request_data = {
|
|
"model": "lorem-tester",
|
|
"input": [{"role": "user", "content": "Use the search tool"}],
|
|
"tools": [{"name": "search", "description": "Search for information", "parameters": {}}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# May or may not have function_call items (random in lorem-tester)
|
|
# But should successfully handle them if present
|
|
output_items = data["output"]
|
|
for item in output_items:
|
|
assert item["type"] in ["message", "reasoning", "function_call"]
|
|
if item["type"] == "function_call":
|
|
assert "name" in item
|
|
assert "arguments" in item
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_pipeline_prefix_stripping(client: TestClient) -> None:
|
|
"""Test that pipeline prefixes are stripped from model names."""
|
|
request_data = {
|
|
"model": "some_pipeline.lorem-tester",
|
|
"input": [{"role": "user", "content": "Hello"}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
# Should successfully strip prefix and find model
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_multiple_pipeline_prefixes(client: TestClient) -> None:
|
|
"""Test multiple dots in model name (only first is prefix)."""
|
|
request_data = {
|
|
"model": "pipeline.sub.lorem-tester",
|
|
"input": [{"role": "user", "content": "Hello"}],
|
|
"stream": False,
|
|
}
|
|
|
|
response = client.post("/v1/responses", json=request_data)
|
|
|
|
# Should strip only first part
|
|
# "pipeline.sub.lorem-tester" -> "sub.lorem-tester"
|
|
# This should fail since "sub.lorem-tester" doesn't exist
|
|
assert response.status_code == 404
|