2026.5.24

【应急演练子系统】测试与质量保障：单元测试、集成测试与Bug修复全记录

一、测试策略概述

1.1 测试金字塔

                    ┌─────────────┐│  E2E测试    │  少量关键场景│  (End to End)│└──────┬──────┘│┌──────┴──────┐│  集成测试    │  API接口测试│  (Integration)│  数据库、Redis└──────┬──────┘│┌────────────┼────────────┐│   ┌───────┴───────┐   ││   │   单元测试    │   │  最大覆盖│   │   (Unit Test) │   │  业务逻辑│   └───────────────┘   │└───────────────────────┘

1.2 测试工具选型

测试类型	工具	用途
单元测试	pytest	Python测试框架
覆盖率	pytest-cov	代码覆盖率统计
API测试	FastAPI TestClient	HTTP接口测试
Mock	pytest-mock	模拟外部依赖

---

二、单元测试编写

2.1 测试项目结构

tests/
├── __init__.py
├── conftest.py          # pytest配置和fixture
├── unit/
│   ├── __init__.py
│   ├── test_user_service.py
│   ├── test_drill_plan.py
│   ├── test_security.py
│   └── test_llm_service.py
├── integration/
│   ├── __init__.py
│   ├── test_auth_api.py
│   ├── test_drill_plan_api.py
│   └── test_ai_api.py
└── fixtures/├── __init__.py└── sample_data.py

2.2 pytest配置和Fixture

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from app.core.database import Base, get_db
from app.main import app# 测试数据库
TEST_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)@pytest.fixture(scope="function")
def db():"""创建测试数据库会话"""Base.metadata.create_all(bind=engine)db = TestingSessionLocal()try:yield dbfinally:db.close()Base.metadata.drop_all(bind=engine)@pytest.fixture(scope="function")
def client(db):"""创建测试客户端"""def override_get_db():try:yield dbfinally:passapp.dependency_overrides[get_db] = override_get_dbwith TestClient(app) as test_client:yield test_clientapp.dependency_overrides.clear()@pytest.fixture
def sample_user(db):"""创建测试用户"""from app.models.user import Userfrom app.core.security import get_password_hashuser = User(username="testuser",password_hash=get_password_hash("testpass123"),real_name="测试用户",status=1)db.add(user)db.commit()db.refresh(user)return user@pytest.fixture
def auth_headers(client, sample_user):"""获取认证后的请求头"""response = client.post("/api/auth/login", json={"username": "testuser","password": "testpass123"})token = response.json()["data"]["access_token"]return {"Authorization": f"Bearer {token}"}

2.3 认证模块单元测试

# tests/unit/test_security.py
import pytest
from app.core.security import (get_password_hash,verify_password,create_access_token,decode_access_token
)
from jose import jwtclass TestPasswordHashing:"""密码哈希测试"""def test_password_hash_consistency(self):"""同一密码多次哈希结果不同（salt）"""password = "my_secure_password"hash1 = get_password_hash(password)hash2 = get_password_hash(password)# bcrypt会生成不同的saltassert hash1 != hash2# 但验证都能通过assert verify_password(password, hash1) is Trueassert verify_password(password, hash2) is Truedef test_wrong_password_rejected(self):"""错误密码被拒绝"""password = "correct_password"wrong_password = "wrong_password"hash_value = get_password_hash(password)assert verify_password(wrong_password, hash_value) is Falsedef test_empty_password(self):"""空密码处理"""hash_value = get_password_hash("")assert verify_password("", hash_value) is Trueclass TestJWTToken:"""JWT Token测试"""def test_token_creation_and_decode(self):"""Token创建和解码"""data = {"sub": "123", "username": "test"}token = create_access_token(data)# 解码验证payload = jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])assert payload["sub"] == "123"assert payload["username"] == "test"assert "exp" in payloadassert "login_time" in payloaddef test_token_expiration(self):"""Token过期测试"""from datetime import timedeltadata = {"sub": "123"}# 创建1秒过期的tokentoken = create_access_token(data, expires_delta=timedelta(seconds=1))import timetime.sleep(2)# 过期后解码应抛出异常with pytest.raises(jwt.ExpiredSignatureError):jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

