You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
135 lines
3.8 KiB
135 lines
3.8 KiB
#!/usr/bin/env python3
|
|
"""
|
|
测试全局异常处理功能
|
|
"""
|
|
import asyncio
|
|
import requests
|
|
from app.core.app.factory import create_development_app
|
|
from app.core.exceptions import (
|
|
AdbConnectionError, AdbProtocolError, AdbDeviceNotFoundError
|
|
)
|
|
from app.core.exceptions import (
|
|
BusinessError, ValidationError, ResourceNotFoundError, PermissionError
|
|
)
|
|
|
|
# 创建测试应用
|
|
app = create_development_app()
|
|
|
|
# 测试客户端
|
|
client = requests.Session()
|
|
base_url = "http://localhost:8000"
|
|
|
|
|
|
def test_adb_exceptions():
|
|
"""测试ADB异常处理"""
|
|
print("=== 测试ADB异常处理 ===")
|
|
|
|
# 模拟ADB连接错误
|
|
try:
|
|
raise AdbConnectionError("无法连接到ADB服务器")
|
|
except AdbConnectionError as e:
|
|
print(f"ADB连接错误: {e}")
|
|
|
|
# 模拟ADB协议错误
|
|
try:
|
|
raise AdbProtocolError("ADB协议错误")
|
|
except AdbProtocolError as e:
|
|
print(f"ADB协议错误: {e}")
|
|
|
|
# 模拟设备未找到错误
|
|
try:
|
|
raise AdbDeviceNotFoundError("设备 device_123 未找到")
|
|
except AdbDeviceNotFoundError as e:
|
|
print(f"设备未找到错误: {e}")
|
|
|
|
|
|
def test_business_exceptions():
|
|
"""测试业务异常处理"""
|
|
print("\n=== 测试业务异常处理 ===")
|
|
|
|
# 模拟数据验证错误
|
|
try:
|
|
raise ValidationError("设备序列号不能为空", field="device_serial")
|
|
except ValidationError as e:
|
|
print(f"数据验证错误: {e.message}, 字段: {e.field}")
|
|
|
|
# 模拟资源未找到错误
|
|
try:
|
|
raise ResourceNotFoundError("应用 com.example.app 未找到", resource_type="application")
|
|
except ResourceNotFoundError as e:
|
|
print(f"资源未找到错误: {e.message}, 资源类型: {e.resource_type}")
|
|
|
|
# 模拟权限错误
|
|
try:
|
|
raise PermissionError("没有权限执行此操作")
|
|
except PermissionError as e:
|
|
print(f"权限错误: {e.message}")
|
|
|
|
|
|
def test_error_response_format():
|
|
"""测试错误响应格式"""
|
|
print("\n=== 测试错误响应格式 ===")
|
|
|
|
from app.core.handlers.exception_handlers import ErrorResponse
|
|
|
|
# 创建标准错误响应
|
|
error_response = ErrorResponse(
|
|
success=False,
|
|
error_code="TEST_ERROR",
|
|
message="这是一个测试错误",
|
|
details={"test_field": "test_value"},
|
|
request_id="req-123"
|
|
)
|
|
|
|
print("标准错误响应格式:")
|
|
print(error_response.to_dict())
|
|
|
|
|
|
def test_api_endpoints():
|
|
"""测试API端点的异常处理"""
|
|
print("\n=== 测试API端点异常处理 ===")
|
|
|
|
# 注意：这需要在应用运行时执行
|
|
print("要测试API端点异常处理，请先启动应用:")
|
|
print("python run.py --env development")
|
|
print("\n然后可以测试以下端点:")
|
|
print("- GET /api/v1/adb/server/info")
|
|
print("- GET /api/v1/adb/devices")
|
|
print("- POST /api/v1/adb/devices/{device_serial}/shell")
|
|
|
|
|
|
def test_middleware():
|
|
"""测试中间件功能"""
|
|
print("\n=== 测试中间件功能 ===")
|
|
|
|
from app.core.middleware import RequestMiddleware
|
|
from fastapi import Request, Response
|
|
from starlette.types import ASGIApp
|
|
|
|
print("中间件功能包括:")
|
|
print("- 自动生成请求ID")
|
|
print("- 记录请求开始和完成时间")
|
|
print("- 添加响应头 (X-Request-ID, X-Process-Time)")
|
|
print("- 记录详细的请求日志")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("全局异常处理测试")
|
|
print("=" * 50)
|
|
|
|
# 测试各种异常类型
|
|
test_adb_exceptions()
|
|
test_business_exceptions()
|
|
test_error_response_format()
|
|
test_api_endpoints()
|
|
test_middleware()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("测试完成！")
|
|
print("\n要启动应用并测试实际API，请运行:")
|
|
print("python run.py --env development")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|