Coverage for apps/workflow/models.py: 87%

1"""Workflow, milestone trackers, key dates, tasks, and automation rules (spec §8.5, §13.2, §16.4).""" 

2 

3from __future__ import annotations 

4 

5import copy 

6from datetime import timedelta 

7from typing import TYPE_CHECKING, Any 

8 

9from django.conf import settings 

10from django.db import models 

11from django.utils import timezone 

12 

13from domain.enums import ( 

14 TrackerStatus, 

15 tracker_status_choices, 

16) 

17from domain.trackers import fresh_trackers 

18 

19if TYPE_CHECKING: 

20 from domain.rules import AutomationRuleDef 

21 

22 

23class LoanTracker(models.Model): 

24 """Operational / compliance milestone tracker associated with a Loan (spec §13.2, §16.4).""" 

25 

26 TRACKER_STATUS_CHOICES = tracker_status_choices() 

27 

28 loan = models.ForeignKey( 

29 "core_los.Loan", 

30 on_delete=models.CASCADE, 

31 related_name="trackers", 

32 ) 

33 key = models.CharField(max_length=50) 

34 label = models.CharField(max_length=100) 

35 category = models.CharField(max_length=50) 

36 status = models.CharField( 

37 max_length=20, 

38 choices=TRACKER_STATUS_CHOICES, 

39 default=TrackerStatus.NOT_STARTED.value, 

40 ) 

41 target_date = models.DateField(null=True, blank=True) 

42 completed_at = models.DateTimeField(null=True, blank=True) 

43 # ADR-002 tracker-mirror seam: read-only supplementary signal from Tireless, populated ONLY 

44 # by TirelessTrackerPort (for trid/title/hoi), never by the staff update endpoint, and it 

45 # never overrides `status`. Empty until the integration is connected. 

46 external_status = models.CharField(max_length=30, null=True, blank=True) 

47 external_updated_at = models.DateTimeField(null=True, blank=True) 

48 external_source = models.CharField(max_length=30, null=True, blank=True) 

49 created_at = models.DateTimeField(auto_now_add=True) 

50 updated_at = models.DateTimeField(auto_now=True) 

51 

52 class Meta: 

53 db_table = "workflow_trackers" 

54 indexes = [ 

55 models.Index(fields=["loan", "key"]), 

56 models.Index(fields=["loan", "status"]), 

57 ] 

58 unique_together = ("loan", "key") 

59 

60 def __str__(self) -> str: 

61 return f"{self.loan_id} - {self.key}: {self.status}" 

62 

63 def complete(self) -> None: 

64 """Advance status to Completed and stamp completion time.""" 

65 self.status = TrackerStatus.COMPLETED.value 

66 self.completed_at = timezone.now() 

67 self.save(update_fields=["status", "completed_at", "updated_at"]) 

68 

69 

70class KeyDates(models.Model): 

71 """TRID and operational statutory milestone dates (1:1 with Loan).""" 

72 

73 loan = models.OneToOneField( 

74 "core_los.Loan", 

75 on_delete=models.CASCADE, 

76 primary_key=True, 

77 related_name="key_dates", 

78 ) 

79 application_date = models.DateField(null=True, blank=True) 

80 le_disclosed_date = models.DateField(null=True, blank=True) 

81 le_signed_date = models.DateField(null=True, blank=True) 

82 intent_to_proceed_date = models.DateField(null=True, blank=True) 

83 appraisal_ordered_date = models.DateField(null=True, blank=True) 

84 appraisal_received_date = models.DateField(null=True, blank=True) 

85 lock_date = models.DateField(null=True, blank=True) 

86 lock_expiration_date = models.DateField(null=True, blank=True) 

87 submitted_to_uw_date = models.DateField(null=True, blank=True) 

88 approved_date = models.DateField(null=True, blank=True) 

89 clear_to_close_date = models.DateField(null=True, blank=True) 

90 cd_disclosed_date = models.DateField(null=True, blank=True) 

91 cd_signed_date = models.DateField(null=True, blank=True) 

