"""KR 名称污染（历史截断值）护栏测试 — 任务 cma-kr-name-pollution-cleanup-20260916 覆盖： 1. 判据单测（sanitize_kr_name / target_number）：污染截断、有意命名、无法判定三类 2. 「JSON/模板名 → krs.title」透传点护栏： - PUT /api/cma/maps/{id}（_sync_objectives_krs） - PUT /api/cma/okr/{objective_id}/krs/sync（画布节点同步） - POST /api/cma/okr-templates/{id}/apply（模板 → 新地图） 3. 阴性对照：合法命名（渠道留存率≥90 / 新客户数≥10）经同样路径不得被改 """ import json import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from app.models import KR, Objective, OKRTemplate, StrategicMap from app.utils.kr_name import sanitize_kr_name, target_number from tests.conftest import create_test_user, get_token_for_user, auth_header # ────────────────────────── 1. 判据单测 ────────────────────────── class TestSanitizeKrName: @pytest.mark.parametrize("name,target,expect", [ ("营业收入≥2", 2500, "营业收入"), # 输入 2500 时截断成首位数字 ("新客户数≥1", 5, "新客户数"), ("营业收入≥7", 780, "营业收入"), ("净利润≥6", 60, "净利润"), ("费用率≤3", 30, "费用率"), ("经营性现金流≥5", 50, "经营性现金流"), ("营业收入≥2", "2500", "营业收入"), # 目标值为字符串数字 ("应收账款天数≤4", "≤45天", "应收账款天数"), # 目标值带单位文本 ]) def test_polluted_tail_stripped(self, name, target, expect): clean, changed = sanitize_kr_name(name, target) assert changed is True assert clean == expect @pytest.mark.parametrize("name,target", [ ("渠道留存率≥90", 90), # 数字==目标值 = 有意命名 ("新客户数≥10", 10), ("A级渠道占比≥30%", 30), ("营业收入≥780", 780), ("费用率≤30", 30), ("现金折扣收益≤收入20%", 20), # 尾部不是「符号+数字」 ("库存周转天数≤45天", "≤45天"), ("数据质量评分≥90", 90.0), ("剔除现金折扣后经营性净利≥0", 0), ]) def test_legal_names_untouched(self, name, target): clean, changed = sanitize_kr_name(name, target) assert changed is False assert clean == name @pytest.mark.parametrize("name,target", [ ("NPS净推荐值≥60", "为正"), # 目标值无数字 → 无法判定，原样保留 ("总资产周转率≥1.0", None), ("自由现金流≥0", ""), ("经营现金流为正", "为正"), # 无尾部符号+数字 ]) def test_unknown_target_untouched(self, name, target): clean, changed = sanitize_kr_name(name, target) assert changed is False assert clean == name def test_empty_base_kept(self): """名称主体为空（只剩符号+数字）时不动，避免产出空标题""" clean, changed = sanitize_kr_name("≥2", 2500) assert changed is False assert clean == "≥2" def test_non_string_input(self): clean, changed = sanitize_kr_name(None, 100) assert changed is False and clean == "" def test_target_number_parsing(self): assert target_number("≥45天") == 45 assert target_number("≥20%") == 20 assert target_number(780) == 780 assert target_number("为正") is None assert target_number(None) is None # ────────────────────────── 2. 透传点护栏 ────────────────────────── def _dims_with_krs(name, target_value): return [ {"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C", "objectives": [{"name": "营收目标", "kpis": [], "krs": [{"name": name, "target_value": target_value, "weight": 70, "operator": ">="}]}]}, {"key": "customer", "name": "客户层", "icon": "🤝", "color": "#67c23a", "objectives": []}, ] class TestPassThroughGuards: def _setup(self, db: Session): user = create_test_user(db) return user def test_map_save_strips_polluted_kr_title(self, client: TestClient, db: Session): """PUT /maps/{id} 的 JSON→krs.title 同步点：污染名被规整（库 + JSON 双清）""" self._setup(db) token = get_token_for_user(client) m = StrategicMap(title="测试地图", status="draft", entity_id=1, dimensions=_dims_with_krs("营业收入≥2", 25000), canvas_data={"connections": []}) db.add(m) db.commit() db.refresh(m) resp = client.put(f"/api/cma/maps/{m.id}", headers=auth_header(token), json={"dimensions": _dims_with_krs("营业收入≥2", 25000), "base_updated_at": m.updated_at.isoformat()}) assert resp.status_code == 200, resp.text kr = db.query(KR).filter(KR.objective_id.isnot(None)).first() assert kr is not None and kr.title == "营业收入" assert float(kr.target_value) == 25000.0 # JSON 回写：展示口径 = 落库口径（回查须先失效身份映射，防读到旧快照） db.expire_all() m2 = db.query(StrategicMap).filter(StrategicMap.id == m.id).first() dims = m2.dimensions dims = json.loads(dims) if isinstance(dims, str) else dims assert dims[0]["objectives"][0]["krs"][0]["name"] == "营业收入" def test_map_save_keeps_legal_kr_title(self, client: TestClient, db: Session): """阴性对照：合法命名（数字==目标值）不得被改""" self._setup(db) token = get_token_for_user(client) m = StrategicMap(title="测试地图", status="draft", entity_id=1, dimensions=_dims_with_krs("渠道留存率≥90", 90), canvas_data={"connections": []}) db.add(m) db.commit() db.refresh(m) resp = client.put(f"/api/cma/maps/{m.id}", headers=auth_header(token), json={"dimensions": _dims_with_krs("渠道留存率≥90", 90), "base_updated_at": m.updated_at.isoformat()}) assert resp.status_code == 200, resp.text kr = db.query(KR).filter(KR.objective_id.isnot(None)).first() assert kr is not None and kr.title == "渠道留存率≥90" def test_kr_sync_endpoint_strips_polluted_title(self, client: TestClient, db: Session): """PUT /okr/{oid}/krs/sync（画布节点同步）同样被护栏覆盖""" self._setup(db) token = get_token_for_user(client) obj = Objective(entity_id=1, title="营收目标", dimension="finance", status="active", quarter="2026Q3") db.add(obj) db.commit() db.refresh(obj) resp = client.put(f"/api/cma/okr/{obj.id}/krs/sync", headers=auth_header(token), json={"krs": [{"title": "新客户数≥1", "operator": ">=", "target_value": 100, "weight": 30}]}) assert resp.status_code == 200, resp.text krs = db.query(KR).filter(KR.objective_id == obj.id).all() assert [k.title for k in krs] == ["新客户数"] def test_apply_template_strips_polluted_kr_name(self, client: TestClient, db: Session): """POST /okr-templates/{id}/apply：存量污染模板 → 新地图 KR 名干净""" self._setup(db) token = get_token_for_user(client) t = OKRTemplate(name="营收目标", description="", dimension="finance", layer="level3", industry_tag="general", source="system", preset_krs=[{"name": "营业收入≥2", "target_value": 2500, "weight": 70}, {"name": "渠道留存率≥90", "target_value": 90, "weight": 30}]) db.add(t) db.commit() db.refresh(t) resp = client.post(f"/api/cma/okr-templates/{t.id}/apply", headers=auth_header(token), json={"title": "套用模板-护栏验证"}) assert resp.status_code == 200, resp.text m = db.query(StrategicMap).filter(StrategicMap.id == resp.json()["map_id"]).first() names = [k["name"] for k in m.dimensions[0]["objectives"][0]["krs"]] assert names == ["营业收入", "渠道留存率≥90"]