536 lines
19 KiB
Python
536 lines
19 KiB
Python
"""
|
|
完整闭环测试用例 - 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
|