713 lines
26 KiB
Python
713 lines
26 KiB
Python
"""
|
||
BSC·OKR·KPI 三位一体 — 集成测试
|
||
覆盖文档中缺失的边缘场景：边界值、并发、权限、归档等
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from datetime import datetime, timedelta, timezone
|
||
from tests.conftest import (
|
||
create_test_user, get_token_for_user, auth_header,
|
||
create_test_kpi, create_test_map,
|
||
)
|
||
from app.models import Objective, ActionPlan, KPIDefinition, BscLayerConfig
|
||
|
||
|
||
# ============================================================
|
||
# 测试数据工厂
|
||
# ============================================================
|
||
|
||
def create_test_objective(db: Session, **kwargs) -> Objective:
|
||
defaults = {
|
||
"title": "测试OKR目标",
|
||
"quarter": "2026Q3",
|
||
"dimension": "finance",
|
||
"owner": "测试管理员",
|
||
"status": "active",
|
||
"progress": 0,
|
||
}
|
||
defaults.update(kwargs)
|
||
obj = Objective(**defaults)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def create_test_kr(db: Session, kpi_id: int, objective_id: int, **kwargs) -> ActionPlan:
|
||
defaults = {
|
||
"kpi_id": kpi_id,
|
||
"objective_id": objective_id,
|
||
"title": "测试KR",
|
||
"assignee": "张三",
|
||
"priority": "medium",
|
||
"status": "pending",
|
||
"progress": 0,
|
||
"due_date": datetime(2026, 9, 15, tzinfo=timezone.utc), # Q3范围内
|
||
"created_by": "testadmin",
|
||
}
|
||
defaults.update(kwargs)
|
||
plan = ActionPlan(**defaults)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
# ============================================================
|
||
# TC01: 边界—KR截止日期设在过去
|
||
# ============================================================
|
||
|
||
class TestKRDeadlineBoundaries:
|
||
"""KR截止日期边界测试"""
|
||
|
||
def test_kr_due_date_in_past(self, client: TestClient, db: Session):
|
||
"""TC01: KR截止日期可以设在过去吗？"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "已过期的KR",
|
||
"kpi_id": kpi.id,
|
||
"due_date": yesterday,
|
||
},
|
||
)
|
||
# 系统应该允许（业务上允许回顾性计划），但最好有个告警
|
||
assert resp.status_code == 200, f"截止日期过去时拒绝：{resp.json()}"
|
||
data = resp.json()
|
||
assert data["due_date"] is not None
|
||
# 验证过期状态
|
||
due = datetime.fromisoformat(data["due_date"].replace("Z", "+00:00") if data["due_date"].endswith("Z") else data["due_date"])
|
||
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
||
due_naive = due.replace(tzinfo=None)
|
||
assert due_naive < now, "应存储为过去日期"
|
||
|
||
def test_kr_due_date_far_future(self, client: TestClient, db: Session):
|
||
"""TC02: KR截止日期设在10年后"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
far_future = (datetime.now(timezone.utc) + timedelta(days=3650)).isoformat()
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "超远期KR",
|
||
"kpi_id": kpi.id,
|
||
"due_date": far_future,
|
||
},
|
||
)
|
||
assert resp.status_code == 200, f"超远期日期被拒绝：{resp.json()}"
|
||
|
||
def test_kr_due_date_year_2025(self, client: TestClient, db: Session):
|
||
"""TC03: 用户手滑设了2025年的日期（已过时）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
wrong_year = "2025-01-01T00:00:00"
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "错误年份KR",
|
||
"kpi_id": kpi.id,
|
||
"due_date": wrong_year,
|
||
},
|
||
)
|
||
# 系统当前没有校验年份一致性，这可能是隐患
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
due = datetime.fromisoformat(data["due_date"])
|
||
assert due.year == 2025, "尽管不合逻辑，系统存储了错误年份"
|
||
|
||
|
||
# ============================================================
|
||
# TC02: KPI删除/禁用后，引用的KR怎么办
|
||
# ============================================================
|
||
|
||
class TestKPIReferencedByKR:
|
||
"""KPI被KR引用后的删除行为"""
|
||
|
||
def test_delete_kpi_referenced_by_kr(self, client: TestClient, db: Session):
|
||
"""TC04: 删除被KR引用的KPI"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="F_REF_001")
|
||
obj = create_test_objective(db)
|
||
kr = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id)
|
||
|
||
# 删除KPI
|
||
resp = client.delete(
|
||
f"/api/cma/kpis/{kpi.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
# KR还在吗？引用的KPI状态变了？
|
||
kr_resp = client.get(
|
||
f"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
)
|
||
assert kr_resp.status_code == 200
|
||
plans = kr_resp.json()["data"]
|
||
matching = [p for p in plans if p["id"] == kr.id]
|
||
assert len(matching) == 1, "KR应该在KPI删除后仍然存在"
|
||
assert matching[0]["kpi_id"] == kpi.id
|
||
|
||
# KPI状态变为disabled
|
||
kpi_resp = client.get(
|
||
f"/api/cma/kpis/{kpi.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert kpi_resp.status_code == 200
|
||
assert kpi_resp.json()["status"] != "active"
|
||
|
||
def test_restore_kpi_updates_kr_context(self, client: TestClient, db: Session):
|
||
"""TC05: 恢复已删除KPI后KR自动恢复"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="F_RESTORE_001")
|
||
obj = create_test_objective(db)
|
||
kr = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id)
|
||
|
||
# 删除
|
||
client.delete(f"/api/cma/kpis/{kpi.id}", headers=auth_header(token))
|
||
|
||
# 恢复
|
||
resp = client.put(
|
||
f"/api/cma/kpis/{kpi.id}/restore",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
# 验证KPI已恢复
|
||
kpi_resp = client.get(
|
||
f"/api/cma/kpis/{kpi.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert kpi_resp.status_code == 200
|
||
assert kpi_resp.json()["status"] == "active", "恢复后应该是active"
|
||
|
||
|
||
# ============================================================
|
||
# TC03: KR创建时目标值与当前值相同 / 极端进度
|
||
# ============================================================
|
||
|
||
class TestKREdgeProgress:
|
||
"""KR进度极端值测试"""
|
||
|
||
def test_kr_initial_progress_already_100(self, client: TestClient, db: Session):
|
||
"""TC06: KR一开始进度就是100%（完成了才创建？）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj = create_test_objective(db)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "已完成但才创建",
|
||
"kpi_id": kpi.id,
|
||
"objective_id": obj.id,
|
||
"progress": 100,
|
||
"status": "completed",
|
||
},
|
||
)
|
||
# 注意：create接口没有透传progress/status参数，创建时固定为0/pending
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["progress"] == 0, "创建时progress应初始化为0"
|
||
assert data["status"] == "pending", "创建时status应为pending"
|
||
|
||
def test_update_progress_boundaries(self, client: TestClient, db: Session):
|
||
"""TC07: 进度值负数/超大数被截断到0-100"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
plan = create_test_kr(db, kpi_id=kpi.id, objective_id=1)
|
||
|
||
# 负数
|
||
resp1 = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"progress": -50},
|
||
)
|
||
assert resp1.status_code == 200
|
||
assert resp1.json()["progress"] == 0, "负数应截断为0"
|
||
|
||
# 超大
|
||
resp2 = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"progress": 999},
|
||
)
|
||
assert resp2.status_code == 200
|
||
assert resp2.json()["progress"] == 100, "超100应截断为100"
|
||
|
||
# 正常值
|
||
resp3 = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"progress": 50},
|
||
)
|
||
assert resp3.status_code == 200
|
||
assert resp3.json()["progress"] == 50
|
||
|
||
def test_kr_with_zero_kpi_target(self, client: TestClient, db: Session):
|
||
"""TC08: KPI目标值为0时，KR进度计算不崩溃"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="F_ZERO_001", target_value=0)
|
||
obj = create_test_objective(db)
|
||
kr = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id)
|
||
|
||
# 验证KPI评分（内部有除零风险）
|
||
score_resp = client.get(
|
||
f"/api/cma/kpis/score?entity_id=1",
|
||
headers=auth_header(token),
|
||
)
|
||
assert score_resp.status_code == 200
|
||
scores = score_resp.json()["kpis"]
|
||
matching = [s for s in scores if s["kpi_code"] == "F_ZERO_001"]
|
||
if matching:
|
||
# 目标值为0时，评分应为None（不崩溃）
|
||
assert matching[0]["score"] is None, "目标0时评分应为空"
|
||
|
||
|
||
# ============================================================
|
||
# TC04: OKR完整生命周期
|
||
# ============================================================
|
||
|
||
class TestOKRFullLifecycle:
|
||
"""OKR从创建→添加KR→更新→归档的完整生命周期"""
|
||
|
||
def test_create_objective(self, client: TestClient, db: Session):
|
||
"""TC09: 创建OKR目标"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.post(
|
||
f"/api/cma/okr",
|
||
headers=auth_header(token),
|
||
json={"title": "2026Q3优化成本", "quarter": "2026Q3", "dimension": "finance", "owner": "任总"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["ok"] is True
|
||
assert data["title"] == "2026Q3优化成本"
|
||
assert "id" in data
|
||
|
||
def test_add_kr_to_objective(self, client: TestClient, db: Session):
|
||
"""TC10: 给OKR添加KR（改善行动计划）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CHANNEL_REBATE", kpi_name="渠补率")
|
||
obj = create_test_objective(db, title="优化成本结构")
|
||
|
||
# 截止日期必须在Q3范围内（Q3=7/1~9/30）
|
||
from datetime import timezone
|
||
due = datetime(2026, 9, 15, tzinfo=timezone.utc).isoformat()
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "渠补率从82.8%降到75%",
|
||
"kpi_id": kpi.id,
|
||
"objective_id": obj.id,
|
||
"assignee": "任总",
|
||
"priority": "high",
|
||
"due_date": due,
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
kr_data = resp.json()
|
||
assert kr_data["kpi_id"] == kpi.id
|
||
assert kr_data["status"] == "pending"
|
||
|
||
def test_get_objective_with_krs(self, client: TestClient, db: Session):
|
||
"""TC11: 查看OKR详情包含关联KR"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CHANNEL_REBATE")
|
||
obj = create_test_objective(db, title="优化成本结构")
|
||
kr = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id,
|
||
title="渠补率降到75%")
|
||
|
||
resp = client.get(
|
||
f"/api/cma/okr/{obj.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["objective"]["title"] == "优化成本结构"
|
||
assert len(data["key_results"]) >= 1
|
||
kr_found = any(k["id"] == kr.id for k in data["key_results"])
|
||
assert kr_found, "KR应出现在OKR详情中"
|
||
|
||
def test_objective_progress_from_krs(self, client: TestClient, db: Session):
|
||
"""TC12: OKR进度随KR进度自动计算"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj = create_test_objective(db)
|
||
kr1 = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id,
|
||
title="KR1", progress=80)
|
||
kr2 = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id,
|
||
title="KR2", progress=40)
|
||
|
||
# 触发progress重算
|
||
resp = client.patch(
|
||
f"/api/cma/okr/{obj.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# progress = (80 + 40) // 2 = 60
|
||
assert data["progress"] == 60, f"OKR进度应为60，实际为{data['progress']}"
|
||
|
||
def test_quarter_end_archive(self, client: TestClient, db: Session):
|
||
"""TC13: 季度结束后归档 — KR应能标记为completed/failed"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj = create_test_objective(db, quarter="2026Q3")
|
||
kr = create_test_kr(db, kpi_id=kpi.id, objective_id=obj.id,
|
||
title="Q3关键结果")
|
||
|
||
# 完成KR
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{kr.id}",
|
||
headers=auth_header(token),
|
||
json={"status": "completed", "progress": 100},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "completed"
|
||
|
||
# 完成Objective
|
||
resp2 = client.patch(
|
||
f"/api/cma/okr/{obj.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp2.status_code == 200
|
||
|
||
# 验证季度过滤
|
||
resp3 = client.get(
|
||
f"/api/cma/okr?quarter=2026Q3",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp3.status_code == 200
|
||
items = resp3.json()["items"]
|
||
obj_found = any(o["id"] == obj.id for o in items)
|
||
assert obj_found, "归档后的OKR在季度过滤中可查"
|
||
|
||
|
||
# ============================================================
|
||
# TC05: 权限测试 — 不同角色的访问控制
|
||
# ============================================================
|
||
|
||
class TestRolePermissions:
|
||
"""不同角色的权限边界"""
|
||
|
||
def test_unauthorized_access(self, client: TestClient, db: Session):
|
||
"""TC14: 未登录访问被拒绝"""
|
||
resp = client.get("/api/cma/okr")
|
||
assert resp.status_code == 403 or resp.status_code == 401
|
||
|
||
def test_business_role_cannot_create_kpi(self, client: TestClient, db: Session):
|
||
"""TC15: business角色不能创建KPI"""
|
||
create_test_user(db, username="bizuser", name="业务员", role="business")
|
||
token = get_token_for_user(client, username="bizuser")
|
||
|
||
resp = client.post(
|
||
"/api/cma/kpis",
|
||
headers=auth_header(token),
|
||
json={"kpi_code": "F_BIZ_001", "kpi_name": "业务创建", "dimension": "finance"},
|
||
)
|
||
# business角色没有写权限
|
||
assert resp.status_code == 403 or resp.status_code == 401
|
||
|
||
def test_business_can_create_action_plan(self, client: TestClient, db: Session):
|
||
"""TC16: business角色可以创建行动计划"""
|
||
create_test_user(db, username="bizuser2", name="业务员", role="business")
|
||
token = get_token_for_user(client, username="bizuser2")
|
||
kpi = create_test_kpi(db)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={"title": "业务员计划", "kpi_id": kpi.id},
|
||
)
|
||
assert resp.status_code == 200, f"business应能创建计划：{resp.json()}"
|
||
|
||
def test_it_role_can_delete_kpi(self, client: TestClient, db: Session):
|
||
"""TC17: IT角色可以删除KPI"""
|
||
create_test_user(db, username="ituser", name="管理员", role="it")
|
||
token = get_token_for_user(client, username="ituser")
|
||
kpi = create_test_kpi(db)
|
||
|
||
resp = client.delete(
|
||
f"/api/cma/kpis/{kpi.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
|
||
# ============================================================
|
||
# TC06: 搜索安全 — 特殊字符/注入
|
||
# ============================================================
|
||
|
||
class TestSearchSecurity:
|
||
"""KPI搜索安全性"""
|
||
|
||
def test_search_with_special_chars(self, client: TestClient, db: Session):
|
||
"""TC18: 搜索含特殊字符的KPI"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
# 创建带特殊字符的KPI
|
||
kpi = create_test_kpi(db, kpi_code="F_XSS_001",
|
||
kpi_name="<script>alert('xss')</script>")
|
||
|
||
# 搜索特殊字符
|
||
resp = client.get(
|
||
"/api/cma/kpis?keyword=<script>",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
# 应该正常返回，不崩溃
|
||
assert isinstance(data, list)
|
||
|
||
# 搜索空字符串
|
||
resp2 = client.get(
|
||
"/api/cma/kpis?keyword=",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp2.status_code == 200
|
||
|
||
# 搜索超长字符串
|
||
long_str = "a" * 1000
|
||
resp3 = client.get(
|
||
f"/api/cma/kpis?keyword={long_str}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp3.status_code == 200
|
||
|
||
def test_raw_sql_injection_kpi_search(self, client: TestClient, db: Session):
|
||
"""TC19: SQL注入尝试"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
injections = [
|
||
"1' OR '1'='1",
|
||
"1; DROP TABLE kpi_definitions--",
|
||
"' UNION SELECT * FROM users--",
|
||
"'; DELETE FROM action_plans; --",
|
||
]
|
||
for inj in injections:
|
||
resp = client.get(
|
||
f"/api/cma/kpis?keyword={inj}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200, f"注入'{inj}'导致异常：{resp.json()}"
|
||
|
||
|
||
# ============================================================
|
||
# TC07: 并发场景 — 快速连续操作
|
||
# ============================================================
|
||
|
||
class TestConcurrency:
|
||
"""模拟高并发操作"""
|
||
|
||
def test_rapid_create_objectives(self, client: TestClient, db: Session):
|
||
"""TC20: 快速连续创建多个OKR"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
ids = []
|
||
for i in range(10):
|
||
resp = client.post(
|
||
"/api/cma/okr",
|
||
headers=auth_header(token),
|
||
json={"title": f"并发目标{i}", "quarter": "2026Q3", "dimension": "finance"},
|
||
)
|
||
assert resp.status_code == 200, f"第{i}个创建失败：{resp.json()}"
|
||
ids.append(resp.json()["id"])
|
||
|
||
assert len(ids) == 10, "应成功创建10个OKR"
|
||
|
||
# 验证列表数
|
||
list_resp = client.get(
|
||
"/api/cma/okr?quarter=2026Q3",
|
||
headers=auth_header(token),
|
||
)
|
||
assert list_resp.status_code == 200
|
||
assert list_resp.json()["total"] == 10
|
||
|
||
def test_rapid_create_delete_kpi(self, client: TestClient, db: Session):
|
||
"""TC21: 快速创建并删除KPI"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
kpis = []
|
||
for i in range(5):
|
||
resp = client.post(
|
||
"/api/cma/kpis",
|
||
headers=auth_header(token),
|
||
json={
|
||
"kpi_code": f"F_CONC_{i:03d}",
|
||
"kpi_name": f"并发KPI_{i}",
|
||
"dimension": "finance",
|
||
"target_value": 100,
|
||
"unit": "%",
|
||
"formula": "实际值/预算值",
|
||
"data_source": "财务系统",
|
||
"data_owner": "测试管理员",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
kpis.append(resp.json()["id"])
|
||
|
||
# 全部删除
|
||
for kid in kpis:
|
||
resp = client.delete(
|
||
f"/api/cma/kpis/{kid}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
|
||
# ============================================================
|
||
# TC08: BSC四层配置集成
|
||
# ============================================================
|
||
|
||
class TestBSCLayerIntegration:
|
||
"""BSC四层与OKR/KPI集成"""
|
||
|
||
def test_bsc_layers_loaded(self, client: TestClient, db: Session):
|
||
"""TC22: BSC四层权重加载"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
# 如果数据库没有BscLayerConfig数据，返回空列表（不崩溃）
|
||
resp = client.get(
|
||
"/api/cma/bsc-layers?entity_id=1",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code in (200, 404)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
assert "layers" in data
|
||
|
||
def test_kpi_dimension_filter(self, client: TestClient, db: Session):
|
||
"""TC23: KPI按BSC维度过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
create_test_kpi(db, kpi_code="F_FIN_001", dimension="finance", kpi_name="财务KPI")
|
||
create_test_kpi(db, kpi_code="C_CUS_001", dimension="customer", kpi_name="客户KPI")
|
||
|
||
resp = client.get(
|
||
"/api/cma/kpis?dimension=finance",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert all(k["dimension"] == "finance" for k in data), "应只返回财务维度KPI"
|
||
assert any(k["kpi_code"] == "F_FIN_001" for k in data)
|
||
|
||
def test_kpi_score_by_layer(self, client: TestClient, db: Session):
|
||
"""TC24: BSC四层评分汇总"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
create_test_kpi(db, kpi_code="F_SCORE_001", kpi_name="财务指标A",
|
||
dimension="finance", target_value=100)
|
||
create_test_kpi(db, kpi_code="C_SCORE_001", kpi_name="客户指标A",
|
||
dimension="customer", target_value=100)
|
||
|
||
resp = client.get(
|
||
"/api/cma/kpis/score?entity_id=1",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "layers" in data
|
||
assert "overall" in data
|
||
|
||
|
||
# ============================================================
|
||
# TC09: KPI五档评分引擎边界
|
||
# ============================================================
|
||
|
||
class TestFiveTierScoring:
|
||
"""五档评分引擎边界测试"""
|
||
|
||
def test_reverse_indicator_scoring(self, client: TestClient, db: Session):
|
||
"""TC25: 反向指标（越低越好）评分正确"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
# 渠补率是反向指标（越低越好）
|
||
create_test_kpi(db, kpi_code="C_REBATE_RATE", kpi_name="渠补率",
|
||
dimension="customer", target_value=75.0)
|
||
|
||
resp = client.get(
|
||
"/api/cma/kpis/score?entity_id=1",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
# 当前值=None → score=None（不崩溃）
|
||
scores = resp.json()["kpis"]
|
||
matching = [s for s in scores if s["kpi_code"] == "C_REBATE_RATE"]
|
||
if matching:
|
||
assert matching[0]["score"] is None, "无实际值时评分应为空"
|
||
|
||
def test_score_with_period_filter(self, client: TestClient, db: Session):
|
||
"""TC26: 按期间过滤评分"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
create_test_kpi(db, kpi_code="F_PERIOD_001", dimension="finance", target_value=100)
|
||
|
||
# 用未来期间过滤
|
||
resp = client.get(
|
||
"/api/cma/kpis/score?entity_id=1&period=2030-Q1",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
# 应该正常返回，只是没有数据
|
||
data = resp.json()
|
||
assert data["kpis"] is not None
|
||
|
||
|
||
# ============================================================
|
||
# TC10: KPI元数据校验
|
||
# ============================================================
|
||
|
||
class TestKPIMetadataValidation:
|
||
"""KPI数据治理校验"""
|
||
|
||
def test_create_kpi_missing_metadata(self, client: TestClient, db: Session):
|
||
"""TC27: 缺少元数据被拒绝"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
# 缺少target_value和unit
|
||
resp = client.post(
|
||
"/api/cma/kpis",
|
||
headers=auth_header(token),
|
||
json={
|
||
"kpi_code": "F_META_001",
|
||
"kpi_name": "缺失元数据",
|
||
"dimension": "finance",
|
||
# 没有 target_value, unit, formula, data_source, data_owner
|
||
},
|
||
)
|
||
assert resp.status_code in (422, 400), f"应拒绝不完整的KPI：{resp.json()}"
|