""" 路线图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