2.4 CRUD模块单元测试

# tests/unit/test_drill_plan.py
import pytest
from app.crud.drill_plan import drill_plan_crud
from app.schemas.drill_plan import DrillPlanCreateclass TestDrillPlanCRUD:"""演练计划CRUD测试"""def test_create_plan(self, db):"""创建演练计划"""plan_data = DrillPlanCreate(plan_no="PLAN-TEST-001",department="安全部",project_name="消防演练",status=0)plan = drill_plan_crud.create(db=db, obj_in=plan_data)assert plan.id is not Noneassert plan.plan_no == "PLAN-TEST-001"assert plan.department == "安全部"assert plan.status == 0def test_get_plan(self, db):"""查询演练计划"""plan_data = DrillPlanCreate(plan_no="PLAN-TEST-002",department="生产部",project_name="地震演练")created_plan = drill_plan_crud.create(db=db, obj_in=plan_data)fetched_plan = drill_plan_crud.get(db=db, id=created_plan.id)assert fetched_plan is not Noneassert fetched_plan.plan_no == "PLAN-TEST-002"def test_update_plan(self, db):"""更新演练计划"""plan_data = DrillPlanCreate(plan_no="PLAN-TEST-003",department="安全部",project_name="泄漏演练")plan = drill_plan_crud.create(db=db, obj_in=plan_data)from app.schemas.drill_plan import DrillPlanUpdateupdate_data = DrillPlanUpdate(department="应急管理部",status=1)updated_plan = drill_plan_crud.update(db=db, db_obj=plan, obj_in=update_data)assert updated_plan.department == "应急管理部"assert updated_plan.status == 1def test_delete_plan(self, db):"""删除演练计划"""plan_data = DrillPlanCreate(plan_no="PLAN-TEST-004",department="安全部",project_name="测试演练")plan = drill_plan_crud.create(db=db, obj_in=plan_data)plan_id = plan.iddrill_plan_crud.remove(db=db, id=plan_id)deleted_plan = drill_plan_crud.get(db=db, id=plan_id)assert deleted_plan is Nonedef test_get_multi_with_pagination(self, db):"""分页查询"""# 创建多条记录for i in range(15):plan_data = DrillPlanCreate(plan_no=f"PLAN-TEST-{i:03d}",department="安全部",project_name=f"演练{i}")drill_plan_crud.create(db=db, obj_in=plan_data)# 第一页page1 = drill_plan_crud.get_multi(db=db, skip=0, limit=10)assert len(page1) == 10# 第二页page2 = drill_plan_crud.get_multi(db=db, skip=10, limit=10)assert len(page2) == 5

---

三、集成测试

3.1 API集成测试

