fullstackandHermes CI Fix
bd4f58aae1
ci/woodpecker/push/main Pipeline failed
- 后端新增 POST /api/cma/action-plans/coso-generate-actions：未通过条目→action_plans， 账套取 token 绑定（忽略 body entity_id）、幂等键 (entity_id,title)、不写 auto_verify_rule - 前端 ActionPlanLibrary.vue 按钮接 @click（含未通过条数提示 + 二次确认 + 生成后刷新并跳列表） - api 客户端补 cosoGenerateActions - 新增 tests/test_coso_generate_actions_20260921.py（生成/幂等/隔离/未认证/归属校验/空集）
279 lines
13 KiB
Python
279 lines
13 KiB
Python
"""COSO内控自检「生成改善行动」端点测试
|
||
|
||
任务：project-20260921-coso-generate-actions
|
||
（自检「未通过」条目 → action_plans 写路径，前端「生成改善行动」按钮的后端实现）
|
||
|
||
本文件覆盖
|
||
① 生成：e1（hanke）会话点击 → 200；created == 未通过条目数（从 COSO_CHECKLIST_DATA
|
||
现算，不写字面量）；新建行 entity_id=1 / status=pending / kpi 属于 e1 /
|
||
title 前缀 [COSO] / description 含「来源: COSO内控自检」/ auto_verify_rule is None；
|
||
passed=True 的条目没有对应行
|
||
② 幂等：同一会话再调一次 → created=0、skipped=上次 created、DB 行数不变
|
||
③ 账套隔离：e2（bohai）会话生成的行 entity_id=2；e1/e2 列表互不可见
|
||
④ 未认证：无 token → 401
|
||
⑤ 显式 kpi_id 归属校验：e1 会话传 e2 的 kpi_id → 404（不得跨账套写）
|
||
⑥ 无未通过条目：monkeypatch 全 passed=True → 200 / created=0 / failed_total=0 / 不建行
|
||
"""
|
||
import copy
|
||
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.api import action_plans as ap
|
||
from app.models import ActionPlan, Entity, KPIDefinition
|
||
from tests.conftest import (
|
||
TEST_USER_PASSWORD,
|
||
auth_header,
|
||
create_test_kpi,
|
||
create_test_user,
|
||
get_token_for_user,
|
||
)
|
||
|
||
E1_KPI_CODE = "ZZ_COSO_E1_KPI_001"
|
||
E2_KPI_CODE = "ZZ_COSO_E2_KPI_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 _login(client: TestClient, entity_id: int, username: str = "testadmin") -> str:
|
||
"""按账套登录并返回 token（e2 会话用；e1 用 conftest 的 get_token_for_user）"""
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": username,
|
||
"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 _failed_items(coso: dict) -> list:
|
||
"""自检未通过条目 [(element, item)]（与端点同一口径：not item.get('passed')）"""
|
||
out = []
|
||
for el in coso["elements"]:
|
||
for item in el.get("items") or []:
|
||
if not item.get("passed"):
|
||
out.append((el, item))
|
||
return out
|
||
|
||
|
||
def _failed_titles(slug: str) -> set:
|
||
"""按端点生成规则计算的标题集合（[COSO] {el}·{item}，截断 200）"""
|
||
coso = ap.COSO_CHECKLIST_DATA[slug]
|
||
return {
|
||
f"[COSO] {el['name']}·{item['text']}"[:200]
|
||
for el, item in _failed_items(coso)
|
||
}
|
||
|
||
|
||
def _seed_e1(db: Session) -> KPIDefinition:
|
||
"""e1（酣客）基础夹具：实体 + ceo 用户 + 1 个本账套 KPI"""
|
||
_ensure_entity(db, 1, "酣客")
|
||
create_test_user(db)
|
||
return create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name=f"{E1_KPI_CODE}-名称", entity_id=1)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ① 生成
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_generate_creates_rows_for_failed_items(client: TestClient, db: Session):
|
||
"""生成：未通过条目各建一行、字段口径全部达标；passed=True 条目不建行
|
||
|
||
断言依据：hanke 自检共 13 条（未通过 10 条）——条数用代码从 COSO_CHECKLIST_DATA
|
||
现算而非写字面量，防自检文案/条目数将来调整时用例失真。
|
||
"""
|
||
kpi = _seed_e1(db)
|
||
token = get_token_for_user(client)
|
||
h = auth_header(token)
|
||
|
||
resp = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={})
|
||
assert resp.status_code == 200, resp.text[:300]
|
||
body = resp.json()
|
||
|
||
coso = ap.COSO_CHECKLIST_DATA["hanke"]
|
||
failed = _failed_items(coso)
|
||
expected_titles = {f"[COSO] {el['name']}·{it['text']}"[:200] for el, it in failed}
|
||
passed_titles = {
|
||
f"[COSO] {el['name']}·{it['text']}"[:200]
|
||
for el in coso["elements"]
|
||
for it in (el.get("items") or [])
|
||
if it.get("passed")
|
||
}
|
||
|
||
assert len(failed) > 0, "夹具失配：hanke 自检应存在未通过条目"
|
||
assert body["failed_total"] == len(failed), body
|
||
assert body["created"] == len(failed), body # 首轮无已存在行，created == 未通过总数
|
||
assert body["skipped"] == 0, body
|
||
assert body["entity_id"] == 1, body
|
||
assert body["entity_slug"] == "hanke", body
|
||
assert body["kpi_id"] == kpi.id, body # 自动选取落在本账套唯一 KPI
|
||
assert body["source"] == "coso", body
|
||
|
||
rows = db.query(ActionPlan).filter(ActionPlan.title.like("[COSO]%")).all()
|
||
assert len(rows) == body["created"] == len(body["created_ids"]), body
|
||
actual_titles = {r.title for r in rows}
|
||
assert actual_titles == expected_titles, "新建标题集合与未通过条目集合不一致"
|
||
assert not (actual_titles & passed_titles), "passed=True 条目不应生成对应行"
|
||
for r in rows:
|
||
assert r.entity_id == 1, r.title # 写库账套只来自 token（get_entity_id）
|
||
assert r.status == "pending", r.title
|
||
assert r.progress == 0, r.title
|
||
assert r.kpi_id == kpi.id, r.title
|
||
# kpi 归属校验：挂的 KPI 必须属于本账套（列表按 KPI 归属做账套 join 的口径）
|
||
kpi_ent = db.query(KPIDefinition).filter(KPIDefinition.id == r.kpi_id).first()
|
||
assert kpi_ent is not None and kpi_ent.entity_id == 1, r.title
|
||
assert r.title.startswith("[COSO]"), r.title
|
||
assert "来源: COSO内控自检" in (r.description or ""), r.title
|
||
assert r.auto_verify_rule is None, r.title # 显式不生成验证规则
|
||
assert r.due_date is not None, r.title # 默认 +30 天
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ② 幂等
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_idempotent_second_call_skips_existing(client: TestClient, db: Session):
|
||
"""幂等：同一会话第二次调用 created=0、skipped=首次 created、DB 行数不变
|
||
|
||
判重键 (entity_id, title)：重复点击「生成改善行动」不得重复造行。
|
||
"""
|
||
_seed_e1(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
|
||
first = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={})
|
||
assert first.status_code == 200
|
||
body1 = first.json()
|
||
count_after_first = db.query(ActionPlan).count()
|
||
assert count_after_first == body1["created"] > 0
|
||
|
||
second = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={})
|
||
assert second.status_code == 200
|
||
body2 = second.json()
|
||
assert body2["created"] == 0, body2
|
||
assert body2["skipped"] == body1["created"], body2
|
||
assert body2["failed_total"] == body1["failed_total"], body2
|
||
assert db.query(ActionPlan).count() == count_after_first, "幂等调用不得新增行"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ③ 账套隔离
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_entity_isolation_e1_e2(client: TestClient, db: Session):
|
||
"""账套隔离：e2 生成的行 entity_id=2 且挂 e2 KPI；e1/e2 列表互不可见
|
||
|
||
两个账套存在同名标题（如「控制环境·授权审批制度」两个账套都未通过）——
|
||
这正好验证判重键带账套过滤：对端账套同标题行不得抑制本账套创建，
|
||
列表也不能把对端行算进来。
|
||
"""
|
||
_ensure_entity(db, 1, "酣客")
|
||
_ensure_entity(db, 2, "博海")
|
||
create_test_user(db)
|
||
k1 = create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name="e1内控KPI", entity_id=1)
|
||
k2 = create_test_kpi(db, kpi_code=E2_KPI_CODE, kpi_name="e2内控KPI", entity_id=2)
|
||
|
||
h1 = auth_header(_login(client, 1))
|
||
h2 = auth_header(_login(client, 2))
|
||
|
||
r1 = client.post("/api/cma/action-plans/coso-generate-actions", headers=h1, json={})
|
||
r2 = client.post("/api/cma/action-plans/coso-generate-actions", headers=h2, json={})
|
||
assert r1.status_code == 200, r1.text[:300]
|
||
assert r2.status_code == 200, r2.text[:300]
|
||
|
||
# e2 落库行：entity_id=2 且挂 e2 自己的 KPI
|
||
e2_rows = db.query(ActionPlan).filter(ActionPlan.entity_id == 2).all()
|
||
assert len(e2_rows) == r2.json()["created"] == len(_failed_items(ap.COSO_CHECKLIST_DATA["bohai"]))
|
||
for row in e2_rows:
|
||
assert row.entity_id == 2, row.title
|
||
assert row.kpi_id == k2.id, row.title
|
||
|
||
# 列表互不可见：各自列表 == 各自自检未通过标题全集（含同名标题，也绝不含对端专属行）
|
||
e1_list = {row["title"] for row in client.get("/api/cma/action-plans", headers=h1).json()["data"]}
|
||
e2_list = {row["title"] for row in client.get("/api/cma/action-plans", headers=h2).json()["data"]}
|
||
assert e1_list == _failed_titles("hanke"), e1_list
|
||
assert e2_list == _failed_titles("bohai"), e2_list
|
||
e1_only = _failed_titles("hanke") - _failed_titles("bohai")
|
||
e2_only = _failed_titles("bohai") - _failed_titles("hanke")
|
||
assert not (e1_list & e2_only), "e1 列表不应出现 e2 专属条目"
|
||
assert not (e2_list & e1_only), "e2 列表不应出现 e1 专属条目"
|
||
# 差异集非空才说明上面的断言有鉴别力（两账套未通过集合确有重叠与差异）
|
||
assert e1_only and e2_only
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ④ 未认证
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_unauthenticated_returns_401(client: TestClient, db: Session):
|
||
"""未认证：无 token 调生成端点 → 401（写路径必须鉴权）"""
|
||
_seed_e1(db)
|
||
resp = client.post("/api/cma/action-plans/coso-generate-actions", json={})
|
||
assert resp.status_code == 401, resp.text[:300]
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ⑤ 显式 kpi_id 归属校验
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_explicit_kpi_id_must_belong_to_session_entity(client: TestClient, db: Session):
|
||
"""显式 kpi_id 归属校验：e1 会话传 e2 的 kpi_id → 404 且不落任何行
|
||
|
||
kpi_id 是本账套隔离的唯一 join 锚点——若允许挂对端 KPI，
|
||
本账套列表会按 KPI 归属 join 而丢失该行，形成「孤儿行」。
|
||
"""
|
||
_ensure_entity(db, 1, "酣客")
|
||
_ensure_entity(db, 2, "博海")
|
||
create_test_user(db)
|
||
create_test_kpi(db, kpi_code=E1_KPI_CODE, kpi_name="e1 KPI", entity_id=1)
|
||
k2 = create_test_kpi(db, kpi_code=E2_KPI_CODE, kpi_name="e2 KPI", entity_id=2)
|
||
|
||
h = auth_header(_login(client, 1))
|
||
resp = client.post("/api/cma/action-plans/coso-generate-actions",
|
||
headers=h, json={"kpi_id": k2.id})
|
||
assert resp.status_code == 404, resp.text[:300]
|
||
assert db.query(ActionPlan).count() == 0, "跨账套 kpi_id 被拒后不得残留任何行"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# ⑥ 无未通过条目
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_no_failed_items_returns_zero_not_error(client: TestClient, db: Session, monkeypatch):
|
||
"""无未通过条目：全 passed=True → 200 / created=0 / failed_total=0 / 不建行
|
||
|
||
monkeypatch 模块级 COSO_CHECKLIST_DATA（深拷贝，测后自动还原）——
|
||
端点必须把它当「空结果」而不是报错。
|
||
"""
|
||
_seed_e1(db)
|
||
original = ap.COSO_CHECKLIST_DATA # patch 前持有原始引用，用于测后还原校验
|
||
patched = copy.deepcopy(ap.COSO_CHECKLIST_DATA)
|
||
for entity_data in patched.values():
|
||
for el in entity_data["elements"]:
|
||
for it in el["items"]:
|
||
it["passed"] = True
|
||
monkeypatch.setattr(ap, "COSO_CHECKLIST_DATA", patched)
|
||
|
||
h = auth_header(get_token_for_user(client))
|
||
resp = client.post("/api/cma/action-plans/coso-generate-actions", headers=h, json={})
|
||
assert resp.status_code == 200, resp.text[:300]
|
||
body = resp.json()
|
||
assert body["failed_total"] == 0, body
|
||
assert body["created"] == 0 and body["skipped"] == 0, body
|
||
assert body["created_ids"] == [], body
|
||
assert db.query(ActionPlan).count() == 0, "无未通过条目时不得新建行"
|
||
|
||
# 还原校验：patch 基于深拷贝，原始数据未被改坏（仍含未通过条目）
|
||
assert len(_failed_items(original["hanke"])) > 0
|