626 lines
27 KiB
Python
626 lines
27 KiB
Python
"""CMA 跨实体隔离残余收口（第五批）账套隔离回归测试 — 2026-09-21
|
||
|
||
任务：project-20260921-cma-entity-isolation-batch5（项目Bot 派单；
|
||
来源 = batch4 执行中反向推送的残余单 fullstack-20260921-batch5-residual-entity-isolation.md）
|
||
|
||
本单范围（逐条）：
|
||
P0 budget.py update_budget_plan / delete_budget_plan / update_deviation_alert
|
||
无账套校验 → 本账套会话可改/删他企预算与预警（跨租户写）
|
||
P1 kpis.py get_kpi_objectives 无账套收口（KPI/StrategicMap/MapObjective 按 id 直取）
|
||
P1 data_quality.py 质量日志族 GET/PUT/DELETE /logs 无账套收口（表无 entity_id 列，
|
||
按选项(1) 经 kpi_id join kpi_definitions 按 kpi_definitions.entity_id 收口）
|
||
P2 budget.py auto_decompose 单 KPI 分支不写 entity_id（落默认 1）
|
||
P2 predict.py growth-quality 硬编码 ENTITY_DATA + 请求体 data.get("entity", data) 回退
|
||
P2 expenses.py 列表假隔离 + 详情/审批无归属校验；tax_compliance.py 演示播种不写 entity_id
|
||
|
||
每条均「跨账套负例 + 本账套正对照」。改前 RED 原文见回报 §三。
|
||
本文件只跑内存 SQLite（tests/conftest.py 机制），零生产库 / 零生产 Redis 连接。
|
||
"""
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import (
|
||
BudgetDeviationAlert,
|
||
BudgetPlan,
|
||
Entity,
|
||
ExpenseReimbursement,
|
||
KPIDefinition,
|
||
KPIValue,
|
||
KpiDataQualityLog,
|
||
MapObjective,
|
||
StrategicMap,
|
||
)
|
||
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
|
||
|
||
BUDGET = "/api/cma/budget"
|
||
KPIS = "/api/cma/kpis"
|
||
DQ = "/api/cma/data-quality"
|
||
PREDICT = "/api/cma/predict"
|
||
EXPENSES = "/api/cma/expenses"
|
||
TAX = "/api/cma/tax"
|
||
|
||
# 两账套资源编码前缀（kpi_code 库内唯一）
|
||
E1_KPI = "ZZ_B5_E1_KPI_001"
|
||
E2_KPI = "ZZ_B5_E2_KPI_001"
|
||
E1_KPI_TREE = "ZZ_B5_E1_KPI_TREE"
|
||
E1_KPI_OTHER_MAP = "ZZ_B5_E1_KPI_XMAP"
|
||
E1_GQ_KPI = "ZZ_B5_E1_GQ"
|
||
E2_GQ_KPI = "ZZ_B5_E2_GQ"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 夹具
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _ensure_entity(db: Session, eid: int, name: str) -> Entity:
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=name, short_name=name, status="active")
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _mk_kpi(db: Session, code: str, eid: int, **kw) -> KPIDefinition:
|
||
kpi = KPIDefinition(
|
||
entity_id=eid,
|
||
kpi_code=code,
|
||
kpi_name=f"{code}-名称",
|
||
dimension="finance",
|
||
category="revenue_growth",
|
||
formula="-",
|
||
data_source_type="excel",
|
||
data_source="Excel导入",
|
||
data_owner="财务部",
|
||
frequency="monthly",
|
||
unit="元",
|
||
target_value=0,
|
||
kpi_level="operational",
|
||
status="active",
|
||
**kw,
|
||
)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
def _mk_plan(db: Session, kpi_id: int, eid: int, period: str = "2026-09",
|
||
value: float = 100.0, status: str = "active") -> BudgetPlan:
|
||
bp = BudgetPlan(entity_id=eid, kpi_id=kpi_id, period=period, budget_value=value,
|
||
budget_year=2026, budget_month=9, version="v1.0", status=status)
|
||
db.add(bp)
|
||
db.commit()
|
||
db.refresh(bp)
|
||
return bp
|
||
|
||
|
||
def _mk_alert(db: Session, kpi_id: int, eid: int, status: str = "open") -> BudgetDeviationAlert:
|
||
a = BudgetDeviationAlert(entity_id=eid, kpi_id=kpi_id, period="2026-09",
|
||
budget_value=100.0, actual_value=130.0,
|
||
deviation_rate=30.0, deviation_value=30.0,
|
||
alert_level="warning", status=status)
|
||
db.add(a)
|
||
db.commit()
|
||
db.refresh(a)
|
||
return a
|
||
|
||
|
||
def _mk_map(db: Session, eid: int, title: str) -> StrategicMap:
|
||
m = StrategicMap(entity_id=eid, title=title, status="draft",
|
||
dimensions=[], canvas_data={"connections": []})
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
return m
|
||
|
||
|
||
def _mk_objective(db: Session, eid: int, map_id: int, name: str, dimension: str = "finance") -> MapObjective:
|
||
o = MapObjective(entity_id=eid, map_id=map_id, dimension_key=dimension, name=name)
|
||
db.add(o)
|
||
db.commit()
|
||
db.refresh(o)
|
||
return o
|
||
|
||
|
||
def _mk_log(db: Session, kpi_id: int, status: str = "open") -> KpiDataQualityLog:
|
||
lg = KpiDataQualityLog(kpi_id=kpi_id, check_type="flat_data", severity="warning",
|
||
detail={"note": "batch5"}, suggestion="检查采集", status=status)
|
||
db.add(lg)
|
||
db.commit()
|
||
db.refresh(lg)
|
||
return lg
|
||
|
||
|
||
def _mk_reimb(db: Session, eid: int, no: str, amount: float = 100.0,
|
||
status: str = "pending") -> ExpenseReimbursement:
|
||
r = ExpenseReimbursement(entity_id=eid, reimb_no=no, applicant="张三", department="财务部",
|
||
expense_type="office", title=f"{no}-报销", amount=amount,
|
||
expense_date=datetime(2026, 9, 10), status=status,
|
||
check_result="pass")
|
||
db.add(r)
|
||
db.commit()
|
||
db.refresh(r)
|
||
return r
|
||
|
||
|
||
def _login(client: TestClient, entity_id: int) -> str:
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin",
|
||
"password": TEST_USER_PASSWORD,
|
||
"entity_id": entity_id,
|
||
})
|
||
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
|
||
return resp.json()["token"]
|
||
|
||
|
||
def _seed_two_entities(db: Session):
|
||
_ensure_entity(db, 1, "测试企业1")
|
||
_ensure_entity(db, 2, "测试企业2")
|
||
create_test_user(db)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P0. budget.py 预算/预警写端点按账套（跨账套 404）
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_budget_update_plan_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话改 e2 预算计划 → 404 且对端计划未被改动"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_plan(db, k2.id, 2, value=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{BUDGET}/plans/{other.id}", headers=auth_header(token1),
|
||
json={"budget_value": 999.0, "status": "inactive"})
|
||
assert r.status_code == 404, f"跨账套改预算应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(BudgetPlan).filter(BudgetPlan.id == other.id).first()
|
||
assert row is not None, "对端预算行被删除"
|
||
assert row.budget_value == 50.0, f"跨账套改预算生效了：budget_value={row.budget_value}"
|
||
assert row.status == "active", f"跨账套改了 status={row.status}"
|
||
|
||
|
||
def test_budget_update_plan_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套改自家预算计划 → 200 且落库"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
mine = _mk_plan(db, k1.id, 1, value=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{BUDGET}/plans/{mine.id}", headers=auth_header(token1),
|
||
json={"budget_value": 88.0})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(BudgetPlan).filter(BudgetPlan.id == mine.id).first().budget_value == 88.0
|
||
|
||
|
||
def test_budget_delete_plan_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话删 e2 预算计划 → 404 且对端计划仍在"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_plan(db, k2.id, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{BUDGET}/plans/{other.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套删预算应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
assert db.query(BudgetPlan).filter(BudgetPlan.id == other.id).first() is not None, \
|
||
"跨账套删除竟生效（对端预算行已消失）"
|
||
|
||
|
||
def test_budget_delete_plan_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套删自家预算计划 → 200 且行消失"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
mine = _mk_plan(db, k1.id, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{BUDGET}/plans/{mine.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(BudgetPlan).filter(BudgetPlan.id == mine.id).first() is None
|
||
|
||
|
||
def test_budget_update_deviation_alert_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话改 e2 偏差预警 → 404 且对端预警状态未变"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_alert(db, k2.id, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{BUDGET}/deviation-alerts/{other.id}", headers=auth_header(token1),
|
||
json={"status": "resolved"})
|
||
assert r.status_code == 404, f"跨账套改预警应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.id == other.id).first()
|
||
assert row is not None, "对端预警行被删除"
|
||
assert row.status == "open", f"跨账套改预警生效了：status={row.status}"
|
||
|
||
|
||
def test_budget_update_deviation_alert_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套改自家预警 → 200 且落库"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
mine = _mk_alert(db, k1.id, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{BUDGET}/deviation-alerts/{mine.id}", headers=auth_header(token1),
|
||
json={"status": "resolved"})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(BudgetDeviationAlert).filter(
|
||
BudgetDeviationAlert.id == mine.id).first().status == "resolved"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P1. kpis.py get_kpi_objectives 账套收口
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_kpi_objectives_requires_auth(client: TestClient, db: Session):
|
||
"""无 token → 401（鉴权面口径）"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI_TREE, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k1.id}/objectives")
|
||
assert r.status_code == 401, f"未鉴权访问 KPI 目标应 401，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
|
||
def test_kpi_objectives_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话取 e2 KPI 的目标 → 404"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k2.id}/objectives", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套取 KPI 目标应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
|
||
def test_kpi_objectives_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套 KPI 的目标/地图正常返回（正对照）"""
|
||
_seed_two_entities(db)
|
||
m1 = _mk_map(db, 1, "e1地图")
|
||
_mk_objective(db, 1, m1.id, "提升收入")
|
||
k1 = _mk_kpi(db, E1_KPI_TREE, 1, map_id=m1.id, objective="提升收入")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k1.id}/objectives", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert body["map"] and body["map"]["id"] == m1.id, f"本账套地图未返回: {body}"
|
||
assert len(body["objectives"]) == 1 and body["objectives"][0]["name"] == "提升收入"
|
||
|
||
|
||
def test_kpi_objectives_map_and_objectives_scoped_by_entity(client: TestClient, db: Session):
|
||
"""本账套 KPI 挂了他账套地图/目标 → 不得带出他账套地图与目标"""
|
||
_seed_two_entities(db)
|
||
m2 = _mk_map(db, 2, "e2地图")
|
||
_mk_objective(db, 2, m2.id, "e2目标")
|
||
# e1 的 KPI，但 map_id 指向 e2 地图（历史脏挂）→ 不应带出 e2 地图/目标
|
||
k1 = _mk_kpi(db, E1_KPI_OTHER_MAP, 1, map_id=m2.id, objective="e2目标")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k1.id}/objectives", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert not body.get("map"), f"带出了他账套战略地图: {body.get('map')}"
|
||
assert body["objectives"] == [], f"带出了他账套战略目标: {body['objectives']}"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P1. data_quality.py 质量日志族按 KPI 归属收口
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_quality_logs_list_cross_entity_invisible(client: TestClient, db: Session):
|
||
"""e1 会话列表看不到 e2 KPI 的质量日志；e2 会话看得到自己的"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
mine = _mk_log(db, k1.id)
|
||
other = _mk_log(db, k2.id)
|
||
token1 = _login(client, 1)
|
||
token2 = _login(client, 2)
|
||
|
||
r1 = client.get(f"{DQ}/logs", headers=auth_header(token1))
|
||
assert r1.status_code == 200, r1.text[:200]
|
||
ids1 = {x["id"] for x in r1.json()["data"]}
|
||
assert other.id not in ids1, f"e1 会话看到 e2 账套质量日志（id={other.id}）"
|
||
assert mine.id in ids1, f"e1 会话未看到本账套日志（id={mine.id}）"
|
||
|
||
r2 = client.get(f"{DQ}/logs", headers=auth_header(token2))
|
||
ids2 = {x["id"] for x in r2.json()["data"]}
|
||
assert other.id in ids2, "e2 会话看不到本账套日志"
|
||
|
||
|
||
def test_quality_logs_list_kpi_filter_cross_entity_empty(client: TestClient, db: Session):
|
||
"""e1 会话用他账套 kpi_id 过滤 → 空结果"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_log(db, k2.id)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{DQ}/logs?kpi_id={k2.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert {x["id"] for x in r.json()["data"]} == set(), \
|
||
f"按他账套 kpi_id 过滤竟返回日志: {r.json()['data']}"
|
||
|
||
# 同一过滤器在 e2 会话下命中（正对照，证明过滤本身有效）
|
||
token2 = _login(client, 2)
|
||
r2 = client.get(f"{DQ}/logs?kpi_id={k2.id}", headers=auth_header(token2))
|
||
assert other.id in {x["id"] for x in r2.json()["data"]}
|
||
|
||
|
||
def test_quality_log_update_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话改 e2 KPI 的质量日志 → 404 且状态未变"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_log(db, k2.id)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{DQ}/logs/{other.id}", headers=auth_header(token1),
|
||
json={"status": "resolved"})
|
||
assert r.status_code == 404, f"跨账套改质量日志应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == other.id).first()
|
||
assert row is not None and row.status == "open", f"跨账套改日志生效了：{row and row.status}"
|
||
|
||
|
||
def test_quality_log_delete_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话删 e2 KPI 的质量日志 → 404 且行仍在"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
other = _mk_log(db, k2.id)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{DQ}/logs/{other.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套删质量日志应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
assert db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == other.id).first() is not None, \
|
||
"跨账套删除竟生效（对端日志行已消失）"
|
||
|
||
|
||
def test_quality_log_update_delete_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套改/删自家质量日志 → 200（正对照）"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
a = _mk_log(db, k1.id)
|
||
b = _mk_log(db, k1.id)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{DQ}/logs/{a.id}", headers=auth_header(token1), json={"status": "resolved"})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == a.id).first().status == "resolved"
|
||
|
||
r2 = client.delete(f"{DQ}/logs/{b.id}", headers=auth_header(token1))
|
||
assert r2.status_code == 200, r2.text[:200]
|
||
db.expire_all()
|
||
assert db.query(KpiDataQualityLog).filter(KpiDataQualityLog.id == b.id).first() is None
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P2. budget.py auto_decompose 单 KPI 分支账套归属
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_auto_decompose_single_kpi_writes_session_entity(client: TestClient, db: Session):
|
||
"""e2 会话单 KPI 分解 → 12 行月度预算归属 e2（改前落默认 1）"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.post(f"{BUDGET}/auto-decompose", headers=auth_header(token2),
|
||
json={"kpi_id": k2.id, "annual_budget": 120000, "year": 2026, "method": "equal"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
|
||
db.expire_all()
|
||
rows = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == k2.id).all()
|
||
assert len(rows) == 12, f"应生成 12 行月度预算，实际 {len(rows)}"
|
||
bad = [x.id for x in rows if x.entity_id != 2]
|
||
assert not bad, f"单 KPI 分支未写会话账套 entity_id（错归属行 id={bad}）"
|
||
|
||
|
||
def test_auto_decompose_single_kpi_own_entity_ok(client: TestClient, db: Session):
|
||
"""e1 会话单 KPI 分解 → 200 且 12 行归属 e1（正对照）"""
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{BUDGET}/auto-decompose", headers=auth_header(token1),
|
||
json={"kpi_id": k1.id, "annual_budget": 120000, "year": 2026, "method": "equal"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
db.expire_all()
|
||
rows = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == k1.id).all()
|
||
assert len(rows) == 12 and all(x.entity_id == 1 for x in rows)
|
||
|
||
|
||
def test_auto_decompose_single_kpi_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话对他账套 KPI 做单 KPI 分解 → 404 且不落任何行"""
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{BUDGET}/auto-decompose", headers=auth_header(token1),
|
||
json={"kpi_id": k2.id, "annual_budget": 120000, "year": 2026, "method": "equal"})
|
||
assert r.status_code == 404, f"跨账套单 KPI 分解应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
assert db.query(BudgetPlan).filter(BudgetPlan.kpi_id == k2.id).count() == 0, \
|
||
"跨账套分解竟写出了预算行"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P2. predict.py growth-quality 按会话账套读库
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _mk_kpi_gq(db: Session, eid: int, code: str, value: float, period: str = "2026-09"):
|
||
kpi = _mk_kpi(db, code, eid)
|
||
db.add(KPIValue(kpi_id=kpi.id, entity_id=eid, period=period, actual_value=value))
|
||
db.commit()
|
||
return kpi
|
||
|
||
|
||
def test_growth_quality_body_entity_ignored(client: TestClient, db: Session):
|
||
"""请求体注入 entity 经营数据不得被采信（原 data.get("entity", data) 回退）"""
|
||
_seed_two_entities(db)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{PREDICT}/growth-quality", headers=auth_header(token1), json={
|
||
"period": "2026-09",
|
||
"entity": {"entity": "伪造企业", "rebateRate": 82.8, "trueGrossMargin": 18.6,
|
||
"cashRatio": 0.6, "expenseGrowthRate": 2.2, "revenueGrowthRate": 1.0,
|
||
"mgmtRatio": 447},
|
||
})
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert body.get("entity_name") != "伪造企业", f"请求体 entity 被采信: {body.get('entity_name')}"
|
||
assert body.get("collected") is False, f"库内无数据却给出评分: {body}"
|
||
assert body.get("dimensions") == {}, f"未采集却输出维度评分: {body.get('dimensions')}"
|
||
assert body.get("overall") is None, f"未采集却输出综合评分: {body.get('overall')}"
|
||
|
||
# e1 已有同名账套实体 → entity_name 来自库内 Entity 表
|
||
ent1 = db.query(Entity).filter(Entity.id == 1).first()
|
||
assert body.get("entity_name") == ent1.name
|
||
|
||
|
||
def test_growth_quality_body_explicit_other_entity_403(client: TestClient, db: Session):
|
||
"""请求体显式 entity_id 与会话账套不一致 → 403"""
|
||
_seed_two_entities(db)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{PREDICT}/growth-quality", headers=auth_header(token1),
|
||
json={"entity_id": 2})
|
||
assert r.status_code == 403, f"跨账套显式 entity_id 应 403，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
|
||
def test_growth_quality_reads_own_entity_kpi(client: TestClient, db: Session):
|
||
"""库内存在本账套 KPI 实际值 → collected=true 且维度评分来自库内数据"""
|
||
_seed_two_entities(db)
|
||
# 库内既有编码口径：C_REBATE_RATE / F_GROSS_MARGIN / F_REVENUE_GROWTH
|
||
_mk_kpi_gq(db, 1, "C_REBATE_RATE", 82.8)
|
||
_mk_kpi_gq(db, 1, "F_GROSS_MARGIN", 18.6)
|
||
_mk_kpi_gq(db, 1, "F_REVENUE_GROWTH", 1.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{PREDICT}/growth-quality", headers=auth_header(token1), json={"period": "2026-09"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert body.get("collected") is True, f"库内有数据却判未采集: {body}"
|
||
assert body.get("overall") is not None, f"未给出综合评分: {body}"
|
||
assert "revenueStructure" in body.get("dimensions", {}), f"营收结构维度缺失: {body.get('dimensions')}"
|
||
|
||
|
||
def test_growth_quality_compare_body_not_trusted(client: TestClient, db: Session):
|
||
"""请求体 compare 数据不得被采信"""
|
||
_seed_two_entities(db)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{PREDICT}/growth-quality", headers=auth_header(token1), json={
|
||
"compare": {"entity": "伪造对比企业", "rebateRate": 0, "mgmtRatio": 1.4},
|
||
})
|
||
assert r.status_code == 200, r.text[:300]
|
||
body = r.json()
|
||
assert not body.get("compare") or body["compare"].get("entity_name") != "伪造对比企业", \
|
||
f"请求体 compare 被采信: {body.get('compare')}"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P2. expenses.py 列表/详情/审批账套归属
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_expenses_list_cross_entity_invisible(client: TestClient, db: Session):
|
||
"""e1 会话列表看不到 e2 报销单（原声明了 entity_id 依赖但未生效）"""
|
||
_seed_two_entities(db)
|
||
mine = _mk_reimb(db, 1, "ZZB5E1BX001")
|
||
other = _mk_reimb(db, 2, "ZZB5E2BX001")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{EXPENSES}/reimbursements", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
ids = {x["id"] for x in r.json()["data"]}
|
||
assert other.id not in ids, f"e1 会话看到 e2 报销单（id={other.id}）"
|
||
assert mine.id in ids, "e1 会话看不到本账套报销单"
|
||
|
||
# 正对照：e2 会话看得到自己的
|
||
token2 = _login(client, 2)
|
||
r2 = client.get(f"{EXPENSES}/reimbursements", headers=auth_header(token2))
|
||
assert other.id in {x["id"] for x in r2.json()["data"]}
|
||
|
||
|
||
def test_expenses_detail_cross_entity_404(client: TestClient, db: Session):
|
||
"""e1 会话读 e2 报销单详情 → 404"""
|
||
_seed_two_entities(db)
|
||
other = _mk_reimb(db, 2, "ZZB5E2BX002")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{EXPENSES}/reimbursements/{other.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套读报销单详情应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
|
||
def test_expenses_detail_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套报销单详情 → 200（正对照）"""
|
||
_seed_two_entities(db)
|
||
mine = _mk_reimb(db, 1, "ZZB5E1BX002")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{EXPENSES}/reimbursements/{mine.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["reimb_no"] == "ZZB5E1BX002"
|
||
|
||
|
||
@pytest.mark.parametrize("action,payload", [
|
||
("approve", {}),
|
||
("reject", {"comment": "不合规"}),
|
||
("return", {"comment": "请补充材料"}),
|
||
])
|
||
def test_expenses_approval_cross_entity_404(client: TestClient, db: Session, action: str, payload: dict):
|
||
"""e1 会话审批 e2 报销单（approve/reject/return）→ 404 且状态未变"""
|
||
_seed_two_entities(db)
|
||
other = _mk_reimb(db, 2, f"ZZB5E2BX_{action.upper()}")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{EXPENSES}/reimbursements/{other.id}/{action}",
|
||
headers=auth_header(token1), json=payload)
|
||
assert r.status_code == 404, f"跨账套{action}应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.id == other.id).first()
|
||
assert row is not None and row.status == "pending", f"跨账套{action}生效了：{row and row.status}"
|
||
|
||
|
||
def test_expenses_approve_own_entity_ok(client: TestClient, db: Session):
|
||
"""本账套审批自家报销单 → 200（正对照）"""
|
||
_seed_two_entities(db)
|
||
mine = _mk_reimb(db, 1, "ZZB5E1BX003")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{EXPENSES}/reimbursements/{mine.id}/approve",
|
||
headers=auth_header(token1), json={})
|
||
assert r.status_code == 200, r.text[:300]
|
||
db.expire_all()
|
||
assert db.query(ExpenseReimbursement).filter(
|
||
ExpenseReimbursement.id == mine.id).first().status == "approved"
|
||
|
||
|
||
def test_tax_demo_seed_reimbursement_entity_id(client: TestClient, db: Session):
|
||
"""tax_compliance 演示播种的报销单须写会话账套 entity_id"""
|
||
_seed_two_entities(db)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.post(f"{TAX}/demo-data", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:300]
|
||
|
||
db.expire_all()
|
||
rows = db.query(ExpenseReimbursement).all()
|
||
assert rows, "演示播种未产生报销单"
|
||
bad = [(x.id, x.entity_id) for x in rows if x.entity_id != 2]
|
||
assert not bad, f"演示报销单未写会话账套（落默认 1）: {bad}"
|