92 closing_date = models.DateField(null=True, blank=True) 

93 funding_date = models.DateField(null=True, blank=True) 

94 first_payment_date = models.DateField(null=True, blank=True) 

95 created_at = models.DateTimeField(auto_now_add=True) 

96 updated_at = models.DateTimeField(auto_now=True) 

97 

98 class Meta: 

99 db_table = "workflow_key_dates" 

100 

101 def __str__(self) -> str: 

102 return f"KeyDates for Loan {self.loan_id}" 

103 

104 

105class LoanTask(models.Model): 

106 """Operational task assigned on a loan with checklist and auto-follow-up support.""" 

107 

108 loan = models.ForeignKey( 

109 "core_los.Loan", 

110 on_delete=models.CASCADE, 

111 related_name="tasks", 

112 ) 

113 title = models.CharField(max_length=200) 

114 description = models.TextField(blank=True) 

115 assignee = models.ForeignKey( 

116 settings.AUTH_USER_MODEL, 

117 on_delete=models.SET_NULL, 

118 null=True, 

119 blank=True, 

120 related_name="workflow_tasks", 

121 ) 

122 due_date = models.DateField(null=True, blank=True) 

123 completed = models.BooleanField(default=False) 

124 completed_at = models.DateTimeField(null=True, blank=True) 

125 checklists = models.JSONField(default=list, blank=True) 

126 follow_up_on_done = models.BooleanField(default=False) 

127 follow_up_offset_days = models.PositiveIntegerField(default=7) 

128 created_at = models.DateTimeField(auto_now_add=True) 

129 updated_at = models.DateTimeField(auto_now=True) 

130 

131 class Meta: 

132 db_table = "workflow_tasks" 

133 indexes = [ 

134 models.Index(fields=["loan", "completed"]), 

135 models.Index(fields=["assignee", "completed"]), 

136 ] 

137 

138 def __str__(self) -> str: 

139 status_label = "DONE" if self.completed else "PENDING" 

140 return f"[{status_label}] {self.title} (Loan {self.loan_id})" 

141 

142 def complete(self) -> LoanTask | None: 

143 """Mark task completed and spawn follow-up task if configured.""" 

144 self.completed = True 

145 if not self.completed_at: 145 ↛ 147line 145 didn't jump to line 147 because the condition on line 145 was always true

146 self.completed_at = timezone.now() 

147 self.save(update_fields=["completed", "completed_at", "updated_at"]) 

148 

149 if self.follow_up_on_done: 

150 base_date = self.completed_at.date() if self.completed_at else timezone.now().date() 

151 follow_up_due = base_date + timedelta(days=self.follow_up_offset_days) 

152 follow_up = LoanTask.objects.create( 

153 loan=self.loan, 

154 title=f"Follow-up: {self.title}", 

155 description=self.description, 

156 assignee=self.assignee, 

157 due_date=follow_up_due, 

158 checklists=[], 

159 follow_up_on_done=False, 

160 follow_up_offset_days=self.follow_up_offset_days, 

161 ) 

162 return follow_up 

163 return None 

164 

165 

166class TaskTemplate(models.Model): 

167 """Reusable task template with checklist and relative due date offset.""" 

168 

169 title = models.CharField(max_length=200) 

170 description = models.TextField(blank=True) 

171 due_offset_days = models.PositiveIntegerField(default=3) 

172 checklists = models.JSONField(default=list, blank=True) 

173 follow_up = models.JSONField(default=dict, blank=True) 

174 active = models.BooleanField(default=True) 

175 created_at = models.DateTimeField(auto_now_add=True) 

176 updated_at = models.DateTimeField(auto_now=True) 

177 

178 class Meta: 

179 db_table = "workflow_task_templates" 

180 

181 def __str__(self) -> str: 

182 return self.title 

183 

184 def apply(self, loan: Any, assignee: Any = None, base_date: Any = None) -> LoanTask: 

185 """Instantiate a new LoanTask from this template for the given loan.""" 

186 if base_date is None: 186 ↛ 188line 186 didn't jump to line 188 because the condition on line 186 was always true

