Hermes CI Fix
6b479bfe7d
- action_plans表加kr_id字段(关联krs.id)，create/update/list支持kr_id读写
- 创建行动选KR时自动继承objective_id；列表附kr_title
- 前端行动方案库加'所属目标/KR'两级联动选择，选中KR自动带出KPI
- OKR详情页KR卡片下显示达成行动列表(O→KR→行动三层)
- MapCanvasDialogs onKpiChange选中KPI自动填kr.name=kpi.kpi_name(中文)
- 编辑回填英文code KR名转中文显示兜底
- maps.py 新增GET /{map_id}/krs端点(MapReview战略回顾会KR进度)
- 数据修复: krs表9条英文code title→中文名(按kpi_definitions映射)
- 数据迁移: 高置信行动挂KR(费用率→kr10,应收→kr14); 地图32空KR清理
- okr list的kr_summary补metric_kpi_id(前端联动用)
- 新增6个kr_id测试(test_action_plans)，全量547通过
342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""
|
||
改善行动计划模块测试
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from datetime import datetime, timedelta
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
from app.models import ActionPlan, Objective, KR
|
||
|
||
|
||
def create_test_objective_kr(db: Session, **kwargs):
|
||
"""创建测试目标+KR（行动挂KR 2026-08-27）"""
|
||
obj = Objective(
|
||
entity_id=kwargs.get("entity_id", 1),
|
||
title=kwargs.get("objective_title", "测试目标"),
|
||
quarter="2026Q3",
|
||
status="active",
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
kr = KR(
|
||
entity_id=kwargs.get("entity_id", 1),
|
||
objective_id=obj.id,
|
||
title=kwargs.get("kr_title", "测试KR"),
|
||
operator=">=",
|
||
weight=50,
|
||
target_value=100,
|
||
)
|
||
db.add(kr)
|
||
db.commit()
|
||
db.refresh(kr)
|
||
return obj, kr
|
||
|
||
|
||
def create_test_plan(db: Session, **kwargs) -> ActionPlan:
|
||
"""创建测试行动计划"""
|
||
defaults = {
|
||
"kpi_id": 1,
|
||
"title": "测试改善计划",
|
||
"assignee": "张三",
|
||
"priority": "medium",
|
||
"status": "pending",
|
||
"progress": 0,
|
||
"created_by": "testadmin",
|
||
}
|
||
defaults.update(kwargs)
|
||
plan = ActionPlan(**defaults)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
class TestActionPlans:
|
||
"""改善行动计划CRUD测试"""
|
||
|
||
def test_list_empty(self, client: TestClient, db: Session):
|
||
"""空列表"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["data"] == []
|
||
|
||
def test_create_plan(self, client: TestClient, db: Session):
|
||
"""创建行动计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "提升营收10%",
|
||
"kpi_id": kpi.id,
|
||
"assignee": "张三",
|
||
"priority": "high",
|
||
"description": "通过增加销售渠道提升营收",
|
||
"due_date": (datetime.now() + timedelta(days=30)).isoformat(),
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["title"] == "提升营收10%"
|
||
assert data["kpi_id"] == kpi.id
|
||
assert data["priority"] == "high"
|
||
assert data["status"] == "pending"
|
||
assert data["progress"] == 0
|
||
|
||
def test_create_plan_missing_required(self, client: TestClient, db: Session):
|
||
"""缺少必填字段"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={"title": "不完整计划"},
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "缺少" in resp.json()["detail"]
|
||
|
||
def test_list_with_data(self, client: TestClient, db: Session):
|
||
"""有数据时列表正常"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
create_test_plan(db, kpi_id=kpi.id, title="计划A")
|
||
create_test_plan(db, kpi_id=kpi.id, title="计划B")
|
||
|
||
resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data["data"]) == 2
|
||
|
||
def test_filter_by_status(self, client: TestClient, db: Session):
|
||
"""按状态过滤"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
create_test_plan(db, kpi_id=kpi.id, title="待开始", status="pending")
|
||
create_test_plan(db, kpi_id=kpi.id, title="进行中", status="in_progress")
|
||
create_test_plan(db, kpi_id=kpi.id, title="已完成", status="completed")
|
||
|
||
resp = client.get("/api/cma/action-plans?status=in_progress", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data["data"]) == 1
|
||
assert data["data"][0]["title"] == "进行中"
|
||
|
||
def test_filter_by_keyword(self, client: TestClient, db: Session):
|
||
"""关键词搜索：keyword 应过滤 title（期望行为，2026-08-27 TDD示范）"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
create_test_plan(db, kpi_id=kpi.id, title="提升营收计划")
|
||
create_test_plan(db, kpi_id=kpi.id, title="降低成本计划")
|
||
|
||
resp = client.get("/api/cma/action-plans?keyword=营收", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
# 期望行为：keyword 过滤生效，只返回标题含"营收"的1条
|
||
assert len(data["data"]) == 1
|
||
assert data["data"][0]["title"] == "提升营收计划"
|
||
|
||
def test_update_plan(self, client: TestClient, db: Session):
|
||
"""更新行动计划"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
plan = create_test_plan(db, kpi_id=kpi.id)
|
||
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "已修改的计划",
|
||
"status": "in_progress",
|
||
"progress": 50,
|
||
"priority": "high",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["title"] == "已修改的计划"
|
||
assert data["status"] == "in_progress"
|
||
assert data["progress"] == 50
|
||
assert data["priority"] == "high"
|
||
|
||
def test_update_plan_not_found(self, client: TestClient, db: Session):
|
||
"""更新不存在的计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.put(
|
||
"/api/cma/action-plans/9999",
|
||
headers=auth_header(token),
|
||
json={"title": "不存在"},
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
def test_update_plan_progress_clamp(self, client: TestClient, db: Session):
|
||
"""进度值被限制在0-100"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
plan = create_test_plan(db, kpi_id=kpi.id)
|
||
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"progress": 150},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["progress"] == 100
|
||
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"progress": -10},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["progress"] == 0
|
||
|
||
def test_delete_plan(self, client: TestClient, db: Session):
|
||
"""删除行动计划"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
plan = create_test_plan(db, kpi_id=kpi.id)
|
||
|
||
resp = client.delete(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "已删除"
|
||
|
||
# 验证已删除
|
||
get_resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
||
assert len(get_resp.json()["data"]) == 0
|
||
|
||
def test_delete_plan_not_found(self, client: TestClient, db: Session):
|
||
"""删除不存在的计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.delete(
|
||
"/api/cma/action-plans/9999",
|
||
headers=auth_header(token),
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
def test_stats(self, client: TestClient, db: Session):
|
||
"""统计接口"""
|
||
user = create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
create_test_plan(db, kpi_id=kpi.id, title="待开始", status="pending")
|
||
create_test_plan(db, kpi_id=kpi.id, title="进行中", status="in_progress")
|
||
create_test_plan(db, kpi_id=kpi.id, title="已完成", status="completed")
|
||
|
||
resp = client.get("/api/cma/action-plans/stats", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 3
|
||
assert data["pending"] == 1
|
||
assert data["in_progress"] == 1
|
||
assert data["completed"] == 1
|
||
|
||
# ── 行动挂KR (2026-08-27) ──
|
||
|
||
def test_create_plan_with_kr_id(self, client: TestClient, db: Session):
|
||
"""创建行动时挂KR：返回kr_id，objective_id从KR自动继承"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj, kr = create_test_objective_kr(db)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={
|
||
"title": "降低费用率行动",
|
||
"kpi_id": kpi.id,
|
||
"kr_id": kr.id,
|
||
"assignee": "张三",
|
||
},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["kr_id"] == kr.id
|
||
# 未显式传objective_id时从KR继承
|
||
assert data["objective_id"] == obj.id
|
||
|
||
def test_create_plan_kr_not_found(self, client: TestClient, db: Session):
|
||
"""kr_id不存在 → 404"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
resp = client.post(
|
||
"/api/cma/action-plans",
|
||
headers=auth_header(token),
|
||
json={"title": "行动", "kpi_id": kpi.id, "kr_id": 99999},
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
def test_list_includes_kr_title(self, client: TestClient, db: Session):
|
||
"""列表返回kr_title"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj, kr = create_test_objective_kr(db)
|
||
create_test_plan(db, kpi_id=kpi.id, kr_id=kr.id, title="挂KR的行动")
|
||
|
||
resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
item = data[0]
|
||
assert item["kr_id"] == kr.id
|
||
assert item["kr_title"] == "测试KR"
|
||
|
||
def test_update_plan_kr_id(self, client: TestClient, db: Session):
|
||
"""更新行动可改挂KR"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj1, kr1 = create_test_objective_kr(db, kr_title="KR一")
|
||
obj2, kr2 = create_test_objective_kr(db, kr_title="KR二")
|
||
plan = create_test_plan(db, kpi_id=kpi.id, kr_id=kr1.id, title="行动")
|
||
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"kr_id": kr2.id},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["kr_id"] == kr2.id
|
||
|
||
def test_update_plan_kr_clear(self, client: TestClient, db: Session):
|
||
"""更新行动可置空kr_id"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
obj, kr = create_test_objective_kr(db)
|
||
plan = create_test_plan(db, kpi_id=kpi.id, kr_id=kr.id, title="行动")
|
||
|
||
resp = client.put(
|
||
f"/api/cma/action-plans/{plan.id}",
|
||
headers=auth_header(token),
|
||
json={"kr_id": None},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["kr_id"] is None
|