509 lines
23 KiB
Python
509 lines
23 KiB
Python
"""
|
||
路线图R1：AI建议→一键落地 测试
|
||
建议CRUD + 应用到KPI/预算/行动方案 + OperationLog留痕 + 已应用/未应用状态
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from datetime import datetime
|
||
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
from app.models import AISuggestion, KPIDefinition, BudgetPlan, ActionPlan, OperationLog, KPIValue
|
||
|
||
|
||
def _create_suggestion(client, token, kpi_id, **kw):
|
||
body = {
|
||
"suggestion_type": "kpi_target",
|
||
"target_type": "kpi",
|
||
"target_id": kpi_id,
|
||
"title": "上调测试KPI目标",
|
||
"content": "达成率超预期",
|
||
"suggestion_data": {"kpi_id": kpi_id, "target_value": 150.0},
|
||
}
|
||
body.update(kw)
|
||
return client.post("/api/cma/ai/suggestions", json=body, headers=auth_header(token))
|
||
|
||
|
||
class TestSuggestionCRUD:
|
||
def test_create_and_list(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
r = _create_suggestion(client, token, kpi.id)
|
||
assert r.status_code == 200, r.text
|
||
data = r.json()["data"]
|
||
assert data["status"] == "unapplied"
|
||
assert data["suggestion_type"] == "kpi_target"
|
||
|
||
# 列表含未应用
|
||
lst = client.get("/api/cma/ai/suggestions", headers=auth_header(token)).json()
|
||
assert lst["total"] == 1
|
||
assert lst["data"][0]["id"] == data["id"]
|
||
|
||
# 详情
|
||
det = client.get(f"/api/cma/ai/suggestions/{data['id']}", headers=auth_header(token)).json()
|
||
assert det["data"]["title"] == "上调测试KPI目标"
|
||
|
||
def test_create_missing_fields(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.post("/api/cma/ai/suggestions", json={"title": "无类型"}, headers=auth_header(token))
|
||
assert r.status_code == 400
|
||
r2 = client.post("/api/cma/ai/suggestions", json={"suggestion_type": "kpi_target"}, headers=auth_header(token))
|
||
assert r2.status_code == 400
|
||
|
||
def test_apply_kpi_target(self, client, db):
|
||
"""应用建议→改KPI目标→操作日志可查"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
|
||
r = _create_suggestion(client, token, kpi.id)
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
# 应用：改KPI目标为150
|
||
app = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "kpi_target", "target_value": 150.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app.status_code == 200, app.text
|
||
app_data = app.json()["data"]
|
||
assert app_data["status"] == "applied"
|
||
assert app_data["applied_by"] == "测试管理员"
|
||
assert app_data["apply_detail"][0]["before"] == 100.0
|
||
assert app_data["apply_detail"][0]["after"] == 150.0
|
||
|
||
# KPI目标已变更
|
||
db.refresh(kpi)
|
||
assert kpi.target_value == 150.0
|
||
|
||
# OperationLog留痕
|
||
logs = db.query(OperationLog).filter(OperationLog.action == "ai_suggestion_apply").all()
|
||
assert len(logs) == 1
|
||
assert logs[0].target_type == "kpi"
|
||
assert logs[0].target_id == kpi.id
|
||
assert logs[0].detail["suggestion_id"] == sug_id
|
||
assert logs[0].detail["before"] == 100.0
|
||
assert logs[0].detail["after"] == 150.0
|
||
|
||
# 重复应用被拒绝
|
||
app2 = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "kpi_target", "target_value": 200.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app2.status_code == 400
|
||
|
||
def test_apply_budget_adjust(self, client, db):
|
||
"""应用建议→调预算（新建/更新BudgetPlan）→操作日志"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
|
||
title="调整预算", suggestion_data={"kpi_id": kpi.id})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
app = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "budget_adjust", "period": "2026-09", "budget_value": 8888.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app.status_code == 200, app.text
|
||
plan = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id, BudgetPlan.period == "2026-09").first()
|
||
assert plan is not None
|
||
assert plan.budget_value == 8888.0
|
||
assert plan.source_type == "ai_suggestion"
|
||
|
||
# 同期间再应用→更新而非新增
|
||
app2 = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "budget_adjust", "period": "2026-09", "budget_value": 9999.0},
|
||
headers=auth_header(token),
|
||
)
|
||
# 已applied被拒；用新建议验证upsert
|
||
r2 = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
|
||
title="调整预算2", suggestion_data={"kpi_id": kpi.id})
|
||
sug_id2 = r2.json()["data"]["id"]
|
||
app3 = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id2}/apply",
|
||
json={"action": "budget_adjust", "period": "2026-09", "budget_value": 9999.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app3.status_code == 200
|
||
plans = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id, BudgetPlan.period == "2026-09").all()
|
||
assert len(plans) == 1
|
||
assert plans[0].budget_value == 9999.0
|
||
assert app3.json()["data"]["apply_detail"][0]["before"] == 8888.0
|
||
|
||
def test_apply_action_plan(self, client, db):
|
||
"""应用建议→建行动方案→操作日志"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="action_plan",
|
||
title="建行动方案", suggestion_data={"kpi_id": kpi.id})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
app = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "action_plan", "title": "营收提升专项", "assignee": "张三",
|
||
"priority": "high", "due_date": "2026-09-30"},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app.status_code == 200, app.text
|
||
plan = db.query(ActionPlan).filter(ActionPlan.kpi_id == kpi.id, ActionPlan.title == "营收提升专项").first()
|
||
assert plan is not None
|
||
assert plan.assignee == "张三"
|
||
assert plan.priority == "high"
|
||
assert plan.created_by == "测试管理员"
|
||
|
||
logs = db.query(OperationLog).filter(OperationLog.action == "ai_suggestion_apply",
|
||
OperationLog.target_type == "action_plan").all()
|
||
assert len(logs) == 1
|
||
assert logs[0].target_id == plan.id
|
||
|
||
def test_dismiss(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
r = _create_suggestion(client, token, kpi.id)
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
d = client.post(f"/api/cma/ai/suggestions/{sug_id}/dismiss", headers=auth_header(token))
|
||
assert d.status_code == 200
|
||
det = client.get(f"/api/cma/ai/suggestions/{sug_id}", headers=auth_header(token)).json()
|
||
assert det["data"]["status"] == "dismissed"
|
||
# 忽略后应用被拒
|
||
app = client.post(f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "kpi_target", "target_value": 1}, headers=auth_header(token))
|
||
assert app.status_code == 400
|
||
|
||
def test_apply_not_found(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
app = client.post("/api/cma/ai/suggestions/9999/apply", json={}, headers=auth_header(token))
|
||
assert app.status_code == 404
|
||
|
||
|
||
class TestRuleSuggestions:
|
||
"""dashboard-analysis 自动生成建议（规则驱动）"""
|
||
|
||
def test_generate_low_ratio_action(self, client, db):
|
||
"""执行率<70% → 生成建行动方案建议"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||
db.commit()
|
||
|
||
# 直接调规则生成
|
||
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
s = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
|
||
assert len(s) >= 1
|
||
assert any(x.suggestion_type == "action_plan" for x in s)
|
||
|
||
# 幂等：再调一次不重复建
|
||
resp2 = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
s2 = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
|
||
assert len(s2) == len(s)
|
||
|
||
def test_generate_high_ratio_target(self, client, db):
|
||
"""执行率>110% → 生成上调目标建议"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=150.0))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
s = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
|
||
assert any(x.suggestion_type == "kpi_target" for x in s)
|
||
assert "suggestions" in resp.json()
|
||
|
||
def test_generate_budget_overrun(self, client, db):
|
||
"""预算执行率>110% → 生成调预算建议"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=200.0))
|
||
db.add(BudgetPlan(entity_id=1, kpi_id=kpi.id, period="2026-08", budget_value=100.0,
|
||
budget_year=2026, budget_month=8, status="active"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
s = db.query(AISuggestion).filter(AISuggestion.suggestion_type == "budget_adjust").all()
|
||
assert len(s) >= 1
|
||
|
||
|
||
class TestSuggestionCategoryPreview:
|
||
"""R1触达修复(2026-08-31)：建议分级(alert/decision) + 列表过滤 + 应用前预览 + 推送开关"""
|
||
|
||
def test_create_marks_category(self, client, db):
|
||
"""手动创建：target_type=alert → category=alert；其余 → decision"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
|
||
r_alert = _create_suggestion(client, token, kpi.id, target_type="alert",
|
||
suggestion_type="action_plan", title="预警类建议")
|
||
assert r_alert.json()["data"]["category"] == "alert"
|
||
|
||
r_decision = _create_suggestion(client, token, kpi.id, title="决策类建议")
|
||
assert r_decision.json()["data"]["category"] == "decision"
|
||
|
||
def test_category_filter(self, client, db):
|
||
"""列表接口 category 过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
_create_suggestion(client, token, kpi.id, target_type="alert",
|
||
suggestion_type="action_plan", title="预警A")
|
||
_create_suggestion(client, token, kpi.id, title="决策B")
|
||
|
||
lst_alert = client.get("/api/cma/ai/suggestions", params={"category": "alert"},
|
||
headers=auth_header(token)).json()
|
||
assert lst_alert["total"] == 1
|
||
assert all(x["category"] == "alert" for x in lst_alert["data"])
|
||
|
||
lst_decision = client.get("/api/cma/ai/suggestions", params={"category": "decision"},
|
||
headers=auth_header(token)).json()
|
||
assert lst_decision["total"] == 1
|
||
assert all(x["category"] == "decision" for x in lst_decision["data"])
|
||
|
||
def test_generate_marks_decision(self, client, db):
|
||
"""规则生成：执行率<70%建议（target_type=kpi）→ category=decision"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||
db.commit()
|
||
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
sug = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).first()
|
||
assert sug is not None
|
||
assert sug.category == "decision"
|
||
|
||
def test_push_disabled_in_test_env(self, client, db):
|
||
"""conftest no-op 推送（monkeypatch）→ 生成决策建议不真推企微，pushed 标记置 1"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||
db.commit()
|
||
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
sug = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).first()
|
||
assert sug is not None
|
||
assert sug.pushed == 1
|
||
|
||
def test_preview_kpi_target(self, client, db):
|
||
"""preview：kpi_target 返回 当前目标 → 新目标"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
r = _create_suggestion(client, token, kpi.id,
|
||
suggestion_data={"kpi_id": kpi.id, "target_value": 150.0})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
|
||
assert pv.status_code == 200, pv.text
|
||
data = pv.json()["data"]
|
||
assert data["type"] == "kpi_target"
|
||
assert data["kpi_name"] == "测试KPI"
|
||
assert data["current_target"] == 100.0
|
||
assert data["new_target"] == 150.0
|
||
|
||
def test_preview_budget_adjust(self, client, db):
|
||
"""preview：budget_adjust 返回 当前预算 → 新预算"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
db.add(BudgetPlan(entity_id=1, kpi_id=kpi.id, period="2026-09", budget_value=8000.0,
|
||
budget_year=2026, budget_month=9, status="active"))
|
||
db.commit()
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
|
||
title="调预算预览", suggestion_data={"kpi_id": kpi.id, "period": "2026-09",
|
||
"budget_value": 9999.0})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
|
||
assert pv.status_code == 200, pv.text
|
||
data = pv.json()["data"]
|
||
assert data["type"] == "budget_adjust"
|
||
assert data["period"] == "2026-09"
|
||
assert data["current_budget"] == 8000.0
|
||
assert data["new_budget"] == 9999.0
|
||
|
||
def test_preview_action_plan(self, client, db):
|
||
"""preview：action_plan 返回计划信息"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="action_plan",
|
||
title="建行动方案预览", suggestion_data={"kpi_id": kpi.id,
|
||
"title": "专项改善", "priority": "high",
|
||
"due_date": "2026-09-30"})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token))
|
||
assert pv.status_code == 200, pv.text
|
||
data = pv.json()["data"]
|
||
assert data["type"] == "action_plan"
|
||
assert data["plan_title"] == "专项改善"
|
||
assert data["priority"] == "high"
|
||
assert data["due_date"] == "2026-09-30"
|
||
|
||
|
||
class TestCategoryAndPreview:
|
||
"""R1触达修复(2026-08-31)：建议分级 + 应用前预览"""
|
||
|
||
def _generate(self, client, db, kpi_id, actual, target=100.0):
|
||
"""造一条KPI数据并触发 dashboard-analysis 规则生成（避开缓存）"""
|
||
db.add(KPIValue(kpi_id=kpi_id, period="2026-07", actual_value=actual))
|
||
db.commit()
|
||
from app.utils.cache import delete as cache_delete
|
||
cache_delete("ai", f"dashboard_analysis:ceo:{kpi_id}")
|
||
resp = client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(get_token_for_user(client)))
|
||
assert resp.status_code == 200
|
||
return resp.json()
|
||
|
||
def test_generate_marks_category(self, client, db, monkeypatch):
|
||
"""生成建议时: target_type=alert → category=alert；其余 → decision"""
|
||
from app.api import ai_analysis
|
||
pushed = []
|
||
ai_analysis._push_decision_suggestion = lambda s: pushed.append(s) or True
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||
db.commit()
|
||
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
|
||
kpi_sugs = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
|
||
assert len(kpi_sugs) >= 1
|
||
for s in kpi_sugs:
|
||
assert s.category == "decision", f"KPI建议应决策类: {s.title}"
|
||
|
||
# 建一条预警 → 规则4生成 alert 类建议
|
||
from app.models import KPIAlert
|
||
db.add(KPIAlert(kpi_id=kpi.id, alert_level="yellow", alert_message="测试预警",
|
||
alert_type="threshold", status="pending"))
|
||
db.commit()
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
alert_sugs = db.query(AISuggestion).filter(AISuggestion.target_type == "alert").all()
|
||
assert len(alert_sugs) >= 1
|
||
for s in alert_sugs:
|
||
assert s.category == "alert", f"预警建议应alert类: {s.title}"
|
||
|
||
def test_alert_not_pushed_decision_pushed(self, client, db, monkeypatch):
|
||
"""推送只发决策类：预警类不推，决策类推且只推一次(pushed=1)"""
|
||
from app.api import ai_analysis
|
||
pushed = []
|
||
ai_analysis._push_decision_suggestion = lambda s: pushed.append(s) or True
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||
db.commit()
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
|
||
kpi_sugs = db.query(AISuggestion).filter(AISuggestion.target_id == kpi.id).all()
|
||
assert len(pushed) >= 1
|
||
assert all(s.category == "decision" for s in pushed)
|
||
for s in pushed:
|
||
assert s.pushed == 1
|
||
|
||
# 预警类建议不在推送流
|
||
from app.models import KPIAlert
|
||
db.add(KPIAlert(kpi_id=kpi.id, alert_level="red", alert_message="测试预警2",
|
||
alert_type="threshold", status="pending"))
|
||
db.commit()
|
||
before = len(pushed)
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
alert_sugs = db.query(AISuggestion).filter(AISuggestion.target_type == "alert").all()
|
||
assert len(alert_sugs) >= 1
|
||
assert len(pushed) == before, "预警类不应触发推送"
|
||
|
||
# 幂等：重复生成不重推（同title建议不重建）
|
||
client.get("/api/cma/ai/dashboard-analysis", headers=auth_header(token))
|
||
assert len(pushed) == before
|
||
|
||
def test_list_category_filter(self, client, db):
|
||
"""列表接口 category 过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
_create_suggestion(client, token, kpi.id, title="决策类A")
|
||
_create_suggestion(client, token, kpi.id, title="决策类B")
|
||
_create_suggestion(client, token, kpi.id, title="预警类C", target_type="alert")
|
||
|
||
lst = client.get("/api/cma/ai/suggestions?category=decision", headers=auth_header(token)).json()
|
||
assert lst["total"] == 2
|
||
assert all(x["category"] == "decision" for x in lst["data"])
|
||
lst2 = client.get("/api/cma/ai/suggestions?category=alert", headers=auth_header(token)).json()
|
||
assert lst2["total"] == 1
|
||
assert lst2["data"][0]["category"] == "alert"
|
||
|
||
def test_preview_kpi_target(self, client, db):
|
||
"""preview: kpi_target 返回 current_target → new_target"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_data={"kpi_id": kpi.id, "target_value": 150.0})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
|
||
assert pv["type"] == "kpi_target"
|
||
assert pv["kpi_name"] == kpi.kpi_name
|
||
assert pv["current_target"] == 100.0
|
||
assert pv["new_target"] == 150.0
|
||
|
||
def test_preview_budget_adjust(self, client, db):
|
||
"""preview: budget_adjust 返回 current_budget → new_budget"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
db.add(BudgetPlan(entity_id=1, kpi_id=kpi.id, period="2026-09", budget_value=5000.0,
|
||
budget_year=2026, budget_month=9, status="active"))
|
||
db.commit()
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="budget_adjust",
|
||
title="调预算", suggestion_data={"kpi_id": kpi.id, "period": "2026-09", "budget_value": 8888.0})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
|
||
assert pv["type"] == "budget_adjust"
|
||
assert pv["current_budget"] == 5000.0
|
||
assert pv["new_budget"] == 8888.0
|
||
assert pv["period"] == "2026-09"
|
||
|
||
def test_preview_action_plan(self, client, db):
|
||
"""preview: action_plan 返回计划参数"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
r = _create_suggestion(client, token, kpi.id, suggestion_type="action_plan",
|
||
title="建行动方案", suggestion_data={"kpi_id": kpi.id, "title": "改善专项",
|
||
"assignee": "李四", "priority": "high", "due_date": "2026-10-01"})
|
||
sug_id = r.json()["data"]["id"]
|
||
|
||
pv = client.get(f"/api/cma/ai/suggestions/{sug_id}/preview", headers=auth_header(token)).json()["data"]
|
||
assert pv["type"] == "action_plan"
|
||
assert pv["plan_title"] == "改善专项"
|
||
assert pv["assignee"] == "李四"
|
||
assert pv["priority"] == "high"
|
||
assert pv["due_date"] == "2026-10-01"
|
||
|
||
def test_preview_not_found(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/ai/suggestions/99999/preview", headers=auth_header(token))
|
||
assert r.status_code == 404
|