# tests/integration/test_drill_plan_api.py
import pytestclass TestDrillPlanAPI:"""演练计划API集成测试"""def test_create_plan_success(self, client, auth_headers):"""创建计划成功"""response = client.post("/api/drill-plan",json={"plan_no": "PLAN-API-001","department": "安全部","project_name": "API测试演练","content": "测试内容","status": 0},headers=auth_headers)assert response.status_code == 200data = response.json()assert data["code"] == 200assert data["data"]["plan_no"] == "PLAN-API-001"def test_create_plan_without_auth(self, client):"""未认证创建计划应失败"""response = client.post("/api/drill-plan",json={"plan_no": "PLAN-API-002","department": "安全部","project_name": "测试演练"})assert response.status_code == 403def test_list_plans(self, client, auth_headers):"""分页查询计划列表"""# 先创建几条数据for i in range(3):client.post("/api/drill-plan",json={"plan_no": f"PLAN-LIST-{i}","department": "安全部","project_name": f"演练{i}"},headers=auth_headers)# 查询列表response = client.get("/api/drill-plan/list",params={"page": 1, "page_size": 10},headers=auth_headers)assert response.status_code == 200data = response.json()assert data["code"] == 200assert "items" in data["data"]assert "total" in data["data"]assert data["data"]["page"] == 1def test_get_plan_detail(self, client, auth_headers):"""获取计划详情"""# 创建计划create_response = client.post("/api/drill-plan",json={"plan_no": "PLAN-DETAIL-001","department": "安全部","project_name": "详情测试演练"},headers=auth_headers)plan_id = create_response.json()["data"]["id"]# 获取详情response = client.get(f"/api/drill-plan/{plan_id}",headers=auth_headers)assert response.status_code == 200data = response.json()assert data["data"]["id"] == plan_idassert data["data"]["plan_no"] == "PLAN-DETAIL-001"def test_update_plan(self, client, auth_headers):"""更新计划"""# 创建计划create_response = client.post("/api/drill-plan",json={"plan_no": "PLAN-UPDATE-001","department": "安全部","project_name": "更新测试演练"},headers=auth_headers)plan_id = create_response.json()["data"]["id"]# 更新response = client.put(f"/api/drill-plan/{plan_id}",json={"department": "生产部", "status": 1},headers=auth_headers)assert response.status_code == 200assert response.json()["data"]["department"] == "生产部"def test_delete_plan(self, client, auth_headers):"""删除计划"""# 创建计划create_response = client.post("/api/drill-plan",json={"plan_no": "PLAN-DELETE-001","department": "安全部","project_name": "删除测试演练"},headers=auth_headers)plan_id = create_response.json()["data"]["id"]# 删除response = client.delete(f"/api/drill-plan/{plan_id}",headers=auth_headers)assert response.status_code == 200# 确认已删除get_response = client.get(f"/api/drill-plan/{plan_id}",headers=auth_headers)assert get_response.json()["code"] == 404

3.2 认证API测试

# tests/integration/test_auth_api.pyclass TestAuthAPI:"""认证API测试"""def test_register_success(self, client):"""注册成功"""response = client.post("/api/auth/register",json={"username": "newuser","password": "password123","real_name": "新用户"})assert response.status_code == 200data = response.json()assert data["code"] == 200assert data["data"]["username"] == "newuser"def test_register_duplicate_username(self, client, sample_user):"""用户名重复注册失败"""response = client.post("/api/auth/register",json={"username": "testuser",  # 已存在"password": "password123","real_name": "另一个用户"})assert response.status_code == 200assert response.json()["code"] == 400assert "已存在" in response.json()["message"]def test_login_success(self, client, sample_user):"""登录成功"""response = client.post("/api/auth/login",json={"username": "testuser","password": "testpass123"})assert response.status_code == 200data = response.json()assert "access_token" in data["data"]assert "refresh_token" in data["data"]def test_login_wrong_password(self, client, sample_user):"""密码错误登录失败"""response = client.post("/api/auth/login",json={"username": "testuser","password": "wrong_password"})assert response.status_code == 200assert response.json()["code"] == 401def test_refresh_token(self, client, sample_user):"""刷新Token"""# 先登录login_response = client.post("/api/auth/login",json={"username": "testuser","password": "testpass123"})refresh_token = login_response.json()["data"]["refresh_token"]# 刷新Tokenresponse = client.post("/api/auth/refresh-token",params={"refresh_token": refresh_token})assert response.status_code == 200assert "access_token" in response.json()["data"]

---

四、Bug修复记录

4.1 Bug #001：任务状态流转校验缺失

严重程度: 高
发现时间: Sprint 1联调阶段
描述: 任务状态可以从"待执行"直接跳转到"已完成"，跳过了"执行中"状态

根因分析:

# 原来的更新逻辑没有状态校验
@router.put("/{task_id}")
def update_task(task_id: int, task_in: DrillTaskUpdate):# 直接更新，缺少状态流转校验db_task.status = task_in.statusdb.commit()

修复方案:

# 添加状态流转校验
VALID_TRANSITIONS = {0: [1, 3],  # 待执行 -> 执行中 或 已取消1: [2, 3],  # 执行中 -> 已完成 或 已取消2: [],       # 已完成不可变更3: []        # 已取消不可变更
}@router.put("/{task_id}")
def update_task(task_id: int, task_in: DrillTaskUpdate):db_task = drill_task_crud.get(db, task_id)if not db_task:return error_response(code=404, message="任务不存在")# 校验状态流转if task_in.status is not None:current = db_task.statusallowed = VALID_TRANSITIONS.get(current, [])if task_in.status not in allowed:return error_response(code=400,message=f"状态流转非法: {current} -> {task_in.status}")# 更新其他字段...

