Spaces:
Sleeping
Sleeping
File size: 12,037 Bytes
7611990 | 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 306 307 308 309 310 311 312 313 314 315 316 317 318 | """
Regression tests for API endpoints
"""
import pytest
import json
from fastapi.testclient import TestClient
from unittest.mock import patch, AsyncMock
from httpx import AsyncClient
@pytest.mark.api
class TestHealthEndpoints:
"""Test health check endpoints"""
def test_health_check(self, client: TestClient):
"""Test basic health check endpoint"""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "timestamp" in data
assert data["service"] == "merchant-api"
assert data["version"] == "1.0.0"
@patch('app.nosql.check_mongodb_health')
@patch('app.nosql.check_redis_health')
def test_readiness_check_healthy(self, mock_redis, mock_mongo, client: TestClient):
"""Test readiness check when databases are healthy"""
mock_mongo.return_value = True
mock_redis.return_value = True
response = client.get("/ready")
assert response.status_code == 200
data = response.json()
assert data["status"] == "ready"
assert data["databases"]["mongodb"] == "healthy"
assert data["databases"]["redis"] == "healthy"
@patch('app.nosql.check_mongodb_health')
@patch('app.nosql.check_redis_health')
def test_readiness_check_unhealthy(self, mock_redis, mock_mongo, client: TestClient):
"""Test readiness check when databases are unhealthy"""
mock_mongo.return_value = False
mock_redis.return_value = True
response = client.get("/ready")
assert response.status_code == 503
data = response.json()
assert "not_ready" in data["detail"]["status"]
class TestMerchantEndpoints:
"""Test merchant-related endpoints"""
@patch('app.services.merchant.get_merchants')
def test_get_merchants_success(self, mock_get_merchants, client: TestClient, sample_merchant_data):
"""Test successful merchant retrieval"""
mock_get_merchants.return_value = [sample_merchant_data]
response = client.get("/api/v1/merchants/")
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["name"] == "Test Hair Salon"
assert data[0]["category"] == "salon"
@patch('app.services.merchant.get_merchant_by_id')
def test_get_merchant_by_id_success(self, mock_get_merchant, client: TestClient, sample_merchant_data):
"""Test successful merchant retrieval by ID"""
mock_get_merchant.return_value = sample_merchant_data
response = client.get("/api/v1/merchants/test_merchant_123")
assert response.status_code == 200
data = response.json()
assert data["_id"] == "test_merchant_123"
assert data["name"] == "Test Hair Salon"
@patch('app.services.merchant.get_merchant_by_id')
def test_get_merchant_by_id_not_found(self, mock_get_merchant, client: TestClient):
"""Test merchant not found scenario"""
mock_get_merchant.return_value = None
response = client.get("/api/v1/merchants/nonexistent_id")
assert response.status_code == 404
@patch('app.services.merchant.search_merchants')
def test_search_merchants_with_location(self, mock_search, client: TestClient, sample_merchant_data):
"""Test merchant search with location parameters"""
mock_search.return_value = [sample_merchant_data]
response = client.get("/api/v1/merchants/search", params={
"latitude": 40.7128,
"longitude": -74.0060,
"radius": 5000,
"category": "salon"
})
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["category"] == "salon"
def test_search_merchants_invalid_coordinates(self, client: TestClient):
"""Test merchant search with invalid coordinates"""
response = client.get("/api/v1/merchants/search", params={
"latitude": 200, # Invalid latitude
"longitude": -74.0060,
"radius": 5000
})
assert response.status_code == 400
class TestHelperEndpoints:
"""Test helper service endpoints"""
@patch('app.services.helper.process_free_text')
def test_process_free_text_success(self, mock_process, client: TestClient):
"""Test successful free text processing"""
mock_process.return_value = {
"query": "find a hair salon",
"extracted_keywords": ["hair", "salon"],
"suggested_category": "salon",
"search_parameters": {"category": "salon"}
}
response = client.post("/api/v1/helpers/process-text", json={
"text": "find a hair salon",
"latitude": 40.7128,
"longitude": -74.0060
})
assert response.status_code == 200
data = response.json()
assert data["suggested_category"] == "salon"
def test_process_free_text_empty_input(self, client: TestClient):
"""Test free text processing with empty input"""
response = client.post("/api/v1/helpers/process-text", json={
"text": "",
"latitude": 40.7128,
"longitude": -74.0060
})
assert response.status_code == 400
def test_process_free_text_too_long(self, client: TestClient):
"""Test free text processing with input too long"""
long_text = "a" * 1001 # Assuming 1000 char limit
response = client.post("/api/v1/helpers/process-text", json={
"text": long_text,
"latitude": 40.7128,
"longitude": -74.0060
})
assert response.status_code == 400
class TestNLPEndpoints:
"""Test NLP demo endpoints"""
@patch('app.services.advanced_nlp.advanced_nlp_pipeline')
def test_analyze_query_success(self, mock_pipeline, client: TestClient, mock_nlp_pipeline):
"""Test successful query analysis"""
mock_pipeline.process_query = mock_nlp_pipeline.process_query
response = client.post("/api/v1/nlp/analyze-query", params={
"query": "find the best hair salon near me",
"latitude": 40.7128,
"longitude": -74.0060
})
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert "analysis" in data
def test_analyze_query_empty_input(self, client: TestClient):
"""Test query analysis with empty input"""
response = client.post("/api/v1/nlp/analyze-query", params={
"query": ""
})
assert response.status_code == 400
def test_get_supported_intents(self, client: TestClient):
"""Test getting supported intents"""
response = client.get("/api/v1/nlp/supported-intents")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert "supported_intents" in data
assert "SEARCH_SERVICE" in data["supported_intents"]
assert "FILTER_QUALITY" in data["supported_intents"]
def test_get_supported_entities(self, client: TestClient):
"""Test getting supported entities"""
response = client.get("/api/v1/nlp/supported-entities")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert "supported_entities" in data
assert "services" in data["supported_entities"]
assert "amenities" in data["supported_entities"]
class TestPerformanceEndpoints:
"""Test performance monitoring endpoints"""
@patch('app.utils.performance_monitor.get_performance_report')
def test_get_performance_report(self, mock_report, client: TestClient):
"""Test performance report endpoint"""
mock_report.return_value = {
"metrics": {
"total_queries": 100,
"average_time": 0.5,
"slow_queries": []
}
}
response = client.get("/api/v1/performance/report")
assert response.status_code == 200
def test_get_metrics(self, client: TestClient):
"""Test metrics endpoint"""
response = client.get("/metrics")
# Should return metrics even if some components fail
assert response.status_code in [200, 500]
class TestSecurityEndpoints:
"""Test security-related functionality"""
def test_cors_headers(self, client: TestClient):
"""Test CORS headers are properly set"""
response = client.options("/api/v1/merchants/", headers={
"Origin": "http://localhost:3000",
"Access-Control-Request-Method": "GET"
})
# Should allow the request
assert response.status_code in [200, 204]
def test_invalid_origin_blocked(self, client: TestClient):
"""Test that invalid origins are blocked"""
response = client.get("/api/v1/merchants/", headers={
"Origin": "http://malicious-site.com"
})
# Should still work but without CORS headers for invalid origin
assert response.status_code == 200
def test_request_size_limit(self, client: TestClient):
"""Test request size limits"""
large_payload = {"data": "x" * (11 * 1024 * 1024)} # 11MB
response = client.post("/api/v1/helpers/process-text", json=large_payload)
# Should be rejected due to size limit
assert response.status_code in [413, 400]
class TestErrorHandling:
"""Test error handling across endpoints"""
def test_404_for_nonexistent_endpoint(self, client: TestClient):
"""Test 404 for non-existent endpoints"""
response = client.get("/api/v1/nonexistent")
assert response.status_code == 404
def test_405_for_wrong_method(self, client: TestClient):
"""Test 405 for wrong HTTP method"""
response = client.delete("/api/v1/merchants/")
assert response.status_code == 405
@patch('app.services.merchant.get_merchants')
def test_500_error_handling(self, mock_get_merchants, client: TestClient):
"""Test 500 error handling"""
mock_get_merchants.side_effect = Exception("Database error")
response = client.get("/api/v1/merchants/")
assert response.status_code == 500
def test_malformed_json(self, client: TestClient):
"""Test handling of malformed JSON"""
response = client.post(
"/api/v1/helpers/process-text",
data="invalid json",
headers={"Content-Type": "application/json"}
)
assert response.status_code == 422
class TestAsyncEndpoints:
"""Test async endpoint functionality"""
@pytest.mark.asyncio
async def test_async_client_health_check(self, async_client: AsyncClient):
"""Test health check with async client"""
response = await async_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
@pytest.mark.asyncio
@patch('app.services.advanced_nlp.advanced_nlp_pipeline')
async def test_async_nlp_processing(self, mock_pipeline, async_client: AsyncClient, mock_nlp_pipeline):
"""Test async NLP processing"""
mock_pipeline.process_query = mock_nlp_pipeline.process_query
response = await async_client.post("/api/v1/nlp/analyze-query", params={
"query": "find a spa"
})
assert response.status_code == 200
data = response.json()
assert data["status"] == "success" |