merge: T7 规则引擎·数据驱动决策表（T-P2-02，TD-10）
This commit is contained in:
@@ -0,0 +1,260 @@
|
||||
"""kb_rule DSL 解析校验与 match 矩阵（TD-10，纯逻辑）。
|
||||
|
||||
核心验收点：**非法结构 100% 拒绝**——未识别的键（顶层/子句/条件字段）全部
|
||||
在解析时抛错，而不是被静默忽略后变成一条永不生效的死规则。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from packages.rules import dsl
|
||||
|
||||
VALID_RULE = {
|
||||
"rule_key": "R-BODY-A-002",
|
||||
"category": "body_shape",
|
||||
"priority": 90,
|
||||
"conditions": {
|
||||
"body_shape": "A型",
|
||||
"features": {"include": ["溜肩"]},
|
||||
"goals": {"include": ["修饰溜肩"]},
|
||||
},
|
||||
"effects": {
|
||||
"recommend": [
|
||||
{"slot": "top", "attrs": {"shoulder": ["垫肩", "自然肩"], "sleeve": ["泡泡袖"]}}
|
||||
],
|
||||
"avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"], "silhouette": ["修身"]}}],
|
||||
"explanation": "溜肩需要在肩部建立结构感",
|
||||
},
|
||||
"is_hard": False,
|
||||
"confidence": 0.95,
|
||||
}
|
||||
|
||||
|
||||
# ── 解析：合法结构 ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_parse_doc_example() -> None:
|
||||
rule = dsl.Rule.model_validate(VALID_RULE)
|
||||
assert rule.rule_key == "R-BODY-A-002"
|
||||
assert rule.priority == 90
|
||||
assert rule.is_hard is False
|
||||
assert rule.confidence == pytest.approx(0.95)
|
||||
# 等值子句解析为 str，集合子句解析为 ConditionClause
|
||||
assert rule.conditions.clauses["body_shape"] == "A型"
|
||||
features = rule.conditions.clauses["features"]
|
||||
assert isinstance(features, dsl.ConditionClause)
|
||||
assert features.include == ("溜肩",)
|
||||
assert rule.effects.recommend[0].slot == "top"
|
||||
assert rule.effects.avoid[0].attrs["shoulder"] == ("落肩",)
|
||||
|
||||
|
||||
def test_parse_min_max_and_exclude_clause() -> None:
|
||||
rule = dsl.Rule.model_validate(
|
||||
{
|
||||
**VALID_RULE,
|
||||
"conditions": {
|
||||
"temperature": {"min": 18, "max": 26},
|
||||
"season": {"exclude": ["夏"]},
|
||||
},
|
||||
}
|
||||
)
|
||||
temp = rule.conditions.clauses["temperature"]
|
||||
assert isinstance(temp, dsl.ConditionClause)
|
||||
assert temp.min == 18.0
|
||||
assert temp.max == 26.0
|
||||
season = rule.conditions.clauses["season"]
|
||||
assert isinstance(season, dsl.ConditionClause)
|
||||
assert season.exclude == ("夏",)
|
||||
|
||||
|
||||
def test_empty_conditions_means_match_all() -> None:
|
||||
# 合规类兜底规则（"全场禁真兽皮"）无条件命中是合法语义。
|
||||
rule = dsl.Rule.model_validate({**VALID_RULE, "conditions": {}})
|
||||
assert rule.conditions.clauses == {}
|
||||
assert dsl.match(rule.conditions, dsl.RuleContext()) is True
|
||||
|
||||
|
||||
def test_explanation_only_effect_is_valid() -> None:
|
||||
rule = dsl.Rule.model_validate(
|
||||
{**VALID_RULE, "effects": {"explanation": "整体偏保守的通勤思路"}}
|
||||
)
|
||||
assert rule.effects.recommend == ()
|
||||
assert rule.effects.avoid == ()
|
||||
|
||||
|
||||
def test_attrs_empty_means_whole_slot() -> None:
|
||||
rule = dsl.Rule.model_validate(
|
||||
{**VALID_RULE, "effects": {"avoid": [{"slot": "swimsuit"}], "explanation": "回避品类"}}
|
||||
)
|
||||
assert rule.effects.avoid[0].attrs == {}
|
||||
|
||||
|
||||
# ── 解析：非法结构 100% 拒绝 ───────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mutation",
|
||||
[
|
||||
{"rule_key": 12345}, # 非 str
|
||||
{"rule_key": "R-" + "x" * 63}, # 超出 64 字符（列宽）
|
||||
{"category": "c" * 33}, # 超出 32 字符（列宽）
|
||||
{"priority": 101}, # 超出分组上限 100
|
||||
{"priority": -1},
|
||||
{"confidence": 1.5},
|
||||
{"confidence": -0.1},
|
||||
],
|
||||
)
|
||||
def test_invalid_top_level_values_rejected(mutation: dict) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
dsl.Rule.model_validate({**VALID_RULE, **mutation})
|
||||
|
||||
|
||||
def test_unknown_top_level_key_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="unexpected"):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "unexpected": 1})
|
||||
|
||||
|
||||
def test_unknown_condition_field_rejected() -> None:
|
||||
# 条件字段名 typo：静默忽略会让这条规则永不命中
|
||||
with pytest.raises(ValidationError, match="未识别的条件字段"):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "conditions": {"body_shap": "A型"}})
|
||||
|
||||
|
||||
def test_unknown_clause_key_rejected() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "conditions": {"features": {"includes": ["溜肩"]}}})
|
||||
|
||||
|
||||
def test_empty_clause_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="至少要包含"):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "conditions": {"features": {}}})
|
||||
|
||||
|
||||
def test_min_greater_than_max_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="min 不能大于 max"):
|
||||
dsl.Rule.model_validate(
|
||||
{**VALID_RULE, "conditions": {"temperature": {"min": 26, "max": 18}}}
|
||||
)
|
||||
|
||||
|
||||
def test_empty_attr_values_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="取值列表不能为空"):
|
||||
dsl.Rule.model_validate(
|
||||
{
|
||||
**VALID_RULE,
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"shoulder": []}}],
|
||||
"explanation": "x",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_empty_attr_dim_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="维度名不能为空"):
|
||||
dsl.Rule.model_validate(
|
||||
{
|
||||
**VALID_RULE,
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"": ["垫肩"]}}],
|
||||
"explanation": "x",
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_empty_effects_rejected() -> None:
|
||||
with pytest.raises(ValidationError, match="至少要包含"):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "effects": {}})
|
||||
|
||||
|
||||
def test_blank_explanation_alone_rejected() -> None:
|
||||
# explanation="" 视同未填，效果为空的规则没有存在意义
|
||||
with pytest.raises(ValidationError, match="至少要包含"):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "effects": {"explanation": ""}})
|
||||
|
||||
|
||||
def test_non_dict_conditions_rejected() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
dsl.Rule.model_validate({**VALID_RULE, "conditions": ["body_shape"]})
|
||||
|
||||
|
||||
def test_effects_missing_rejected() -> None:
|
||||
payload = dict(VALID_RULE)
|
||||
del payload["effects"]
|
||||
with pytest.raises(ValidationError):
|
||||
dsl.Rule.model_validate(payload)
|
||||
|
||||
|
||||
# ── match 矩阵 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _cond(**clauses: object) -> dsl.RuleCondition:
|
||||
return dsl.RuleCondition.model_validate(clauses)
|
||||
|
||||
|
||||
def test_match_equality_hit_and_miss() -> None:
|
||||
cond = _cond(body_shape="A型")
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape="A型")) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape="H型")) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape=None)) is False # 缺字段
|
||||
|
||||
|
||||
def test_match_equality_on_collection_field_is_membership() -> None:
|
||||
cond = _cond(user_taboos="露脐")
|
||||
assert dsl.match(cond, dsl.RuleContext(user_taboos=("露脐", "豹纹"))) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(user_taboos=())) is False
|
||||
|
||||
|
||||
def test_match_include_requires_all_values() -> None:
|
||||
cond = _cond(features={"include": ["溜肩", "宽胯"]})
|
||||
assert dsl.match(cond, dsl.RuleContext(features=("溜肩", "宽胯", "小腿粗"))) is True
|
||||
# 子集语义：只命中一半不算命中
|
||||
assert dsl.match(cond, dsl.RuleContext(features=("溜肩",))) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(features=())) is False
|
||||
|
||||
|
||||
def test_match_include_on_scalar_field() -> None:
|
||||
cond = _cond(occasion={"include": ["面试", "通勤"]})
|
||||
assert dsl.match(cond, dsl.RuleContext(occasion="面试")) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(occasion="约会")) is False
|
||||
|
||||
|
||||
def test_match_exclude() -> None:
|
||||
cond = _cond(features={"exclude": ["溜肩"]})
|
||||
assert dsl.match(cond, dsl.RuleContext(features=("宽胯",))) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(features=("宽胯", "溜肩"))) is False
|
||||
|
||||
|
||||
def test_match_min_max_closed_interval() -> None:
|
||||
cond = _cond(temperature={"min": 18, "max": 26})
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=18.0)) is True # 闭区间含端点
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=26.0)) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=21.5)) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=17.9)) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=26.1)) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(temperature=None)) is False # 缺字段
|
||||
|
||||
|
||||
def test_match_min_only_and_max_only() -> None:
|
||||
assert dsl.match(_cond(temperature={"min": 10}), dsl.RuleContext(temperature=10.0))
|
||||
assert not dsl.match(_cond(temperature={"min": 10}), dsl.RuleContext(temperature=9.9))
|
||||
assert dsl.match(_cond(temperature={"max": 10}), dsl.RuleContext(temperature=10.0))
|
||||
assert not dsl.match(_cond(temperature={"max": 10}), dsl.RuleContext(temperature=10.1))
|
||||
|
||||
|
||||
def test_match_clauses_are_anded() -> None:
|
||||
cond = _cond(body_shape="A型", features={"include": ["溜肩"]})
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape="A型", features=("溜肩",))) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape="A型", features=())) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(body_shape="H型", features=("溜肩",))) is False
|
||||
|
||||
|
||||
def test_match_include_and_min_max_combined_in_one_clause() -> None:
|
||||
# 同一子句内 include 与区间也按 AND 处理
|
||||
cond = _cond(goals={"include": ["显瘦"], "exclude": ["显壮"]})
|
||||
assert dsl.match(cond, dsl.RuleContext(goals=("显瘦",))) is True
|
||||
assert dsl.match(cond, dsl.RuleContext(goals=("显瘦", "显壮"))) is False
|
||||
assert dsl.match(cond, dsl.RuleContext(goals=("显高",))) is False
|
||||
@@ -0,0 +1,322 @@
|
||||
"""纯域求值引擎：排序、硬约束拦截、用户禁忌压过专业推荐、性能基准（T-P2-02）。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from packages.rules import dsl, engine
|
||||
|
||||
|
||||
def _rule(
|
||||
rule_key: str,
|
||||
priority: int,
|
||||
*,
|
||||
category: str = "test",
|
||||
confidence: float = 0.9,
|
||||
is_hard: bool = False,
|
||||
conditions: dict[str, Any] | None = None,
|
||||
recommend: tuple[dict[str, Any], ...] = (),
|
||||
avoid: tuple[dict[str, Any], ...] = (),
|
||||
explanation: str | None = None,
|
||||
) -> dsl.Rule:
|
||||
effects: dict[str, Any] = {}
|
||||
if recommend:
|
||||
effects["recommend"] = list(recommend)
|
||||
if avoid:
|
||||
effects["avoid"] = list(avoid)
|
||||
if explanation:
|
||||
effects["explanation"] = explanation
|
||||
if not effects:
|
||||
effects["explanation"] = "测试规则" # 空效果本身非法，测试助手兜底
|
||||
return dsl.Rule.model_validate(
|
||||
{
|
||||
"rule_key": rule_key,
|
||||
"category": category,
|
||||
"priority": priority,
|
||||
"conditions": conditions or {},
|
||||
"effects": effects,
|
||||
"is_hard": is_hard,
|
||||
"confidence": confidence,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── 排序与命中链 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_hits_sorted_by_priority_then_confidence_then_key() -> None:
|
||||
rules = [
|
||||
_rule("R-LOW", 50),
|
||||
_rule("R-TIE-B", 90, confidence=0.9),
|
||||
_rule("R-TOP", 95, confidence=0.5),
|
||||
_rule("R-TIE-A", 90, confidence=0.9),
|
||||
_rule("R-TIE-C", 90, confidence=0.8),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert [h.rule_key for h in verdict.rule_hits] == [
|
||||
"R-TOP",
|
||||
"R-TIE-A",
|
||||
"R-TIE-B",
|
||||
"R-TIE-C",
|
||||
"R-LOW",
|
||||
]
|
||||
assert [h.priority for h in verdict.rule_hits] == [95, 90, 90, 90, 50]
|
||||
|
||||
|
||||
def test_rule_hit_carries_metadata_and_explanations_in_priority_order() -> None:
|
||||
rules = [
|
||||
_rule("R-LOW", 50, explanation="趋势解释", category="trend", confidence=0.7),
|
||||
_rule("R-HIGH", 90, explanation="体型解释", category="body_shape", confidence=0.95),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
top = verdict.rule_hits[0]
|
||||
assert (top.rule_key, top.priority, top.category, top.confidence) == (
|
||||
"R-HIGH",
|
||||
90,
|
||||
"body_shape",
|
||||
0.95,
|
||||
)
|
||||
assert top.is_hard is False
|
||||
assert verdict.explanations == ("体型解释", "趋势解释")
|
||||
assert verdict.explanation == "体型解释；趋势解释"
|
||||
|
||||
|
||||
def test_no_rules_or_no_hits_yield_empty_verdict() -> None:
|
||||
assert engine.evaluate([], dsl.RuleContext()).recommend == ()
|
||||
verdict = engine.evaluate(
|
||||
[_rule("R-X", 90, conditions={"body_shape": "A型"})],
|
||||
dsl.RuleContext(body_shape="H型"),
|
||||
)
|
||||
assert verdict.rule_hits == ()
|
||||
assert verdict.blocked is False
|
||||
|
||||
|
||||
# ── 硬约束 ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_hard_rule_with_avoid_populates_hard_blocks() -> None:
|
||||
avoid = {"slot": "shoes", "attrs": {"material": ["真皮"]}}
|
||||
rules = [_rule("R-HARD", 100, is_hard=True, avoid=(avoid,), explanation="合规红线")]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert verdict.blocked is True
|
||||
assert len(verdict.hard_blocks) == 1
|
||||
block = verdict.hard_blocks[0]
|
||||
assert block.rule_key == "R-HARD"
|
||||
assert block.avoid[0].slot == "shoes"
|
||||
assert block.avoid[0].attrs["material"] == ("真皮",)
|
||||
assert block.explanation == "合规红线"
|
||||
|
||||
|
||||
def test_hard_rule_without_avoid_does_not_block() -> None:
|
||||
# 拦截语义只由 avoid 表达：is_hard 规则只写 recommend 就无物可拦
|
||||
rules = [
|
||||
_rule(
|
||||
"R-HARD-REC",
|
||||
100,
|
||||
is_hard=True,
|
||||
recommend=({"slot": "top", "attrs": {"collar": ["衬衫领"]}},),
|
||||
)
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert verdict.hard_blocks == ()
|
||||
assert verdict.blocked is False
|
||||
|
||||
|
||||
def test_hard_rule_not_fired_does_not_block() -> None:
|
||||
rules = [
|
||||
_rule(
|
||||
"R-HARD",
|
||||
100,
|
||||
is_hard=True,
|
||||
conditions={"occasion": "面试"},
|
||||
avoid=({"slot": "top", "attrs": {"neckline": ["深V"]}},),
|
||||
)
|
||||
]
|
||||
assert engine.evaluate(rules, dsl.RuleContext(occasion="通勤")).blocked is False
|
||||
assert engine.evaluate(rules, dsl.RuleContext(occasion="面试")).blocked is True
|
||||
|
||||
|
||||
def test_soft_rule_with_avoid_never_blocks() -> None:
|
||||
rules = [_rule("R-SOFT", 90, avoid=({"slot": "top", "attrs": {"silhouette": ["修身"]}},))]
|
||||
assert engine.evaluate(rules, dsl.RuleContext()).hard_blocks == ()
|
||||
|
||||
|
||||
# ── 冲突仲裁：高优先级 avoid 剔除严格低优先级的冲突 recommend ────────────────
|
||||
|
||||
|
||||
def test_user_taboo_overrides_professional_recommend() -> None:
|
||||
# 用户禁忌（95）与专业推荐（90）同 slot 同维度取值相交 → 推荐被剔除
|
||||
rules = [
|
||||
_rule(
|
||||
"R-PRO",
|
||||
dsl.PRIORITY_BODY_SHAPE,
|
||||
recommend=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-TABOO",
|
||||
dsl.PRIORITY_USER_TABOO,
|
||||
conditions={"user_taboos": {"include": ["oversize"]}},
|
||||
avoid=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext(user_taboos=("oversize",)))
|
||||
assert verdict.recommend == ()
|
||||
assert verdict.avoid != () # 禁忌约束保留在 avoid 里供下游过滤
|
||||
assert [h.rule_key for h in verdict.rule_hits] == ["R-TABOO", "R-PRO"]
|
||||
|
||||
|
||||
def test_taboo_keeps_non_conflicting_recommend() -> None:
|
||||
# 不同维度（length vs shoulder）不构成冲突：专业推荐保留
|
||||
rules = [
|
||||
_rule(
|
||||
"R-PRO",
|
||||
90,
|
||||
recommend=({"slot": "top", "attrs": {"shoulder": ["垫肩"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-TABOO",
|
||||
95,
|
||||
avoid=({"slot": "top", "attrs": {"length": ["短款"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert [item.slot for item in verdict.recommend] == ["top"]
|
||||
assert verdict.recommend[0].attrs["shoulder"] == ("垫肩",)
|
||||
|
||||
|
||||
def test_taboo_does_not_override_higher_priority_recommend() -> None:
|
||||
# 合规（100）> 用户禁忌（95）：只有"严格更低"优先级的推荐才会被剔除
|
||||
rules = [
|
||||
_rule(
|
||||
"R-COMPLIANCE",
|
||||
dsl.PRIORITY_COMPLIANCE,
|
||||
recommend=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-TABOO",
|
||||
dsl.PRIORITY_USER_TABOO,
|
||||
avoid=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert len(verdict.recommend) == 1
|
||||
|
||||
|
||||
def test_lower_priority_recommend_vetoed_by_higher_soft_avoid() -> None:
|
||||
# 通用优先级语义：体型 avoid（90）同样压过趋势 recommend（50）
|
||||
rules = [
|
||||
_rule(
|
||||
"R-TREND",
|
||||
dsl.PRIORITY_TREND,
|
||||
recommend=({"slot": "top", "attrs": {"silhouette": ["修身"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-BODY",
|
||||
dsl.PRIORITY_BODY_SHAPE,
|
||||
avoid=({"slot": "top", "attrs": {"silhouette": ["修身"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert verdict.recommend == ()
|
||||
|
||||
|
||||
def test_same_priority_avoid_does_not_veto_recommend() -> None:
|
||||
# 同优先级不仲裁：推荐与回避并存，由下游按场景取舍
|
||||
rules = [
|
||||
_rule(
|
||||
"R-A",
|
||||
90,
|
||||
recommend=({"slot": "top", "attrs": {"sleeve": ["泡泡袖"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-B",
|
||||
90,
|
||||
avoid=({"slot": "top", "attrs": {"sleeve": ["泡泡袖"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert len(verdict.recommend) == 1
|
||||
|
||||
|
||||
def test_different_slot_never_conflicts() -> None:
|
||||
rules = [
|
||||
_rule(
|
||||
"R-REC",
|
||||
50,
|
||||
recommend=({"slot": "bottom", "attrs": {"silhouette": ["紧身"]}},),
|
||||
),
|
||||
_rule(
|
||||
"R-AVOID",
|
||||
100,
|
||||
is_hard=True,
|
||||
avoid=({"slot": "top", "attrs": {"silhouette": ["紧身"]}},),
|
||||
),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert len(verdict.recommend) == 1
|
||||
assert verdict.blocked is True # 拦截归拦截，推荐归推荐
|
||||
|
||||
|
||||
# ── 汇总与去重 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_identical_recommend_deduped_and_order_follows_priority() -> None:
|
||||
item = {"slot": "top", "attrs": {"collar": ["衬衫领"]}}
|
||||
rules = [
|
||||
_rule("R-STYLE", 70, recommend=(item,)),
|
||||
_rule("R-OCCASION", 80, recommend=(item,)),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert len(verdict.recommend) == 1
|
||||
assert len(verdict.avoid) == 0
|
||||
|
||||
|
||||
def test_identical_avoid_deduped() -> None:
|
||||
avoid = {"slot": "top", "attrs": {"silhouette": ["修身"]}}
|
||||
rules = [
|
||||
_rule("R-A", 90, avoid=(avoid,)),
|
||||
_rule("R-B", 60, avoid=(avoid,)),
|
||||
]
|
||||
verdict = engine.evaluate(rules, dsl.RuleContext())
|
||||
assert len(verdict.avoid) == 1
|
||||
|
||||
|
||||
# ── 性能基准（T-P2-02：p95 < 50ms @ 400 条规则）─────────────────────────────
|
||||
|
||||
|
||||
def test_benchmark_400_rules_well_under_50ms_budget() -> None:
|
||||
"""400 条规则 × 1000 次求值。
|
||||
|
||||
断言均值 < 5ms/次（对 50ms 预算留 10 倍裕量，CI 机器差异不至于 flaky），
|
||||
并直接断言 p95 < 50ms 对应验收指标。求值是纯内存操作，规则从缓存来，
|
||||
不含任何 I/O。
|
||||
"""
|
||||
shapes = ("A型", "H型", "X型", "O型")
|
||||
rules = [
|
||||
_rule(
|
||||
f"R-BENCH-{i:03d}",
|
||||
50 + (i % 6) * 10,
|
||||
category="bench",
|
||||
confidence=0.5 + (i % 10) / 20,
|
||||
conditions={"body_shape": shapes[i % 4], "season": {"exclude": ["夏"]}},
|
||||
recommend=({"slot": "top", "attrs": {"collar": [f"领型{i % 5}"]}},),
|
||||
avoid=({"slot": "bottom", "attrs": {"silhouette": ["紧身"]}},),
|
||||
explanation=f"基准规则 {i}",
|
||||
)
|
||||
for i in range(400)
|
||||
]
|
||||
assert len(rules) == 400
|
||||
ctx = dsl.RuleContext(body_shape="A型", season="冬", temperature=18.0)
|
||||
|
||||
engine.evaluate(rules, ctx) # 预热：首次调用的函数定位等开销不摊给被测对象
|
||||
timings: list[float] = []
|
||||
for _ in range(1000):
|
||||
start = time.perf_counter()
|
||||
engine.evaluate(rules, ctx)
|
||||
timings.append(time.perf_counter() - start)
|
||||
|
||||
mean = sum(timings) / len(timings)
|
||||
p95 = sorted(timings)[int(len(timings) * 0.95) - 1]
|
||||
assert mean < 0.005, f"平均单次求值 {mean * 1000:.2f}ms，超出 5ms 基线"
|
||||
assert p95 < 0.05, f"p95 单次求值 {p95 * 1000:.2f}ms，超出 50ms 预算"
|
||||
@@ -0,0 +1,330 @@
|
||||
"""规则引擎黄金集雏形（架构 §1081）：固定画像 → 期望输出。
|
||||
|
||||
改引擎行为必须先过黄金集——这是"专业可信"的回归防线。必含两类场景：
|
||||
硬禁忌（用户禁忌压过专业推荐）与多规则叠加。规则库是知识库文档 §3.2⑩ 的
|
||||
迷你缩影（体型/场景/温度/风格/趋势/用户禁忌/合规硬约束各至少一条）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
from typing import Any
|
||||
|
||||
from packages.rules import dsl, engine
|
||||
|
||||
# ── 固定规则库（8 条）───────────────────────────────────────────────────────
|
||||
|
||||
_GOLDEN_RULES = [
|
||||
# 体型（90）：文档 §3.2⑩ 原始示例规则
|
||||
{
|
||||
"rule_key": "R-BODY-A-002",
|
||||
"category": "body_shape",
|
||||
"priority": 90,
|
||||
"confidence": 0.95,
|
||||
"conditions": {
|
||||
"body_shape": "A型",
|
||||
"features": {"include": ["溜肩"]},
|
||||
"goals": {"include": ["修饰溜肩"]},
|
||||
},
|
||||
"effects": {
|
||||
"recommend": [
|
||||
{"slot": "top", "attrs": {"shoulder": ["垫肩", "自然肩"], "sleeve": ["泡泡袖"]}}
|
||||
],
|
||||
"avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"], "silhouette": ["修身"]}}],
|
||||
"explanation": "溜肩需要用垫肩与泡泡袖在肩部建立结构感。",
|
||||
},
|
||||
},
|
||||
# 体型（90，同优先级低置信度：验证二级排序）
|
||||
{
|
||||
"rule_key": "R-BODY-A-010",
|
||||
"category": "body_shape",
|
||||
"priority": 90,
|
||||
"confidence": 0.90,
|
||||
"conditions": {"body_shape": "A型"},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "bottom", "attrs": {"silhouette": ["直筒", "阔腿"]}}],
|
||||
"avoid": [{"slot": "bottom", "attrs": {"silhouette": ["紧身"]}}],
|
||||
"explanation": "A型下半身宜直线轮廓，避免紧身放大胯部。",
|
||||
},
|
||||
},
|
||||
# 场景（80）
|
||||
{
|
||||
"rule_key": "R-OCCASION-INTERVIEW-001",
|
||||
"category": "occasion",
|
||||
"priority": 80,
|
||||
"confidence": 0.90,
|
||||
"conditions": {"occasion": "面试"},
|
||||
"effects": {
|
||||
"recommend": [
|
||||
{"slot": "top", "attrs": {"collar": ["衬衫领"]}},
|
||||
{"slot": "bottom", "attrs": {"silhouette": ["直筒"]}},
|
||||
],
|
||||
"avoid": [{"slot": "top", "attrs": {"neckline": ["深V"]}}],
|
||||
"explanation": "面试场合需要干练可信的轮廓。",
|
||||
},
|
||||
},
|
||||
# 温度（75）：闭区间
|
||||
{
|
||||
"rule_key": "R-TEMP-COMFORT-18-26",
|
||||
"category": "temperature",
|
||||
"priority": 75,
|
||||
"confidence": 0.85,
|
||||
"conditions": {"temperature": {"min": 18, "max": 26}},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "outer", "attrs": {"thickness": ["薄款"]}}],
|
||||
"avoid": [{"slot": "outer", "attrs": {"thickness": ["厚款"]}}],
|
||||
"explanation": "18-26℃ 单层薄外套最舒适。",
|
||||
},
|
||||
},
|
||||
# 风格（70）：与场景规则的推荐完全相同 → 验证去重
|
||||
{
|
||||
"rule_key": "R-STYLE-OL-001",
|
||||
"category": "style",
|
||||
"priority": 70,
|
||||
"confidence": 0.80,
|
||||
"conditions": {"style": "通勤"},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"collar": ["衬衫领"]}}],
|
||||
"explanation": "通勤风格以衬衫领立住精神气。",
|
||||
},
|
||||
},
|
||||
# 趋势（50）：推荐恰与用户禁忌冲突 → 必须被压过
|
||||
{
|
||||
"rule_key": "R-TREND-SWEET-001",
|
||||
"category": "trend",
|
||||
"priority": 50,
|
||||
"confidence": 0.70,
|
||||
"conditions": {"style": "甜酷"},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"length": ["短款"]}}],
|
||||
"explanation": "短款上衣是当季甜酷风的核心单品。",
|
||||
},
|
||||
},
|
||||
# 用户禁忌（95）：avoid 永远压过专业推荐
|
||||
{
|
||||
"rule_key": "R-TABOO-NAVEL-001",
|
||||
"category": "user_taboo",
|
||||
"priority": 95,
|
||||
"confidence": 1.0,
|
||||
"conditions": {"user_taboos": {"include": ["露脐"]}},
|
||||
"effects": {
|
||||
"avoid": [{"slot": "top", "attrs": {"length": ["短款", "露脐"]}}],
|
||||
"explanation": "用户不接受露脐，已剔除相关单品。",
|
||||
},
|
||||
},
|
||||
# 合规（100）：is_hard，avoid 命中即硬拦截
|
||||
{
|
||||
"rule_key": "R-HARD-EXOTIC-001",
|
||||
"category": "compliance",
|
||||
"priority": 100,
|
||||
"confidence": 1.0,
|
||||
"is_hard": True,
|
||||
"conditions": {"occasion": {"include": ["红毯", "晚宴"]}},
|
||||
"effects": {
|
||||
"avoid": [
|
||||
{"slot": "accessory", "attrs": {"material": ["真兽皮"]}},
|
||||
{"slot": "shoes", "attrs": {"material": ["真皮"]}},
|
||||
],
|
||||
"explanation": "合规红线：不推荐真兽皮材质。",
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
_RULES = [dsl.Rule.model_validate(r) for r in _GOLDEN_RULES]
|
||||
|
||||
|
||||
# ── 断言辅助：把 SlotAttrs 规整成可比较的集合 ────────────────────────────────
|
||||
|
||||
|
||||
#: 单个 recommend/avoid 条目规整后的形状。
|
||||
NormItem = tuple[str, tuple[tuple[str, tuple[str, ...]], ...]]
|
||||
|
||||
|
||||
def _item(slot: str, **dims: tuple[str, ...]) -> NormItem:
|
||||
# 与 _norm 相同的规整规则（维度与取值都排序），期望值不必手写排序
|
||||
return (slot, tuple(sorted((d, tuple(sorted(v))) for d, v in dims.items())))
|
||||
|
||||
|
||||
def _norm(items: Iterable[dsl.SlotAttrs]) -> frozenset[NormItem]:
|
||||
return frozenset(
|
||||
(i.slot, tuple(sorted((d, tuple(sorted(v))) for d, v in i.attrs.items()))) for i in items
|
||||
)
|
||||
|
||||
|
||||
def _evaluate(**ctx: Any) -> engine.RuleVerdict:
|
||||
return engine.evaluate(_RULES, dsl.RuleContext(**ctx))
|
||||
|
||||
|
||||
def _hit_keys(verdict: engine.RuleVerdict) -> list[str]:
|
||||
return [h.rule_key for h in verdict.rule_hits]
|
||||
|
||||
|
||||
# ── 1. 基础命中 ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_a_shape_with_sloped_shoulders() -> None:
|
||||
verdict = _evaluate(body_shape="A型", features=("溜肩",), goals=("修饰溜肩",))
|
||||
assert _hit_keys(verdict) == ["R-BODY-A-002", "R-BODY-A-010"]
|
||||
assert _norm(verdict.recommend) == {
|
||||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||||
}
|
||||
assert _norm(verdict.avoid) == {
|
||||
_item("top", shoulder=("落肩",), silhouette=("修身",)),
|
||||
_item("bottom", silhouette=("紧身",)),
|
||||
}
|
||||
assert verdict.hard_blocks == ()
|
||||
assert verdict.blocked is False
|
||||
|
||||
|
||||
def test_missing_body_shape_does_not_fire_shape_rules() -> None:
|
||||
# 缺字段 = 不匹配该子句：只有 features/goals 时体型规则不命中
|
||||
verdict = _evaluate(features=("溜肩",), goals=("修饰溜肩",))
|
||||
assert _hit_keys(verdict) == []
|
||||
assert verdict.recommend == ()
|
||||
assert verdict.avoid == ()
|
||||
assert verdict.explanation == ""
|
||||
|
||||
|
||||
def test_include_is_subset_semantics() -> None:
|
||||
# features 是 include 列表的超集 → 命中
|
||||
hit = _evaluate(body_shape="A型", features=("溜肩", "宽胯"), goals=("修饰溜肩",))
|
||||
assert "R-BODY-A-002" in _hit_keys(hit)
|
||||
# goals 缺一个 → 不命中
|
||||
miss = _evaluate(body_shape="A型", features=("溜肩",), goals=("显瘦",))
|
||||
assert "R-BODY-A-002" not in _hit_keys(miss)
|
||||
assert _hit_keys(miss) == ["R-BODY-A-010"]
|
||||
|
||||
|
||||
# ── 2. 多规则叠加 ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_body_plus_temperature_plus_occasion_stack() -> None:
|
||||
verdict = _evaluate(
|
||||
body_shape="A型",
|
||||
features=("溜肩",),
|
||||
goals=("修饰溜肩",),
|
||||
temperature=20.0,
|
||||
occasion="面试",
|
||||
)
|
||||
assert _hit_keys(verdict) == [
|
||||
"R-BODY-A-002",
|
||||
"R-BODY-A-010",
|
||||
"R-OCCASION-INTERVIEW-001",
|
||||
"R-TEMP-COMFORT-18-26",
|
||||
]
|
||||
# 四条规则的推荐全部叠加，互不冲突（不同维度/不同 slot；两条 bottom 推荐
|
||||
# 的属性集不同，各自保留）
|
||||
assert _norm(verdict.recommend) == {
|
||||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||||
_item("top", collar=("衬衫领",)),
|
||||
_item("bottom", silhouette=("直筒",)),
|
||||
_item("outer", thickness=("薄款",)),
|
||||
}
|
||||
assert _norm(verdict.avoid) == {
|
||||
_item("top", shoulder=("落肩",), silhouette=("修身",)),
|
||||
_item("bottom", silhouette=("紧身",)),
|
||||
_item("top", neckline=("深V",)),
|
||||
_item("outer", thickness=("厚款",)),
|
||||
}
|
||||
assert verdict.blocked is False
|
||||
|
||||
|
||||
def test_same_priority_sorted_by_confidence_and_dedupe_collapsed() -> None:
|
||||
# A-002 不命中（缺 features/goals），A-010 与面试/通勤规则叠加；
|
||||
# 场景(80)与风格(70)推荐同一件衬衫领 → 去重为一条
|
||||
verdict = _evaluate(body_shape="A型", occasion="面试", style="通勤")
|
||||
assert _hit_keys(verdict) == [
|
||||
"R-BODY-A-010",
|
||||
"R-OCCASION-INTERVIEW-001",
|
||||
"R-STYLE-OL-001",
|
||||
]
|
||||
assert _norm(verdict.recommend) == {
|
||||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||||
_item("bottom", silhouette=("直筒",)),
|
||||
_item("top", collar=("衬衫领",)),
|
||||
}
|
||||
assert verdict.explanations == (
|
||||
"A型下半身宜直线轮廓，避免紧身放大胯部。",
|
||||
"面试场合需要干练可信的轮廓。",
|
||||
"通勤风格以衬衫领立住精神气。",
|
||||
)
|
||||
|
||||
|
||||
def test_temperature_closed_interval_boundaries() -> None:
|
||||
assert _hit_keys(_evaluate(temperature=18.0)) == ["R-TEMP-COMFORT-18-26"]
|
||||
assert _hit_keys(_evaluate(temperature=26.0)) == ["R-TEMP-COMFORT-18-26"]
|
||||
assert _hit_keys(_evaluate(temperature=17.9)) == []
|
||||
assert _hit_keys(_evaluate(temperature=26.1)) == []
|
||||
|
||||
|
||||
# ── 3. 用户禁忌压过专业推荐（硬性场景）──────────────────────────────────────
|
||||
|
||||
|
||||
def test_user_taboo_overrides_trend_recommendation() -> None:
|
||||
# 趋势(50)推荐短款上衣，用户禁忌(95)回避短款/露脐 → 推荐被剔除
|
||||
verdict = _evaluate(style="甜酷", user_taboos=("露脐",))
|
||||
assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-TREND-SWEET-001"]
|
||||
assert verdict.recommend == ()
|
||||
assert _norm(verdict.avoid) == {_item("top", length=("露脐", "短款"))}
|
||||
# 禁忌不是 is_hard：不构成硬拦截，但推荐侧已经干净
|
||||
assert verdict.hard_blocks == ()
|
||||
assert verdict.explanation.startswith("用户不接受露脐")
|
||||
|
||||
|
||||
def test_user_taboo_keeps_unrelated_slot_recommendations() -> None:
|
||||
# 禁忌只作用于 top.length：体型对 bottom 的推荐不受影响
|
||||
verdict = _evaluate(body_shape="A型", user_taboos=("露脐",))
|
||||
assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-BODY-A-010"]
|
||||
assert _norm(verdict.recommend) == {_item("bottom", silhouette=("阔腿", "直筒"))}
|
||||
|
||||
|
||||
# ── 4. 硬约束拦截（is_hard）─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_hard_compliance_rule_blocks_gala_occasion() -> None:
|
||||
verdict = _evaluate(occasion="红毯")
|
||||
assert _hit_keys(verdict) == ["R-HARD-EXOTIC-001"]
|
||||
assert verdict.blocked is True
|
||||
assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"]
|
||||
# 调用方用 hard_blocks.avoid 对候选做硬过滤：约束内容完整带出
|
||||
assert _norm(verdict.hard_blocks[0].avoid) == {
|
||||
_item("accessory", material=("真兽皮",)),
|
||||
_item("shoes", material=("真皮",)),
|
||||
}
|
||||
assert verdict.hard_blocks[0].explanation == "合规红线：不推荐真兽皮材质。"
|
||||
|
||||
|
||||
def test_hard_rule_not_fired_on_other_occasions() -> None:
|
||||
assert _evaluate(occasion="通勤").hard_blocks == ()
|
||||
assert _evaluate(occasion="通勤").blocked is False
|
||||
|
||||
|
||||
# ── 5. 综合叠加：优先级链 + 禁忌压制 + 硬拦截同时发生 ────────────────────────
|
||||
|
||||
|
||||
def test_composite_profile_full_priority_chain() -> None:
|
||||
verdict = _evaluate(
|
||||
body_shape="A型",
|
||||
features=("溜肩",),
|
||||
goals=("修饰溜肩",),
|
||||
style="甜酷",
|
||||
user_taboos=("露脐",),
|
||||
occasion="红毯",
|
||||
)
|
||||
# 命中链严格按 priority 降序、同优先级按 confidence 降序
|
||||
assert _hit_keys(verdict) == [
|
||||
"R-HARD-EXOTIC-001", # 100
|
||||
"R-TABOO-NAVEL-001", # 95
|
||||
"R-BODY-A-002", # 90 / 0.95
|
||||
"R-BODY-A-010", # 90 / 0.90
|
||||
"R-TREND-SWEET-001", # 50
|
||||
]
|
||||
# 体型推荐保留（与禁忌不同维度）；趋势推荐短款被禁忌压掉
|
||||
assert _norm(verdict.recommend) == {
|
||||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||||
}
|
||||
assert verdict.blocked is True
|
||||
assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"]
|
||||
@@ -0,0 +1,122 @@
|
||||
"""service 层纯逻辑：状态机矩阵、DSL 写入校验、ORM 行 → 域规则转换。
|
||||
|
||||
不连库——DB 路径由 tests/integration/test_rules_engine.py 覆盖。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.rules import service
|
||||
from packages.shared.errors import AppError, ErrorCode
|
||||
from packages.shared.models import KbRule
|
||||
|
||||
VALID_PAYLOAD = {
|
||||
"rule_key": "R-TEST-001",
|
||||
"category": "body_shape",
|
||||
"priority": 90,
|
||||
"conditions": {"body_shape": "A型"},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"shoulder": ["垫肩"]}}],
|
||||
"avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"]}}],
|
||||
"explanation": "测试规则",
|
||||
},
|
||||
"is_hard": False,
|
||||
"confidence": 0.95,
|
||||
}
|
||||
|
||||
ALL_STATUSES = ("draft", "published", "archived")
|
||||
|
||||
|
||||
# ── 状态机全跃迁矩阵 ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_legal_transitions_accepted() -> None:
|
||||
for current, target in (
|
||||
("draft", "published"), # 发布
|
||||
("published", "archived"), # 下线
|
||||
("draft", "archived"), # 撤稿
|
||||
):
|
||||
service.ensure_transition(current, target) # 不抛即通过
|
||||
|
||||
|
||||
@pytest.mark.parametrize("current", ALL_STATUSES)
|
||||
@pytest.mark.parametrize("target", ALL_STATUSES)
|
||||
def test_illegal_transitions_rejected(current: str, target: str) -> None:
|
||||
if (current, target) in service.LEGAL_TRANSITIONS:
|
||||
pytest.skip("合法跃迁另测")
|
||||
with pytest.raises(AppError) as excinfo:
|
||||
service.ensure_transition(current, target)
|
||||
assert excinfo.value.code == ErrorCode.STATE_CONFLICT
|
||||
assert excinfo.value.data == {"current": current, "target": target}
|
||||
|
||||
|
||||
# ── DSL 写入校验 ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_validate_rule_payload_accepts_valid_dsl() -> None:
|
||||
rule = service.validate_rule_payload(VALID_PAYLOAD)
|
||||
assert rule.rule_key == "R-TEST-001"
|
||||
|
||||
|
||||
def test_validate_rule_payload_rejects_unknown_keys_with_details() -> None:
|
||||
with pytest.raises(AppError) as excinfo:
|
||||
service.validate_rule_payload({**VALID_PAYLOAD, "conclusions": []})
|
||||
assert excinfo.value.code == ErrorCode.VALIDATION_FAILED
|
||||
assert excinfo.value.data["errors"] # 拒绝原因必须可读，不能只说"不合法"
|
||||
assert any("conclusions" in err for err in excinfo.value.data["errors"])
|
||||
|
||||
|
||||
def test_validate_rule_payload_rejects_unknown_condition_field() -> None:
|
||||
with pytest.raises(AppError) as excinfo:
|
||||
service.validate_rule_payload({**VALID_PAYLOAD, "conditions": {"body_shap": "A型"}})
|
||||
assert excinfo.value.code == ErrorCode.VALIDATION_FAILED
|
||||
|
||||
|
||||
# ── ORM 行 → 域规则 ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _row(**overrides: object) -> KbRule:
|
||||
values: dict[str, object] = {
|
||||
"rule_id": 1,
|
||||
"rule_key": "R-ROW-001",
|
||||
"category": "body_shape",
|
||||
"version": 1,
|
||||
"status": "published",
|
||||
"priority": 90,
|
||||
"conditions": {"body_shape": "A型"},
|
||||
"effects": {
|
||||
"recommend": [{"slot": "top", "attrs": {"shoulder": ["垫肩"]}}],
|
||||
"explanation": "行转换",
|
||||
},
|
||||
"is_hard": False,
|
||||
"confidence": Decimal("0.950"),
|
||||
}
|
||||
values.update(overrides)
|
||||
return KbRule(**values) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_row_to_rule_parses_stored_jsonb() -> None:
|
||||
rule = service.row_to_rule(_row())
|
||||
assert rule.rule_key == "R-ROW-001"
|
||||
assert rule.confidence == pytest.approx(0.95) # Numeric(4,3) → float
|
||||
assert rule.conditions.clauses["body_shape"] == "A型"
|
||||
assert rule.effects.recommend[0].attrs["shoulder"] == ("垫肩",)
|
||||
|
||||
|
||||
def test_row_to_rule_rejects_corrupted_stored_dsl() -> None:
|
||||
# 库里数据坏了是 5xxxx 知识库数据错误，必须响亮失败而不是静默跳过
|
||||
with pytest.raises(AppError) as excinfo:
|
||||
service.row_to_rule(_row(conditions={"mood": "开心"}))
|
||||
assert excinfo.value.code == ErrorCode.CONFIG_ERROR
|
||||
assert excinfo.value.data == {"rule_key": "R-ROW-001"}
|
||||
|
||||
|
||||
def test_published_rules_cache_invalidate_resets() -> None:
|
||||
cache = service.PublishedRules()
|
||||
assert cache._rules is None
|
||||
cache._rules = ()
|
||||
cache.invalidate()
|
||||
assert cache._rules is None
|
||||
Reference in New Issue
Block a user