576 lines
25 KiB
Python
576 lines
25 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))
|
||
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
|