import pytest from lead_scoring.schemas.dynamic_rule import ( DynamicRule, DynamicRuleOperator, DynamicRuleTarget, ) from lead_scoring.schemas.scoring_config import ScoringConfig from lead_scoring.scoring import ( RuleEvaluationStatus, evaluate_education_rules, ) from test_configurable_demo_backend import make_education_snapshot def test_dynamic_rule_operators(): snap = make_education_snapshot( { "student_age": 12, "budget": 6000000, "branch": "Quan 3, TP. Ho Chi Minh", "urgency": "high", }, conversations=[ { "conversation_id": "conv-1", "channel": "zalo", "messages": [ { "message_id": "m1", "sender_type": "contact", "sent_at": "2026-08-28T11:00:00Z", "content": "Toi can tim lop tieng Anh gap", } ], } ], ) rules = ( DynamicRule( id="rule_budget_gt", name="Ngân sách trên 5 triệu", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="budget", operator=DynamicRuleOperator.GT, value=5000000, points=15, ), DynamicRule( id="rule_branch_contains", name="Cơ sở Quận 3", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="branch", operator=DynamicRuleOperator.CONTAINS, value="Quận 3", points=10, ), DynamicRule( id="rule_urgency_in", name="Mức độ cấp bách cao hoặc khẩn", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="urgency", operator=DynamicRuleOperator.IN, value=["high", "urgent"], points=10, ), DynamicRule( id="rule_age_gte", name="Tuổi từ 10 trở lên", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.GTE, value=10, points=5, ), ) config = ScoringConfig( version="dynamic-test-v1", base_score=0, confidence=0.85, confidence_threshold=0.8, hitl_policy_version="1.0.0", education_demo_enabled=True, dynamic_rules=rules, ) score, breakdown, _, _ = evaluate_education_rules(snap, config) # Fixed rules: # student_age=12 -> age_primary (40) # recency -> hot (40) # lead_source -> missing (0) # Fixed subtotal = 80 # Dynamic rules: # budget (15) + branch (10) + urgency (10) + age_gte (5) = 40 # Total = 80 + 40 = 120 -> clamped to 100 assert score == 100 dyn_items = [b for b in breakdown if b.rule_name.startswith("rule_")] assert len(dyn_items) == 4 assert all(item.status == RuleEvaluationStatus.MATCHED for item in dyn_items) assert sum(item.points_awarded for item in dyn_items) == 40 def test_dynamic_rule_missing_and_disabled(): snap = make_education_snapshot( {"student_age": 12}, conversations=[], ) rules = ( # Missing custom field DynamicRule( id="rule_missing_vip", name="Khách VIP", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="vip_status", operator=DynamicRuleOperator.EQ, value="gold", points=20, ), # Disabled rule DynamicRule( id="rule_disabled", name="Luật tạm dừng", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.EQ, value=12, points=30, enabled=False, ), ) config = ScoringConfig( version="dynamic-test-missing-v1", base_score=0, confidence=0.85, confidence_threshold=0.8, hitl_policy_version="1.0.0", education_demo_enabled=True, dynamic_rules=rules, ) score, breakdown, _, _ = evaluate_education_rules(snap, config) missing_item = next(b for b in breakdown if b.rule_name == "rule_missing_vip") assert missing_item.status == RuleEvaluationStatus.SKIPPED_MISSING_DATA assert missing_item.points_awarded == 0 disabled_item = next(b for b in breakdown if b.rule_name == "rule_disabled") assert disabled_item.status == RuleEvaluationStatus.NOT_MATCHED assert disabled_item.points_awarded == 0 def test_rule_overrides_validation(): import pytest from pydantic import ValidationError from lead_scoring.schemas.scoring_run import CreateScoringRunRequest base_snapshot = make_education_snapshot({"student_age": 12, "course_interest": "English"}) # 1. Duplicate rule ID with pytest.raises(ValidationError, match="Duplicate dynamic rule ID"): CreateScoringRunRequest( tenant_id="tenant_1", subject_ref=base_snapshot.lead_profile.ems_lead_id, scoring_config_version="1.0.0", rule_overrides=[ DynamicRule( id="rule_1", name="Rule 1", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.EQ, value=12, points=10, ), DynamicRule( id="rule_1", name="Rule 1 Duplicate", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.EQ, value=12, points=20, ), ], snapshot=base_snapshot, ) # 2. Custom field without custom_field_key with pytest.raises( ValidationError, match="custom_field_key is required when target is 'custom_field'" ): CreateScoringRunRequest( tenant_id="tenant_1", subject_ref=base_snapshot.lead_profile.ems_lead_id, scoring_config_version="1.0.0", rule_overrides=[ DynamicRule( id="rule_cf", name="Missing Key", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="", operator=DynamicRuleOperator.EQ, value="test", points=10, ) ], snapshot=base_snapshot, ) # 3. String value > 1000 chars with pytest.raises(ValidationError, match="exceeds maximum allowed size"): CreateScoringRunRequest( tenant_id="tenant_1", subject_ref=base_snapshot.lead_profile.ems_lead_id, scoring_config_version="1.0.0", rule_overrides=[ DynamicRule( id="rule_long", name="Long String", target=DynamicRuleTarget.LEAD_SOURCE, operator=DynamicRuleOperator.EQ, value="x" * 1001, points=10, ) ], snapshot=base_snapshot, ) # 4. List value > 100 items with pytest.raises(ValidationError, match="exceeds maximum allowed length"): CreateScoringRunRequest( tenant_id="tenant_1", subject_ref=base_snapshot.lead_profile.ems_lead_id, scoring_config_version="1.0.0", rule_overrides=[ DynamicRule( id="rule_list_long", name="Long List", target=DynamicRuleTarget.LEAD_SOURCE, operator=DynamicRuleOperator.IN, value=["item"] * 101, points=10, ) ], snapshot=base_snapshot, ) # 5. Over 30 rules with pytest.raises(ValidationError): CreateScoringRunRequest( tenant_id="tenant_1", subject_ref=base_snapshot.lead_profile.ems_lead_id, scoring_config_version="1.0.0", rule_overrides=[ DynamicRule( id=f"rule_{i}", name=f"Rule {i}", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.EQ, value=i, points=1, ) for i in range(31) ], snapshot=base_snapshot, ) def test_rule_overrides_overlay_and_replay(): snap = make_education_snapshot( { "student_age": 10, "target_campus": "Disctrict 1", "course_interest": "IELTS", }, conversations=[ { "conversation_id": "conv-overlay", "channel": "zalo", "messages": [ { "message_id": "m1", "sender_type": "contact", "sent_at": "2026-08-28T11:00:00Z", "content": "Toi can dang ky IELTS", } ], } ], ) base_config = ScoringConfig( version="base-overlay-v1", base_score=10, confidence=0.90, confidence_threshold=0.80, hitl_policy_version="1.0.0", dynamic_rules=(), ) overrides = [ DynamicRule( id="rule_campus_d1", name="Cơ sở Quận 1", target=DynamicRuleTarget.CUSTOM_FIELD, custom_field_key="target_campus", operator=DynamicRuleOperator.CONTAINS, value="District 1", points=25, ), DynamicRule( id="rule_age_10", name="Tuổi đúng 10", target=DynamicRuleTarget.STUDENT_AGE, operator=DynamicRuleOperator.EQ, value=10, points=15, ), ] effective_config = base_config.model_copy(update={"dynamic_rules": tuple(overrides)}) assert len(effective_config.dynamic_rules) == 2 score, breakdown, _, _ = evaluate_education_rules(snap, effective_config) age_bd = next(b for b in breakdown if b.rule_name == "rule_age_10") assert age_bd.status == RuleEvaluationStatus.MATCHED assert age_bd.points_awarded == 15 @pytest.mark.asyncio async def test_api_rule_overrides_overlay_and_replay(temp_postgres_db: str): import asyncio from pathlib import Path from alembic import command from alembic.config import Config from httpx import ASGITransport, AsyncClient from pydantic import SecretStr from lead_scoring.config import ServiceCallerConfig, Settings from lead_scoring.main import create_app from lead_scoring.schemas.scoring_config import ScoringConfig root_dir = Path(__file__).resolve().parent.parent ini_path = root_dir / "alembic.ini" alembic_cfg = Config(str(ini_path)) alembic_cfg.attributes["sqlalchemy_url"] = temp_postgres_db await asyncio.to_thread(command.upgrade, alembic_cfg, "head") base_config = ScoringConfig( version="overlay-demo-v1", base_score=20, rule_weight=1.0, ai_weight=0.0, confidence=0.90, confidence_threshold=0.80, hitl_policy_version="1.0.0", dynamic_rules=(), ) settings = Settings( database_url=temp_postgres_db, scoring_configs=[base_config], tenant_allowlist=["tenant_dyn"], service_callers=[ ServiceCallerConfig( caller_id="test_caller", token=SecretStr("valid_token"), allowed_apps=["omni"], allowed_features=["scoring"], allowed_tenants=["tenant_dyn"], ) ], ) app = create_app(settings=settings) snapshot_data = { "version": "1.0.0", "scope": "linked_contact_history", "captured_at": "2026-08-27T12:00:00Z", "lead_profile": { "ems_lead_id": "lead_overlay_1", "name": "Student A", "custom_fields": { "course_interest": "Python Bootcamp", "campus": "District 1", }, }, "conversations": [], } rule_overrides = [ { "id": "dyn_campus_d1", "name": "District 1 Campus", "target": "custom_field", "custom_field_key": "campus", "operator": "eq", "value": "District 1", "points": 30, "enabled": True, } ] payload = { "tenant_id": "tenant_dyn", "subject_ref": "lead_overlay_1", "scoring_config_version": "overlay-demo-v1", "rule_overrides": rule_overrides, "snapshot": snapshot_data, } headers = { "Authorization": "Bearer valid_token", "Idempotency-Key": "idem-overlay-101", "X-DotB-App": "omni", "X-DotB-Feature": "scoring", } async with app.router.lifespan_context(app): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: # 1. Admit run with rule_overrides resp = await client.post("/v1/scoring-runs", json=payload, headers=headers) assert resp.status_code == 202 run_id = resp.json()["run_id"] # 2. Wait for completion get_resp = None for _ in range(50): await asyncio.sleep(0.1) get_resp = await client.get( f"/v1/scoring-runs/{run_id}", headers={"Authorization": "Bearer valid_token", "X-Tenant-Id": "tenant_dyn"}, ) if get_resp.status_code == 200 and get_resp.json()["status"] in { "completed", "failed", }: break assert get_resp is not None run_data = get_resp.json() assert run_data["status"] == "completed" assert run_data["rule_score"] == 50 # 20 base + 30 rule assert run_data["final_score"] == 50 assert run_data["breakdown"] is not None assert any(b["rule_name"] == "dyn_campus_d1" for b in run_data["breakdown"]) # 3. Replay with identical payload and idempotency key replay_resp = await client.post("/v1/scoring-runs", json=payload, headers=headers) assert replay_resp.status_code == 202 assert replay_resp.json()["run_id"] == run_id # 4. Check internal trace endpoint contains effective_config rule_breakdown trace_resp = await client.get( f"/v1/scoring-runs/{run_id}/trace", headers={"Authorization": "Bearer valid_token", "X-Tenant-Id": "tenant_dyn"}, ) assert trace_resp.status_code == 200 trace_data = trace_resp.json() assert any(b["rule_name"] == "dyn_campus_d1" for b in trace_data["rule_breakdown"])