Coverage for apps/disclosures/api.py: 92%
117 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Disclosures router: order-gated audit, generate, send, and per-recipient signing
2(under /loans)."""
4from uuid import UUID
6from django.utils import timezone
7from ninja import Router
9from apps.accounts.auth import ensure_staff_scope, requires
10from apps.application_1003.models import Borrower
11from apps.core_los.models import Loan
12from apps.disclosures.models import Disclosure, DisclosureRecipient
13from apps.disclosures.providers import get_docmagic_provider
14from apps.disclosures.schemas import (
15 AuditResultOut,
16 ChangeOfCircumstanceIn,
17 ChangeOfCircumstanceOut,
18 DisclosureOut,
19 PackageTypeIn,
20 SendIn,
21 SignIn,
22)
23from apps.fees_trid.models import Fee
24from apps.fees_trid.services import replace_fees
25from apps.workflow.models import KeyDates
26from domain.disclosures import can_generate_package
27from domain.enums import LoanStage, ToleranceBucket
28from domain.fees import FeeInput, compute_tolerance_cure
29from domain.lifecycle import assert_transition
30from domain.result import AppError, DomainError
32router = Router(tags=["disclosures"])
34_BUCKET_ORDER = (ToleranceBucket.ZERO, ToleranceBucket.TEN_PERCENT, ToleranceBucket.UNLIMITED)
36_DISCLOSED_DATE_FIELD = {
37 "LoanEstimate": "le_disclosed_date",
38 "ClosingDisclosure": "cd_disclosed_date",
39}
40_SIGNED_DATE_FIELD = {
41 "LoanEstimate": "le_signed_date",
42 "ClosingDisclosure": "cd_signed_date",
43}
46def _scoped_loan(request, guid: UUID) -> Loan:
47 loan = Loan.objects.filter(guid=guid).first()
48 if loan is None: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true
49 raise DomainError([AppError("NOT_FOUND", "Loan not found")])
50 ensure_staff_scope(request.principal, loan)
51 return loan
54def _scoped_disclosure(loan: Loan, disclosure_id: int) -> Disclosure:
55 disclosure = Disclosure.objects.filter(pk=disclosure_id, loan=loan).first()
56 if disclosure is None: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 raise DomainError([AppError("NOT_FOUND", "Disclosure not found on this loan")])
58 return disclosure
61def _check_order(loan: Loan, package_type: str) -> None:
62 result = can_generate_package(loan.stage, list(loan.disclosures.all()), package_type)
63 if not result.ok:
64 raise DomainError(result.errors)
67@router.get("/{guid}/disclosures", response=list[DisclosureOut], auth=requires("LOAN_VIEW"))
68def list_disclosures(request, guid: UUID):
69 loan = _scoped_loan(request, guid)
70 return list(loan.disclosures.all())
73@router.post("/{guid}/disclosures/audit", response=AuditResultOut, auth=requires("LOAN_VIEW"))
74def audit_disclosure(request, guid: UUID, payload: PackageTypeIn):
75 loan = _scoped_loan(request, guid)
76 _check_order(loan, payload.package_type)
77 provider = get_docmagic_provider(org=loan.org_unit)
78 findings = provider.audit(loan, payload.package_type)
79 return {
80 "findings": [
81 {"code": f.code, "severity": f.severity, "message": f.message} for f in findings
82 ]
83 }
86@router.post(
87 "/{guid}/disclosures/generate", response={201: DisclosureOut}, auth=requires("LOAN_EDIT")
88)
89def generate_disclosure(request, guid: UUID, payload: PackageTypeIn):
90 loan = _scoped_loan(request, guid)
91 _check_order(loan, payload.package_type)
92 provider = get_docmagic_provider(org=loan.org_unit)
93 findings = provider.audit(loan, payload.package_type)
94 fatal = [f for f in findings if f.severity == "fatal"]
95 if fatal:
96 raise DomainError([AppError("DISC-AUDIT", f.message) for f in fatal])
97 provider.generate(loan, payload.package_type)
98 disclosure = Disclosure.objects.create(
99 loan=loan,
100 package_type=payload.package_type,
101 status="Generated",
102 audit_findings=[{"code": f.code, "severity": f.severity, "message": f.message}
103 for f in findings],
104 generated_at=timezone.now(),
105 )
106 return 201, disclosure
109@router.post(
110 "/{guid}/disclosures/{disclosure_id}/send", response=DisclosureOut, auth=requires("LOAN_EDIT")
111)
112def send_disclosure(request, guid: UUID, disclosure_id: int, payload: SendIn):
113 loan = _scoped_loan(request, guid)
114 disclosure = _scoped_disclosure(loan, disclosure_id)
115 for r in payload.recipients:
116 borrower = Borrower.objects.filter(pk=r.borrower_id, loan=loan).first()
117 if borrower is None: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 raise DomainError([AppError("NOT_FOUND", "Borrower not found on this loan")])
119 DisclosureRecipient.objects.create(
120 disclosure=disclosure, borrower=borrower, email=r.email
121 )
122 disclosure.status = "SentForEsign"
123 disclosure.sent_at = timezone.now()
124 disclosure.save(update_fields=["status", "sent_at"])
126 disclosed_field = _DISCLOSED_DATE_FIELD.get(disclosure.package_type)
127 if disclosed_field: 127 ↛ 132line 127 didn't jump to line 132 because the condition on line 127 was always true
128 key_dates, _ = KeyDates.objects.get_or_create(loan=loan)
129 setattr(key_dates, disclosed_field, timezone.now().date())
130 key_dates.save(update_fields=[disclosed_field, "updated_at"])
132 if disclosure.package_type == "LoanEstimate" and loan.stage == LoanStage.LOAN_SETUP.value:
133 transition = assert_transition(LoanStage.LOAN_SETUP, LoanStage.DISCLOSURE_SENT)
134 if transition.ok: 134 ↛ 138line 134 didn't jump to line 138 because the condition on line 134 was always true
135 loan.stage = LoanStage.DISCLOSURE_SENT.value
136 loan.save(update_fields=["stage", "updated_at"])
138 return disclosure
141@router.patch(
142 "/{guid}/disclosures/{disclosure_id}/recipients/{recipient_id}",
143 response=DisclosureOut,
144 auth=requires("LOAN_EDIT"),
145)
146def sign_recipient(request, guid: UUID, disclosure_id: int, recipient_id: int, payload: SignIn):
147 loan = _scoped_loan(request, guid)
148 disclosure = _scoped_disclosure(loan, disclosure_id)
149 recipient = disclosure.recipients.filter(pk=recipient_id).first()
150 if recipient is None: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 raise DomainError([AppError("NOT_FOUND", "Recipient not found on this disclosure")])
153 recipient.signed_at = timezone.now() if payload.signed else None
154 recipient.save(update_fields=["signed_at"])
156 if disclosure.recipients.filter(signed_at__isnull=True).exists():
157 return disclosure
159 disclosure.status = "Completed"
160 disclosure.completed_at = timezone.now()
161 disclosure.save(update_fields=["status", "completed_at"])
163 signed_field = _SIGNED_DATE_FIELD.get(disclosure.package_type)
164 if signed_field: 164 ↛ 169line 164 didn't jump to line 169 because the condition on line 164 was always true
165 key_dates, _ = KeyDates.objects.get_or_create(loan=loan)
166 setattr(key_dates, signed_field, timezone.now().date())
167 key_dates.save(update_fields=[signed_field, "updated_at"])
169 return disclosure
172@router.post(
173 "/{guid}/disclosures/change-of-circumstance",
174 response={201: ChangeOfCircumstanceOut},
175 auth=requires("LOAN_EDIT"),
176)
177def change_of_circumstance(request, guid: UUID, payload: ChangeOfCircumstanceIn):
178 loan = _scoped_loan(request, guid)
179 if not payload.reason.strip():
180 raise DomainError([AppError("VALIDATION", "A Change of Circumstance reason is required.")])
182 # The loan's current fees ARE the disclosed baseline — set_fees is gated, so
183 # nothing else could have changed them since the last disclosure (design.md D3).
184 baseline = list(loan.fees.all())
185 revised_fees = replace_fees(loan, payload.fees)
186 revised = [
187 FeeInput(
188 section=f.section, total_amt=f.total_amt, tolerance_bucket=f.tolerance_bucket,
189 poc_ind=f.poc_ind, name=f.name, paid_by=f.paid_by, paid_to=f.paid_to,
190 )
191 for f in revised_fees
192 ]
193 cure = compute_tolerance_cure(baseline, revised)
195 lender_credit = None
196 if cure.total_cure > 0:
197 lender_credit = Fee.objects.create(
198 loan=loan, section="A", name="Lender Credit (COC tolerance cure)",
199 total_amt=-cure.total_cure, tolerance_bucket="ZERO",
200 paid_by="Lender", paid_to="Lender", poc_ind=False,
201 )
203 discovered_date = payload.discovered_date or timezone.now().date()
204 disclosure = Disclosure.objects.create(
205 loan=loan,
206 package_type="ChangeOfCircumstance",
207 status="Generated",
208 coc_reason=f"{payload.reason} (discovered {discovered_date})",
209 coc_discovered_date=discovered_date,
210 generated_at=timezone.now(),
211 )
213 return 201, {
214 "disclosure": disclosure,
215 "tolerance_cure": {
216 "by_bucket": [
217 {
218 "bucket": b.value,
219 "baseline": cure.by_bucket[b].baseline,
220 "revised": cure.by_bucket[b].revised,
221 "variance": cure.by_bucket[b].variance,
222 "cure": cure.by_bucket[b].cure,
223 }
224 for b in _BUCKET_ORDER
225 ],
226 "total_cure": cure.total_cure,
227 },
228 "lender_credit": lender_credit,
229 }