sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e82a6af59c68868a357a3087f4962c9b122ff06b
cma-management/backend/tests/test_bsc_okr_kpi_integration.py
T
Hermes CI Fix 61936302b2 feat(KR): KR完整修复 — krs表打通+方向符号operator+权重自由输入 
- 后端okr.py: KR读取从ActionPlan改为krs表, 新增KR CRUD API(POST/PUT/DELETE /okr/{objective_id}/krs) + 批量sync
- operator方向符号: krs表加operator/tolerance/weight/sort_order/monthly_milestones列
- progress方向感知计算(>=/>: current/target, <=/<: target/current, =: 容差), 达成→status=achieved
- 关联KPI自动继承方向(threshold_green解析: F_COST_RATIO<=18等)
- maps.py: 保存地图时自动同步objectives+krs表, JSON→krs数据迁移脚本
- 前端: 权重下拉改自由数字输入(可小数33.33) + 方向选择器(≥/≤/>/</=) + 自动平分按钮 + KPI方向继承提示
- pytest: 10个新测试(krs CRUD/方向感知/权重校验/多租户隔离) + 更新旧KR测试
- 迁移: 现有strategic_maps JSON 12条KR已写入krs表
2026-08-27 10:59:56 +08:00

724 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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, KR
# ============================================================
# 测试数据工厂
# ============================================================
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（KR完整修复2026-08-27: 从krs表读取）"""
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落krs表（KR完整修复后，ActionPlan不再是KR存储）
kr = db.query(KR).filter(KR.objective_id == obj.id).first()
if not kr:
from app.models import KR as KRModel
kr = KRModel(entity_id=1, objective_id=obj.id, title="渠补率降到75%",
metric_kpi_id=kpi.id, operator="<=", target_value=75,
weight=33, status="pending", progress=0)
db.add(kr)
db.commit()
db.refresh(kr)
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进度自动计算（KR完整修复2026-08-27: krs表）"""
create_test_user(db)
token = get_token_for_user(client)
kpi = create_test_kpi(db)
obj = create_test_objective(db)
from app.models import KR as KRModel
kr1 = KRModel(entity_id=1, objective_id=obj.id, title="KR1",
metric_kpi_id=kpi.id, progress=80, weight=50)
kr2 = KRModel(entity_id=1, objective_id=obj.id, title="KR2",
metric_kpi_id=kpi.id, progress=40, weight=50)
db.add_all([kr1, kr2])
db.commit()
# 触发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()}"
Reference in New Issue View Git Blame Copy Permalink