f9f4560459
- FastAPI backend with SQLModel, Alembic migrations, AgentScope agents - Next.js 15 frontend with React 19, Tailwind, Zustand, React Flow - Multi-provider AI system (DashScope, Kling, MiniMax, Volcengine, OpenAI, etc.) - All HTTP clients migrated from sync requests to async httpx - Admin-managed API keys via environment variables - SSRF vulnerability fixed in ensure_url()
790 lines
31 KiB
Python
790 lines
31 KiB
Python
"""
|
|
Property-Based Tests for Error Handling System
|
|
|
|
This module contains property-based tests that verify correctness properties
|
|
of the error handling system across all possible inputs.
|
|
|
|
Properties tested:
|
|
- Property 4: Exception type correctness
|
|
- Property 5: Error response standardization
|
|
- Property 6: Error log completeness
|
|
- Property 7: Error response structure consistency
|
|
"""
|
|
import pytest
|
|
import logging
|
|
import json
|
|
from datetime import datetime
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from hypothesis import given, strategies as st, assume, settings
|
|
from hypothesis.strategies import composite
|
|
from fastapi import Request, FastAPI
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from src.utils.errors import (
|
|
AppException,
|
|
BusinessException,
|
|
SystemException,
|
|
ErrorCode,
|
|
InvalidParameterException,
|
|
ResourceNotFoundException,
|
|
ProjectNotFoundException,
|
|
TaskNotFoundException,
|
|
TaskTimeoutException,
|
|
TaskQueueFullException,
|
|
ModelNotFoundException,
|
|
GenerationFailedException,
|
|
StorageException,
|
|
RateLimitExceededException,
|
|
UnauthorizedException,
|
|
ForbiddenException,
|
|
ConflictException,
|
|
AssetNotFoundException,
|
|
CanvasNotFoundException,
|
|
EpisodeNotFoundException,
|
|
StoryboardNotFoundException,
|
|
ProjectCreateFailedException,
|
|
ProjectUpdateFailedException,
|
|
ProjectDeleteFailedException,
|
|
TaskExecutionFailedException,
|
|
TaskCancelledException,
|
|
ModelNotAvailableException,
|
|
QuotaExceededException,
|
|
InvalidPromptException,
|
|
ProviderErrorException,
|
|
UploadFailedException,
|
|
DownloadFailedException,
|
|
FileTooLargeException,
|
|
InvalidFileTypeException,
|
|
)
|
|
from src.middlewares.error_handler import ErrorResponse, error_handler_middleware
|
|
|
|
|
|
# ============================================================================
|
|
# Hypothesis Strategies for Generating Test Data
|
|
# ============================================================================
|
|
|
|
@composite
|
|
def error_codes(draw):
|
|
"""Generate valid error codes"""
|
|
return draw(st.sampled_from(list(ErrorCode)))
|
|
|
|
|
|
@composite
|
|
def error_messages(draw):
|
|
"""Generate error messages"""
|
|
return draw(st.text(min_size=1, max_size=200))
|
|
|
|
|
|
@composite
|
|
def error_details(draw):
|
|
"""Generate error details dictionaries"""
|
|
# Generate simple dictionaries with string keys and various value types
|
|
keys = draw(st.lists(st.text(min_size=1, max_size=20), min_size=0, max_size=5, unique=True))
|
|
values = []
|
|
for _ in keys:
|
|
value = draw(st.one_of(
|
|
st.text(max_size=100),
|
|
st.integers(),
|
|
st.floats(allow_nan=False, allow_infinity=False),
|
|
st.booleans(),
|
|
st.none()
|
|
))
|
|
values.append(value)
|
|
return dict(zip(keys, values))
|
|
|
|
|
|
@composite
|
|
def business_exception_types(draw):
|
|
"""Generate business exception classes"""
|
|
exception_classes = [
|
|
InvalidParameterException,
|
|
ResourceNotFoundException,
|
|
ProjectNotFoundException,
|
|
TaskNotFoundException,
|
|
ModelNotFoundException,
|
|
RateLimitExceededException,
|
|
UnauthorizedException,
|
|
ForbiddenException,
|
|
ConflictException,
|
|
AssetNotFoundException,
|
|
CanvasNotFoundException,
|
|
EpisodeNotFoundException,
|
|
StoryboardNotFoundException,
|
|
TaskCancelledException,
|
|
ModelNotAvailableException,
|
|
QuotaExceededException,
|
|
InvalidPromptException,
|
|
]
|
|
return draw(st.sampled_from(exception_classes))
|
|
|
|
|
|
@composite
|
|
def system_exception_types(draw):
|
|
"""Generate system exception classes"""
|
|
exception_classes = [
|
|
TaskTimeoutException,
|
|
TaskQueueFullException,
|
|
GenerationFailedException,
|
|
StorageException,
|
|
ProjectCreateFailedException,
|
|
ProjectUpdateFailedException,
|
|
ProjectDeleteFailedException,
|
|
TaskExecutionFailedException,
|
|
ProviderErrorException,
|
|
UploadFailedException,
|
|
DownloadFailedException,
|
|
]
|
|
return draw(st.sampled_from(exception_classes))
|
|
|
|
|
|
@composite
|
|
def create_business_exception(draw, exception_class):
|
|
"""Create a business exception instance with appropriate parameters"""
|
|
# Generate parameters based on exception type
|
|
if exception_class == InvalidParameterException:
|
|
field = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(field=field, reason=reason)
|
|
|
|
elif exception_class == ResourceNotFoundException:
|
|
resource_type = draw(st.text(min_size=1, max_size=50))
|
|
resource_id = draw(st.text(min_size=1, max_size=50))
|
|
return exception_class(resource_type=resource_type, resource_id=resource_id)
|
|
|
|
elif exception_class in [ProjectNotFoundException, TaskNotFoundException, ModelNotFoundException,
|
|
AssetNotFoundException, CanvasNotFoundException, EpisodeNotFoundException,
|
|
StoryboardNotFoundException, TaskCancelledException]:
|
|
id_value = draw(st.text(min_size=1, max_size=50))
|
|
# Get the parameter name from the exception class
|
|
if exception_class == ProjectNotFoundException:
|
|
return exception_class(project_id=id_value)
|
|
elif exception_class == TaskNotFoundException:
|
|
return exception_class(task_id=id_value)
|
|
elif exception_class == ModelNotFoundException:
|
|
return exception_class(model_id=id_value)
|
|
elif exception_class == AssetNotFoundException:
|
|
return exception_class(asset_id=id_value)
|
|
elif exception_class == CanvasNotFoundException:
|
|
return exception_class(canvas_id=id_value)
|
|
elif exception_class == EpisodeNotFoundException:
|
|
return exception_class(episode_id=id_value)
|
|
elif exception_class == StoryboardNotFoundException:
|
|
return exception_class(storyboard_id=id_value)
|
|
elif exception_class == TaskCancelledException:
|
|
return exception_class(task_id=id_value)
|
|
|
|
elif exception_class == RateLimitExceededException:
|
|
limit = draw(st.integers(min_value=1, max_value=10000))
|
|
window = draw(st.integers(min_value=1, max_value=3600))
|
|
return exception_class(limit=limit, window=window)
|
|
|
|
elif exception_class == UnauthorizedException:
|
|
reason = draw(st.one_of(st.none(), st.text(min_size=1, max_size=100)))
|
|
return exception_class(reason=reason)
|
|
|
|
elif exception_class == ForbiddenException:
|
|
reason = draw(st.one_of(st.none(), st.text(min_size=1, max_size=100)))
|
|
return exception_class(reason=reason)
|
|
|
|
elif exception_class == ConflictException:
|
|
resource_type = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(resource_type=resource_type, reason=reason)
|
|
|
|
elif exception_class == ModelNotAvailableException:
|
|
model_id = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.one_of(st.none(), st.text(min_size=1, max_size=100)))
|
|
return exception_class(model_id=model_id, reason=reason)
|
|
|
|
elif exception_class == QuotaExceededException:
|
|
resource = draw(st.text(min_size=1, max_size=50))
|
|
limit = draw(st.integers(min_value=1, max_value=1000000))
|
|
return exception_class(resource=resource, limit=limit)
|
|
|
|
elif exception_class == InvalidPromptException:
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(reason=reason)
|
|
|
|
# Default fallback
|
|
return exception_class(project_id="test_id")
|
|
|
|
|
|
@composite
|
|
def create_system_exception(draw, exception_class):
|
|
"""Create a system exception instance with appropriate parameters"""
|
|
if exception_class == TaskTimeoutException:
|
|
task_id = draw(st.text(min_size=1, max_size=50))
|
|
timeout = draw(st.integers(min_value=1, max_value=3600))
|
|
return exception_class(task_id=task_id, timeout=timeout)
|
|
|
|
elif exception_class == TaskQueueFullException:
|
|
queue_size = draw(st.integers(min_value=1, max_value=10000))
|
|
return exception_class(queue_size=queue_size)
|
|
|
|
elif exception_class == GenerationFailedException:
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
provider = draw(st.one_of(st.none(), st.text(min_size=1, max_size=50)))
|
|
return exception_class(reason=reason, provider=provider)
|
|
|
|
elif exception_class == StorageException:
|
|
operation = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(operation=operation, reason=reason)
|
|
|
|
elif exception_class == ProjectCreateFailedException:
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(reason=reason)
|
|
|
|
elif exception_class in [ProjectUpdateFailedException, ProjectDeleteFailedException]:
|
|
project_id = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(project_id=project_id, reason=reason)
|
|
|
|
elif exception_class == TaskExecutionFailedException:
|
|
task_id = draw(st.text(min_size=1, max_size=50))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(task_id=task_id, reason=reason)
|
|
|
|
elif exception_class == ProviderErrorException:
|
|
provider = draw(st.text(min_size=1, max_size=50))
|
|
error_message = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(provider=provider, error_message=error_message)
|
|
|
|
elif exception_class == UploadFailedException:
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(reason=reason)
|
|
|
|
elif exception_class == DownloadFailedException:
|
|
url = draw(st.text(min_size=1, max_size=100))
|
|
reason = draw(st.text(min_size=1, max_size=100))
|
|
return exception_class(url=url, reason=reason)
|
|
|
|
# Default fallback
|
|
return exception_class(reason="test reason")
|
|
|
|
|
|
# ============================================================================
|
|
# Property 4: Exception Type Correctness
|
|
# ============================================================================
|
|
|
|
class TestProperty4ExceptionTypeCorrectness:
|
|
"""
|
|
Property 4: 异常类型正确性
|
|
|
|
验证错误条件抛出正确的异常类型
|
|
Validates: Requirements 3.2
|
|
"""
|
|
|
|
@given(exc=st.one_of([
|
|
create_business_exception(InvalidParameterException),
|
|
create_business_exception(ResourceNotFoundException),
|
|
create_business_exception(ProjectNotFoundException),
|
|
create_business_exception(TaskNotFoundException),
|
|
create_business_exception(ModelNotFoundException),
|
|
create_business_exception(RateLimitExceededException),
|
|
create_business_exception(UnauthorizedException),
|
|
create_business_exception(ForbiddenException),
|
|
create_business_exception(ConflictException),
|
|
create_business_exception(AssetNotFoundException),
|
|
create_business_exception(CanvasNotFoundException),
|
|
create_business_exception(EpisodeNotFoundException),
|
|
create_business_exception(StoryboardNotFoundException),
|
|
create_business_exception(TaskCancelledException),
|
|
create_business_exception(ModelNotAvailableException),
|
|
create_business_exception(QuotaExceededException),
|
|
create_business_exception(InvalidPromptException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
def test_business_exceptions_are_business_exception_type(self, exc):
|
|
"""
|
|
Property: All business error conditions should throw BusinessException subclasses
|
|
|
|
For any business exception class, instances should be BusinessException type
|
|
"""
|
|
# Verify it's a BusinessException
|
|
assert isinstance(exc, BusinessException), \
|
|
f"{type(exc).__name__} should be a BusinessException"
|
|
|
|
# Verify it's also an AppException
|
|
assert isinstance(exc, AppException), \
|
|
f"{type(exc).__name__} should be an AppException"
|
|
|
|
# Verify status code is 4xx (except for rate limit which is 429)
|
|
if isinstance(exc, RateLimitExceededException):
|
|
assert exc.status_code == 429
|
|
elif isinstance(exc, UnauthorizedException):
|
|
assert exc.status_code == 401
|
|
elif isinstance(exc, ForbiddenException):
|
|
assert exc.status_code == 403
|
|
elif isinstance(exc, ConflictException):
|
|
assert exc.status_code == 409
|
|
else:
|
|
assert 400 <= exc.status_code < 500, \
|
|
f"{type(exc).__name__} should have 4xx status code"
|
|
|
|
@given(exc=st.one_of([
|
|
create_system_exception(TaskTimeoutException),
|
|
create_system_exception(TaskQueueFullException),
|
|
create_system_exception(GenerationFailedException),
|
|
create_system_exception(StorageException),
|
|
create_system_exception(ProjectCreateFailedException),
|
|
create_system_exception(ProjectUpdateFailedException),
|
|
create_system_exception(ProjectDeleteFailedException),
|
|
create_system_exception(TaskExecutionFailedException),
|
|
create_system_exception(ProviderErrorException),
|
|
create_system_exception(UploadFailedException),
|
|
create_system_exception(DownloadFailedException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
def test_system_exceptions_are_system_exception_type(self, exc):
|
|
"""
|
|
Property: All system error conditions should throw SystemException subclasses
|
|
|
|
For any system exception class, instances should be SystemException type
|
|
"""
|
|
# Verify it's a SystemException
|
|
assert isinstance(exc, SystemException), \
|
|
f"{type(exc).__name__} should be a SystemException"
|
|
|
|
# Verify it's also an AppException
|
|
assert isinstance(exc, AppException), \
|
|
f"{type(exc).__name__} should be an AppException"
|
|
|
|
# Verify status code is 500
|
|
assert exc.status_code == 500, \
|
|
f"{type(exc).__name__} should have 500 status code"
|
|
|
|
@given(
|
|
code=error_codes(),
|
|
message=error_messages(),
|
|
details=error_details()
|
|
)
|
|
@settings(max_examples=100, deadline=None)
|
|
def test_app_exception_preserves_error_information(self, code, message, details):
|
|
"""
|
|
Property: AppException should preserve all error information
|
|
|
|
For any error code, message, and details, the exception should store them correctly
|
|
"""
|
|
exc = AppException(code=code, message=message, details=details)
|
|
|
|
# Verify all information is preserved
|
|
assert exc.code == code
|
|
assert exc.message == message
|
|
assert exc.details == details
|
|
|
|
# Verify to_dict includes all information
|
|
exc_dict = exc.to_dict()
|
|
assert exc_dict["code"] == code.value
|
|
assert exc_dict["message"] == message
|
|
assert exc_dict["details"] == details
|
|
|
|
|
|
# ============================================================================
|
|
# Property 5: Error Response Standardization
|
|
# ============================================================================
|
|
|
|
class TestProperty5ErrorResponseStandardization:
|
|
"""
|
|
Property 5: 错误响应标准化
|
|
|
|
验证所有错误响应格式一致
|
|
Validates: Requirements 3.3
|
|
"""
|
|
|
|
@given(exc=st.one_of([
|
|
create_business_exception(InvalidParameterException),
|
|
create_business_exception(ResourceNotFoundException),
|
|
create_business_exception(ProjectNotFoundException),
|
|
create_business_exception(TaskNotFoundException),
|
|
create_business_exception(ModelNotFoundException),
|
|
create_business_exception(RateLimitExceededException),
|
|
create_business_exception(UnauthorizedException),
|
|
create_business_exception(ForbiddenException),
|
|
create_business_exception(ConflictException),
|
|
create_business_exception(AssetNotFoundException),
|
|
create_business_exception(CanvasNotFoundException),
|
|
create_business_exception(EpisodeNotFoundException),
|
|
create_business_exception(StoryboardNotFoundException),
|
|
create_business_exception(TaskCancelledException),
|
|
create_business_exception(ModelNotAvailableException),
|
|
create_business_exception(QuotaExceededException),
|
|
create_business_exception(InvalidPromptException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
async def test_business_exceptions_produce_standard_response(self, exc):
|
|
"""
|
|
Property: All business exceptions should produce standardized JSON responses
|
|
|
|
For any business exception, the error handler should convert it to standard format
|
|
"""
|
|
# Create mock request
|
|
app = FastAPI()
|
|
request = Request(scope={
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/test",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
})
|
|
request.state.request_id = "test_request_id"
|
|
request.state.timestamp = "2024-01-01T00:00:00Z"
|
|
|
|
# Create mock call_next that raises the exception
|
|
async def mock_call_next(req):
|
|
raise exc
|
|
|
|
# Call error handler middleware
|
|
response = await error_handler_middleware(request, mock_call_next)
|
|
|
|
# Verify response is JSONResponse
|
|
assert isinstance(response, JSONResponse)
|
|
|
|
# Parse response content
|
|
content = json.loads(response.body.decode())
|
|
|
|
# Verify standard format
|
|
assert "code" in content
|
|
assert "message" in content
|
|
assert "details" in content
|
|
assert "request_id" in content
|
|
assert "timestamp" in content
|
|
|
|
# Verify values match exception
|
|
assert content["code"] == (exc.code.value if isinstance(exc.code, ErrorCode) else exc.code)
|
|
assert content["message"] == exc.message
|
|
assert content["details"] == exc.details
|
|
|
|
@given(exc=st.one_of([
|
|
create_system_exception(TaskTimeoutException),
|
|
create_system_exception(TaskQueueFullException),
|
|
create_system_exception(GenerationFailedException),
|
|
create_system_exception(StorageException),
|
|
create_system_exception(ProjectCreateFailedException),
|
|
create_system_exception(ProjectUpdateFailedException),
|
|
create_system_exception(ProjectDeleteFailedException),
|
|
create_system_exception(TaskExecutionFailedException),
|
|
create_system_exception(ProviderErrorException),
|
|
create_system_exception(UploadFailedException),
|
|
create_system_exception(DownloadFailedException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
async def test_system_exceptions_produce_standard_response(self, exc):
|
|
"""
|
|
Property: All system exceptions should produce standardized JSON responses
|
|
|
|
For any system exception, the error handler should convert it to standard format
|
|
"""
|
|
# Create mock request
|
|
app = FastAPI()
|
|
request = Request(scope={
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/test",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
})
|
|
request.state.request_id = "test_request_id"
|
|
request.state.timestamp = "2024-01-01T00:00:00Z"
|
|
|
|
# Create mock call_next that raises the exception
|
|
async def mock_call_next(req):
|
|
raise exc
|
|
|
|
# Call error handler middleware
|
|
response = await error_handler_middleware(request, mock_call_next)
|
|
|
|
# Verify response is JSONResponse
|
|
assert isinstance(response, JSONResponse)
|
|
|
|
# Parse response content
|
|
content = json.loads(response.body.decode())
|
|
|
|
# Verify standard format
|
|
assert "code" in content
|
|
assert "message" in content
|
|
assert "details" in content
|
|
assert "request_id" in content
|
|
assert "timestamp" in content
|
|
|
|
# Verify values match exception
|
|
assert content["code"] == (exc.code.value if isinstance(exc.code, ErrorCode) else exc.code)
|
|
assert content["message"] == exc.message
|
|
assert content["details"] == exc.details
|
|
|
|
|
|
# ============================================================================
|
|
# Property 6: Error Log Completeness
|
|
# ============================================================================
|
|
|
|
class TestProperty6ErrorLogCompleteness:
|
|
"""
|
|
Property 6: 错误日志完整性
|
|
|
|
验证错误日志包含必要信息
|
|
Validates: Requirements 3.4
|
|
"""
|
|
|
|
@given(exc=st.one_of([
|
|
create_business_exception(InvalidParameterException),
|
|
create_business_exception(ResourceNotFoundException),
|
|
create_business_exception(ProjectNotFoundException),
|
|
create_business_exception(TaskNotFoundException),
|
|
create_business_exception(ModelNotFoundException),
|
|
create_business_exception(RateLimitExceededException),
|
|
create_business_exception(UnauthorizedException),
|
|
create_business_exception(ForbiddenException),
|
|
create_business_exception(ConflictException),
|
|
create_business_exception(AssetNotFoundException),
|
|
create_business_exception(CanvasNotFoundException),
|
|
create_business_exception(EpisodeNotFoundException),
|
|
create_business_exception(StoryboardNotFoundException),
|
|
create_business_exception(TaskCancelledException),
|
|
create_business_exception(ModelNotAvailableException),
|
|
create_business_exception(QuotaExceededException),
|
|
create_business_exception(InvalidPromptException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
async def test_business_exceptions_log_with_warning_level(self, exc):
|
|
"""
|
|
Property: Business exceptions should be logged with WARNING level
|
|
|
|
For any business exception, the log should use WARNING severity
|
|
"""
|
|
# Create mock request
|
|
request = Request(scope={
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/test",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
})
|
|
request.state.request_id = "test_request_id"
|
|
request.state.timestamp = "2024-01-01T00:00:00Z"
|
|
|
|
# Create mock call_next that raises the exception
|
|
async def mock_call_next(req):
|
|
raise exc
|
|
|
|
# Mock logger to capture log calls
|
|
with patch('src.middlewares.error_handler.logger') as mock_logger:
|
|
# Call error handler middleware
|
|
response = await error_handler_middleware(request, mock_call_next)
|
|
|
|
# Verify logger.log was called with WARNING level
|
|
mock_logger.log.assert_called_once()
|
|
call_args = mock_logger.log.call_args
|
|
|
|
# First argument should be WARNING level
|
|
assert call_args[0][0] == logging.WARNING, \
|
|
f"Business exception should log with WARNING level, got {call_args[0][0]}"
|
|
|
|
# Verify log includes necessary context
|
|
extra = call_args[1].get('extra', {})
|
|
assert 'request_id' in extra
|
|
assert 'timestamp' in extra
|
|
assert 'path' in extra
|
|
assert 'method' in extra
|
|
assert 'error_code' in extra
|
|
assert 'details' in extra
|
|
assert 'exception_type' in extra
|
|
|
|
@given(exc=st.one_of([
|
|
create_system_exception(TaskTimeoutException),
|
|
create_system_exception(TaskQueueFullException),
|
|
create_system_exception(GenerationFailedException),
|
|
create_system_exception(StorageException),
|
|
create_system_exception(ProjectCreateFailedException),
|
|
create_system_exception(ProjectUpdateFailedException),
|
|
create_system_exception(ProjectDeleteFailedException),
|
|
create_system_exception(TaskExecutionFailedException),
|
|
create_system_exception(ProviderErrorException),
|
|
create_system_exception(UploadFailedException),
|
|
create_system_exception(DownloadFailedException),
|
|
]))
|
|
@settings(max_examples=50, deadline=None)
|
|
async def test_system_exceptions_log_with_error_level(self, exc):
|
|
"""
|
|
Property: System exceptions should be logged with ERROR level
|
|
|
|
For any system exception, the log should use ERROR severity
|
|
"""
|
|
# Create mock request
|
|
request = Request(scope={
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/test",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
})
|
|
request.state.request_id = "test_request_id"
|
|
request.state.timestamp = "2024-01-01T00:00:00Z"
|
|
|
|
# Create mock call_next that raises the exception
|
|
async def mock_call_next(req):
|
|
raise exc
|
|
|
|
# Mock logger to capture log calls
|
|
with patch('src.middlewares.error_handler.logger') as mock_logger:
|
|
# Call error handler middleware
|
|
response = await error_handler_middleware(request, mock_call_next)
|
|
|
|
# Verify logger.log was called with ERROR level
|
|
mock_logger.log.assert_called_once()
|
|
call_args = mock_logger.log.call_args
|
|
|
|
# First argument should be ERROR level
|
|
assert call_args[0][0] == logging.ERROR, \
|
|
f"System exception should log with ERROR level, got {call_args[0][0]}"
|
|
|
|
# Verify log includes necessary context
|
|
extra = call_args[1].get('extra', {})
|
|
assert 'request_id' in extra
|
|
assert 'timestamp' in extra
|
|
assert 'path' in extra
|
|
assert 'method' in extra
|
|
assert 'error_code' in extra
|
|
assert 'details' in extra
|
|
assert 'exception_type' in extra
|
|
|
|
# Verify exc_info is True for system exceptions (includes stack trace)
|
|
assert call_args[1].get('exc_info') == True, \
|
|
"System exceptions should include stack trace (exc_info=True)"
|
|
|
|
|
|
# ============================================================================
|
|
# Property 7: Error Response Structure Consistency
|
|
# ============================================================================
|
|
|
|
class TestProperty7ErrorResponseStructureConsistency:
|
|
"""
|
|
Property 7: 错误响应结构一致性
|
|
|
|
验证错误响应JSON结构
|
|
Validates: Requirements 3.5
|
|
"""
|
|
|
|
@given(
|
|
code=error_codes(),
|
|
message=error_messages(),
|
|
details=error_details()
|
|
)
|
|
@settings(max_examples=100, deadline=None)
|
|
def test_error_response_has_consistent_structure(self, code, message, details):
|
|
"""
|
|
Property: All error responses should have consistent JSON structure
|
|
|
|
For any error code, message, and details, the response structure should be identical
|
|
"""
|
|
# Create ErrorResponse
|
|
error_response = ErrorResponse(
|
|
code=code.value,
|
|
message=message,
|
|
details=details,
|
|
request_id="test_request_id",
|
|
timestamp="2024-01-01T00:00:00Z"
|
|
)
|
|
|
|
# Convert to dict
|
|
response_dict = error_response.to_dict()
|
|
|
|
# Verify structure has exactly these keys
|
|
expected_keys = {"code", "message", "details", "request_id", "timestamp"}
|
|
assert set(response_dict.keys()) == expected_keys, \
|
|
f"Response should have exactly {expected_keys}, got {set(response_dict.keys())}"
|
|
|
|
# Verify types
|
|
assert isinstance(response_dict["code"], str)
|
|
assert isinstance(response_dict["message"], str)
|
|
assert isinstance(response_dict["details"], dict)
|
|
assert isinstance(response_dict["request_id"], str)
|
|
assert isinstance(response_dict["timestamp"], str)
|
|
|
|
# Verify values match input
|
|
assert response_dict["code"] == code.value
|
|
assert response_dict["message"] == message
|
|
assert response_dict["details"] == details
|
|
|
|
@given(exc=st.one_of([
|
|
# Business exceptions
|
|
create_business_exception(InvalidParameterException),
|
|
create_business_exception(ResourceNotFoundException),
|
|
create_business_exception(ProjectNotFoundException),
|
|
create_business_exception(TaskNotFoundException),
|
|
create_business_exception(ModelNotFoundException),
|
|
create_business_exception(RateLimitExceededException),
|
|
create_business_exception(UnauthorizedException),
|
|
create_business_exception(ForbiddenException),
|
|
create_business_exception(ConflictException),
|
|
create_business_exception(AssetNotFoundException),
|
|
create_business_exception(CanvasNotFoundException),
|
|
create_business_exception(EpisodeNotFoundException),
|
|
create_business_exception(StoryboardNotFoundException),
|
|
create_business_exception(TaskCancelledException),
|
|
create_business_exception(ModelNotAvailableException),
|
|
create_business_exception(QuotaExceededException),
|
|
create_business_exception(InvalidPromptException),
|
|
# System exceptions
|
|
create_system_exception(TaskTimeoutException),
|
|
create_system_exception(TaskQueueFullException),
|
|
create_system_exception(GenerationFailedException),
|
|
create_system_exception(StorageException),
|
|
create_system_exception(ProjectCreateFailedException),
|
|
create_system_exception(ProjectUpdateFailedException),
|
|
create_system_exception(ProjectDeleteFailedException),
|
|
create_system_exception(TaskExecutionFailedException),
|
|
create_system_exception(ProviderErrorException),
|
|
create_system_exception(UploadFailedException),
|
|
create_system_exception(DownloadFailedException),
|
|
]))
|
|
@settings(max_examples=100, deadline=None)
|
|
async def test_all_exceptions_produce_same_response_structure(self, exc):
|
|
"""
|
|
Property: All exception types should produce responses with identical structure
|
|
|
|
For any exception type, the response structure should be consistent
|
|
"""
|
|
# Create mock request
|
|
request = Request(scope={
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/test",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
})
|
|
request.state.request_id = "test_request_id"
|
|
request.state.timestamp = "2024-01-01T00:00:00Z"
|
|
|
|
# Create mock call_next that raises the exception
|
|
async def mock_call_next(req):
|
|
raise exc
|
|
|
|
# Call error handler middleware
|
|
response = await error_handler_middleware(request, mock_call_next)
|
|
|
|
# Parse response content
|
|
content = json.loads(response.body.decode())
|
|
|
|
# Verify structure is consistent
|
|
expected_keys = {"code", "message", "details", "request_id", "timestamp"}
|
|
assert set(content.keys()) == expected_keys, \
|
|
f"All responses should have structure {expected_keys}, got {set(content.keys())}"
|
|
|
|
# Verify types are consistent
|
|
assert isinstance(content["code"], str)
|
|
assert isinstance(content["message"], str)
|
|
assert isinstance(content["details"], dict)
|
|
assert isinstance(content["request_id"], str)
|
|
assert isinstance(content["timestamp"], str)
|
|
|
|
# Verify timestamp is ISO format
|
|
try:
|
|
datetime.fromisoformat(content["timestamp"].replace("Z", "+00:00"))
|
|
except ValueError:
|
|
pytest.fail(f"Timestamp should be ISO format, got {content['timestamp']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|