add backend source code
This commit is contained in:
535
backend/tests/test_e2e.py
Normal file
535
backend/tests/test_e2e.py
Normal file
@@ -0,0 +1,535 @@
|
||||
"""
|
||||
完整闭环测试用例 - End-to-End Tests
|
||||
|
||||
测试覆盖完整的业务流程：
|
||||
1. 用户登录 -> 获取Token
|
||||
2. 科室管理 -> CRUD操作
|
||||
3. 员工管理 -> CRUD操作
|
||||
4. 考核指标管理 -> CRUD操作
|
||||
5. 考核流程 -> 创建、提交、审核、确认
|
||||
6. 满意度调查 -> 创建问卷、提交回答、统计
|
||||
7. 评分方法计算 -> 各种评分方法验证
|
||||
8. BSC维度权重 -> 配置和计算
|
||||
"""
|
||||
import pytest
|
||||
import httpx
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
# API基础URL
|
||||
BASE_URL = "http://localhost:8000/api/v1"
|
||||
|
||||
|
||||
class TestAuth:
|
||||
"""认证测试"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_success(self):
|
||||
"""测试登录成功"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login-json",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
assert "access_token" in data["data"]
|
||||
return data["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_failure(self):
|
||||
"""测试登录失败"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login-json",
|
||||
json={"username": "admin", "password": "wrong_password"}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestDepartmentManagement:
|
||||
"""科室管理闭环测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
"""获取认证Token"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login-json",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_department_list(self, auth_token):
|
||||
"""测试获取科室列表"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/departments",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
assert isinstance(data["data"], list)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_and_delete_department(self, auth_token):
|
||||
"""测试创建和删除科室"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
headers = {"Authorization": f"Bearer {auth_token}"}
|
||||
|
||||
# 创建科室
|
||||
create_response = await client.post(
|
||||
f"{BASE_URL}/departments",
|
||||
headers=headers,
|
||||
json={
|
||||
"name": "测试科室E2E",
|
||||
"code": "TEST_E2E_001",
|
||||
"dept_type": "clinical_surgical",
|
||||
"description": "E2E测试用科室"
|
||||
}
|
||||
)
|
||||
|
||||
assert create_response.status_code == 200
|
||||
created = create_response.json()
|
||||
assert created["code"] == 200
|
||||
dept_id = created["data"]["id"]
|
||||
|
||||
# 获取科室详情
|
||||
get_response = await client.get(
|
||||
f"{BASE_URL}/departments/{dept_id}",
|
||||
headers=headers
|
||||
)
|
||||
assert get_response.status_code == 200
|
||||
|
||||
# 删除科室
|
||||
delete_response = await client.delete(
|
||||
f"{BASE_URL}/departments/{dept_id}",
|
||||
headers=headers
|
||||
)
|
||||
assert delete_response.status_code == 200
|
||||
|
||||
|
||||
class TestStaffManagement:
|
||||
"""员工管理闭环测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.fixture
|
||||
async def department_id(self, auth_token):
|
||||
"""获取一个科室ID用于测试"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/departments",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
departments = response.json()["data"]
|
||||
if departments:
|
||||
return departments[0]["id"]
|
||||
return None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_staff_list(self, auth_token):
|
||||
"""测试获取员工列表"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/staff",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
|
||||
|
||||
class TestIndicatorManagement:
|
||||
"""考核指标管理闭环测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_indicator_list(self, auth_token):
|
||||
"""测试获取指标列表"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/indicators",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_indicator(self, auth_token):
|
||||
"""测试创建指标"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/indicators",
|
||||
headers={"Authorization": f"Bearer {auth_token}"},
|
||||
json={
|
||||
"name": "E2E测试指标",
|
||||
"code": "E2E_TEST_001",
|
||||
"indicator_type": "quality",
|
||||
"bs_dimension": "internal_process",
|
||||
"weight": 1.0,
|
||||
"max_score": 100,
|
||||
"target_value": 95,
|
||||
"target_unit": "%",
|
||||
"calculation_method": "测试计算方法",
|
||||
"assessment_method": "interval_high"
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
|
||||
# 清理：删除创建的指标
|
||||
indicator_id = data["data"]["id"]
|
||||
async with httpx.AsyncClient() as client:
|
||||
await client.delete(
|
||||
f"{BASE_URL}/indicators/{indicator_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
|
||||
class TestAssessmentWorkflow:
|
||||
"""考核流程闭环测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_assessment_list(self, auth_token):
|
||||
"""测试获取考核列表"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/assessments",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
|
||||
|
||||
class TestSurveyManagement:
|
||||
"""满意度调查闭环测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_survey_list(self, auth_token):
|
||||
"""测试获取问卷列表"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/surveys",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
assert isinstance(data["data"], list)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_survey_detail(self, auth_token):
|
||||
"""测试获取问卷详情"""
|
||||
# 先获取问卷列表
|
||||
async with httpx.AsyncClient() as client:
|
||||
list_response = await client.get(
|
||||
f"{BASE_URL}/surveys",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
surveys = list_response.json()["data"]
|
||||
|
||||
if surveys:
|
||||
survey_id = surveys[0]["id"]
|
||||
detail_response = await client.get(
|
||||
f"{BASE_URL}/surveys/{survey_id}",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
assert detail_response.status_code == 200
|
||||
data = detail_response.json()
|
||||
assert data["code"] == 200
|
||||
assert data["data"]["id"] == survey_id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_and_close_survey(self, auth_token):
|
||||
"""测试创建和关闭问卷"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
headers = {"Authorization": f"Bearer {auth_token}"}
|
||||
|
||||
# 创建问卷
|
||||
create_response = await client.post(
|
||||
f"{BASE_URL}/surveys",
|
||||
headers=headers,
|
||||
json={
|
||||
"survey_name": "E2E测试问卷",
|
||||
"survey_code": "E2E_TEST_SURVEY_001",
|
||||
"survey_type": "inpatient",
|
||||
"description": "E2E测试用问卷",
|
||||
"is_anonymous": True,
|
||||
"questions": [
|
||||
{
|
||||
"question_text": "您对服务是否满意？",
|
||||
"question_type": "score",
|
||||
"score_max": 5,
|
||||
"is_required": True
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert create_response.status_code == 200
|
||||
created = create_response.json()
|
||||
assert created["code"] == 200
|
||||
survey_id = created["data"]["id"]
|
||||
|
||||
# 发布问卷
|
||||
publish_response = await client.post(
|
||||
f"{BASE_URL}/surveys/{survey_id}/publish",
|
||||
headers=headers
|
||||
)
|
||||
assert publish_response.status_code == 200
|
||||
|
||||
# 关闭问卷
|
||||
close_response = await client.post(
|
||||
f"{BASE_URL}/surveys/{survey_id}/close",
|
||||
headers=headers
|
||||
)
|
||||
assert close_response.status_code == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_department_satisfaction(self, auth_token):
|
||||
"""测试获取科室满意度统计"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/surveys/stats/department",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["code"] == 200
|
||||
|
||||
|
||||
class TestScoringMethodIntegration:
|
||||
"""评分方法集成测试"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_target_reference_method(self):
|
||||
"""测试目标参照法（单元测试集成）"""
|
||||
from app.services.scoring_service import ScoringService, ScoringMethod, ScoringParams
|
||||
|
||||
params = ScoringParams(weight=12.6, target_value=15.0)
|
||||
result = ScoringService.calculate(
|
||||
ScoringMethod.TARGET_REFERENCE,
|
||||
actual_value=18.0,
|
||||
params=params
|
||||
)
|
||||
|
||||
assert result.score == 15.12 # 12.6 * (18/15)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interval_high_method(self):
|
||||
"""测试区间法-趋高指标"""
|
||||
from app.services.scoring_service import ScoringService, ScoringMethod, ScoringParams
|
||||
|
||||
params = ScoringParams(
|
||||
weight=12.6,
|
||||
best_value=100,
|
||||
baseline_value=80,
|
||||
worst_value=0
|
||||
)
|
||||
result = ScoringService.calculate(
|
||||
ScoringMethod.INTERVAL_HIGH,
|
||||
actual_value=100,
|
||||
params=params
|
||||
)
|
||||
|
||||
assert result.score == 12.6 # 达到最佳值，满分
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_interval_low_method(self):
|
||||
"""测试区间法-趋低指标"""
|
||||
from app.services.scoring_service import ScoringService, ScoringMethod, ScoringParams
|
||||
|
||||
params = ScoringParams(weight=12.6, target_value=25.0)
|
||||
result = ScoringService.calculate(
|
||||
ScoringMethod.INTERVAL_LOW,
|
||||
actual_value=30.0,
|
||||
params=params
|
||||
)
|
||||
|
||||
# 12.6 * (25/30) = 10.5
|
||||
assert result.score == 10.5
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deduction_method(self):
|
||||
"""测试扣分法"""
|
||||
from app.services.scoring_service import ScoringService, ScoringMethod, ScoringParams
|
||||
|
||||
params = ScoringParams(weight=3.2, deduction_per_unit=5.0)
|
||||
result = ScoringService.calculate(
|
||||
ScoringMethod.DEDUCTION,
|
||||
actual_value=1, # 1次违规
|
||||
params=params
|
||||
)
|
||||
|
||||
assert result.score == 0 # 3.2 - 5 = -1.8 -> 0 (不计负分)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bonus_method(self):
|
||||
"""测试加分法"""
|
||||
from app.services.scoring_service import ScoringService, ScoringMethod, ScoringParams
|
||||
|
||||
params = ScoringParams(
|
||||
weight=4.5,
|
||||
bonus_per_unit=2.0,
|
||||
max_bonus_ratio=0.5
|
||||
)
|
||||
result = ScoringService.calculate(
|
||||
ScoringMethod.BONUS,
|
||||
actual_value=2, # 2项加分
|
||||
params=params
|
||||
)
|
||||
|
||||
assert result.score == 8.5 # 4.5 + 2*2 = 8.5
|
||||
|
||||
|
||||
class TestDimensionWeight:
|
||||
"""BSC维度权重测试"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_weights(self):
|
||||
"""测试默认权重配置"""
|
||||
from app.services.dimension_weight_service import DimensionWeightService, DEFAULT_WEIGHTS
|
||||
from app.models.models import DeptType
|
||||
|
||||
# 测试手术临床科室权重
|
||||
weights = DimensionWeightService.get_dimension_weights(DeptType.CLINICAL_SURGICAL)
|
||||
assert weights["financial"] == 0.60
|
||||
assert weights["customer"] == 0.15
|
||||
assert weights["internal_process"] == 0.20
|
||||
assert weights["learning_growth"] == 0.05
|
||||
|
||||
# 测试护理单元权重
|
||||
weights = DimensionWeightService.get_dimension_weights(DeptType.NURSING)
|
||||
assert weights["financial"] == 0.20
|
||||
assert weights["internal_process"] == 0.50 # 护理单元内部流程权重最高
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_weight_sum_to_one(self):
|
||||
"""测试所有权重总和为1"""
|
||||
from app.services.dimension_weight_service import DEFAULT_WEIGHTS
|
||||
|
||||
for dept_type, weights in DEFAULT_WEIGHTS.items():
|
||||
total = (
|
||||
weights["financial"] +
|
||||
weights["customer"] +
|
||||
weights["internal_process"] +
|
||||
weights["learning_growth"]
|
||||
)
|
||||
assert abs(total - 1.0) < 0.01, f"{dept_type} 权重总和不为1"
|
||||
|
||||
|
||||
class TestStatsAPI:
|
||||
"""统计报表API测试"""
|
||||
|
||||
@pytest.fixture
|
||||
async def auth_token(self):
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={"username": "admin", "password": "admin123"}
|
||||
)
|
||||
return response.json()["data"]["access_token"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_bsc_dimension_stats(self, auth_token):
|
||||
"""测试获取BSC维度统计"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/stats/bsc-dimension",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_department_stats(self, auth_token):
|
||||
"""测试获取科室统计"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/stats/department",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_ranking_stats(self, auth_token):
|
||||
"""测试获取排名统计"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{BASE_URL}/stats/ranking",
|
||||
headers={"Authorization": f"Bearer {auth_token}"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
class TestHealthCheck:
|
||||
"""健康检查测试"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_health_endpoint(self):
|
||||
"""测试健康检查端点"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get("http://localhost:8000/health")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "healthy"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_api_docs_available(self):
|
||||
"""测试API文档可访问"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(f"{BASE_URL}/docs")
|
||||
|
||||
assert response.status_code == 200
|
||||
Reference in New Issue
Block a user