feat(bot-bridge): 新增会计科目写入端点 POST /api/cma/bot/subjects — 行级全量校验+按subject_code幂等UPSERT(dry_run/update_existing开关)+L3留痕, 绝不DELETE (dama-cma-subjects-write-endpoint-20260912)

backend/tests/test_bot_bridge.py

@@ -16,7 +16,7 @@ from sqlalchemy.orm import Session
 
 from app.models import (
     User, KPIDefinition, KPIValue, KPIAlert, StrategicMap, MapObjective,
-    ActionPlan, OrgNode, DataSourceConfig, Objective,
+    ActionPlan, OrgNode, DataSourceConfig, Objective, Subject,
 )
 from app.models.budget_plan import BudgetPlan
 from app.models.cost_model import StandardCost, ActualCost
@@ -416,3 +416,142 @@ class TestDictZxtAccountWrite:
         resp = client.post(self.URL, json={"rows": [{"bad_field": 1}]}, headers=BOT_KEY)
         assert resp.status_code == 400
         assert "非白名单字段" in str(resp.json())
+
+
+class TestSubjectsWrite:
+    """会计科目写入端点（POST /api/cma/bot/subjects）— 鉴权/校验/dry_run/幂等/更新
+
+    实现为纯 ORM（无 MySQL 专有 SQL），故正向/幂等路径可在 SQLite 测试库完整覆盖。
+    """
+
+    URL = "/api/cma/bot/subjects"
+
+    @staticmethod
+    def _row(code, name, **kw):
+        row = {"subject_code": code, "subject_name": name, "parent_code": None,
+               "level": 1, "category": "equity", "is_active": 1}
+        row.update(kw)
+        return row
+
+    # ── 鉴权 ──
+    def test_no_key_401(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]})
+        assert resp.status_code == 401
+
+    def test_invalid_key_401(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目")]},
+                           headers={"X-BOT-KEY": "wrong-key"})
+        assert resp.status_code == 401
+
+    # ── 参数校验（全批拒绝、零落库）──
+    def test_level_out_of_range_400(self, client: TestClient, db: Session):
+        resp = client.post(self.URL, json={"subjects": [self._row("9001", "验证科目", level=9)]},
+                           headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "level" in str(resp.json())
+        assert db.query(Subject).filter(Subject.subject_code == "9001").count() == 0
+
+    def test_batch_duplicate_code_400(self, client: TestClient, db: Session):
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("9002", "验证科目A"), self._row("9002", "验证科目B")]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "重复" in str(resp.json())
+        assert db.query(Subject).filter(Subject.subject_code == "9002").count() == 0
+
+    def test_parent_not_exists_400(self, client: TestClient, db: Session):
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("900301", "验证子科目", parent_code="999999", level=2)]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "parent_code" in str(resp.json())
+        assert db.query(Subject).filter(Subject.subject_code == "900301").count() == 0
+
+    def test_illegal_category_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("9003", "验证科目", category="not_a_category")]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "category" in str(resp.json())
+
+    def test_illegal_is_active_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("9004", "验证科目", is_active=7)]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "is_active" in str(resp.json())
+
+    def test_empty_subjects_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": []}, headers=BOT_KEY)
+        assert resp.status_code == 400
+
+    def test_non_whitelist_field_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("9005", "验证科目", entity_id=1)]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "非白名单字段" in str(resp.json())
+
+    def test_missing_name_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"subjects": [
+            {"subject_code": "9006", "level": 1}]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "subject_name" in str(resp.json())
+
+    # ── 落库语义 ──
+    def test_dry_run_zero_write(self, client: TestClient, db: Session):
+        before = db.query(Subject).count()
+        resp = client.post(self.URL, json={"subjects": [self._row("9101", "验证-干跑")],
+                                           "dry_run": True}, headers=BOT_KEY)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["dry_run"] is True and data["applied"] is False
+        assert data["inserted"] == 0 and data["upserted"] == 0
+        assert data["would_insert"] == 1
+        db.expire_all()
+        assert db.query(Subject).count() == before
+        assert db.query(Subject).filter(Subject.subject_code == "9101").count() == 0
+
+    def test_insert_skip_counts_and_idempotent(self, client: TestClient, db: Session):
+        db.add(Subject(subject_code="9201", subject_name="既有科目", level=1,
+                       category="equity", is_active=1))
+        db.commit()
+        before = db.query(Subject).count()
+
+        payload = {"subjects": [self._row("9201", "既有科目"), self._row("9202", "新增科目A"),
+                                self._row("9203", "新增科目B")]}
+        resp = client.post(self.URL, json=payload, headers=BOT_KEY)
+        assert resp.status_code == 200
+        d1 = resp.json()
+        assert (d1["inserted"], d1["updated"], d1["skipped"]) == (2, 0, 1), d1
+        assert d1["after"] == before + 2
+
+        db.expire_all()
+        assert db.query(Subject).filter(Subject.subject_code.in_(["9202", "9203"])).count() == 2
+
+        # 幂等：同 payload 二次调用 → 全部 skip，行数不变、无重复
+        resp2 = client.post(self.URL, json=payload, headers=BOT_KEY)
+        assert resp2.status_code == 200
+        d2 = resp2.json()
+        assert (d2["inserted"], d2["updated"], d2["skipped"]) == (0, 0, 3), d2
+        assert d2["before"] == d2["after"] == before + 2
+        db.expire_all()
+        assert db.query(Subject).count() == before + 2
+        assert db.query(Subject).filter(Subject.subject_code == "9202").count() == 1
+
+    def test_update_existing_switch(self, client: TestClient, db: Session):
+        db.add(Subject(subject_code="9301", subject_name="旧名称", level=1,
+                       category="equity", is_active=1))
+        db.commit()
+        resp = client.post(self.URL, json={
+            "subjects": [self._row("9301", "新名称", category="cost", level=2)],
+            "update_existing": True}, headers=BOT_KEY)
+        assert resp.status_code == 200
+        d = resp.json()
+        assert (d["inserted"], d["updated"], d["skipped"]) == (0, 1, 0), d
+        db.expire_all()
+        row = db.query(Subject).filter(Subject.subject_code == "9301").first()
+        assert row.subject_name == "新名称" and row.category == "cost" and row.level == 2
+
+    def test_batch_parent_within_batch_ok(self, client: TestClient, db: Session):
+        """父科目在本批次内（层级同批导入）允许；父科目不存在则拒绝"""
+        resp = client.post(self.URL, json={"subjects": [
+            self._row("9401", "父科目"), self._row("940101", "子科目", parent_code="9401", level=2)],
+        }, headers=BOT_KEY)
+        assert resp.status_code == 200, resp.json()
+        assert resp.json()["inserted"] == 2

