"""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