测试用例:

def test_task_status_transition_invalid(db, client, auth_headers):"""测试非法状态流转"""# 创建任务（状态0）response = client.post("/api/drill-task", ...)task_id = response.json()["data"]["id"]# 尝试直接跳转到已完成（非法）response = client.put(f"/api/drill-task/{task_id}",json={"status": 2},  # 从0跳到2，非法headers=auth_headers)assert response.status_code == 200assert response.json()["code"] == 400assert "非法" in response.json()["message"]

4.2 Bug #002：分页参数边界问题

严重程度: 中
发现时间: Sprint 1测试阶段
描述: 当page_size为0或负数时，系统报错

修复方案:

@router.get("/list")
def list_drill_plans(page: int = Query(1, ge=1, description="页码"),      # ge=1 保证最小为1page_size: int = Query(10, ge=1, le=100, description="每页条数"),  # ge=1 le=100
):skip = (page - 1) * page_size...

4.3 Bug #003：AI问答无API Key时崩溃

严重程度: 高
发现时间: Sprint 2测试阶段
描述: 当LLM_API_KEY未配置时，调用AI问答接口直接抛出500错误

根因分析: LLM服务初始化时没有处理API Key为空的情况

修复方案:

def _get_llm(self) -> Optional[ChatOpenAI]:if not self._initialized:api_key = settings.llm_api_keyif not api_key or api_key.strip() == "":# API Key为空，返回None，由调用方处理self._initialized = Truereturn Noneself._llm = ChatOpenAI(...)self._initialized = Truereturn self._llmdef answer_with_knowledge(self, question: str, context: str):llm = self._get_llm()if llm is None:# 降级处理：返回后备回答return {"answer": self._fallback_answer(question, context),"source_type": "knowledge_base","is_knowledge_based": True}# 正常流程...

4.4 Bug #004：文件上传大小时机读取问题

严重程度: 低
发现时间: Sprint 2测试阶段
描述: 大文件上传时，由于先读取整个文件到内存导致内存溢出

修复方案: 使用流式读取

# 原来的实现
content = await file.read()  # 一次性读取全部内容
if len(content) > MAX_SIZE:raise BusinessException("文件过大")with open(save_path, "wb") as f:f.write(content)# 修复后：流式读取
file_size = 0
with open(save_path, "wb") as f:while chunk := file.file.read(8192):  # 分块读取file_size += len(chunk)if file_size > MAX_SIZE:os.remove(save_path)  # 删除已写入的部分raise BusinessException("文件超过50MB限制")f.write(chunk)

---

五、测试覆盖率报告

5.1 当前覆盖率

模块	语句覆盖	分支覆盖	行数
CRUD层	92%	85%	450
Service层	78%	70%	380
API层	85%	75%	280
Security	95%	88%	120
总计	86%	79%	1230

5.2 测试运行命令

# 运行所有测试
pytest tests/ -v# 运行单元测试
pytest tests/unit/ -v# 运行集成测试
pytest tests/integration/ -v# 生成覆盖率报告
pytest tests/ --cov=app --cov-report=html --cov-report=term# 查看HTML报告
open htmlcov/index.html

---

六、质量保障总结

6.1 测试流程

开发阶段│├── 编写单元测试（同步）│▼
代码提交│├── 运行单元测试├── 运行集成测试├── 代码覆盖率检查│▼
代码审查│├── 代码风格检查├── 安全审查└── 逻辑审查│▼
合并到主分支

6.2 质量目标达成

指标	目标	实际	状态
代码覆盖率	80%	86%	达成
关键路径测试	100%	100%	达成
高优先级Bug修复	100%	100%	达成
中优先级Bug修复	90%	95%	达成
测试通过率	95%	98%	达成