Spaces:
Sleeping
Sleeping
| """ | |
| Tests for API Response module. | |
| Tests the standardized API response format, exception handlers, | |
| and helper functions. | |
| """ | |
| import pytest | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.testclient import TestClient | |
| from fastapi.exceptions import RequestValidationError | |
| from pydantic import BaseModel, Field | |
| from core.api_response import ( | |
| ErrorCode, | |
| ErrorDetail, | |
| ApiErrorResponse, | |
| ApiSuccessResponse, | |
| APIError, | |
| success_response, | |
| error_response, | |
| status_to_error_code, | |
| ) | |
| # ============================================================================= | |
| # Unit Tests - Helper Functions | |
| # ============================================================================= | |
| class TestSuccessResponse: | |
| """Test success_response helper function.""" | |
| def test_basic_success(self): | |
| """success_response returns correct format.""" | |
| result = success_response() | |
| assert result == {"success": True} | |
| def test_success_with_data(self): | |
| """success_response includes data when provided.""" | |
| data = {"job_id": "123", "status": "queued"} | |
| result = success_response(data=data) | |
| assert result["success"] is True | |
| assert result["data"] == data | |
| assert "message" not in result | |
| def test_success_with_message(self): | |
| """success_response includes message when provided.""" | |
| result = success_response(message="Job created") | |
| assert result["success"] is True | |
| assert result["message"] == "Job created" | |
| assert "data" not in result | |
| def test_success_with_data_and_message(self): | |
| """success_response includes both data and message.""" | |
| data = {"id": 1} | |
| result = success_response(data=data, message="Success!") | |
| assert result["success"] is True | |
| assert result["data"] == data | |
| assert result["message"] == "Success!" | |
| class TestErrorResponse: | |
| """Test error_response helper function.""" | |
| def test_basic_error(self): | |
| """error_response returns correct format.""" | |
| result = error_response( | |
| code=ErrorCode.NOT_FOUND, | |
| message="Job not found" | |
| ) | |
| assert result["success"] is False | |
| assert result["error"]["code"] == "NOT_FOUND" | |
| assert result["error"]["message"] == "Job not found" | |
| assert "details" not in result["error"] | |
| def test_error_with_details(self): | |
| """error_response includes details when provided.""" | |
| result = error_response( | |
| code=ErrorCode.INSUFFICIENT_CREDITS, | |
| message="Not enough credits", | |
| details={"required": 10, "available": 5} | |
| ) | |
| assert result["success"] is False | |
| assert result["error"]["code"] == "INSUFFICIENT_CREDITS" | |
| assert result["error"]["details"]["required"] == 10 | |
| assert result["error"]["details"]["available"] == 5 | |
| class TestStatusToErrorCode: | |
| """Test status_to_error_code mapping function.""" | |
| def test_401_maps_to_unauthorized(self): | |
| assert status_to_error_code(401) == ErrorCode.UNAUTHORIZED | |
| def test_402_maps_to_payment_required(self): | |
| assert status_to_error_code(402) == ErrorCode.PAYMENT_REQUIRED | |
| def test_404_maps_to_not_found(self): | |
| assert status_to_error_code(404) == ErrorCode.NOT_FOUND | |
| def test_429_maps_to_rate_limited(self): | |
| assert status_to_error_code(429) == ErrorCode.RATE_LIMITED | |
| def test_500_maps_to_server_error(self): | |
| assert status_to_error_code(500) == ErrorCode.SERVER_ERROR | |
| def test_unknown_maps_to_server_error(self): | |
| """Unknown status codes default to SERVER_ERROR.""" | |
| assert status_to_error_code(418) == ErrorCode.SERVER_ERROR | |
| # ============================================================================= | |
| # Unit Tests - APIError Exception | |
| # ============================================================================= | |
| class TestAPIError: | |
| """Test APIError custom exception.""" | |
| def test_api_error_attributes(self): | |
| """APIError stores all attributes correctly.""" | |
| error = APIError( | |
| code=ErrorCode.INSUFFICIENT_CREDITS, | |
| message="Need more credits", | |
| status_code=402, | |
| details={"needed": 10} | |
| ) | |
| assert error.code == ErrorCode.INSUFFICIENT_CREDITS | |
| assert error.message == "Need more credits" | |
| assert error.status_code == 402 | |
| assert error.details == {"needed": 10} | |
| def test_api_error_default_status(self): | |
| """APIError defaults to 400 status code.""" | |
| error = APIError( | |
| code=ErrorCode.BAD_REQUEST, | |
| message="Invalid input" | |
| ) | |
| assert error.status_code == 400 | |
| assert error.details is None | |
| def test_api_error_to_dict(self): | |
| """APIError.to_dict() returns correct format.""" | |
| error = APIError( | |
| code=ErrorCode.NOT_FOUND, | |
| message="Resource not found", | |
| details={"id": "xyz"} | |
| ) | |
| result = error.to_dict() | |
| assert result["success"] is False | |
| assert result["error"]["code"] == "NOT_FOUND" | |
| assert result["error"]["message"] == "Resource not found" | |
| assert result["error"]["details"]["id"] == "xyz" | |
| def test_api_error_is_exception(self): | |
| """APIError can be raised and caught as Exception.""" | |
| with pytest.raises(APIError) as exc_info: | |
| raise APIError(code="TEST", message="Test error") | |
| assert exc_info.value.code == "TEST" | |
| assert str(exc_info.value) == "Test error" | |
| # ============================================================================= | |
| # Unit Tests - Pydantic Models | |
| # ============================================================================= | |
| class TestPydanticModels: | |
| """Test Pydantic response models.""" | |
| def test_error_detail_model(self): | |
| """ErrorDetail model validates correctly.""" | |
| detail = ErrorDetail( | |
| code="TEST_ERROR", | |
| message="Test message", | |
| details={"key": "value"} | |
| ) | |
| assert detail.code == "TEST_ERROR" | |
| assert detail.message == "Test message" | |
| assert detail.details == {"key": "value"} | |
| def test_error_detail_optional_details(self): | |
| """ErrorDetail allows missing details.""" | |
| detail = ErrorDetail(code="TEST", message="Test") | |
| assert detail.details is None | |
| def test_api_success_response_model(self): | |
| """ApiSuccessResponse model validates correctly.""" | |
| response = ApiSuccessResponse( | |
| success=True, | |
| message="Done", | |
| data={"result": 42} | |
| ) | |
| assert response.success is True | |
| assert response.message == "Done" | |
| assert response.data == {"result": 42} | |
| def test_api_error_response_model(self): | |
| """ApiErrorResponse model validates correctly.""" | |
| response = ApiErrorResponse( | |
| success=False, | |
| error=ErrorDetail(code="ERR", message="Error occurred") | |
| ) | |
| assert response.success is False | |
| assert response.error.code == "ERR" | |
| # ============================================================================= | |
| # Integration Tests - Exception Handlers | |
| # ============================================================================= | |
| class TestExceptionHandlers: | |
| """Test FastAPI exception handlers produce correct responses.""" | |
| def client(self): | |
| """Create test client with exception handlers.""" | |
| from app import app | |
| return TestClient(app, raise_server_exceptions=False) | |
| def test_http_exception_format(self, client): | |
| """HTTPException returns standardized format.""" | |
| # Access protected route without auth | |
| response = client.get("/gemini/jobs") | |
| assert response.status_code == 401 | |
| data = response.json() | |
| assert data["success"] is False | |
| assert "error" in data | |
| assert data["error"]["code"] == "UNAUTHORIZED" | |
| assert "message" in data["error"] | |
| def test_404_error_format(self, client): | |
| """404 errors return standardized format.""" | |
| response = client.get("/nonexistent-endpoint") | |
| assert response.status_code == 404 | |
| data = response.json() | |
| assert data["success"] is False | |
| assert data["error"]["code"] == "NOT_FOUND" | |
| class TestErrorCodeConstants: | |
| """Test ErrorCode constants are defined correctly.""" | |
| def test_auth_error_codes(self): | |
| assert ErrorCode.UNAUTHORIZED == "UNAUTHORIZED" | |
| assert ErrorCode.TOKEN_EXPIRED == "TOKEN_EXPIRED" | |
| assert ErrorCode.FORBIDDEN == "FORBIDDEN" | |
| def test_payment_error_codes(self): | |
| assert ErrorCode.INSUFFICIENT_CREDITS == "INSUFFICIENT_CREDITS" | |
| assert ErrorCode.PAYMENT_REQUIRED == "PAYMENT_REQUIRED" | |
| def test_validation_error_codes(self): | |
| assert ErrorCode.VALIDATION_ERROR == "VALIDATION_ERROR" | |
| assert ErrorCode.BAD_REQUEST == "BAD_REQUEST" | |
| def test_server_error_codes(self): | |
| assert ErrorCode.SERVER_ERROR == "SERVER_ERROR" | |
| assert ErrorCode.SERVICE_UNAVAILABLE == "SERVICE_UNAVAILABLE" | |