187 base_date = timezone.now().date() 

188 due_date = base_date + timedelta(days=self.due_offset_days) if self.due_offset_days is not None else None 

189 

190 follow_up_cfg = self.follow_up or {} 

191 create_on_done = bool(follow_up_cfg.get("createOnDone") or follow_up_cfg.get("create_on_done", False)) 

192 offset_days = int(follow_up_cfg.get("dueOffsetDays") or follow_up_cfg.get("due_offset_days", 7)) 

193 

194 return LoanTask.objects.create( 

195 loan=loan, 

196 title=self.title, 

197 description=self.description, 

198 assignee=assignee, 

199 due_date=due_date, 

200 checklists=copy.deepcopy(self.checklists), 

201 follow_up_on_done=create_on_done, 

202 follow_up_offset_days=offset_days, 

203 ) 

204 

205 

206class AutomationRule(models.Model): 

207 """Configurable automation rule executed on loan lifecycle triggers (spec §8.5).""" 

208 

209 rule_id = models.CharField(max_length=20, unique=True) 

210 name = models.CharField(max_length=200) 

211 trigger = models.CharField(max_length=50) 

212 conditions = models.JSONField(default=list, blank=True) 

213 actions = models.JSONField(default=list, blank=True) 

214 is_active = models.BooleanField(default=True) 

215 created_at = models.DateTimeField(auto_now_add=True) 

216 updated_at = models.DateTimeField(auto_now=True) 

217 

218 class Meta: 

219 db_table = "workflow_automation_rules" 

220 

221 def __str__(self) -> str: 

222 return f"{self.rule_id}: {self.name} ({self.trigger})" 

223 

224 def to_domain(self) -> AutomationRuleDef: 

225 """Convert persistence model to pure domain AutomationRuleDef.""" 

226 from domain.rules import AutomationRuleDef, RuleAction, RuleCondition 

227 

228 conds = [] 

229 for c in self.conditions: 

230 conds.append( 

231 RuleCondition( 

232 kind=c.get("kind", ""), 

233 field_name=c.get("field", "") or c.get("field_name", ""), 

234 value=c.get("value"), 

235 values=tuple(c.get("values", ())), 

236 ) 

237 ) 

238 

239 acts = [] 

240 for a in self.actions: 

241 action_type = a.get("type", "") or a.get("action_type", "") or a.get("actionType", "") 

242 # Canonicalize actionType (e.g. "createTask" -> "CreateTask") 

243 if action_type.lower() == "createtask": 243 ↛ 245line 243 didn't jump to line 245 because the condition on line 243 was always true

244 action_type = "CreateTask" 

245 elif action_type.lower() == "requirefields": 

246 action_type = "RequireFields" 

247 elif action_type.lower() == "senddocument": 

248 action_type = "SendDocument" 

249 elif action_type.lower() == "stop": 

250 action_type = "Stop" 

251 

252 acts.append( 

253 RuleAction( 

254 action_type=action_type, 

255 message=a.get("message", ""), 

256 fields=tuple(a.get("fields", ())), 

257 document_classification_id=a.get("documentClassificationId", "") 

258 or a.get("document_classification_id", ""), 

259 task_title=a.get("taskTitle", "") or a.get("task_title", ""), 

260 ) 

261 ) 

262 

263 return AutomationRuleDef( 

264 id=self.rule_id, 

265 name=self.name, 

266 trigger=self.trigger, 

267 conditions=tuple(conds), 

268 actions=tuple(acts), 

269 active=self.is_active, 

270 ) 

271 

272 

273def seed_trackers(loan: Any) -> list[LoanTracker]: 

274 """Seed 16 initial milestone trackers for a loan.""" 

275 rows = fresh_trackers(loan.id) 

276 trackers = [ 

277 LoanTracker( 

278 loan=loan, 

279 key=row.key, 

280 label=row.label, 

281 category=row.category, 

282 status=row.status.value, 

283 ) 

284 for row in rows 

285 ] 

286 return LoanTracker.objects.bulk_create(trackers, ignore_conflicts=True)