Hermes CI Fix
a64f184525
- subjects/kpi_definitions 新增 important_flag/data_level/data_category 列 - 新增 /api/cma/data-classification/* API: inventory/stats/标记/批量/export/industry-reference - 数据级别: core核心/important重要/general一般（GB/T 43697 简化） - 前端 DataClassification.vue: 统计概览+清单+打标+导出+行业参考目录 - pytest 17例覆盖（含多租户隔离验证）
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""数据分类分级 API 测试 — 重要数据标记 + 资产清单 + 导出"""
|
||
import hashlib
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Subject, KPIDefinition, Entity
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
|
||
def create_test_subject(db: Session, **kwargs) -> Subject:
|
||
"""创建测试科目"""
|
||
defaults = {
|
||
"subject_code": "1001",
|
||
"subject_name": "库存现金",
|
||
"level": 1,
|
||
"is_active": 1,
|
||
}
|
||
defaults.update(kwargs)
|
||
s = Subject(**defaults)
|
||
db.add(s)
|
||
db.commit()
|
||
db.refresh(s)
|
||
return s
|
||
|
||
|
||
def create_test_kpi(db: Session, **kwargs) -> KPIDefinition:
|
||
"""创建测试KPI（带entity_id）"""
|
||
defaults = {
|
||
"entity_id": 1,
|
||
"kpi_code": "F_TEST_REVENUE",
|
||
"kpi_name": "营业收入(万元)",
|
||
"dimension": "finance",
|
||
"status": "active",
|
||
}
|
||
defaults.update(kwargs)
|
||
k = KPIDefinition(**defaults)
|
||
db.add(k)
|
||
db.commit()
|
||
db.refresh(k)
|
||
return k
|
||
|
||
|
||
class TestInventory:
|
||
def test_inventory_empty(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert data["total"] == 0
|
||
assert data["stats"]["marked"] == 0
|
||
|
||
def test_inventory_contains_subject_and_kpi(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
types = {i["type"] for i in data["items"]}
|
||
assert types == {"subject", "kpi"}
|
||
assert data["total"] == 2
|
||
# 未标记时级别默认 general
|
||
assert data["stats"]["by_level"]["general"] == 2
|
||
|
||
def test_filter_by_data_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory",
|
||
params={"data_type": "kpi"}, headers=auth_header(token))
|
||
data = r.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["type"] == "kpi"
|
||
|
||
def test_filter_by_level(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db, data_level="important", important_flag=1)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory",
|
||
params={"data_level": "important"}, headers=auth_header(token))
|
||
data = r.json()
|
||
assert data["total"] == 1
|
||
assert data["items"][0]["name"] == "库存现金"
|
||
|
||
def test_filter_important_only(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db, data_level="important", important_flag=1)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory",
|
||
params={"important": 1}, headers=auth_header(token))
|
||
data = r.json()
|
||
assert data["total"] == 1
|
||
|
||
def test_invalid_level_400(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/inventory",
|
||
params={"data_level": "secret"}, headers=auth_header(token))
|
||
assert r.status_code == 400
|
||
|
||
|
||
class TestMarkSubject:
|
||
def test_mark_subject(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
s = create_test_subject(db)
|
||
token = get_token_for_user(client)
|
||
r = client.put(f"/api/cma/data-classification/subjects/{s.id}",
|
||
params={"data_level": "important", "data_category": "财务数据"},
|
||
headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
# 重要级别自动视为重要数据
|
||
assert body["important_flag"] == 1
|
||
assert body["data_level"] == "important"
|
||
assert body["data_category"] == "财务数据"
|
||
|
||
def test_mark_subject_general_clears(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
s = create_test_subject(db, data_level="important", important_flag=1)
|
||
token = get_token_for_user(client)
|
||
r = client.put(f"/api/cma/data-classification/subjects/{s.id}",
|
||
params={"data_level": "general", "important_flag": 0},
|
||
headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
assert r.json()["data_level"] == "general"
|
||
assert r.json()["important_flag"] == 0
|
||
|
||
def test_mark_subject_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.put("/api/cma/data-classification/subjects/9999",
|
||
params={"data_level": "important"}, headers=auth_header(token))
|
||
assert r.status_code == 404
|
||
|
||
|
||
class TestMarkKpi:
|
||
def test_mark_kpi(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
k = create_test_kpi(db, entity_id=1)
|
||
token = get_token_for_user(client)
|
||
r = client.put(f"/api/cma/data-classification/kpis/{k.id}",
|
||
params={"data_level": "core", "data_category": "财务数据"},
|
||
headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
assert r.json()["important_flag"] == 1
|
||
assert r.json()["data_level"] == "core"
|
||
|
||
def test_kpi_entity_isolation(self, client: TestClient, db: Session):
|
||
"""多租户隔离：entity 2 的KPI，entity 1 的token不可见、不可标记"""
|
||
create_test_user(db)
|
||
k2 = create_test_kpi(db, entity_id=2, kpi_code="F_OTHER_ENTITY")
|
||
token = get_token_for_user(client) # token 绑定 entity 1
|
||
# 不可标记
|
||
r = client.put(f"/api/cma/data-classification/kpis/{k2.id}",
|
||
params={"data_level": "important"}, headers=auth_header(token))
|
||
assert r.status_code == 404
|
||
# 清单中不可见
|
||
r2 = client.get("/api/cma/data-classification/inventory",
|
||
params={"data_type": "kpi"}, headers=auth_header(token))
|
||
data = r2.json()
|
||
assert all(i["code"] != "F_OTHER_ENTITY" for i in data["items"])
|
||
|
||
|
||
class TestBatch:
|
||
def test_batch_mark_subjects(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
s1 = create_test_subject(db, subject_code="1001", subject_name="库存现金")
|
||
s2 = create_test_subject(db, subject_code="1002", subject_name="银行存款")
|
||
token = get_token_for_user(client)
|
||
r = client.put("/api/cma/data-classification/batch",
|
||
params={"data_type": "subject", "ids": [s1.id, s2.id],
|
||
"data_level": "important", "data_category": "财务数据"},
|
||
headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
assert r.json()["updated_count"] == 2
|
||
# 回查确认生效
|
||
db.expire_all()
|
||
assert db.query(Subject).get(s1.id).important_flag == 1
|
||
assert db.query(Subject).get(s2.id).data_level == "important"
|
||
|
||
def test_batch_invalid_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.put("/api/cma/data-classification/batch",
|
||
params={"data_type": "bad", "ids": [1]}, headers=auth_header(token))
|
||
assert r.status_code == 400
|
||
|
||
|
||
class TestExport:
|
||
def test_export_csv(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db, data_level="important", important_flag=1)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/export", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
assert "text/csv" in r.headers["content-type"]
|
||
text = r.text
|
||
# 表头 + 类型 + 中文标签
|
||
assert "类型" in text and "数据级别" in text
|
||
assert "库存现金" in text
|
||
assert "科目" in text and "KPI" in text
|
||
assert "重要数据" in text
|
||
|
||
def test_export_filtered(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db, data_level="important", important_flag=1)
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/export",
|
||
params={"important": 1}, headers=auth_header(token))
|
||
text = r.text
|
||
assert "库存现金" in text
|
||
assert "营业收入(万元)" not in text
|
||
|
||
|
||
class TestStats:
|
||
def test_stats(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
create_test_subject(db, data_level="important", important_flag=1)
|
||
create_test_subject(db, subject_code="1002", subject_name="银行存款")
|
||
create_test_kpi(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/stats", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert data["subjects"]["total"] == 2
|
||
assert data["subjects"]["marked"] == 1
|
||
assert data["kpis"]["total"] == 1
|
||
assert data["marked_total"] == 1
|
||
assert "policy_note" in data
|
||
|
||
def test_industry_reference(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r = client.get("/api/cma/data-classification/industry-reference", headers=auth_header(token))
|
||
assert r.status_code == 200
|
||
items = r.json()["items"]
|
||
assert len(items) >= 5
|
||
industries = {i["industry"] for i in items}
|
||
assert "金融" in industries and "医疗" in industries
|