Coverage for domain/rules.py: 84%
83 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-05 19:46 +0000
1"""Automation rules engine (spec §8.5): trigger -> condition -> action.
3Evaluates rules on loan lifecycle events (LoanCreated, StatusChanged, TridDateSet).
4Pure domain logic: no direct DB queries or side effects.
5"""
7from __future__ import annotations
9from collections.abc import Sequence
10from dataclasses import dataclass, field
11from typing import Any
13from domain.result import AppError
16@dataclass(frozen=True)
17class RuleAction:
18 action_type: str # "Stop" | "RequireFields" | "SendDocument" | "CreateTask"
19 message: str = ""
20 fields: tuple[str, ...] = ()
21 document_classification_id: str = ""
22 task_title: str = ""
25@dataclass(frozen=True)
26class RuleCondition:
27 kind: str # "fieldIn" | "fieldEq" | "licenseInvalidForOccupancy"
28 field_name: str = ""
29 value: Any = None
30 values: tuple[Any, ...] = ()
33@dataclass(frozen=True)
34class AutomationRuleDef:
35 id: str # e.g. "ORG-00001"
36 name: str
37 trigger: str # "LoanCreated" | "StatusChanged" | "TridDateSet"
38 conditions: tuple[RuleCondition, ...] = ()
39 actions: tuple[RuleAction, ...] = ()
40 active: bool = True
43@dataclass(frozen=True)
44class RuleEvent:
45 event_type: str # "LoanCreated" | "StatusChanged" | "TridDateSet"
46 loan: Any
47 target_stage: str | None = None
48 date_key: str | None = None
49 lo: Any = None
52@dataclass(frozen=True)
53class RuleEffect:
54 rule_id: str
55 action: RuleAction
58@dataclass(frozen=True)
59class RuleOutcome:
60 errors: list[AppError] = field(default_factory=list)
61 effects: list[RuleEffect] = field(default_factory=list)
63 @property
64 def ok(self) -> bool:
65 return len(self.errors) == 0
68def _get_path(obj: Any, path: str) -> Any:
69 for part in path.split("."):
70 if obj is None: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 return None
72 if isinstance(obj, dict): 72 ↛ 74line 72 didn't jump to line 74 because the condition on line 72 was always true
73 obj = obj.get(part)
74 elif hasattr(obj, part):
75 obj = getattr(obj, part)
76 else:
77 return None
78 return obj
81def condition_holds(c: RuleCondition, ev: RuleEvent) -> bool:
82 if c.kind == "fieldIn":
83 v = ev.target_stage if c.field_name == "targetStage" else _get_path(ev.loan, c.field_name)
84 return v in c.values
85 if c.kind == "fieldEq": 85 ↛ 88line 85 didn't jump to line 88 because the condition on line 85 was always true
86 v = ev.target_stage if c.field_name == "targetStage" else _get_path(ev.loan, c.field_name)
87 return v == c.value
88 if c.kind == "dateKeyEq":
89 return ev.date_key == c.field_name
90 return True
93def evaluate_rules(rules: Sequence[AutomationRuleDef], ev: RuleEvent) -> RuleOutcome:
94 """Evaluate active automation rules against a loan event, returning errors and effects."""
95 errors: list[AppError] = []
96 effects: list[RuleEffect] = []
98 for r in rules:
99 if not r.active or r.trigger != ev.event_type:
100 continue
101 if not all(condition_holds(c, ev) for c in r.conditions): 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 continue
104 for a in r.actions:
105 if a.action_type == "Stop": 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 errors.append(AppError(code=r.id, message=a.message))
107 elif a.action_type == "RequireFields":
108 for f in a.fields:
109 val = _get_path(ev.loan, f)
110 if val is None or val == "":
111 errors.append(
112 AppError(
113 code=r.id,
114 message=f"Field {f} required by rule {r.id}",
115 field=f,
116 )
117 )
118 else:
119 effects.append(RuleEffect(rule_id=r.id, action=a))
121 return RuleOutcome(errors=errors, effects=effects)
124SEED_RULES: tuple[AutomationRuleDef, ...] = (
125 AutomationRuleDef(
126 id="ORG-00001",
127 name="Require property state & occupancy on loan creation",
128 trigger="LoanCreated",
129 conditions=(),
130 actions=(RuleAction(action_type="RequireFields", fields=("stateCode", "occupancy")),),
131 active=True,
132 ),
133 AutomationRuleDef(
134 id="ORG-00002",
135 name="Block unlicensed activity (primary & second homes)",
136 trigger="LoanCreated",
137 conditions=(
138 RuleCondition(kind="fieldIn", field_name="occupancy", values=("PrimaryResidence", "SecondHome")),
139 RuleCondition(kind="licenseInvalidForOccupancy"),
140 ),
141 actions=(
142 RuleAction(
143 action_type="Stop",
144 message="LO must have active and sponsored state license to create loan for primary residence / second home. Please contact admin / update your licenses.",
145 ),
146 ),
147 active=True,
148 ),
149 AutomationRuleDef(
150 id="ORG-00003",
151 name="Required fields when moving Purchase loan to processing",
152 trigger="StatusChanged",
153 conditions=(
154 RuleCondition(kind="fieldIn", field_name="targetStage", values=("LOAN_SETUP", "DISCLOSURE_SENT")),
155 RuleCondition(kind="fieldEq", field_name="purpose", value="Purchase"),
156 ),
157 actions=(
158 RuleAction(
159 action_type="RequireFields",
160 fields=("county", "postalCode", "stateCode", "transaction.BaseLoanAmt", "transaction.PurchasePriceAmt"),
161 ),
162 ),
163 active=True,
164 ),
165 AutomationRuleDef(
166 id="ORG-00004",
167 name="Required fields when moving Refinance loan to processing",
168 trigger="StatusChanged",
169 conditions=(
170 RuleCondition(
171 kind="fieldIn",
172 field_name="targetStage",
173 values=("LOAN_SETUP", "DISCLOSURE_SENT", "UNDERWRITING_SUBMITTED"),
174 ),
175 RuleCondition(kind="fieldEq", field_name="purpose", value="Refinance"),
176 ),
177 actions=(
178 RuleAction(
179 action_type="RequireFields",
180 fields=("county", "postalCode", "stateCode", "transaction.BaseLoanAmt", "transaction.AppraisedValueAmt"),
181 ),
182 ),
183 active=True,
184 ),
185 AutomationRuleDef(
186 id="ORG-00005",
187 name="Send privacy policy to borrowers on TRID",
188 trigger="TridDateSet",
189 conditions=(),
190 actions=(RuleAction(action_type="SendDocument", document_classification_id="LOAN_ORIGINATION_AGREEMENT"),),
191 active=False,
192 ),
193)