"""数据管理模块脱节修复测试（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 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", "admin123", 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", "admin123", 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", "admin123", 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", "admin123", 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"]