Coverage for domain/validation.py: 97%

1"""Create + transition validations incl. the ORG-xxxx business rules, ported from 

2``arive-spec/src/domain/validation.ts``. 

3 

4Pure: functions operate on plain dataclasses/dicts and return ``Result`` objects; the 

5API layer maps results to HTTP. ``REQUIRED_TO_PROCESSING`` is adapted to the fields this 

6platform's models currently carry (the oracle also required postalCode/RefinanceType, 

7which arrive with the 1003 sub-models later). 

8""" 

9 

10from __future__ import annotations 

11 

12import re 

13from dataclasses import dataclass, field 

14from typing import Any 

15 

16from .enums import OccupancyType 

17from .result import AppError, Result, err, ok 

18from .state_compliance import STATE_COMPLIANCE 

19 

20VALID_STATE_CODES = frozenset(r["state_code"] for r in STATE_COMPLIANCE) 

21EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]{2,}$") 

22 

23 

24@dataclass 

25class LicenseInput: 

26 state_code: str 

27 active: bool = False 

28 sponsored: bool = False 

29 

30 

31@dataclass 

32class LoanOfficerInput: 

33 licenses: list[LicenseInput] = field(default_factory=list) 

34 

35 

36@dataclass 

37class BorrowerInput: 

38 first_name: str | None = None 

39 last_name: str | None = None 

40 email: str | None = None 

41 

42 

43@dataclass 

44class CreateLoanInput: 

45 purpose: str | None = None 

46 occupancy: str | None = None 

47 state_code: str | None = None 

48 borrower: BorrowerInput | None = None 

49 

50 

51def validate_licensing(lo: LoanOfficerInput, state_code: str, occupancy: str) -> Result: 

52 """ORG-00002: LO needs an active, sponsored state license for primary/second homes.""" 

53 if occupancy == OccupancyType.INVESTMENT.value: 

54 return ok(True) 

55 lic = next((lic for lic in lo.licenses if lic.state_code == state_code), None) 

56 if lic is None or not lic.active or not lic.sponsored: 

57 return err( 

58 AppError( 

59 code="ORG-00002", 

60 message=( 

61 "LO must have active and sponsored state license to create loan for " 

62 "primary residence / second home. Please contact admin / update licenses." 

63 ), 

64 field="loanOfficerId", 

65 ) 

66 ) 

67 return ok(True) 

68 

69 

70def validate_loan_header( 

71 state_code: str | None, occupancy: str | None, lo: LoanOfficerInput 

72) -> Result: 

73 """ORG-00001 (state/occupancy required) + ORG-00002 (licensing) for a loan header 

74 created without borrower identity. `validate_create_loan` is the one-call, 

75 borrower-included shape `POST /api/loans` actually uses.""" 

76 errors: list[AppError] = [] 

77 if not state_code: 

78 errors.append( 

79 AppError("ORG-00001", "Required field missing: Subject Property State.", "stateCode") 

80 ) 

81 elif state_code.upper() not in VALID_STATE_CODES: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 errors.append( 

83 AppError( 

84 "ORG-00001", 

85 f"Unrecognized Subject Property State: '{state_code}'. Must be a valid US state or territory.", 

86 "stateCode", 

87 ) 

88 ) 

89 if not occupancy: 

90 errors.append(AppError("ORG-00001", "Required field missing: Occupancy.", "occupancy")) 

91 if errors: 

92 return err(*errors) 

93 lic = validate_licensing(lo, state_code or "", occupancy or "") 

94 if not lic.ok: 

95 return lic 

96 return ok(True) 

97 

98 

99def validate_create_loan(inp: CreateLoanInput, lo: LoanOfficerInput) -> Result: 

100 """ORG-00001: require Subject Property State and Occupancy (+ borrower identity).""" 

101 errors: list[AppError] = [] 

102 if not inp.state_code: 

103 errors.append( 

104 AppError("ORG-00001", "Required field missing: Subject Property State.", "stateCode") 

105 ) 

106 elif inp.state_code.upper() not in VALID_STATE_CODES: 

107 errors.append( 

108 AppError( 

109 "ORG-00001", 

110 f"Unrecognized Subject Property State: '{inp.state_code}'. Must be a valid US state or territory.", 

111 "stateCode", 

112 ) 

113 ) 

114 if not inp.occupancy: 

115 errors.append(AppError("ORG-00001", "Required field missing: Occupancy.", "occupancy")) 

116 b = inp.borrower 

117 if not b or not b.first_name or not b.last_name: 

118 errors.append( 

119 AppError("VALIDATION", "Borrower first and last name are required.", "borrower") 

120 ) 

121 elif b.email: 

122 clean_email = b.email.strip() 

123 if not EMAIL_RE.match(clean_email): 

124 errors.append( 

125 AppError("VALIDATION", f"Invalid borrower email address: '{b.email}'.", "email") 

126 ) 

127 if errors: 

128 return err(*errors) 

129 lic = validate_licensing(lo, inp.state_code or "", inp.occupancy or "") 

130 if not lic.ok: 

131 return lic 

132 return ok(inp) 

133 

134 

135# ORG-00003 / ORG-00004: required fields before moving to a processing stage, by purpose. 

136REQUIRED_TO_PROCESSING: dict[str, list[str]] = { 

137 "Purchase": [ 

138 "county", "state_code", 

139 "transaction.purchase_price_amt", "transaction.base_loan_amt", 

140 ], 

141 "Refinance": [ 

142 "county", "state_code", 

143 "transaction.appraised_value_amt", "transaction.base_loan_amt", 

144 ], 

145} 

146PROCESSING_STAGES: frozenset[str] = frozenset( 

147 { 

148 "LOAN_SETUP", "DISCLOSURE_SENT", "UNDERWRITING_SUBMITTED", 

149 "APPROVED_WITH_CONDITION", "RE_SUBMITTAL", 

150 } 

151) 

152 

153 

154def _get_path(obj: Any, path: str) -> Any: 

155 cur = obj 

156 for key in path.split("."): 

157 if cur is None: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 return None 

159 cur = cur.get(key) if isinstance(cur, dict) else getattr(cur, key, None) 

160 return cur 

161 

162 

163def validate_stage_required_fields(loan: dict, target: str) -> Result: 

164 """Enforce required fields when a loan enters a processing stage.""" 

165 if target not in PROCESSING_STAGES: 

166 return ok(True) 

167 purpose = loan.get("purpose") 

168 code = "ORG-00003" if purpose == "Purchase" else "ORG-00004" 

169 required = REQUIRED_TO_PROCESSING.get(purpose or "", []) 

170 missing = [p for p in required if _get_path(loan, p) in (None, "")] 

171 if missing: 

172 joined = ", ".join(missing) 

173 return err( 

174 AppError( 

175 code=code, 

176 message=f"Required fields missing before processing ({purpose}): {joined}.", 

177 ) 

178 ) 

179 return ok(True) 

180 

181 

182def validate_borrower_delete(is_primary: bool) -> Result: 

183 """Primary borrower may not be deleted (spec §15.1).""" 

184 if is_primary: 

185 return err(AppError("VALIDATION", "The primary borrower cannot be deleted.")) 

186 return ok(True)