fix(cma): KR 名称污染存量清理 + 透传点护栏 (cma-kr-name-pollution-cleanup-20260916)
ci/woodpecker/push/main Pipeline was canceled

- kr_name.py/krName.ts: sanitize_kr_name 判据(尾部≥/≤数字≠目标值→截断;==目标值→保留)
- maps.py/okr.py/okr_templates.py: 模板名/JSON→krs.title 透传点护栏
- test: 28 passed(独立复核)
- 原会话因迭代上限停手,本提交代收口(工作已复核完成)
This commit is contained in:
Hermes CI Fix
2026-09-16 14:48:06 +08:00
parent 09fb11d86f
commit 06c23927aa
8 changed files with 300 additions and 3 deletions
@@ -0,0 +1,178 @@
"""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)})
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)})
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"]