backend/tests/test_risk_levels.py

@@ -49,12 +49,12 @@ class TestRiskMapCoverage:
         assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}"
 
     def test_risk_level_counts(self):
-        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=4(+字典写入)、L4=0"""
+        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=5(+字典写入+科目写入)、L4=0"""
         from collections import Counter
         counts = Counter(API_RISK_MAP.values())
         assert counts["L1"] == 21, counts
         assert counts["L2"] == 6, counts
-        assert counts["L3"] == 4, counts
+        assert counts["L3"] == 5, counts
         assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）"
 
     def test_no_l4_bot_endpoints(self):
@@ -92,11 +92,12 @@ class TestRiskLevelsEndpoint:
         assert apis["POST /api/cma/bot/import"]["risk_level"] == "L3"
         assert apis["POST /api/cma/bot/import"]["handling"]  # 处理方式非空
         assert apis["POST /api/cma/bot/dict/zxt-account"]["risk_level"] == "L3"
+        assert apis["POST /api/cma/bot/subjects"]["risk_level"] == "L3"
 
         # 分级统计
         assert data["summary"]["L1"] == 21
         assert data["summary"]["L2"] == 6
-        assert data["summary"]["L3"] == 4
+        assert data["summary"]["L3"] == 5
         assert data["summary"]["L4"] == 0