sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): retire check-governance + scope /reports/dupont to session entity

Browse Source 
- data_quality.py: remove GET /check-governance (no caller repo-wide; its
  entity_id param had no session auth -> cross-tenant read; duplicate of
  governance-check) -> the path now returns 404
- reports.py: /reports/dupont resolves the analysis target from the session
  account (assert_entity_slug_scope / resolve_session_slug); explicit
  cross-entity entity param -> 403, illegal value -> 400; the 2026-08-31
  cross-entity exemption is revoked (docstring updated)
- frontend: drop the checkGovernance api definition; DupontAnalysis.vue
  entity radio group -> read-only session entity tag, request without param
- tests: TestCheckGovernanceRetired (404 x2), TestDupontSessionScope (3),
  TestDupont reworked to account-bound tokens (8 tests)
- docs/api-verification-split.md: drop the retired row, note 334/404 -> 333/403

pytest: 10 failed / 957 passed / 1 xfailed vs HEAD baseline 10 failed /
951 passed / 1 xfailed (identical failure set -> zero regression)

(cma-residual-cross-entity-cleanup-20260914)
This commit is contained in:
Hermes CI Fix
2026-09-14 12:09:04 +08:00
parent 72c1bbce16
commit aac1bd1df1
7 changed files with 147 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_reports.py
+67 -29
View File
@@ -647,16 +647,37 @@ class TestStatutory:
class TestDupont:
"""杜邦分析"""
"""杜邦分析 — 多租户口径（cma-residual-cross-entity-cleanup-20260914）
分析对象=会话账套：显式 entity 与会话不一致 → 403；非法值 → 400；
旧「跨实体对比」豁免已撤销。
"""
BASE = "/api/cma/reports/dupont"
@staticmethod
def _login_entity(client: TestClient, db: Session, entity_id: int) -> str:
"""按账套登录（1=酣客 / 2=博海）：确保企业存在 + 授权 + 取 token"""
from app.models import Entity, UserEntity
ent = db.query(Entity).filter(Entity.id == entity_id).first()
if not ent:
db.add(Entity(id=entity_id, name=f"企业{entity_id}", short_name=f"E{entity_id}", status="active"))
db.commit()
user = create_test_user(db)
if not db.query(UserEntity).filter(UserEntity.user_id == user.id, UserEntity.entity_id == entity_id).first():
db.add(UserEntity(user_id=user.id, entity_id=entity_id))
db.commit()
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
})
assert resp.status_code == 200, f"账套登录失败({entity_id}): {resp.status_code} {resp.text[:200]}"
return resp.json()["token"]
def test_bohai_fallback(self, client: TestClient, db: Session):
"""博海无DB数据 → 回退文档常量"""
create_test_user(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?entity=bohai", headers=auth_header(token))
assert resp.status_code == 200
"""博海会话 + 无 entity 参数 → 会话账套=博海；无DB数据回退文档常量"""
token = self._login_entity(client, db, 2)
resp = client.get(self.BASE, headers=auth_header(token))
assert resp.status_code == 200, resp.text[:300]
data = resp.json()
assert data["entity_name"] == "陕西博海科技（IT服务）"
assert data["roe"] > 0
@@ -665,52 +686,69 @@ class TestDupont:
def test_bohai_with_db_data(self, client: TestClient, db: Session):
"""博海(entity 2) DB有verified KPI → 用DB值"""
create_test_user(db)
token = get_token_for_user(client)
token = self._login_entity(client, db, 2)
k1 = mk_kpi(db, "F_NET_PROFIT", entity_id=2, target=0)
mk_val(db, k1, "2026-H1", 20.0)
k2 = mk_kpi(db, "F_REVENUE", entity_id=2, target=0)
mk_val(db, k2, "2026-H1", 400.0)
resp = client.get(f"{self.BASE}?entity=bohai", headers=auth_header(token))
assert resp.status_code == 200
resp = client.get(self.BASE, headers=auth_header(token))
assert resp.status_code == 200, resp.text[:300]
data = resp.json()
assert data["raw_data"]["net_profit"] == 20.0
assert data["raw_data"]["revenue"] == 400.0
assert data["roe"] > 0
def test_hanke_no_data(self, client: TestClient, db: Session):
"""酣客无数据 → roe None，不报错"""
create_test_user(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
assert resp.status_code == 200
def test_hanke_session_default_is_hanke(self, client: TestClient, db: Session):
"""酣客会话 + 无 entity 参数 → 返回酣客（不再默认博海）；酣客无数据 → roe None"""
token = self._login_entity(client, db, 1)
resp = client.get(self.BASE, headers=auth_header(token))
assert resp.status_code == 200, resp.text[:300]
data = resp.json()
assert data["entity"] == "hanke"
assert data["entity_name"] == "陕西酣客文化传媒（白酒经销）"
assert data["roe"] is None
assert data["factors"]["net_profit_margin"]["value"] is None
def test_hanke_cross_entity_bohai_403(self, client: TestClient, db: Session):
"""酣客会话显式请求 entity=bohai（博海真实KPI）→ 403（本单核心验收点）"""
token = self._login_entity(client, db, 1)
resp = client.get(f"{self.BASE}?entity=bohai", headers=auth_header(token))
assert resp.status_code == 403, f"跨实体应 403，实际 {resp.status_code}: {resp.text[:200]}"
assert "无权访问企业 entity_id=2（当前账套: 1）" in resp.text
def test_bohai_cross_entity_hanke_403(self, client: TestClient, db: Session):
"""博海会话显式请求 entity=hanke → 403"""
token = self._login_entity(client, db, 2)
resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
assert resp.status_code == 403, f"跨实体应 403，实际 {resp.status_code}: {resp.text[:200]}"
def test_same_entity_explicit_200(self, client: TestClient, db: Session):
"""显式 entity 与会话账套一致 → 200"""
token = self._login_entity(client, db, 1)
resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
assert resp.status_code == 200, resp.text[:200]
assert resp.json()["entity"] == "hanke"
def test_illegal_entity_400(self, client: TestClient, db: Session):
"""非法实体值 → 400 显式报错（不再静默回落）"""
token = self._login_entity(client, db, 1)
resp = client.get(f"{self.BASE}?entity=xxx", headers=auth_header(token))
assert resp.status_code == 400, f"非法实体应 400，实际 {resp.status_code}: {resp.text[:200]}"
assert "非法实体" in resp.text
def test_hanke_with_loss_data(self, client: TestClient, db: Session):
"""酣客亏损数据 → 负ROE"""
create_test_user(db)
token = get_token_for_user(client)
"""酣客会话亏损数据 → 负ROE"""
token = self._login_entity(client, db, 1)
k1 = mk_kpi(db, "F_REVENUE", entity_id=1, target=0)
mk_val(db, k1, "2026-H1", 100.0)
k2 = mk_kpi(db, "F_NET_PROFIT", entity_id=1, target=0)
mk_val(db, k2, "2026-H1", -10.0)
resp = client.get(f"{self.BASE}?entity=hanke", headers=auth_header(token))
assert resp.status_code == 200
resp = client.get(self.BASE, headers=auth_header(token))
assert resp.status_code == 200, resp.text[:300]
data = resp.json()
assert data["roe"] == -10.0
assert data["factors"]["net_profit_margin"]["status"] == "🔴"
def test_unknown_entity(self, client: TestClient, db: Session):
"""未知实体 → error"""
create_test_user(db)
token = get_token_for_user(client)
resp = client.get(f"{self.BASE}?entity=xxx", headers=auth_header(token))
assert resp.status_code == 200
assert "error" in resp.json()
class TestGenerateReports:
"""自动报告生成（周报/月报/专项）+ 历史"""