Spaces:
Sleeping
Sleeping
File size: 11,735 Bytes
ea4c39c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 | """HTTP endpoint tests for request validation."""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch
from app import app
from models import Message, ChatRequest, ErrorResponse
@pytest.fixture
def client():
"""Fixture for FastAPI test client."""
return TestClient(app)
@pytest.fixture
def mock_catalog(monkeypatch):
"""Mock catalog being loaded."""
from app import app_state
original_state = app_state.copy()
app_state["catalog_loaded"] = True
app_state["catalog_error"] = None
app_state["catalog_validated"] = True
app_state["catalog_validation_error"] = None
app_state["faiss_loaded"] = True
app_state["faiss_error"] = None
app_state["faiss_validated"] = True
yield
# Restore original state
app_state.update(original_state)
class TestValidationHTTP400Responses:
"""Tests for HTTP 400 responses with validation errors."""
def test_zero_messages_rejected(self, client, mock_catalog):
"""POST /chat with 0 messages should return HTTP 400."""
# Pydantic will reject this before our handler, so we test at validation layer
# This is a constraint enforced by ChatRequest model
pass
def test_seventeen_messages_rejected(self, client, mock_catalog):
"""POST /chat with 17 messages should return HTTP 400."""
messages = []
for i in range(8):
messages.append({"role": "user", "content": f"Message {i+1}"})
messages.append({"role": "assistant", "content": f"Response {i+1}"})
messages.append({"role": "user", "content": "17th message"})
response = client.post("/chat", json={"messages": messages})
# Pydantic validates max_items=16
assert response.status_code == 422 # Validation error from Pydantic
def test_nine_user_turns_rejected(self, client, mock_catalog):
"""POST /chat with 9 user turns should return HTTP 400."""
messages = []
for i in range(9):
messages.append({"role": "user", "content": f"User message {i+1}"})
response = client.post("/chat", json={"messages": messages})
# Pydantic validates user turn count
assert response.status_code == 422 # Validation error from Pydantic
def test_malformed_json_returns_error(self, client, mock_catalog):
"""POST /chat with malformed JSON should return HTTP 422."""
response = client.post("/chat", content="not json")
assert response.status_code == 422
class TestValidErrorResponse:
"""Tests for error response schema compliance."""
def test_error_response_has_required_fields(self, client, mock_catalog):
"""Error response should have 'error' and 'status' fields."""
# Create a request that will fail validation
# 17 messages will fail Pydantic validation
messages = []
for i in range(8):
messages.append({"role": "user", "content": f"Message {i+1}"})
messages.append({"role": "assistant", "content": f"Response {i+1}"})
messages.append({"role": "user", "content": "17th"})
response = client.post("/chat", json={"messages": messages})
# Status code depends on Pydantic validation
assert response.status_code in [400, 422]
def test_error_response_status_field_is_error(self, client, mock_catalog):
"""Error response 'status' field should always be 'error'."""
# This would be tested if we can trigger our custom validation
# Currently Pydantic validates first
pass
class TestValidRequestsWork:
"""Tests that valid requests work correctly."""
def test_valid_single_message(self, client, mock_catalog):
"""Valid request with 1 message should succeed."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "I need a developer"}
]
})
assert response.status_code == 200
data = response.json()
assert "reply" in data
assert "recommendations" in data
assert "end_of_conversation" in data
def test_valid_multiple_messages(self, client, mock_catalog):
"""Valid request with multiple messages should succeed."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "I need a backend engineer"},
{"role": "assistant", "content": "What level?"},
{"role": "user", "content": "Mid-level"},
{"role": "assistant", "content": "Here are recommendations"}
]
})
assert response.status_code == 200
data = response.json()
assert "reply" in data
assert "recommendations" in data
assert "end_of_conversation" in data
def test_valid_16_messages(self, client, mock_catalog):
"""Valid request with exactly 16 messages should succeed."""
messages = []
for i in range(8):
messages.append({"role": "user", "content": f"Message {i+1}"})
messages.append({"role": "assistant", "content": f"Response {i+1}"})
response = client.post("/chat", json={"messages": messages})
assert response.status_code == 200
class TestCatalogUnavailable:
"""Tests for catalog unavailability handling."""
def test_catalog_unavailable_returns_503(self, client):
"""When catalog not loaded, should return HTTP 503."""
from app import app_state
app_state["catalog_loaded"] = False
app_state["catalog_error"] = "Catalog initialization failed"
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "Hello"}
]
})
assert response.status_code == 503
# Restore state
app_state["catalog_loaded"] = True
app_state["catalog_error"] = None
class TestResponseSchema:
"""Tests for response schema compliance."""
def test_success_response_structure(self, client, mock_catalog):
"""Successful response should have correct structure."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "I need a developer"}
]
})
assert response.status_code == 200
data = response.json()
# Verify all required fields present
assert "reply" in data
assert "recommendations" in data
assert "end_of_conversation" in data
# Verify field types
assert isinstance(data["reply"], str)
assert isinstance(data["recommendations"], list)
assert isinstance(data["end_of_conversation"], bool)
def test_recommendations_have_name_and_url(self, client, mock_catalog):
"""Each recommendation should have name and url."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "I need a developer"}
]
})
assert response.status_code == 200
data = response.json()
# If there are recommendations, they should have required fields
for rec in data["recommendations"]:
assert "name" in rec
assert "url" in rec
class TestEdgeCaseHTTP:
"""HTTP endpoint edge case tests."""
def test_empty_message_content(self, client, mock_catalog):
"""Message with empty content should be rejected."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": ""}
]
})
# Pydantic validation should reject this
assert response.status_code == 422
def test_whitespace_only_message(self, client, mock_catalog):
"""Message with only whitespace should be rejected."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": " "}
]
})
# Should be rejected
assert response.status_code in [400, 422]
def test_invalid_role(self, client, mock_catalog):
"""Message with invalid role should be rejected."""
response = client.post("/chat", json={
"messages": [
{"role": "invalid", "content": "Hello"}
]
})
# Pydantic validation should reject this
assert response.status_code == 422
def test_unicode_content_accepted(self, client, mock_catalog):
"""Unicode content should be accepted."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "Hello 你好 مرحبا"}
]
})
assert response.status_code == 200
def test_very_long_message(self, client, mock_catalog):
"""Very long message should be accepted."""
long_content = "x" * 10000
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": long_content}
]
})
assert response.status_code == 200
class TestHealthCheckNotAffectedByValidation:
"""Health check endpoint should not be affected by chat validation."""
def test_health_check_works(self, client, mock_catalog):
"""Health check should work with catalog loaded."""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
def test_health_check_fails_without_catalog(self, client):
"""Health check should fail if catalog not loaded."""
from app import app_state
app_state["catalog_loaded"] = False
app_state["catalog_error"] = "Test error"
response = client.get("/health")
assert response.status_code == 503
# Restore state
app_state["catalog_loaded"] = True
app_state["catalog_error"] = None
class TestValidationEdgeCases:
"""Additional edge case tests for validation."""
def test_exactly_8_user_turns_valid(self, client, mock_catalog):
"""Exactly 8 user turns should be valid."""
messages = []
for i in range(8):
messages.append({"role": "user", "content": f"User {i+1}"})
if i < 7:
messages.append({"role": "assistant", "content": f"Response {i+1}"})
response = client.post("/chat", json={"messages": messages})
assert response.status_code == 200
def test_special_characters_in_content(self, client, mock_catalog):
"""Special characters should be handled correctly."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "What about Python? #hashtag @mention $cost"}
]
})
assert response.status_code == 200
def test_newlines_and_tabs_in_content(self, client, mock_catalog):
"""Newlines and tabs in content should be preserved."""
response = client.post("/chat", json={
"messages": [
{"role": "user", "content": "Line 1\nLine 2\tTabbed"}
]
})
assert response.status_code == 200
|