跳到主要内容

邮轮穿舱件管理系统后台 - API接口和系统集成测试方法

文档概述

本文档详细记录了邮轮穿舱件管理系统后台的API接口测试方法和系统集成测试策略。基于对代码库的深入分析,提供了全面的测试框架、测试用例设计和集成测试方案。

系统架构概述

技术栈

  • 后端框架: FastAPI
  • 数据库ORM: Tortoise-ORM
  • 认证机制: JWT Token + OAuth2
  • 测试框架: Pytest + FastAPI TestClient
  • 代码覆盖: Pytest-cov

核心组件架构

flowchart TD
A[FastAPI应用] --> B[认证授权模块]
A --> C[业务路由模块]
A --> D[数据服务层]
A --> E[数据模型层]

B --> B1[JWT Token验证]
B --> B2[权限范围检查]

C --> C1[用户管理路由]
C --> C2[工件管理路由]
C --> C3[工单管理路由]
C --> C4[图片管理路由]
C --> C5[微信小程序路由]

D --> D1[用户服务]
D --> D2[工件服务]
D --> D3[图片服务]

E --> E1[用户模型]
E --> E2[工件模型]
E --> E3[图片模型]

测试环境配置

测试配置文件

参考文件: pytest.ini

[pytest]
addopts = --cov=app/ --cov-report=term-missing
asyncio_mode = auto
env =
ENV=test
MPLBACKEND=Agg
DISPLAY=

测试数据库配置

参考文件: test/conftest.py

@pytest.fixture(scope="session", autouse=True)
async def setup_database() -> Generator[None, None, None]:
"""设置测试数据库"""
await Tortoise.init(config=settings.tortoise_orm_config_dict)
await Tortoise.generate_schemas()
yield
await Tortoise.close_connections()

API接口测试方法

1. 用户管理API测试

用户创建和认证测试

参考文件: test/test_user_router.py

def test_user_create_and_list_user(client: TestClient, get_admin_token) -> None:
client.headers.update({
"Authorization": f"Bearer {get_admin_token}"
})
response = client.post("/users/", json={
"username": "testuser",
"password": "123456",
"email": "testuser@example.com",
"sms": "13800000000"
})
assert response.status_code == status.HTTP_201_CREATED

用户认证流程测试

sequenceDiagram
participant T as 测试客户端
participant A as 认证API
participant S as 用户服务
participant D as 数据库

T->>A: POST /token (用户名密码)
A->>S: 验证用户凭据
S->>D: 查询用户信息
D-->>S: 返回用户数据
S-->>A: 返回用户对象
A->>A: 生成JWT Token
A-->>T: 返回访问令牌

2. 权限和范围测试

权限依赖关系

参考文件: app/routers/user_router.py

user_router = APIRouter(
prefix="/users",
tags=["用户管理"],
dependencies=[Depends(require_scopes(["system:read"], mode="AND")), Depends(verify_jwt_token)],
)

权限测试用例设计

def test_permission_validation():
# 测试无权限访问
response = client.get("/users/", headers={"Authorization": "Bearer invalid_token"})
assert response.status_code == 401

# 测试权限不足
response = client.get("/users/", headers={"Authorization": f"Bearer {user_token}"})
assert response.status_code == 403

3. 错误处理测试

全局异常处理测试

参考文件: test/test_global_error_handling.py

def test_value_error_handling(client: TestClient) -> None:
response = client.get("/error-test/value-error")
assert response.status_code == status.HTTP_400_BAD_REQUEST
data = response.json()
assert data["error"] == "Bad Request"

错误处理架构

flowchart TD
A[API请求] --> B{是否异常?}
B -->|是| C[异常捕获]
B -->|否| D[正常处理]

C --> C1[HTTPException]
C --> C2[ValueError]
C --> C3[PermissionError]
C --> C4[其他异常]

C1 --> E1[返回对应状态码]
C2 --> E2[返回400错误]
C3 --> E3[返回403错误]
C4 --> E4[返回500错误]

