e39829fc4e
批次1: 项目结构 + SQLite 存储层 + 数据模型 批次2: REST API (http.server) 批次3: LLM 编译器 (支持 OpenAI/Anthropic) 批次4: RestrictedPython 规则执行器 批次5: 规则匹配器 + LLM Callback 兜底 批次6: 冲突检测器 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
"""Tests for rule matcher."""
|
||
import pytest
|
||
import os
|
||
import tempfile
|
||
|
||
from rule_engine.matcher import RuleMatcher
|
||
from rule_engine.store import RuleStore
|
||
from rule_engine.executor import RuleExecutor
|
||
|
||
|
||
@pytest.fixture
|
||
def temp_db():
|
||
fd, path = tempfile.mkstemp(suffix=".db")
|
||
os.close(fd)
|
||
yield path
|
||
os.unlink(path)
|
||
|
||
|
||
@pytest.fixture
|
||
def store(temp_db):
|
||
return RuleStore(temp_db)
|
||
|
||
|
||
@pytest.fixture
|
||
def executor():
|
||
return RuleExecutor(timeout_ms=100)
|
||
|
||
|
||
@pytest.fixture
|
||
def matcher(store, executor):
|
||
return RuleMatcher(store, executor)
|
||
|
||
|
||
def test_find_matching_rule_by_id(matcher, store):
|
||
"""验证按 ID 查找匹配规则。"""
|
||
rule = store.create_rule(
|
||
name="test",
|
||
condition_template="premium",
|
||
code='def rule(facts):\n if facts.get("sub") == "premium":\n return {"action": "ok"}\n return None'
|
||
)
|
||
|
||
# 匹配
|
||
rule_id, result = matcher.find_matching_rule({"sub": "premium"}, rule_id=rule.id)
|
||
assert rule_id == rule.id
|
||
assert result == {"action": "ok"}
|
||
|
||
# 不匹配
|
||
rule_id, result = matcher.find_matching_rule({"sub": "basic"}, rule_id=rule.id)
|
||
assert rule_id is None
|
||
|
||
|
||
def test_find_matching_rule_all_active(matcher, store):
|
||
"""验证匹配所有活跃规则。"""
|
||
# 创建两条规则
|
||
rule1 = store.create_rule(
|
||
name="rule1",
|
||
condition_template="c1",
|
||
code='def rule(facts):\n if facts.get("type") == "a":\n return {"action": "a"}\n return None',
|
||
priority=1
|
||
)
|
||
rule2 = store.create_rule(
|
||
name="rule2",
|
||
condition_template="c2",
|
||
code='def rule(facts):\n if facts.get("type") == "b":\n return {"action": "b"}\n return None',
|
||
priority=2
|
||
)
|
||
|
||
# 匹配 rule2（优先级更高）
|
||
rule_id, result = matcher.find_matching_rule({"type": "b"})
|
||
assert rule_id == rule2.id
|
||
assert result["action"] == "b"
|
||
|
||
# 匹配 rule1
|
||
rule_id, result = matcher.find_matching_rule({"type": "a"})
|
||
assert rule_id == rule1.id
|
||
|
||
|
||
def test_find_matching_rule_no_match(matcher, store):
|
||
"""验证无匹配时返回 None。"""
|
||
store.create_rule(
|
||
name="nomatch",
|
||
condition_template="c1",
|
||
code='def rule(facts):\n return None'
|
||
)
|
||
|
||
rule_id, result = matcher.find_matching_rule({"type": "c"})
|
||
assert rule_id is None
|
||
assert result is None
|
||
|
||
|
||
def test_find_matching_rule_inactive_rule(matcher, store):
|
||
"""验证不匹配非活跃规则。"""
|
||
rule = store.create_rule(
|
||
name="inactive",
|
||
condition_template="c1",
|
||
code='def rule(facts):\n return {"action": "ok"}'
|
||
)
|
||
# 手动设置为非活跃（通过直接操作）
|
||
store.delete_rule(rule.id)
|
||
|
||
rule_id, result = matcher.find_matching_rule({"type": "anything"})
|
||
assert rule_id is None
|