feat: 数据分类分级 — 重要数据标记+资产清单+CSV导出（8/20评估办法政策驱动）

- 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例覆盖（含多租户隔离验证）
2026-08-26 07:51:39 +08:00
parent 6af4366a84
commit a64f184525
7 changed files with 1025 additions and 5 deletions
@@ -0,0 +1,247 @@
+"""数据分类分级 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