Hermes CI Fix
06c23927aa
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（独立复核） - 原会话因迭代上限停手，本提交代收口（工作已复核完成）
179 lines
8.4 KiB
Python
179 lines
8.4 KiB
Python
"""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"]
|