Hermes CI Fix
40d522cd34
- Login.vue: no prefilled default creds; no anonymous entity enumeration on mount - deploy.sh: readiness polling (0.5s/30s cap) replaces fixed sleep 3; login smoke reads same-source 600 secret file, missing file = explicit abort - 19 verify/check/gen_report scripts: shared _admin_creds() loader (CMA_ADMIN_ENV -> ops .secrets -> profile copy); no hardcoded creds - tests: single TEST_USER_PASSWORD fixture constant; docs: pointer to 600 secret file; removed 4 legacy .bak + pycache residue
231 lines
9.8 KiB
Python
231 lines
9.8 KiB
Python
"""数据管理模块脱节修复测试（2026-09-08 P0，project-20260908-data-manage-fix）
|
||
|
||
覆盖：
|
||
- Excel 导入不跨账套匹配（import-excel / import-excel-smart 双端点）
|
||
- 数据源管理端点 entity 隔离（list/create/update/delete + 跨账套 404）
|
||
- sync-kpis 已停用（410）
|
||
- channel-status 真实通道状态端点
|
||
"""
|
||
import io
|
||
import pandas as pd
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import KPIDefinition, KPIValue, DataSourceConfig, Entity
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
|
||
|
||
DATA = "/api/cma/data"
|
||
|
||
|
||
def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity:
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=name or f"企业{eid}", short_name=f"E{eid}", status="active")
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _login(client: TestClient, username: str, password: str, entity_id: int) -> str:
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": username, "password": password, "entity_id": entity_id,
|
||
})
|
||
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
|
||
return resp.json()["token"]
|
||
|
||
|
||
def _excel_bytes(df: pd.DataFrame) -> io.BytesIO:
|
||
buf = io.BytesIO()
|
||
df.to_excel(buf, index=False)
|
||
buf.seek(0)
|
||
return buf
|
||
|
||
|
||
class TestExcelImportEntityIsolation:
|
||
"""Excel 导入只匹配当前账套 KPI，防跨账套写入（P0 安全）"""
|
||
|
||
def _mk_kpi(self, db: Session, code: str, eid: int) -> KPIDefinition:
|
||
kpi = KPIDefinition(
|
||
entity_id=eid, kpi_code=code, kpi_name=f"{code}-名称",
|
||
dimension="finance", category="financial_report", 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 test_import_excel_entity2_does_not_write_entity1_kpi(self, client: TestClient, db: Session):
|
||
"""entity2 登录导入 F_REVENUE → entity1 的 F_REVENUE 零写入（旧实现会全表匹配串写）"""
|
||
_ensure_entity(db, 2)
|
||
k1 = self._mk_kpi(db, "F_REVENUE", 1)
|
||
_ensure_entity(db, 2)
|
||
create_test_user(db)
|
||
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
|
||
before = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count()
|
||
|
||
df = pd.DataFrame({
|
||
"kpi_code": ["F_REVENUE"],
|
||
"period": ["2026-08"],
|
||
"actual_value": [999.0],
|
||
})
|
||
resp = client.post(
|
||
f"{DATA}/import-excel",
|
||
headers=auth_header(token2),
|
||
files={"file": ("t.xlsx", _excel_bytes(df),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||
)
|
||
assert resp.status_code == 200
|
||
# entity2 无 F_REVENUE → 匹配不到 → 跳过 0 条（不是跨账套写入 entity1）
|
||
assert "0 条" in resp.json()["message"]
|
||
assert "不存在" in resp.json()["message"]
|
||
after = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count()
|
||
assert after == before, "entity2 导入不得写入 entity1 的 KPI"
|
||
|
||
def test_import_excel_smart_creates_kpi_in_current_entity_only(self, client: TestClient, db: Session):
|
||
"""entity2 智能导入不存在的 F_REVENUE → 自动创建在 entity2（不匹配 entity1 的同名KPI）"""
|
||
_ensure_entity(db, 2)
|
||
k1 = self._mk_kpi(db, "F_REVENUE", 1)
|
||
create_test_user(db)
|
||
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
|
||
|
||
df = pd.DataFrame({
|
||
"kpi_code": ["F_REVENUE"],
|
||
"period": ["2026-08"],
|
||
"actual_value": [999.0],
|
||
})
|
||
resp = client.post(
|
||
f"{DATA}/import-excel-smart",
|
||
headers=auth_header(token2),
|
||
files={"file": ("测试利润表.xlsx", _excel_bytes(df),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
# entity2 自动创建 KPI（PL_ 前缀因为文件名含"利润表"），不写 entity1 的 F_REVENUE
|
||
assert "自动创建1个新KPI" in body["message"]
|
||
created = db.query(KPIDefinition).filter(
|
||
KPIDefinition.entity_id == 2,
|
||
KPIDefinition.kpi_code.like("PL_%"),
|
||
).first()
|
||
assert created is not None, "entity2 导入应自动创建 PL_xxx KPI 在 entity2 账套"
|
||
# entity1 的 F_REVENUE 及其 KPIValue 未被触碰
|
||
assert db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count() == 0
|
||
|
||
def test_import_excel_entity1_own_kpi_still_works(self, client: TestClient, db: Session):
|
||
"""正常路径：entity1 导入自己的 F_REVENUE 成功"""
|
||
k1 = self._mk_kpi(db, "F_REVENUE", 1)
|
||
create_test_user(db)
|
||
token1 = get_token_for_user(client) # entity1
|
||
df = pd.DataFrame({
|
||
"kpi_code": ["F_REVENUE"],
|
||
"period": ["2026-08"],
|
||
"actual_value": [100.0],
|
||
})
|
||
resp = client.post(
|
||
f"{DATA}/import-excel",
|
||
headers=auth_header(token1),
|
||
files={"file": ("t.xlsx", _excel_bytes(df),
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert "导入成功 1 条" in resp.json()["message"]
|
||
row = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).first()
|
||
assert row is not None and row.actual_value == 100.0
|
||
|
||
|
||
class TestSourceCRUDEntityIsolation:
|
||
"""数据源管理端点 entity 隔离（P1）"""
|
||
|
||
def _mk_entity2_source(self, client, db, token2) -> int:
|
||
resp = client.post(
|
||
f"{DATA}/sources",
|
||
headers=auth_header(token2),
|
||
json={"name": "entity2数据源", "source_type": "manual"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["entity_id"] == 2
|
||
return resp.json()["data"]["id"]
|
||
|
||
def test_list_sources_isolated_by_entity(self, client: TestClient, db: Session):
|
||
"""entity2 登录 list_sources 看不到 entity1 的数据源"""
|
||
_ensure_entity(db, 2)
|
||
create_test_user(db)
|
||
token1 = get_token_for_user(client)
|
||
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
|
||
|
||
r1 = client.post(f"{DATA}/sources", headers=auth_header(token1),
|
||
json={"name": "entity1数据源", "source_type": "manual"})
|
||
assert r1.status_code == 200
|
||
|
||
# entity2 登录看不到 entity1 的源
|
||
resp2 = client.get(f"{DATA}/sources", headers=auth_header(token2))
|
||
assert resp2.status_code == 200
|
||
names2 = [s["name"] for s in resp2.json()["data"]]
|
||
assert "entity1数据源" not in names2
|
||
|
||
# entity1 登录能看到自己的
|
||
resp1 = client.get(f"{DATA}/sources", headers=auth_header(token1))
|
||
names1 = [s["name"] for s in resp1.json()["data"]]
|
||
assert "entity1数据源" in names1
|
||
|
||
def test_cross_entity_update_delete_404(self, client: TestClient, db: Session):
|
||
"""entity2 不能改/删 entity1 的数据源（404 越权防护）"""
|
||
_ensure_entity(db, 2)
|
||
create_test_user(db)
|
||
token1 = get_token_for_user(client)
|
||
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
|
||
|
||
r1 = client.post(f"{DATA}/sources", headers=auth_header(token1),
|
||
json={"name": "entity1的源", "source_type": "manual"})
|
||
e1_source_id = r1.json()["data"]["id"]
|
||
|
||
# entity2 尝试更新 entity1 的源 → 404（隔离生效，不再允许越权操作）
|
||
resp = client.put(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token2),
|
||
json={"name": "篡改"})
|
||
assert resp.status_code == 404
|
||
resp = client.delete(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token2))
|
||
assert resp.status_code == 404
|
||
# entity1 自己仍可删除
|
||
resp = client.delete(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token1))
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestSyncKpisDecommissioned:
|
||
"""sync-kpis 收口：ERP 直连通道 2026-08-31 停用，端点返回 410（下线说明）"""
|
||
|
||
def test_sync_kpis_returns_410(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{DATA}/sync-kpis", headers=auth_header(token))
|
||
assert resp.status_code == 410
|
||
assert "2026-08-31" in resp.json()["detail"]
|
||
assert "zxt" in resp.json()["detail"]
|
||
|
||
def test_admin_erp_sync_returns_410(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post("/api/cma/admin/erp-sync", headers=auth_header(token))
|
||
assert resp.status_code == 410
|
||
resp2 = client.get("/api/cma/admin/erp-sync/dry-run", headers=auth_header(token))
|
||
assert resp2.status_code == 410
|
||
|
||
|
||
class TestChannelStatus:
|
||
"""channel-status 真实取数链路状态端点"""
|
||
|
||
def test_channel_status_shape(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{DATA}/channel-status", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
d = resp.json()
|
||
assert "real_channel" in d and "erp_direct" in d
|
||
assert d["real_channel"]["channel"].startswith("zxt")
|
||
assert d["erp_direct"]["enabled"] is False
|
||
assert "2026-08-31" in d["erp_direct"]["reason"]
|