Hermes CI Fix
50baa16f09
list_plans enriched kr_title via KR.id only -> legacy rows whose kr_id points at another entity's KR echoed the foreign KR title. Add KR.entity_id filter (read-side sibling of batch3 item 4). Local commit only; no push (push triggers Gitea deploy hook).
466 lines
20 KiB
Python
466 lines
20 KiB
Python
"""CMA 跨实体隔离收口（第三批）账套隔离回归测试 — 2026-09-18
|
||
|
||
任务：project-20260918-entity-isolation-batch3（项目Bot 派单）
|
||
缺陷（本单范围，均与 batch2 同类残留）：
|
||
|
||
A. 写型未接账套
|
||
1. POST /api/cma/cost/abc/activities 函数签名无 get_entity_id 依赖；AbcActivity(...) 不写归属
|
||
→ 落模型默认 entity_id=1（非 e1 账套新建的作业中心落进 e1）
|
||
2. POST /api/cma/cost/abc/allocate 无账套依赖；cost_engine.allocate_cost() 内 AbcAllocation(...)
|
||
不写 entity_id → 分配记录全落 e1；activity_id 也不校验归属
|
||
|
||
B. 读型未接账套
|
||
3. GET /api/cma/cost/abc/allocations 查询只按 period/product_code 过滤 → 跨账套读泄漏
|
||
（与 batch2 已修的 GET /abc/activities 同族）
|
||
4. GET /api/cma/action-plans/stats 统计 db.query(ActionPlan) 无账套过滤
|
||
→ 非 e1 账套用户统计数含其它账套计划（同文件既有先例=join KPIDefinition）
|
||
|
||
C. 归属校验缺失（写型）
|
||
5. PUT /api/cma/action-plans/{plan_id} plan_id 已按账套校验（batch2），但 kr_id 分支
|
||
db.query(KR).filter(KR.id == data["kr_id"]) 无账套校验
|
||
→ 可把本账套计划挂到对端账套的 KR（同文件 create_plan:158 有先例）
|
||
|
||
本用例覆盖（每条均双向 + 同账套正对照；改前必失败条目见回报 §三 RED 原文）：
|
||
- 写型：会话账套新建 ABC 作业中心 → 库内该行 entity_id = 会话账套；对端账套看不到；
|
||
- 写型：本账套 activity_id 分配 → 200 且新建 abc_allocations 行 entity_id = 会话账套；
|
||
对端 activity_id → 404 且不产生任何分配行；
|
||
- 读型：GET /abc/allocations 只返回本账套行；
|
||
- 归属：PUT /{plan_id} 传对端 kr_id → 404 且对端 KR/本账套计划均未被改动；同账套改挂 → 200；
|
||
- 统计：GET /action-plans/stats 只统计本账套计划；
|
||
- 鉴权底线：无 token → 401（改前改后都应通过）。
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import (
|
||
AbcActivity,
|
||
AbcAllocation,
|
||
ActionPlan,
|
||
Entity,
|
||
KR,
|
||
KPIDefinition,
|
||
Objective,
|
||
)
|
||
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
|
||
|
||
COST = "/api/cma/cost"
|
||
PLANS = "/api/cma/action-plans"
|
||
|
||
# 账套 1/2 各自的资源编码前缀（activity_code 库内唯一）
|
||
E1_KPI = "ZZ_B3_E1_KPI_001"
|
||
E2_KPI = "ZZ_B3_E2_KPI_001"
|
||
E1_ACT = "ZZ_B3_E1_ACT_001"
|
||
E2_ACT = "ZZ_B3_E2_ACT_001"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 夹具
|
||
# ──────────────────────────────────────────────
|
||
|
||
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) -> 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",
|
||
)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
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 _mk_activity(db: Session, code: str, eid: int) -> AbcActivity:
|
||
"""直插作业中心（显式 entity_id，用于构造「对端账套既有数据」）"""
|
||
act = AbcActivity(
|
||
entity_id=eid, activity_code=code, activity_name=f"{code}-作业",
|
||
cost_driver="动因", driver_unit="次", total_cost=1000.0, driver_volume=10,
|
||
driver_rate=100.0, status="active",
|
||
)
|
||
db.add(act)
|
||
db.commit()
|
||
db.refresh(act)
|
||
return act
|
||
|
||
|
||
def _mk_allocation(db: Session, activity: AbcActivity, eid: int, product_code: str) -> AbcAllocation:
|
||
alloc = AbcAllocation(
|
||
entity_id=eid, period="2026-06", activity_id=activity.id,
|
||
product_code=product_code, product_name=f"{product_code}-产品",
|
||
driver_consumed=3, allocated_cost=300.0,
|
||
)
|
||
db.add(alloc)
|
||
db.commit()
|
||
db.refresh(alloc)
|
||
return alloc
|
||
|
||
|
||
def _mk_plan(db: Session, kpi_id: int, eid: int, title: str, **kw) -> ActionPlan:
|
||
plan = ActionPlan(entity_id=eid, kpi_id=kpi_id, kr_id=kw.get("kr_id"), title=title,
|
||
assignee=kw.get("assignee", "张三"), priority="medium",
|
||
status=kw.get("status", "pending"), progress=0)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
def _mk_objective_kr(db: Session, eid: int, title: str) -> KR:
|
||
obj = Objective(entity_id=eid, title=f"{title}-目标", quarter="2026Q3", status="active")
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
kr = KR(entity_id=eid, objective_id=obj.id, title=title, operator=">=",
|
||
weight=50, target_value=100)
|
||
db.add(kr)
|
||
db.commit()
|
||
db.refresh(kr)
|
||
return kr
|
||
|
||
|
||
def _seed_two_entities(db: Session):
|
||
_ensure_entity(db, 1, "酣客")
|
||
_ensure_entity(db, 2, "博海")
|
||
create_test_user(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
return k1, k2
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# A1. 写型：POST /cost/abc/activities 按会话账套写入
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_create_abc_activity_writes_session_entity(client: TestClient, db: Session):
|
||
"""博海(e2)会话新建作业中心 → 库内该行 entity_id=2；酣客(e1)会话看不到它"""
|
||
_seed_two_entities(db)
|
||
|
||
token2 = _login(client, 2)
|
||
r = client.post(f"{COST}/abc/activities", headers=auth_header(token2), json={
|
||
"activity_code": E2_ACT,
|
||
"activity_name": "e2 新建作业",
|
||
"cost_driver": "动因",
|
||
"driver_unit": "次",
|
||
"total_cost": 2000.0,
|
||
"driver_volume": 20,
|
||
})
|
||
assert r.status_code == 200, r.text
|
||
new_id = r.json()["id"]
|
||
|
||
db.expire_all()
|
||
row = db.query(AbcActivity).filter(AbcActivity.id == new_id).first()
|
||
assert row is not None, "作业中心未落库"
|
||
assert row.entity_id == 2, (
|
||
f"会话账套=2 新建的作业中心落库归属 entity_id={row.entity_id}（缺陷：未写归属→模型默认 1）"
|
||
)
|
||
|
||
# e1 会话看不到 e2 的作业中心；e2 会话能看到
|
||
codes1 = _abc_codes(client, _login(client, 1))
|
||
codes2 = _abc_codes(client, _login(client, 2))
|
||
assert E2_ACT not in codes1, f"酣客(e1)会话看到了博海(e2)新建的作业中心: {sorted(codes1)}"
|
||
assert E2_ACT in codes2, f"博海(e2)会话看不到自己新建的作业中心: {sorted(codes2)}"
|
||
|
||
|
||
def test_create_abc_activity_e1_still_defaults_to_own_entity(client: TestClient, db: Session):
|
||
"""酣客(e1)会话新建 → entity_id=1 且对端(e2)看不到（同账套正对照，防「一律写死某账套」）"""
|
||
_seed_two_entities(db)
|
||
|
||
token1 = _login(client, 1)
|
||
r = client.post(f"{COST}/abc/activities", headers=auth_header(token1), json={
|
||
"activity_code": E1_ACT,
|
||
"activity_name": "e1 新建作业",
|
||
"cost_driver": "动因",
|
||
"driver_unit": "次",
|
||
"total_cost": 1000.0,
|
||
"driver_volume": 10,
|
||
})
|
||
assert r.status_code == 200, r.text
|
||
new_id = r.json()["id"]
|
||
|
||
db.expire_all()
|
||
row = db.query(AbcActivity).filter(AbcActivity.id == new_id).first()
|
||
assert row.entity_id == 1, f"会话账套=1 新建行归属 {row.entity_id}"
|
||
|
||
codes2 = _abc_codes(client, _login(client, 2))
|
||
assert E1_ACT not in codes2, f"博海(e2)会话看到了酣客(e1)作业中心: {sorted(codes2)}"
|
||
|
||
|
||
def _abc_codes(client: TestClient, token: str) -> set:
|
||
r = client.get(f"{COST}/abc/activities", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
return {row["activity_code"] for row in r.json()["data"]}
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# A2. 写型：POST /cost/abc/allocate 按会话账套 + activity_id 归属校验
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_allocate_cross_entity_activity_rejected(client: TestClient, db: Session):
|
||
"""酣客(e1)会话用博海(e2)的 activity_id 分配 → 404，且不产生任何分配行"""
|
||
_seed_two_entities(db)
|
||
other = _mk_activity(db, E2_ACT, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{COST}/abc/allocate", headers=auth_header(token1), json={
|
||
"activity_id": other.id,
|
||
"period": "2026-06",
|
||
"product_code": "ZZ_B3_E1_PROD",
|
||
"product_name": "跨账套产品",
|
||
"driver_consumed": 5,
|
||
})
|
||
assert r.status_code == 404, (
|
||
f"跨账套 allocate 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
)
|
||
|
||
db.expire_all()
|
||
assert db.query(AbcAllocation).count() == 0, "跨账套 allocate 竟写入了分配记录"
|
||
# 对端作业中心未被改动（分配率计算会回写 driver_rate）
|
||
db.refresh(other)
|
||
assert other.driver_rate == 100.0, "跨账套 allocate 竟改动了对端作业中心"
|
||
|
||
|
||
def test_allocate_writes_session_entity(client: TestClient, db: Session):
|
||
"""双向：e1/e2 各自用本账套 activity_id 分配 → 200 且新建行 entity_id = 会话账套"""
|
||
_seed_two_entities(db)
|
||
act1 = _mk_activity(db, E1_ACT, 1)
|
||
act2 = _mk_activity(db, E2_ACT, 2)
|
||
|
||
# e2 会话分配本账套作业中心
|
||
token2 = _login(client, 2)
|
||
r = client.post(f"{COST}/abc/allocate", headers=auth_header(token2), json={
|
||
"activity_id": act2.id,
|
||
"period": "2026-06",
|
||
"product_code": "ZZ_B3_E2_PROD",
|
||
"product_name": "e2 产品",
|
||
"driver_consumed": 4,
|
||
})
|
||
assert r.status_code == 200, f"同账套 allocate 应 200，实际 {r.status_code} {r.text[:200]}"
|
||
assert r.json()["allocated_cost"] == 400.0 # 4 × 100.0
|
||
|
||
db.expire_all()
|
||
row2 = (db.query(AbcAllocation)
|
||
.filter(AbcAllocation.product_code == "ZZ_B3_E2_PROD").first())
|
||
assert row2 is not None, "e2 会话分配未落库"
|
||
assert row2.entity_id == 2, (
|
||
f"e2 会话分配行归属 entity_id={row2.entity_id}（缺陷：AbcAllocation 未写归属→默认 1）"
|
||
)
|
||
|
||
# e1 会话分配本账套作业中心（反向对照）
|
||
token1 = _login(client, 1)
|
||
r = client.post(f"{COST}/abc/allocate", headers=auth_header(token1), json={
|
||
"activity_id": act1.id,
|
||
"period": "2026-06",
|
||
"product_code": "ZZ_B3_E1_PROD",
|
||
"product_name": "e1 产品",
|
||
"driver_consumed": 2,
|
||
})
|
||
assert r.status_code == 200, r.text
|
||
|
||
db.expire_all()
|
||
row1 = (db.query(AbcAllocation)
|
||
.filter(AbcAllocation.product_code == "ZZ_B3_E1_PROD").first())
|
||
assert row1 is not None and row1.entity_id == 1, f"e1 分配行归属异常: {row1.entity_id if row1 else None}"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# B3. 读型：GET /cost/abc/allocations 按会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _alloc_products(client: TestClient, token: str) -> set:
|
||
r = client.get(f"{COST}/abc/allocations", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
return {row["product_code"] for row in r.json()["data"]}
|
||
|
||
|
||
def test_allocations_list_isolated_by_entity(client: TestClient, db: Session):
|
||
"""酣客(e1)会话只返本账套分配行，不含博海(e2)行；e2 会话同理（双向）"""
|
||
_seed_two_entities(db)
|
||
act1 = _mk_activity(db, E1_ACT, 1)
|
||
act2 = _mk_activity(db, E2_ACT, 2)
|
||
_mk_allocation(db, act1, 1, "ZZ_B3_E1_ALLOC")
|
||
_mk_allocation(db, act2, 2, "ZZ_B3_E2_ALLOC")
|
||
|
||
prods1 = _alloc_products(client, _login(client, 1))
|
||
prods2 = _alloc_products(client, _login(client, 2))
|
||
|
||
assert prods1 == {"ZZ_B3_E1_ALLOC"}, f"e1 会话分配记录集应为本账套，实际 {sorted(prods1)}"
|
||
assert prods2 == {"ZZ_B3_E2_ALLOC"}, f"e2 会话分配记录集应为本账套，实际 {sorted(prods2)}"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# C5. PUT /action-plans/{plan_id} 的 kr_id 归属校验
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_update_plan_kr_id_cross_entity_404(client: TestClient, db: Session):
|
||
"""酣客(e1)计划改挂博海(e2)KR → 404，且计划 kr_id 未变、对端 KR 未被改动"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_plan(db, k1.id, 1, "e1 计划")
|
||
kr2 = _mk_objective_kr(db, 2, "e2 KR")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{PLANS}/{mine.id}", headers=auth_header(token1),
|
||
json={"kr_id": kr2.id})
|
||
assert r.status_code == 404, (
|
||
f"跨账套 kr_id 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
)
|
||
|
||
db.expire_all()
|
||
after = db.query(ActionPlan).filter(ActionPlan.id == mine.id).first()
|
||
assert after.kr_id is None, f"跨账套 kr_id 竟被写入: kr_id={after.kr_id}"
|
||
assert after.objective_id is None, f"跨账套改挂竟改动了 objective_id={after.objective_id}"
|
||
|
||
db.refresh(kr2)
|
||
assert kr2.entity_id == 2 and kr2.title == "e2 KR", "对端 KR 被改动"
|
||
|
||
|
||
def test_update_plan_kr_id_same_entity_200(client: TestClient, db: Session):
|
||
"""同账套正对照：酣客(e1)计划改挂本账套 KR → 200 且 kr_id/objective_id 生效"""
|
||
k1, _ = _seed_two_entities(db)
|
||
mine = _mk_plan(db, k1.id, 1, "e1 计划")
|
||
kr1 = _mk_objective_kr(db, 1, "e1 KR")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{PLANS}/{mine.id}", headers=auth_header(token1),
|
||
json={"kr_id": kr1.id})
|
||
assert r.status_code == 200, f"同账套 kr_id 应 200，实际 {r.status_code} {r.text[:200]}"
|
||
body = r.json()
|
||
assert body["kr_id"] == kr1.id
|
||
assert body["objective_id"] == kr1.objective_id
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# B4. GET /action-plans/stats 按会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _stats(client: TestClient, token: str) -> dict:
|
||
r = client.get(f"{PLANS}/stats", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
def test_plan_stats_isolated_by_entity(client: TestClient, db: Session):
|
||
"""酣客(e1)会话统计只含本账套计划（e2 造 2 条做对照）"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
_mk_plan(db, k1.id, 1, "e1 待开始")
|
||
_mk_plan(db, k1.id, 1, "e1 已完成", status="completed")
|
||
_mk_plan(db, k2.id, 2, "e2 待开始")
|
||
_mk_plan(db, k2.id, 2, "e2 进行中", status="in_progress")
|
||
|
||
s1 = _stats(client, _login(client, 1))
|
||
assert s1["total"] == 2, f"e1 统计应只含本账套 2 条，实际 {s1}"
|
||
assert s1["pending"] == 1 and s1["completed"] == 1 and s1["in_progress"] == 0
|
||
|
||
|
||
def test_plan_stats_reverse_direction(client: TestClient, db: Session):
|
||
"""博海(e2)会话统计只含本账套计划（防止「一律统计 e1」或全局口径）"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
_mk_plan(db, k1.id, 1, "e1 待开始")
|
||
_mk_plan(db, k1.id, 1, "e1 待开始2")
|
||
_mk_plan(db, k2.id, 2, "e2 待开始")
|
||
|
||
s2 = _stats(client, _login(client, 2))
|
||
assert s2["total"] == 1, f"e2 统计应只含本账套 1 条，实际 {s2}"
|
||
assert s2["pending"] == 1
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# D6. GET /action-plans 的 KR 富化按会话账套（batch3 读侧同族残余）
|
||
# ──────────────────────────────────────────────
|
||
# 缺陷：list_plans 富化 kr_title 的查询 db.query(KR).filter(KR.id == p.kr_id) 无账套校验
|
||
# → 存量脏行（action_plans.kr_id 指向对端账套 KR）会让列表跨账套回显对端 KR 标题。
|
||
# 写入侧已在 batch3 修（update_plan/create_plan 校验 kr_id 归属），本用例覆盖读侧。
|
||
|
||
def _list_plans(client: TestClient, token: str) -> dict:
|
||
r = client.get(PLANS, headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
return {row["id"]: row for row in r.json()["data"]}
|
||
|
||
|
||
def test_list_plans_kr_enrichment_isolated_by_entity(client: TestClient, db: Session):
|
||
"""双向：本账套计划挂对端 KR（存量脏行）→ 列表不回显对端 KR 标题；挂本账套 KR → 正常回显"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
kr1 = _mk_objective_kr(db, 1, "e1 KR")
|
||
kr2 = _mk_objective_kr(db, 2, "e2 KR")
|
||
|
||
# e1 账套：一条挂本账套 KR（正对照）+ 一条脏行挂 e2 的 KR
|
||
good1 = _mk_plan(db, k1.id, 1, "zz e1 挂本账套 KR", kr_id=kr1.id)
|
||
dirty1 = _mk_plan(db, k1.id, 1, "zz e1 脏行挂对端 KR", kr_id=kr2.id)
|
||
# e2 账套：反向同理
|
||
good2 = _mk_plan(db, k2.id, 2, "zz e2 挂本账套 KR", kr_id=kr2.id)
|
||
dirty2 = _mk_plan(db, k2.id, 2, "zz e2 脏行挂对端 KR", kr_id=kr1.id)
|
||
|
||
rows1 = _list_plans(client, _login(client, 1))
|
||
assert good1.id in rows1 and dirty1.id in rows1, f"e1 会话看不到本账套计划: {sorted(rows1)}"
|
||
assert rows1[good1.id]["kr_title"] == "e1 KR", (
|
||
f"e1 本账套 KR 标题未回显: {rows1[good1.id].get('kr_title')!r}"
|
||
)
|
||
assert rows1[dirty1.id]["kr_title"] is None, (
|
||
f"e1 会话跨账套回显了 e2 的 KR 标题 {rows1[dirty1.id]['kr_title']!r}"
|
||
"（缺陷：list_plans 富化 kr 查询缺 KR.entity_id 校验）"
|
||
)
|
||
assert rows1[dirty1.id]["kr_metric_kpi_id"] is None, (
|
||
f"e1 会话跨账套回显了 e2 KR 的 metric_kpi_id = {rows1[dirty1.id]['kr_metric_kpi_id']!r}"
|
||
)
|
||
|
||
rows2 = _list_plans(client, _login(client, 2))
|
||
assert good2.id in rows2 and dirty2.id in rows2, f"e2 会话看不到本账套计划: {sorted(rows2)}"
|
||
assert rows2[good2.id]["kr_title"] == "e2 KR", (
|
||
f"e2 本账套 KR 标题未回显: {rows2[good2.id].get('kr_title')!r}"
|
||
)
|
||
assert rows2[dirty2.id]["kr_title"] is None, (
|
||
f"e2 会话跨账套回显了 e1 的 KR 标题 {rows2[dirty2.id]['kr_title']!r}"
|
||
)
|
||
assert rows2[dirty2.id]["kr_metric_kpi_id"] is None
|
||
|
||
# 计划本体归属未被本改动影响：e1 列表不含 e2 的两条计划
|
||
assert good2.id not in rows1 and dirty2.id not in rows1, "e1 列表混入了 e2 账套计划"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 鉴权底线（改前改后都应通过）
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_batch3_endpoints_without_token_401(client: TestClient, db: Session):
|
||
"""无 Bearer token → 401（四个端点鉴权底线不变）"""
|
||
_seed_two_entities(db)
|
||
assert client.post(f"{COST}/abc/activities", json={
|
||
"activity_code": "ZZ_B3_NOTOKEN", "activity_name": "x",
|
||
"cost_driver": "d", "total_cost": 1, "driver_volume": 1,
|
||
}).status_code == 401
|
||
assert client.post(f"{COST}/abc/allocate", json={
|
||
"activity_id": 1, "product_code": "p", "driver_consumed": 1,
|
||
}).status_code == 401
|
||
assert client.get(f"{COST}/abc/allocations").status_code == 401
|
||
assert client.get(f"{PLANS}/stats").status_code == 401
|