Hermes CI Fix
a13a080381
- 新增8个测试文件(bot_bridge/kpi_causality/cash/predict/reports/tax_compliance/expenses/probe_cost) - 增强 budget/auth/users + conftest账套模式适配 - 测试驱动修复: bot_bridge导入batch_id→source_batch; cash_forecast extra空dict - 全量: 451 passed, 1 xfailed; 报告 docs/cma-test-coverage-report.md
367 lines
15 KiB
Python
367 lines
15 KiB
Python
"""BOT桥接层测试 — CMA供财务/研学Bot调用的主通道（X-BOT-KEY鉴权）
|
||
|
||
覆盖 bot_bridge.py 全部18个端点：
|
||
ping / overview / kpis / kpis{id}/history / strategic-maps / alerts /
|
||
budget/plans / cost/standard / cost/actual / actions / organization /
|
||
data-sources / users / query / import / okr/create / okr/list / nlp
|
||
"""
|
||
import io
|
||
import hashlib
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from openpyxl import Workbook
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import (
|
||
User, KPIDefinition, KPIValue, KPIAlert, StrategicMap, MapObjective,
|
||
ActionPlan, OrgNode, DataSourceConfig, Objective,
|
||
)
|
||
from app.models.budget_plan import BudgetPlan
|
||
from app.models.cost_model import StandardCost, ActualCost
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
|
||
|
||
|
||
def _make_excel(kpi_code: str, period: str, value: float) -> bytes:
|
||
"""生成Excel导入文件（列: kpi_code, period, actual_value）"""
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["kpi_code", "period", "actual_value"])
|
||
ws.append([kpi_code, period, value])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
return buf.getvalue()
|
||
|
||
|
||
def _seed_kpi(db: Session, **kwargs) -> KPIDefinition:
|
||
defaults = dict(
|
||
kpi_code="BH_REVENUE",
|
||
kpi_name="营业收入",
|
||
dimension="finance",
|
||
status="active",
|
||
target_value=100.0,
|
||
unit="万元",
|
||
frequency="monthly",
|
||
)
|
||
defaults.update(kwargs)
|
||
kpi = KPIDefinition(**defaults)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
class TestPingAndAuth:
|
||
def test_ping_no_key(self, client: TestClient):
|
||
"""ping 无需鉴权"""
|
||
resp = client.get("/api/cma/bot/ping")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "ok"
|
||
|
||
def test_invalid_bot_key(self, client: TestClient):
|
||
"""无效BOT Key → 401"""
|
||
resp = client.get("/api/cma/bot/overview", headers={"X-BOT-KEY": "wrong-key"})
|
||
assert resp.status_code == 401
|
||
|
||
def test_missing_bot_key(self, client: TestClient):
|
||
"""缺BOT Key → 401"""
|
||
resp = client.get("/api/cma/bot/overview")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
class TestOverviewAndKpis:
|
||
def test_overview_stats(self, client: TestClient, db: Session):
|
||
"""总览统计：造数后计数正确"""
|
||
_seed_kpi(db)
|
||
db.add(KPIAlert(kpi_id=1, alert_level="red", alert_message="收入下滑", status="pending"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/overview", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["bot"]["name"] == "财务BOT"
|
||
assert data["stats"]["kpis_total"] == 1
|
||
assert data["stats"]["alerts_open"] == 1
|
||
|
||
def test_kpis_filter_by_dimension(self, client: TestClient, db: Session):
|
||
"""KPI列表：按维度过滤 + 关联最新实际值"""
|
||
k1 = _seed_kpi(db, kpi_code="BH_REVENUE", dimension="finance")
|
||
_seed_kpi(db, kpi_code="BH_CUSTOMER", dimension="customer")
|
||
db.add(KPIValue(kpi_id=k1.id, period="2026-06", actual_value=88.0, data_status="verified"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/kpis?dimension=finance", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["code"] == "BH_REVENUE"
|
||
assert data["items"][0]["latest_value"] == 88.0
|
||
assert data["items"][0]["latest_period"] == "2026-06"
|
||
|
||
def test_kpis_status_filter(self, client: TestClient, db: Session):
|
||
"""KPI列表：status过滤（默认active，inactive被过滤）"""
|
||
_seed_kpi(db, kpi_code="BH_ACTIVE")
|
||
_seed_kpi(db, kpi_code="BH_INACTIVE", status="inactive")
|
||
resp = client.get("/api/cma/bot/kpis", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
codes = {i["code"] for i in resp.json()["items"]}
|
||
assert "BH_ACTIVE" in codes
|
||
assert "BH_INACTIVE" not in codes
|
||
|
||
def test_kpi_history(self, client: TestClient, db: Session):
|
||
"""KPI历史值"""
|
||
k = _seed_kpi(db, kpi_code="BH_REVENUE")
|
||
db.add(KPIValue(kpi_id=k.id, period="2026-07", actual_value=95.0, source_type="manual", data_status="verified"))
|
||
db.add(KPIValue(kpi_id=k.id, period="2026-06", actual_value=88.0))
|
||
db.commit()
|
||
|
||
resp = client.get(f"/api/cma/bot/kpis/{k.id}/history", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["kpi"]["code"] == "BH_REVENUE"
|
||
# 按期间倒序，最新在前
|
||
assert data["values"][0]["period"] == "2026-07"
|
||
assert len(data["values"]) == 2
|
||
|
||
def test_kpi_history_not_found(self, client: TestClient):
|
||
"""不存在的KPI → 404"""
|
||
resp = client.get("/api/cma/bot/kpis/99999/history", headers=BOT_KEY)
|
||
assert resp.status_code == 404
|
||
|
||
|
||
class TestMapsAlertsBudgetCost:
|
||
def test_strategic_maps(self, client: TestClient, db: Session):
|
||
"""战略地图列表（含目标）"""
|
||
m = StrategicMap(title="博海战略地图", version="v1.0", status="published",
|
||
dimensions=[{"key": "finance", "name": "财务"}])
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
db.add(MapObjective(map_id=m.id, dimension_key="finance", name="提升收入"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/strategic-maps", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["title"] == "博海战略地图"
|
||
assert data["items"][0]["objectives"]["finance"][0]["name"] == "提升收入"
|
||
|
||
def test_alerts_filter(self, client: TestClient, db: Session):
|
||
"""预警列表：按状态/等级过滤"""
|
||
_seed_kpi(db)
|
||
db.add(KPIAlert(kpi_id=1, alert_level="red", alert_message="严重", status="pending"))
|
||
db.add(KPIAlert(kpi_id=1, alert_level="yellow", alert_message="关注", status="resolved"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/alerts?status=pending&level=red", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["level"] == "red"
|
||
assert data["items"][0]["message"] == "严重"
|
||
|
||
def test_budget_plans(self, client: TestClient, db: Session):
|
||
"""预算计划（按年过滤）"""
|
||
k = _seed_kpi(db)
|
||
db.add(BudgetPlan(kpi_id=k.id, period="2026-06", budget_value=50000.0,
|
||
budget_year=2026, budget_month=6, status="active"))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/budget/plans?year=2026", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["budget_value"] == 50000.0
|
||
|
||
def test_budget_plans_year_filter(self, client: TestClient, db: Session):
|
||
"""预算计划：其他年份被过滤"""
|
||
k = _seed_kpi(db)
|
||
db.add(BudgetPlan(kpi_id=k.id, period="2025-12", budget_value=100.0,
|
||
budget_year=2025, budget_month=12, status="active"))
|
||
db.commit()
|
||
resp = client.get("/api/cma/bot/budget/plans?year=2024", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] == 0
|
||
|
||
def test_cost_standard(self, client: TestClient, db: Session):
|
||
"""标准成本"""
|
||
db.add(StandardCost(product_code="P001", product_name="产品A", cost_type="material",
|
||
item_name="原料", standard_quantity=2.0, unit="kg",
|
||
standard_price=10.0, standard_cost=20.0, status="active"))
|
||
db.commit()
|
||
resp = client.get("/api/cma/bot/cost/standard", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] == 1
|
||
assert resp.json()["items"][0]["standard_cost"] == 20.0
|
||
|
||
def test_cost_actual_period(self, client: TestClient, db: Session):
|
||
"""实际成本（按期间过滤）"""
|
||
db.add(ActualCost(period="2026-06", product_code="P001", product_name="产品A",
|
||
cost_type="material", item_name="原料",
|
||
actual_quantity=3.0, actual_price=12.0, actual_cost=36.0))
|
||
db.commit()
|
||
resp = client.get("/api/cma/bot/cost/actual?period=2026-06", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] == 1
|
||
assert resp.json()["items"][0]["actual_cost"] == 36.0
|
||
|
||
|
||
class TestActionsOrgSourcesUsers:
|
||
def test_actions(self, client: TestClient, db: Session):
|
||
"""行动方案列表（按状态过滤）"""
|
||
k = _seed_kpi(db)
|
||
db.add(ActionPlan(kpi_id=k.id, title="提升收入", status="in_progress", priority="high", progress=50))
|
||
db.add(ActionPlan(kpi_id=k.id, title="已关闭", status="completed", priority="low", progress=100))
|
||
db.commit()
|
||
|
||
resp = client.get("/api/cma/bot/actions?status=in_progress", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["title"] == "提升收入"
|
||
|
||
def test_organization(self, client: TestClient, db: Session):
|
||
"""组织架构"""
|
||
db.add(OrgNode(name="测试组织", code="TEST_ORG_001", level=1, sort_order=1, enabled=1))
|
||
db.commit()
|
||
resp = client.get("/api/cma/bot/organization", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
codes = [i["code"] for i in resp.json()["items"]]
|
||
assert "TEST_ORG_001" in codes
|
||
|
||
def test_data_sources(self, client: TestClient, db: Session):
|
||
"""数据源"""
|
||
db.add(DataSourceConfig(name="ERP", source_type="erp", api_endpoint="http://erp",
|
||
sync_type="batch", status="active"))
|
||
db.commit()
|
||
resp = client.get("/api/cma/bot/data-sources", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] == 1
|
||
assert resp.json()["items"][0]["name"] == "ERP"
|
||
|
||
def test_users(self, client: TestClient, db: Session):
|
||
"""用户列表（不返回密码等敏感字段）"""
|
||
create_test_user(db)
|
||
resp = client.get("/api/cma/bot/users", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] >= 1
|
||
user = resp.json()["items"][0]
|
||
assert "username" in user
|
||
assert "password" not in user
|
||
|
||
|
||
class TestUnifiedQuery:
|
||
def test_query_overview(self, client: TestClient, db: Session):
|
||
"""统一查询 overview"""
|
||
_seed_kpi(db)
|
||
resp = client.get("/api/cma/bot/query?q=overview", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["overview"]["kpis"] == 1
|
||
|
||
def test_query_kpis_and_budget(self, client: TestClient, db: Session):
|
||
"""统一查询 kpis / budget"""
|
||
k = _seed_kpi(db)
|
||
db.add(BudgetPlan(kpi_id=k.id, period="2026-06", budget_value=10.0,
|
||
budget_year=2026, budget_month=6, status="active"))
|
||
db.commit()
|
||
|
||
r1 = client.get("/api/cma/bot/query?q=kpis", headers=BOT_KEY)
|
||
assert r1.status_code == 200
|
||
assert len(r1.json()["kpis"]) == 1
|
||
|
||
r2 = client.get("/api/cma/bot/query?q=budget", headers=BOT_KEY)
|
||
assert r2.status_code == 200
|
||
assert len(r2.json()["budget"]) == 1
|
||
|
||
def test_query_all(self, client: TestClient, db: Session):
|
||
"""统一查询 all：返回全部分组"""
|
||
_seed_kpi(db)
|
||
resp = client.get("/api/cma/bot/query?q=all", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "overview" in data and "kpis" in data and "alerts" in data
|
||
assert "maps" in data and "budget" in data and "costs" in data
|
||
assert "actions" in data and "okr" in data
|
||
|
||
|
||
class TestImport:
|
||
def test_import_excel(self, client: TestClient, db: Session):
|
||
"""Excel导入KPI实际值"""
|
||
_seed_kpi(db, kpi_code="BH_REVENUE")
|
||
files = {"file": ("kpi.xlsx", _make_excel("BH_REVENUE", "2026-08", 99.5),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
||
resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["ok"] is True
|
||
assert data["imported"] == 1
|
||
|
||
# 验证入库
|
||
val = db.query(KPIValue).filter(KPIValue.period == "2026-08").first()
|
||
assert val is not None and val.actual_value == 99.5
|
||
|
||
def test_import_excel_unknown_kpi(self, client: TestClient, db: Session):
|
||
"""导入不存在的KPI编码 → 跳过并记录错误"""
|
||
files = {"file": ("kpi.xlsx", _make_excel("NO_SUCH_KPI", "2026-08", 10.0),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
||
resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["imported"] == 0
|
||
assert data["errors"] == 1
|
||
|
||
def test_import_bad_file(self, client: TestClient):
|
||
"""非Excel文件 → 400"""
|
||
files = {"file": ("bad.txt", b"not an excel", "text/plain")}
|
||
resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files)
|
||
assert resp.status_code == 400
|
||
|
||
def test_import_missing_value_col(self, client: TestClient, db: Session):
|
||
"""缺少数值列 → 400"""
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["kpi_code"])
|
||
ws.append(["BH_REVENUE"])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
files = {"file": ("kpi.xlsx", buf.getvalue(),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
||
resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
class TestOkrAndNlp:
|
||
def test_okr_create_and_list(self, client: TestClient, db: Session):
|
||
"""Bot创建OKR目标 + 列表"""
|
||
resp = client.post("/api/cma/bot/okr/create?title=提升净利润&quarter=2026Q3&dimension=finance",
|
||
headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
obj_id = resp.json()["id"]
|
||
assert obj_id > 0
|
||
|
||
list_resp = client.get("/api/cma/bot/okr/list?quarter=2026Q3", headers=BOT_KEY)
|
||
assert list_resp.status_code == 200
|
||
assert list_resp.json()["total"] == 1
|
||
assert list_resp.json()["items"][0]["title"] == "提升净利润"
|
||
|
||
def test_okr_create_missing_quarter(self, client: TestClient):
|
||
"""缺quarter → 422"""
|
||
resp = client.post("/api/cma/bot/okr/create?title=无季度目标", headers=BOT_KEY)
|
||
assert resp.status_code == 422
|
||
|
||
def test_nlp_intent_mapping(self, client: TestClient, db: Session):
|
||
"""自然语言意图映射"""
|
||
_seed_kpi(db)
|
||
# 中文意图 → 映射到预算
|
||
resp = client.get("/api/cma/bot/nlp?intent=预算", headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
assert "budget" in resp.json()
|
||
|
||
resp2 = client.get("/api/cma/bot/nlp?intent=总览", headers=BOT_KEY)
|
||
assert resp2.status_code == 200
|
||
assert "overview" in resp2.json()
|