Hermes CI Fix
fb4609be14
ci/woodpecker/push/main Pipeline failed
- ① kpi_templates instantiate 构造点显式传会话账套（原缺 entity_id ⇒ 静默落 e1），kpi_code 判重按账套收紧
- ② kpi_subject_map 3 个读取点加护栏注释（不加列、不改语义）
- ③ subjects 分层：读口 union（NULL=全局标准准则科目 / N=账套私有），UPSERT 唯一键改 (entity_id, subject_code)
- ④ analysis_results 账套化：读/写/删三端点按会话账套判定归属（列未落地时 kpi_code JOIN 止血），
无归属行进 unattributed 登记、删除一律 404
- 模型层按现网 DDL 补齐：KPIDefinition(template_id/is_system + uk_entity_kpi)、AnalysisResult.entity_id、Subject.entity_id
- 新增 migrate_{subjects,analysis_results}_entity_id_20260923.py（--check/--apply/--down）、
22 条对账用例 test_entity_dimension_convergence_20260923.py
DDL 已由 ops 受控通道落地（subjects 21:31:56 / analysis_results 21:31:59 CST，审批 1e9b4d9a/5ffb1640）
本提交为切片提交，未 push（发布须另取放行）
652 lines
30 KiB
Python
652 lines
30 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, Subject,
|
||
)
|
||
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):
|
||
"""用户列表（不返回密码等敏感字段）"""
|
||
user = create_test_user(db)
|
||
# 多租户隔离（2026-08-31）：bot_users 按 user_entities 授权表过滤，需先授权
|
||
from app.models import UserEntity
|
||
db.add(UserEntity(user_id=user.id, entity_id=1))
|
||
db.commit()
|
||
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()
|
||
|
||
|
||
class TestDictZxtAccountWrite:
|
||
"""智享通科目字典写入端点（POST /api/cma/bot/dict/zxt-account）负例
|
||
|
||
说明: 正向 UPSERT 走 MySQL 专有 ON DUPLICATE KEY UPDATE，测试库为 SQLite
|
||
无法覆盖——正向/幂等/落库一致性由生产 curl 实测（见任务验收记录），
|
||
此处覆盖鉴权与参数校验（校验先于任何 DB 访问，故 SQLite 可测）。
|
||
"""
|
||
URL = "/api/cma/bot/dict/zxt-account"
|
||
|
||
def test_no_key_401(self, client: TestClient):
|
||
resp = client.post(self.URL, json=[{"CwAccountID": "999", "subject_code": "1"}])
|
||
assert resp.status_code == 401
|
||
|
||
def test_invalid_key_401(self, client: TestClient):
|
||
resp = client.post(self.URL, json=[{"CwAccountID": "999"}],
|
||
headers={"X-BOT-KEY": "wrong-key"})
|
||
assert resp.status_code == 401
|
||
|
||
def test_non_whitelist_field_400(self, client: TestClient):
|
||
resp = client.post(self.URL,
|
||
json=[{"CwAccountID": "999", "entity_id": 1}], headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "非白名单字段" in str(resp.json())
|
||
|
||
def test_missing_required_field_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json=[{"subject_code": "1122"}], headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "CwAccountID" in str(resp.json())
|
||
|
||
def test_illegal_value_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json=[{"CwAccountID": "9 99;DROP"}], headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
|
||
def test_empty_array_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json=[], headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
|
||
def test_non_array_json_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"CwAccountID": "999"}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
|
||
def test_rows_wrapper_accepted_shape(self, client: TestClient):
|
||
"""{"rows":[...]} 包装形态被解析（校验失败仍 400，证明已进入校验而非形态拒绝）"""
|
||
resp = client.post(self.URL, json={"rows": [{"bad_field": 1}]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "非白名单字段" in str(resp.json())
|
||
|
||
|
||
class TestSubjectsWrite:
|
||
"""会计科目写入端点（POST /api/cma/bot/subjects）— 鉴权/校验/dry_run/幂等/更新
|
||
|
||
实现为纯 ORM（无 MySQL 专有 SQL），故正向/幂等路径可在 SQLite 测试库完整覆盖。
|
||
"""
|
||
|
||
URL = "/api/cma/bot/subjects"
|
||
|
||
@staticmethod
|
||
def _row(code, name, **kw):
|
||
row = {"subject_code": code, "subject_name": name, "parent_code": None,
|
||
"level": 1, "category": "equity", "is_active": 1}
|
||
row.update(kw)
|
||
return row
|
||
|
||
# ── 鉴权 ──
|
||
def test_no_key_401(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]})
|
||
assert resp.status_code == 401
|
||
|
||
def test_invalid_key_401(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]},
|
||
headers={"X-BOT-KEY": "wrong-key"})
|
||
assert resp.status_code == 401
|
||
|
||
# ── 参数校验（全批拒绝、零落库）──
|
||
def test_level_out_of_range_400(self, client: TestClient, db: Session):
|
||
resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目", level=9)]},
|
||
headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "level" in str(resp.json())
|
||
assert db.query(Subject).filter(Subject.subject_code == "9001").count() == 0
|
||
|
||
def test_batch_duplicate_code_400(self, client: TestClient, db: Session):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9002", "验证科目A"), self._row("9002", "验证科目B")]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "重复" in str(resp.json())
|
||
assert db.query(Subject).filter(Subject.subject_code == "9002").count() == 0
|
||
|
||
def test_parent_not_exists_400(self, client: TestClient, db: Session):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("900301", "验证子科目", parent_code="999999", level=2)]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "parent_code" in str(resp.json())
|
||
assert db.query(Subject).filter(Subject.subject_code == "900301").count() == 0
|
||
|
||
def test_illegal_category_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9003", "验证科目", category="not_a_category")]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "category" in str(resp.json())
|
||
|
||
def test_illegal_is_active_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9004", "验证科目", is_active=7)]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "is_active" in str(resp.json())
|
||
|
||
def test_empty_subjects_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": []}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
|
||
def test_non_whitelist_field_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9005", "验证科目", entity_id=1)]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "非白名单字段" in str(resp.json())
|
||
|
||
def test_missing_name_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
{"subject_code": "9006", "level": 1}]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400
|
||
assert "subject_name" in str(resp.json())
|
||
|
||
def test_code_too_long_400(self, client: TestClient, db: Session):
|
||
"""编码长度上限 = min(任务规格 32, DB varchar(20)) = 20
|
||
|
||
生产实测（2026-09-12）：按 32 放行时 21 字符编码穿过校验 → MySQL 1406 Data too long
|
||
→ 整批 500；收敛到 DB 列宽后应 400 拒绝且零落库。
|
||
"""
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9" * 21, "验证科目")]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400, resp.json()
|
||
assert "超长" in str(resp.json())
|
||
assert db.query(Subject).filter(Subject.subject_code == "9" * 21).count() == 0
|
||
|
||
def test_parent_code_too_long_400(self, client: TestClient):
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9402", "验证子科目", parent_code="9" * 21, level=2)]}, headers=BOT_KEY)
|
||
assert resp.status_code == 400, resp.json()
|
||
assert "超长" in str(resp.json())
|
||
|
||
# ── 落库语义 ──
|
||
def test_dry_run_zero_write(self, client: TestClient, db: Session):
|
||
before = db.query(Subject).count()
|
||
resp = client.post(self.URL, json={"subjects": [self._row("9101", "验证-干跑")],
|
||
"dry_run": True}, headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["dry_run"] is True and data["applied"] is False
|
||
assert data["inserted"] == 0 and data["upserted"] == 0
|
||
assert data["would_insert"] == 1
|
||
db.expire_all()
|
||
assert db.query(Subject).count() == before
|
||
assert db.query(Subject).filter(Subject.subject_code == "9101").count() == 0
|
||
|
||
def test_insert_skip_counts_and_idempotent(self, client: TestClient, db: Session):
|
||
db.add(Subject(subject_code="9201", subject_name="既有科目", level=1,
|
||
category="equity", is_active=1))
|
||
db.commit()
|
||
before = db.query(Subject).count()
|
||
|
||
payload = {"subjects": [self._row("9201", "既有科目"), self._row("9202", "新增科目A"),
|
||
self._row("9203", "新增科目B")]}
|
||
resp = client.post(self.URL, json=payload, headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
d1 = resp.json()
|
||
assert (d1["inserted"], d1["updated"], d1["skipped"]) == (2, 0, 1), d1
|
||
assert d1["after"] == before + 2
|
||
|
||
db.expire_all()
|
||
assert db.query(Subject).filter(Subject.subject_code.in_(["9202", "9203"])).count() == 2
|
||
|
||
# 幂等：同 payload 二次调用 → 全部 skip，行数不变、无重复
|
||
resp2 = client.post(self.URL, json=payload, headers=BOT_KEY)
|
||
assert resp2.status_code == 200
|
||
d2 = resp2.json()
|
||
assert (d2["inserted"], d2["updated"], d2["skipped"]) == (0, 0, 3), d2
|
||
assert d2["before"] == d2["after"] == before + 2
|
||
db.expire_all()
|
||
assert db.query(Subject).count() == before + 2
|
||
assert db.query(Subject).filter(Subject.subject_code == "9202").count() == 1
|
||
|
||
def test_update_existing_switch(self, client: TestClient, db: Session):
|
||
db.add(Subject(subject_code="9301", subject_name="旧名称", level=1,
|
||
category="equity", is_active=1, entity_id=1)) # 账套维度：本账套私有行（分层后 NULL=全局标准科目）
|
||
db.commit()
|
||
resp = client.post(self.URL, json={
|
||
"subjects": [self._row("9301", "新名称", category="cost", level=2)],
|
||
"update_existing": True}, headers=BOT_KEY)
|
||
assert resp.status_code == 200
|
||
d = resp.json()
|
||
assert (d["inserted"], d["updated"], d["skipped"]) == (0, 1, 0), d
|
||
db.expire_all()
|
||
row = db.query(Subject).filter(Subject.subject_code == "9301").first()
|
||
assert row.subject_name == "新名称" and row.category == "cost" and row.level == 2
|
||
|
||
def test_batch_parent_within_batch_ok(self, client: TestClient, db: Session):
|
||
"""父科目在本批次内（层级同批导入）允许；父科目不存在则拒绝"""
|
||
resp = client.post(self.URL, json={"subjects": [
|
||
self._row("9401", "父科目"), self._row("940101", "子科目", parent_code="9401", level=2)],
|
||
}, headers=BOT_KEY)
|
||
assert resp.status_code == 200, resp.json()
|
||
assert resp.json()["inserted"] == 2
|
||
|
||
# ── category 白名单 vs 字典实况（2026-09-12 dama-cma-subjects-category-whitelist）──
|
||
# 生产库 SELECT DISTINCT category FROM subjects（207 行，2026-09-12 只读核验）：
|
||
DB_CATEGORY_CORPUS_20260912 = ("asset", "cost", "detail", "equity", "expense",
|
||
"income", "liability", "profit_loss", "revenue")
|
||
|
||
def test_whitelist_covers_dictionary_corpus(self):
|
||
"""防漂移：库内真实取值若被白名单漏掉 → 本测试失败
|
||
|
||
原缺陷（DAMA 复核）：白名单 8 值漏 revenue / profit_loss，库内合法值被 400 拒。
|
||
"""
|
||
from app.api.bot_bridge import _SUBJ_CATEGORIES
|
||
missing = sorted(set(self.DB_CATEGORY_CORPUS_20260912) - set(_SUBJ_CATEGORIES))
|
||
assert not missing, f"库内真实取值被端点白名单漏掉: {missing}"
|
||
|
||
@pytest.mark.parametrize("cat", DB_CATEGORY_CORPUS_20260912)
|
||
def test_dictionary_category_accepted(self, client: TestClient, db: Session, cat: str):
|
||
"""字典实况每个取值都能通过端点校验并落库（不再出现「库内合法、端点 400」）"""
|
||
code = "95" + f"{self.DB_CATEGORY_CORPUS_20260912.index(cat) + 1:02d}"
|
||
resp = client.post(self.URL, json={"subjects": [self._row(code, f"验证-{cat}", category=cat)]},
|
||
headers=BOT_KEY)
|
||
assert resp.status_code == 200, resp.json()
|
||
assert resp.json()["inserted"] == 1
|
||
db.expire_all()
|
||
row = db.query(Subject).filter(Subject.subject_code == code).first()
|
||
assert row is not None and row.category == cat
|
||
|
||
def test_revenue_existing_row_update_existing_true(self, client: TestClient, db: Session):
|
||
"""原 400 复现用例：库内既有 revenue 行（6001 主营业务收入）+ update_existing=true"""
|
||
db.add(Subject(subject_code="6001", subject_name="主营业务收入", level=1,
|
||
category="revenue", is_active=1, entity_id=1)) # 本账套私有行（分层口径）
|
||
other = Subject(subject_code="6051", subject_name="其他业务收入", level=1,
|
||
category="revenue", is_active=1, entity_id=1)
|
||
db.add(other)
|
||
db.commit()
|
||
before = db.query(Subject).count()
|
||
resp = client.post(self.URL, json={
|
||
"subjects": [self._row("6001", "主营业务收入（新30号）", category="revenue", level=1)],
|
||
"update_existing": True}, headers=BOT_KEY)
|
||
assert resp.status_code == 200, resp.json()
|
||
d = resp.json()
|
||
assert (d["inserted"], d["updated"], d["skipped"]) == (0, 1, 0), d
|
||
db.expire_all()
|
||
row = db.query(Subject).filter(Subject.subject_code == "6001").first()
|
||
assert row.subject_name == "主营业务收入（新30号）" and row.category == "revenue"
|
||
# 仅目标行变化：行数不变，邻行逐字段未动
|
||
assert db.query(Subject).count() == before
|
||
nb = db.query(Subject).filter(Subject.subject_code == "6051").first()
|
||
assert (nb.subject_name, nb.category, nb.level) == ("其他业务收入", "revenue", 1)
|
||
|
||
def test_profit_loss_existing_row_update_existing_true(self, client: TestClient, db: Session):
|
||
db.add(Subject(subject_code="6901", subject_name="终止经营损益", level=1,
|
||
category="profit_loss", is_active=1, entity_id=1)) # 本账套私有行（分层口径）
|
||
db.commit()
|
||
resp = client.post(self.URL, json={
|
||
"subjects": [self._row("6901", "终止经营损益（V2）", category="profit_loss")],
|
||
"update_existing": True}, headers=BOT_KEY)
|
||
assert resp.status_code == 200, resp.json()
|
||
assert resp.json()["updated"] == 1
|
||
db.expire_all()
|
||
row = db.query(Subject).filter(Subject.subject_code == "6901").first()
|
||
assert row.subject_name == "终止经营损益（V2）" and row.category == "profit_loss"
|
||
|
||
def test_revenue_existing_row_update_existing_false_skips(self, client: TestClient, db: Session):
|
||
"""update_existing=false：revenue 行正常 skip，逐字段未被改写"""
|
||
db.add(Subject(subject_code="6011", subject_name="利息收入（银行存款）", level=1,
|
||
category="revenue", is_active=1))
|
||
db.commit()
|
||
resp = client.post(self.URL, json={
|
||
"subjects": [self._row("6011", "不应生效的名称", category="income")]}, headers=BOT_KEY)
|
||
assert resp.status_code == 200, resp.json()
|
||
d = resp.json()
|
||
assert (d["inserted"], d["updated"], d["skipped"]) == (0, 0, 1), d
|
||
db.expire_all()
|
||
row = db.query(Subject).filter(Subject.subject_code == "6011").first()
|
||
assert (row.subject_name, row.category) == ("利息收入（银行存款）", "revenue")
|