E1 --> F[统一错误响应格式]
E2 --> F
E3 --> F
E4 --> F

系统集成测试策略

1. 数据库集成测试

测试数据库生命周期管理

@pytest.fixture(scope="module")
async def test_database_transaction():
"""测试数据库事务完整性"""
async with database.transaction():
# 执行测试操作
yield
# 自动回滚事务

2. 认证系统集成测试

JWT Token验证流程

参考文件: app/routers/login.py

def test_jwt_token_flow():
# 获取Token
token_response = client.post("/token", data={
"username": "admin",
"password": "123456"
})

# 使用Token访问受保护端点
users_response = client.get("/users/", headers={
"Authorization": f"Bearer {token_response.json()['access_token']}"
})

3. 文件上传和图片处理集成测试

图片管理API测试

def test_image_upload_and_processing():
# 上传图片
with open("test_image.jpg", "rb") as f:
upload_response = client.post("/images/", files={"file": f})

# 验证图片处理
processing_response = client.post(f"/images/{image_id}/process")

测试用例设计模式

1. CRUD操作测试模板

class CRUDTestTemplate:
def test_create_entity(self):
"""测试创建实体"""
pass

def test_read_entity(self):
"""测试读取实体"""
pass

def test_update_entity(self):
"""测试更新实体"""
pass

def test_delete_entity(self):
"""测试删除实体"""
pass

2. 权限测试模板

class PermissionTestTemplate:
def test_public_access(self):
"""测试公开端点访问"""
pass

def test_authenticated_access(self):
"""测试认证用户访问"""
pass

def test_authorized_access(self):
"""测试授权用户访问"""
pass

性能测试方法

1. API响应时间测试

import time

def test_api_response_time():
start_time = time.time()
response = client.get("/users/")
end_time = time.time()

assert response.status_code == 200
assert (end_time - start_time) < 1.0 # 响应时间小于1秒

2. 并发访问测试

import asyncio

async def test_concurrent_access():
tasks = []
for i in range(10):
task = asyncio.create_task(client.get("/users/"))
tasks.append(task)

responses = await asyncio.gather(*tasks)
for response in responses:
assert response.status_code == 200

安全测试方法

1. SQL注入防护测试

def test_sql_injection_protection():
# 测试SQL注入尝试
malicious_input = "'; DROP TABLE users; --"
response = client.get(f"/users/?username={malicious_input}")
assert response.status_code != 500 # 不应该导致服务器错误

2. XSS攻击防护测试

def test_xss_protection():
xss_payload = "<script>alert('XSS')</script>"
response = client.post("/users/", json={
"username": xss_payload,
"password": "123456"
})
# 验证响应中不包含原始脚本标签
assert "<script>" not in response.text

持续集成测试流程

1. 测试执行流程

flowchart LR
A[代码提交] --> B[运行单元测试]
B --> C[运行集成测试]
C --> D[生成测试报告]
D --> E[代码覆盖率分析]
E --> F[测试结果反馈]

2. 测试覆盖率目标

  • 行覆盖率: ≥ 80%
  • 分支覆盖率: ≥ 75%
  • 函数覆盖率: ≥ 85%

测试数据管理

1. 测试数据夹具

@pytest.fixture
def test_user_data():
return {
"username": "test_user",
"password": "test_password",
"email": "test@example.com",
"sms": "13800000000"
}

2. 测试数据清理

@pytest.fixture(autouse=True)
async def cleanup_test_data():
yield
# 测试完成后清理数据
await cleanup_test_users()

索引

本文档提供了邮轮穿舱件管理系统后台的完整API接口和系统集成测试方法。通过结合单元测试、集成测试和端到端测试,确保系统的可靠性、安全性和性能。测试策略覆盖了认证授权、数据操作、错误处理、安全防护等关键方面,为系统的质量保障提供了坚实基础。

关键测试文件参考: