feat(bot-bridge): 新增智享通科目字典写入端点 POST /api/cma/bot/dict/zxt-account — 幂等UPSERT(JSON/CSV)+L3分级+audit留痕, 解除 ops 管道 API 迁移阻塞 (ops-cma-dict-write-endpoint-20260912)

backend/tests/test_bot_bridge.py

@@ -368,3 +368,51 @@ class TestOkrAndNlp:
         resp2 = client.get("/api/cma/bot/nlp?intent=总览", headers=BOT_KEY)
         assert resp2.status_code == 200
         assert "overview" in resp2.json()
+
+
+class TestDictZxtAccountWrite:
+    """智享通科目字典写入端点（POST /api/cma/bot/dict/zxt-account）负例
+
+    说明: 正向 UPSERT 走 MySQL 专有 ON DUPLICATE KEY UPDATE，测试库为 SQLite
+    无法覆盖——正向/幂等/落库一致性由生产 curl 实测（见任务验收记录），
+    此处覆盖鉴权与参数校验（校验先于任何 DB 访问，故 SQLite 可测）。
+    """
+    URL = "/api/cma/bot/dict/zxt-account"
+
+    def test_no_key_401(self, client: TestClient):
+        resp = client.post(self.URL, json=[{"CwAccountID": "999", "subject_code": "1"}])
+        assert resp.status_code == 401
+
+    def test_invalid_key_401(self, client: TestClient):
+        resp = client.post(self.URL, json=[{"CwAccountID": "999"}],
+                           headers={"X-BOT-KEY": "wrong-key"})
+        assert resp.status_code == 401
+
+    def test_non_whitelist_field_400(self, client: TestClient):
+        resp = client.post(self.URL,
+                           json=[{"CwAccountID": "999", "entity_id": 1}], headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "非白名单字段" in str(resp.json())
+
+    def test_missing_required_field_400(self, client: TestClient):
+        resp = client.post(self.URL, json=[{"subject_code": "1122"}], headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "CwAccountID" in str(resp.json())
+
+    def test_illegal_value_400(self, client: TestClient):
+        resp = client.post(self.URL, json=[{"CwAccountID": "9 99;DROP"}], headers=BOT_KEY)
+        assert resp.status_code == 400
+
+    def test_empty_array_400(self, client: TestClient):
+        resp = client.post(self.URL, json=[], headers=BOT_KEY)
+        assert resp.status_code == 400
+
+    def test_non_array_json_400(self, client: TestClient):
+        resp = client.post(self.URL, json={"CwAccountID": "999"}, headers=BOT_KEY)
+        assert resp.status_code == 400
+
+    def test_rows_wrapper_accepted_shape(self, client: TestClient):
+        """{"rows":[...]} 包装形态被解析（校验失败仍 400，证明已进入校验而非形态拒绝）"""
+        resp = client.post(self.URL, json={"rows": [{"bad_field": 1}]}, headers=BOT_KEY)
+        assert resp.status_code == 400
+        assert "非白名单字段" in str(resp.json())

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=3、L4=0"""
+        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=4(+字典写入)、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"] == 3, counts
+        assert counts["L3"] == 4, counts
         assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）"
 
     def test_no_l4_bot_endpoints(self):
@@ -91,11 +91,12 @@ class TestRiskLevelsEndpoint:
         assert apis["POST /api/cma/bot/kpi-value-with-check"]["risk_level"] == "L2"
         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 data["summary"]["L1"] == 21
         assert data["summary"]["L2"] == 6
-        assert data["summary"]["L3"] == 3
+        assert data["summary"]["L3"] == 4
         assert data["summary"]["L4"] == 0