58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
import unittest
|
|
|
|
from core.quote_state_machine import QuoteStateMachine
|
|
from core.rules import Rule, RuleContext, RuleEngine, RuleResult
|
|
from services.risk_service import RiskService
|
|
|
|
|
|
class _StubState:
|
|
def __init__(self):
|
|
self.pending_image_urls = []
|
|
self.pending_requirements = []
|
|
self.quote_phase = "idle"
|
|
self.quote_ready_turns = 0
|
|
|
|
|
|
class RuleEngineTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_rule_engine_priority_and_stop(self):
|
|
async def pred_true(_ctx):
|
|
return True
|
|
|
|
async def act_first(_ctx):
|
|
return RuleResult(matched=True, stop=True, action="first", payload={"x": 1})
|
|
|
|
async def act_second(_ctx):
|
|
return RuleResult(matched=True, stop=True, action="second", payload={"x": 2})
|
|
|
|
engine = RuleEngine(
|
|
[
|
|
Rule(name="r2", priority=20, predicate=pred_true, action=act_second),
|
|
Rule(name="r1", priority=10, predicate=pred_true, action=act_first),
|
|
]
|
|
)
|
|
out = await engine.run(RuleContext())
|
|
self.assertTrue(out.matched)
|
|
self.assertEqual(out.action, "first")
|
|
self.assertEqual(out.payload["x"], 1)
|
|
|
|
def test_quote_state_machine_transitions(self):
|
|
sm = QuoteStateMachine(delay_turns=1)
|
|
st = _StubState()
|
|
st.pending_image_urls = ["u1"]
|
|
sm.refresh(st)
|
|
self.assertEqual(st.quote_phase, "collecting")
|
|
self.assertTrue(sm.should_defer_batch_quote(st, mark_ready=True))
|
|
self.assertEqual(st.quote_phase, "ready_to_quote")
|
|
self.assertEqual(st.quote_ready_turns, 0)
|
|
self.assertFalse(sm.should_defer_batch_quote(st, mark_ready=False))
|
|
|
|
def test_risk_service_reject_text(self):
|
|
svc = RiskService()
|
|
self.assertIn("地图", svc.build_reject_text("map"))
|
|
self.assertIn("政治", svc.build_reject_text("political"))
|
|
self.assertIn("涉黄", svc.build_reject_text("sexual"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|