浅念
60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""管理域评分配置/引擎判别测试(批次B)。"""
|
|
import json
|
|
|
|
import sqlalchemy as sa
|
|
|
|
from app.db.models import EmployeeScoreConfig
|
|
from app.management.metric_definitions import (
|
|
DEFAULT_SCORE_CONFIG,
|
|
METRIC_DEFINITIONS,
|
|
validate_config,
|
|
)
|
|
|
|
|
|
def test_config_weights_sum_and_structure():
|
|
dims = DEFAULT_SCORE_CONFIG["dimensions"]
|
|
assert sum(d["weight"] for d in dims.values()) == 100
|
|
# 一期 pending_data 的维度不参与试算
|
|
available = [d for d in dims.values() if d.get("status") != "pending_data"]
|
|
assert sum(d["weight"] for d in available) == 70 # 30+25+15
|
|
assert dims["action"]["status"] == "pending_data"
|
|
assert dims["risk"]["status"] == "pending_data"
|
|
|
|
|
|
def test_config_normalization_declared():
|
|
assert DEFAULT_SCORE_CONFIG["normalization"] == "available_only_rescale"
|
|
|
|
|
|
def test_validate_config_rejects_bad_weight_sum():
|
|
bad = json.loads(json.dumps(DEFAULT_SCORE_CONFIG)) # deep copy
|
|
bad["dimensions"]["service"]["weight"] = 26 # 总和 101
|
|
problems = validate_config(bad)
|
|
assert any("权重" in p for p in problems)
|
|
|
|
|
|
def test_metric_definitions_cover_scored_metrics():
|
|
# 每个参与评分的 metric_code 都有元数据(名称/单位/方向)
|
|
for dim in DEFAULT_SCORE_CONFIG["dimensions"].values():
|
|
for code in dim.get("metrics", {}):
|
|
md = METRIC_DEFINITIONS[code if code in METRIC_DEFINITIONS else "active_share"]
|
|
# band 型指标的 key 带 _band 后缀,spec 里有 metric 字段:
|
|
spec = dim["metrics"][code]
|
|
src = spec.get("metric", code)
|
|
assert src in METRIC_DEFINITIONS
|
|
m = METRIC_DEFINITIONS[src]
|
|
assert m["name"] and m["unit"] and m["direction"] in (1, -1)
|
|
|
|
|
|
async def test_seed_and_idempotent(make_session):
|
|
from app.sync.cli import _seed_score_config # 纯会话函数
|
|
s = make_session
|
|
v1 = await _seed_score_config(s)
|
|
await s.commit()
|
|
assert v1 == "v1.0"
|
|
rows = (await s.execute(sa.select(EmployeeScoreConfig))).scalars().all()
|
|
assert len(rows) == 1 and rows[0].status == "active"
|
|
# 幂等:已存在同版本返回 None 不新建
|
|
assert await _seed_score_config(s) is None
|
|
await s.commit()
|
|
assert len((await s.execute(sa.select(EmployeeScoreConfig))).scalars().all()) == 1
|