Coverage for apps/borrower_portal/pos_api.py: 92%

1"""The signed-in borrower's own application — ``/api/pos/*`` (add-borrower-portal-api spec 

2borrower-application; design.md D6–D9). 

3 

4Shapes follow borrower-pos ``services/pos/pos.types.ts`` (``PosMy`` / ``PosApplication``). Every 

5endpoint acts on the caller's own loan (``borrower_loan``); staff tokens get 403 (``PortalAuth``). 

6""" 

7 

8from __future__ import annotations 

9 

10from datetime import date 

11from decimal import Decimal 

12 

13from django.db import transaction 

14from django.utils import timezone 

15from ninja import Router 

16 

17from apps.accounts.admin_api import _get_company, _pos_config_out 

18from apps.accounts.auth import borrower_loan, portal 

19from apps.accounts.models import User, UserRole 

20from apps.application_1003.models import ( 

21 Asset, 

22 Borrower, 

23 Declaration, 

24 Demographics, 

25 Employer, 

26 Income, 

27 Liability, 

28 RealEstate, 

29) 

30from apps.borrower_portal.models import PosApplication 

31from apps.borrower_portal.ports import get_address_lookup 

32from apps.borrower_portal.schemas import ( 

33 AddressSuggestionOut, 

34 DeleteStepItemInput, 

35 PortalConfigOut, 

36 PosMyOut, 

37 SaveStepInput, 

38 StartApplicationInput, 

39) 

40from apps.client_needs.services import generate_standard_needs 

41from apps.core_los.models import Loan, LoanStageHistory, LoanTransaction 

42from apps.integrations.encryption import decrypt_secret, encrypt_secret 

43from domain import pos as rules 

44from domain.enums import LoanStage 

45from domain.lifecycle import assert_transition 

46from domain.result import AppError, DomainError 

47 

48router = Router(tags=["borrower-pos"], auth=portal()) 

49geocode_router = Router(tags=["borrower-pos"], auth=portal()) 

50 

51NO_APPLICATION = "No application found." 

52 

53 

54# --- reading ---------------------------------------------------------------------- 

55def _num(value: Decimal | None): 

56 """Whole-dollar amounts come back as ints, others as floats (the portal sends numbers).""" 

57 if value is None: 

58 return None 

59 return int(value) if value == value.to_integral_value() else float(value) 

60 

61 

62def _amount(value: Decimal | None): 

63 """An amount the borrower gave; None for unset and for the $0 placeholders a new loan has.""" 

64 if value is None or value == 0: 

65 return None 

66 return _num(value) 

67 

68 

69def _primary(loan: Loan) -> Borrower | None: 

70 return ( 

71 loan.borrowers.filter(is_primary=True).order_by("pk").first() 

72 or loan.borrowers.order_by("pk").first() 

73 ) 

74 

75 

76def _state(loan: Loan) -> PosApplication: 

77 state, _ = PosApplication.objects.get_or_create(loan=loan) 

78 return state 

79 

80 

81def _transaction_out(loan: Loan) -> dict: 

82 tx = LoanTransaction.objects.filter(loan=loan).first() 

83 if tx is None: 

84 return {} 

85 out = { 

86 "purchase_price_amt": _amount(tx.purchase_price_amt), 

87 "appraised_value_amt": _amount(tx.appraised_value_amt), 

88 "base_loan_amt": _amount(tx.base_loan_amt), 

89 } 

90 return {k: v for k, v in out.items() if v is not None} 

91 

92 

93def _advisor(loan: Loan) -> dict: 

94 lo = loan.loan_officer 

95 if lo is None: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 return {"name": "", "title": "Loan Officer", "nmls_id": ""} 

97 return { 

98 "name": f"{lo.first_name} {lo.last_name}".strip(), 

99 "title": lo.title or "Loan Officer", 

100 "nmls_id": lo.nmls_id or "", 

101 } 

102 

103 

104def _me_borrower(request, loan: Loan) -> Borrower | None: 

105 """The Borrower row for the signed-in account on ``loan`` (matched by email), else the primary.""" 

106 email = getattr(getattr(request, "principal", None), "borrower_email", None) 

107 me = ( 

108 loan.borrowers.filter(email__iexact=email).order_by("-is_primary", "pk").first() 

109 if email 

110 else None 

111 ) 

112 return me or _primary(loan) 

113 

114 

115def _clean(row: dict) -> dict: 

116 return {k: v for k, v in row.items() if v is not None and v != ""} 

117 

118 

119def _borrower_block(loan: Loan, b: Borrower, *, editable: bool) -> dict: 

120 """One borrower's own data (pos.types.ts): personal, employment, income, declarations, 

121 demographics and credit consent, plus whether the signed-in borrower may edit it.""" 

122 ssn = decrypt_secret(b.ssn_encrypted) if b.ssn_encrypted else None 

123 demo = b.demographics_answers or {} 

124 block = { 

125 "id": b.pk, 

126 "is_primary": b.is_primary, 

127 "first_name": b.first_name, 

128 "last_name": b.last_name, 

129 "name": f"{b.first_name} {b.last_name}".strip(), 

130 "editable": editable, 

131 "declarations": dict(b.declaration_answers or {}), 

132 "demographics": { 

133 **({"ethnicity": demo["ethnicity"]} if demo.get("ethnicity") else {}), 

134 **({"sex": demo["sex"]} if demo.get("sex") else {}), 

135 "race": list(demo.get("race") or []), 

136 }, 

137 "credit_report_auth_ind": bool(b.credit_auth_ind), 

138 "employers": [ 

139 _clean( 

140 { 

141 "id": e.pk, 

142 "employer_name": e.employer_name, 

143 "position_title": e.position_title, 

144 "monthly_income_amt": _num(e.monthly_income_amt), 

145 "years_in_line_of_work": _num(e.years_in_line_of_work), 

146 "self_employed_ind": e.self_employed_ind, 

147 } 

148 ) 

149 for e in Employer.objects.filter(loan=loan, borrower=b).order_by("pk") 

150 ], 

151 "incomes": [ 

152 {"id": i.pk, "income_type": i.income_type, "monthly_amt": _num(i.monthly_amt)} 

153 for i in Income.objects.filter(loan=loan, borrower=b).order_by("pk") 

154 ], 

155 } 

156 if ssn: 

157 block["ssn"] = rules.mask_ssn(ssn) 

158 if b.birth_date: 

159 block["birth_date"] = b.birth_date.isoformat() 

160 if b.email: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true

161 block["email"] = b.email 

162 if b.cell_phone: 

163 block["cell_phone"] = b.cell_phone 

164 return block 

165 

166 

167def _snapshot(loan: Loan, me: Borrower | None = None) -> dict: 

168 """The ``PosApplication`` JSON for ``loan`` (pos.types.ts), keyed to the signed-in borrower 

169 ``me``: common loan data once, plus ``me`` and every borrower's own data in ``borrowers``.""" 

170 primary = _primary(loan) 

171 if me is None: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 me = primary 

173 can_edit_all = bool(me and me.is_primary) 

174 borrowers = [ 

175 _borrower_block(loan, b, editable=can_edit_all or (me is not None and b.pk == me.pk)) 

176 for b in loan.borrowers.order_by("-is_primary", "pk") 

177 ] 

178 me_block = next((bl for bl in borrowers if me and bl["id"] == me.pk), None) or {} 

179 co_borrowers = [ 

180 _clean( 

181 { 

182 "id": b["id"], 

183 "first_name": b["first_name"], 

184 "last_name": b["last_name"], 

185 "email": b.get("email"), 

186 "cell_phone": b.get("cell_phone"), 

187 } 

188 ) 

189 for b in borrowers 

190 if not (me and b["id"] == me.pk) 

191 ] 

192 assets = [ 

193 { 

194 "id": a.pk, 

195 "asset_type": a.asset_type, 

196 "institution_name": a.institution_name, 

197 "cash_or_market_value_amt": _num(a.cash_or_market_value_amt), 

198 } 

199 for a in Asset.objects.filter(loan=loan).order_by("pk") 

200 ] 

201 liabilities = [ 

202 { 

203 "id": li.pk, 

204 "liability_type": li.liability_type, 

205 "creditor_name": li.creditor_name, 

206 "monthly_payment_amt": _num(li.monthly_payment_amt), 

207 "unpaid_balance_amt": _num(li.unpaid_balance_amt), 

208 } 

209 for li in Liability.objects.filter(loan=loan).order_by("pk") 

210 ] 

211 real_estates = [ 

212 { 

213 "id": r.pk, 

214 "property_usage_type": r.property_usage_type or "PrimaryResidence", 

215 "disposition_status_type": r.disposition_status_type or "Retain", 

216 "property_value_amt": _num(r.property_value_amt), 

217 } 

218 for r in RealEstate.objects.filter(loan=loan, subject_property_ind=False).order_by("pk") 

219 ] 

220 return { 

221 "loan": { 

222 "loan_no": loan.arive_loan_no, 

223 "state_code": loan.state_code, 

224 "purpose": loan.purpose, 

225 "occupancy": loan.occupancy, 

226 "stage": loan.stage, 

227 }, 

228 "advisor": _advisor(loan), 

229 "transaction": _transaction_out(loan), 

230 "me": ( 

231 {"id": me.pk, "is_primary": bool(me.is_primary), "can_edit_all": can_edit_all} 

232 if me 

233 else None 

234 ), 

235 "borrowers": borrowers, 

236 # Compatibility aliases — the signed-in borrower's own data. 

237 "borrower": me_block, 

238 "co_borrowers": co_borrowers, 

239 "employers": me_block.get("employers", []), 

240 "incomes": me_block.get("incomes", []), 

241 "assets": assets, 

242 "liabilities": liabilities, 

243 "real_estates": real_estates, 

244 } 

245 

246 

247def _pos_my(request, loan: Loan | None) -> dict: 

248 me = _me_borrower(request, loan) if loan is not None else None 

249 application = _snapshot(loan, me) if loan is not None else None 

250 steps = rules.steps_of(application) 

251 return { 

252 "application": application, 

253 "steps": steps, 

254 "percent_complete": rules.percent_complete_of(application) if application else 0, 

255 } 

256 

257 

258def _require_loan(request) -> Loan: 

259 loan = borrower_loan(request) 

260 if loan is None: 

261 raise DomainError([AppError("NOT_FOUND", NO_APPLICATION)]) 

262 return loan 

263 

264 

265def _fail(errors: dict[str, str]) -> None: 

266 problem = rules.first_error(errors) 

267 if problem: 

268 raise DomainError([problem]) 

269 

270 

271@router.get("/my", response=PosMyOut) 

272def pos_my(request): 

273 return _pos_my(request, borrower_loan(request)) 

274 

275 

276# --- starting ----------------------------------------------------------------------- 

277def _pick_loan_officer(state_code: str | None = None) -> User | None: 

278 """``pos_config.default_loan_officer_id`` when set and active, else the first active user 

279 holding "Loan Officer" at a headquarters org-unit (design.md D8). 

280 When ``state_code`` is provided, prioritizes active loan officers who hold an active and 

281 sponsored license in that state so origination licensing checks pass.""" 

282 clean_state = (state_code or "").strip().upper() 

283 from apps.accounts.models import License 

284 

285 if clean_state: 285 ↛ 328line 285 didn't jump to line 328 because the condition on line 285 was always true

286 config = _get_company().pos_config or {} 

287 configured = config.get("default_loan_officer_id") 

288 if configured: 

289 user = User.objects.filter(pk=configured, is_active=True).first() 

290 if user is not None: 290 ↛ 297line 290 didn't jump to line 297 because the condition on line 290 was always true

291 has_lic = License.objects.filter( 

292 user=user, state_code=clean_state, active=True, sponsored=True 

293 ).exists() 

294 if has_lic: 294 ↛ 297line 294 didn't jump to line 297 because the condition on line 294 was always true

295 return user 

296 

297 licensed_lo = ( 

298 UserRole.objects.filter( 

299 role__name="Loan Officer", 

300 org_unit__is_headquarters=True, 

301 user__is_active=True, 

302 user__licenses__state_code=clean_state, 

303 user__licenses__active=True, 

304 user__licenses__sponsored=True, 

305 ) 

306 .select_related("user") 

307 .order_by("user_id") 

308 .first() 

309 ) 

310 if licensed_lo is not None: 

311 return licensed_lo.user 

312 

313 licensed_lo_any = ( 

314 UserRole.objects.filter( 

315 role__name="Loan Officer", 

316 user__is_active=True, 

317 user__licenses__state_code=clean_state, 

318 user__licenses__active=True, 

319 user__licenses__sponsored=True, 

320 ) 

321 .select_related("user") 

322 .order_by("user_id") 

323 .first() 

324 ) 

325 if licensed_lo_any is not None: 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true

326 return licensed_lo_any.user 

327 

328 config = _get_company().pos_config or {} 

329 configured = config.get("default_loan_officer_id") 

330 if configured: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true

331 user = User.objects.filter(pk=configured, is_active=True).first() 

332 if user is not None: 

333 return user 

334 binding = ( 

335 UserRole.objects.filter( 

336 role__name="Loan Officer", org_unit__is_headquarters=True, user__is_active=True 

337 ) 

338 .select_related("user") 

339 .order_by("user_id") 

340 .first() 

341 ) 

342 return binding.user if binding else None 

343 

344 

345@router.post("/applications", response={201: PosMyOut}) 

346def start_application(request, payload: StartApplicationInput): 

347 existing = borrower_loan(request) 

348 if existing is not None: 

349 return 201, _pos_my(request, existing) 

350 

351 data = payload.model_dump() 

352 _fail(rules.validate_start(data)) 

353 officer = _pick_loan_officer(data.get("state_code")) 

354 if officer is None: 

355 raise DomainError( 

356 [AppError("VALIDATION", "No loan officer is available to take your application.")] 

357 ) 

358 # Imported here: core_los.api pulls in the whole loan stack. 

359 from apps.core_los.api import _build_loan 

360 

361 account = request.borrower_account 

362 parts = account.name.split() 

363 first_name = parts[0] if parts else account.email 

364 last_name = " ".join(parts[1:]) or first_name 

365 with transaction.atomic(): 

366 # The shared origination path (licensing rules ORG-00001/00002 included). 

367 loan = _build_loan( 

368 loan_officer_id=officer.pk, 

369 state_code=data["state_code"].strip().upper(), 

370 occupancy=data["occupancy"], 

371 purpose=data["purpose"], 

372 first_name=first_name, 

373 last_name=last_name, 

374 email=account.email, 

375 ) 

376 PosApplication.objects.create(loan=loan, started_by=account) 

377 if account.phone: 377 ↛ 379line 377 didn't jump to line 379

378 loan.borrowers.filter(is_primary=True).update(cell_phone=account.phone) 

379 return 201, _pos_my(request, loan) 

380 

381 

382# --- saving a section --------------------------------------------------------------- 

383PER_BORROWER_SECTIONS = frozenset( 

384 {"personal", "employment", "income", "declarations", "additional", "demographics", "credit"} 

385) 

386 

387 

388def _may_edit(me: Borrower | None, borrower: Borrower | None) -> bool: 

389 """The primary may edit any borrower; anyone else only themselves.""" 

390 return bool(me and borrower and (me.is_primary or me.pk == borrower.pk)) 

391 

392 

393def _resolve_target( 

394 loan: Loan, me: Borrower | None, section: str, borrower_id: int | None 

395) -> Borrower | None: 

396 """The borrower a save targets. Common sections act on the loan, so they use the editor 

397 (``me``) only for joint ownership; per-borrower sections may name another borrower, subject 

398 to ``_may_edit``.""" 

399 if section not in PER_BORROWER_SECTIONS or borrower_id is None: 

400 return me 

401 target = loan.borrowers.filter(pk=borrower_id).first() 

402 if target is None: 

403 raise DomainError([AppError("NOT_FOUND", "Borrower not found on this loan.")]) 

404 if not _may_edit(me, target): 

405 raise DomainError([AppError("FORBIDDEN", "You cannot edit another borrower's details.")]) 

406 return target 

407 

408 

409def _current_for_validation(loan: Loan, target: Borrower | None = None) -> dict: 

410 borrower = target or _primary(loan) 

411 birth = borrower.birth_date if borrower else None 

412 return { 

413 "transaction": _transaction_out(loan), 

414 "borrower": { 

415 "has_ssn": bool(borrower and borrower.ssn_encrypted), 

416 "birth_date": birth.isoformat() if birth else None, 

417 }, 

418 } 

419 

420 

421def _dec(value) -> Decimal | None: 

422 return None if value is None else Decimal(str(value)) 

423 

424 

425def _text(data: dict, key: str) -> str: 

426 return (data.get(key) or "").strip() 

427 

428 

429_ETHNICITY_TO_HMDA = {"Hispanic": "HispanicOrLatino", "NotHispanic": "NotHispanicOrLatino"} 

430 

431 

432def _write_declarations(primary: Borrower, answers: dict) -> None: 

433 declaration, _ = Declaration.objects.get_or_create(borrower=primary) 

434 for letter, value in answers.items(): 

435 field = rules.DECLARATION_FIELDS.get(letter) 

436 if field is None: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true

437 continue 

438 if field == "intent_to_occupy_type": 

439 setattr(declaration, field, "Yes" if value else "No") 

440 else: 

441 setattr(declaration, field, bool(value)) 

442 declaration.save() 

443 

444 

445def _write_demographics(primary: Borrower, data: dict) -> None: 

446 demo, _ = Demographics.objects.get_or_create(borrower=primary) 

447 ethnicity, sex, race = data.get("ethnicity"), data.get("sex"), list(data.get("race") or []) 

448 demo.ethnicity = [_ETHNICITY_TO_HMDA[ethnicity]] if ethnicity in _ETHNICITY_TO_HMDA else [] 

449 demo.ethnicity_refusal_ind = ethnicity == "NoProvide" 

450 demo.sex = sex if sex in ("Male", "Female") else "" 

451 demo.sex_refusal_ind = sex == "NoProvide" 

452 demo.race = [r for r in race if r != "NoProvide"] 

453 demo.race_refusal_ind = "NoProvide" in race 

454 demo.save() 

455 

456 

457def _apply_step( 

458 loan: Loan, section: str, data: dict, target: Borrower | None, item_id: int | None = None 

459) -> None: 

460 primary = _primary(loan) 

461 if primary is None or target is None: 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true

462 raise DomainError([AppError("NOT_FOUND", NO_APPLICATION)]) 

463 state = _state(loan) 

464 target_is_primary = bool(target.is_primary) 

465 

466 if section == "mortgage": 

467 tx, _ = LoanTransaction.objects.get_or_create( 

468 loan=loan, 

469 defaults={"base_loan_amt": Decimal("0.00"), "note_amt": Decimal("0.00")}, 

470 ) 

471 for key in ("purchase_price_amt", "appraised_value_amt"): 

472 if key in data and data[key] is not None: 

473 setattr(tx, key, _dec(data[key])) 

474 if data.get("base_loan_amt") is not None: 474 ↛ 477line 474 didn't jump to line 477 because the condition on line 474 was always true

475 tx.base_loan_amt = _dec(data["base_loan_amt"]) 

476 tx.note_amt = tx.base_loan_amt 

477 tx.save() 

478 

479 elif section == "personal": 

480 ssn = data.get("ssn") 

481 if ssn and not rules.is_masked_ssn(ssn): 

482 target.ssn_encrypted = encrypt_secret(rules.digits(ssn)) 

483 if data.get("birth_date"): 483 ↛ 485line 483 didn't jump to line 485 because the condition on line 483 was always true

484 target.birth_date = date.fromisoformat(data["birth_date"]) 

485 target.save() 

486 

487 elif section == "coborrower": 

488 Borrower.objects.create( 

489 loan=loan, 

490 is_primary=False, 

491 first_name=_text(data, "first_name"), 

492 last_name=_text(data, "last_name"), 

493 email=_text(data, "email").lower(), 

494 cell_phone=_text(data, "cell_phone"), 

495 ) 

496 

497 elif section == "employment": 

498 fields = dict( 

499 employer_name=_text(data, "employer_name"), 

500 position_title=_text(data, "position_title"), 

501 monthly_income_amt=_dec(data.get("monthly_income_amt")), 

502 years_in_line_of_work=_dec(data.get("years_in_line_of_work")), 

503 self_employed_ind=bool(data.get("self_employed_ind")), 

504 ) 

505 if item_id: 

506 Employer.objects.filter(loan=loan, borrower=target, pk=item_id).update(**fields) 

507 else: 

508 Employer.objects.create(loan=loan, borrower=target, **fields) 

509 

510 elif section == "income": 

511 # A borrower's employer income auto-syncs a canonical Base income 

512 # (sync_employer_income); reconcile with it here rather than creating a 

513 # duplicate Base record that would double-count the borrower's income. 

514 amount = _dec(data["monthly_amt"]) 

515 if item_id: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true

516 Income.objects.filter(loan=loan, borrower=target, pk=item_id).update(monthly_amt=amount) 

517 else: 

518 existing = Income.objects.filter( 

519 loan=loan, borrower=target, income_type="Base" 

520 ).first() 

521 if existing: 

522 if existing.monthly_amt != amount: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true

523 existing.monthly_amt = amount 

524 existing.save(update_fields=["monthly_amt"]) 

525 else: 

526 Income.objects.create( 

527 loan=loan, borrower=target, income_type="Base", monthly_amt=amount 

528 ) 

529 

530 elif section == "assets": 

531 fields = dict( 

532 institution_name=_text(data, "institution_name"), 

533 cash_or_market_value_amt=_dec(data["cash_or_market_value_amt"]), 

534 ) 

535 if item_id: 

536 Asset.objects.filter(loan=loan, pk=item_id).update(**fields) 

537 else: 

538 asset = Asset.objects.create(loan=loan, asset_type="Checking", **fields) 

539 asset.borrowers.add(target) 

540 

541 elif section == "liabilities": 

542 fields = dict( 

543 creditor_name=_text(data, "creditor_name"), 

544 monthly_payment_amt=_dec(data["monthly_payment_amt"]), 

545 unpaid_balance_amt=_dec(data["unpaid_balance_amt"]), 

546 ) 

547 if item_id: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true

548 Liability.objects.filter(loan=loan, pk=item_id).update(**fields) 

549 else: 

550 liability = Liability.objects.create(loan=loan, liability_type="Installment", **fields) 

551 liability.borrowers.add(target) 

552 

553 elif section == "reo": 

554 fields = dict( 

555 property_usage_type=data.get("property_usage_type") or "PrimaryResidence", 

556 disposition_status_type=data.get("disposition_status_type") or "Retain", 

557 property_value_amt=_dec(data["property_value_amt"]), 

558 ) 

559 if item_id: 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true

560 RealEstate.objects.filter(loan=loan, pk=item_id, subject_property_ind=False).update( 

561 **fields 

562 ) 

563 else: 

564 reo = RealEstate.objects.create(loan=loan, **fields) 

565 reo.borrowers.add(target) 

566 

567 elif section == "declarations": 

568 answers = {k: bool(data["declarations"][k]) for k in rules.DECLARATION_KEYS} 

569 target.declaration_answers = answers 

570 target.save(update_fields=["declaration_answers"]) 

571 if target_is_primary: # keep the loan-level cache as a read fallback 

572 state.declaration_answers = answers 

573 state.save(update_fields=["declaration_answers", "updated_at"]) 

574 _write_declarations(target, answers) 

575 

576 elif section == "additional": 

577 merged = {**(target.declaration_answers or {})} 

578 merged.update({k: bool(data["answers"][k]) for k in rules.MORE_QUESTION_KEYS}) 

579 target.declaration_answers = merged 

580 target.save(update_fields=["declaration_answers"]) 

581 if target_is_primary: 

582 state.declaration_answers = merged 

583 state.save(update_fields=["declaration_answers", "updated_at"]) 

584 _write_declarations(target, merged) 

585 

586 elif section == "demographics": 

587 cleaned = { 

588 k: v for k, v in (("ethnicity", data.get("ethnicity")), ("sex", data.get("sex"))) if v 

589 } 

590 cleaned["race"] = list(data.get("race") or []) 

591 target.demographics_answers = cleaned 

592 target.save(update_fields=["demographics_answers"]) 

593 if target_is_primary: 

594 state.demographics = cleaned 

595 state.save(update_fields=["demographics", "updated_at"]) 

596 _write_demographics(target, cleaned) 

597 

598 elif section == "credit": 598 ↛ exitline 598 didn't return from function '_apply_step' because the condition on line 598 was always true

599 consent = bool(data["consent"]) 

600 target.credit_auth_ind = consent 

601 target.save(update_fields=["credit_auth_ind"]) 

602 if target_is_primary: 

603 state.e_consent_at = timezone.now() if consent else None 

604 state.save(update_fields=["e_consent_at", "updated_at"]) 

605 

606 

607@router.patch("/step", response=PosMyOut) 

608def save_step(request, payload: SaveStepInput): 

609 loan = _require_loan(request) 

610 me = _me_borrower(request, loan) 

611 target = _resolve_target(loan, me, payload.section, payload.borrower_id) 

612 _fail( 

613 rules.validate_step( 

614 payload.section, 

615 payload.data, 

616 _current_for_validation(loan, target), 

617 timezone.localdate(), 

618 ) 

619 ) 

620 with transaction.atomic(): 

621 _apply_step(loan, payload.section, payload.data, target, payload.item_id) 

622 return _pos_my(request, Loan.objects.get(pk=loan.pk)) 

623 

624 

625def _authorize_item(me: Borrower | None, item) -> None: 

626 """A per-borrower record (employer/income) may be removed only by someone who may edit 

627 the borrower it belongs to.""" 

628 if item is not None and not _may_edit(me, item.borrower): 

629 raise DomainError([AppError("FORBIDDEN", "You cannot edit another borrower's details.")]) 

630 

631 

632@router.delete("/step", response=PosMyOut) 

633def delete_step_item(request, payload: DeleteStepItemInput): 

634 loan = _require_loan(request) 

635 me = _me_borrower(request, loan) 

636 section = payload.section.lower().strip() 

637 try: 

638 item_pk = int(payload.item_id) 

639 except (ValueError, TypeError): 

640 raise DomainError([AppError("VALIDATION", "Invalid item ID.")]) from None 

641 

642 with transaction.atomic(): 

643 if section in ("income", "incomes"): 

644 item = Income.objects.filter(loan=loan, pk=item_pk).first() 

645 _authorize_item(me, item) 

646 if item: 

647 item.delete() 

648 elif section in ("employment", "employers", "employer"): 

649 item = Employer.objects.filter(loan=loan, pk=item_pk).first() 

650 _authorize_item(me, item) 

651 if item: 651 ↛ 665line 651 didn't jump to line 665

652 item.delete() 

653 elif section in ("assets", "asset"): 

654 Asset.objects.filter(loan=loan, pk=item_pk).delete() 

655 elif section in ("liabilities", "debts", "liability"): 

656 Liability.objects.filter(loan=loan, pk=item_pk).delete() 

657 elif section in ("reo", "real_estates", "property"): 

658 RealEstate.objects.filter(loan=loan, pk=item_pk, subject_property_ind=False).delete() 

659 elif section in ("coborrower", "co_borrowers"): 

660 if me and me.is_primary and item_pk != me.pk: 660 ↛ 665line 660 didn't jump to line 665

661 Borrower.objects.filter(loan=loan, pk=item_pk, is_primary=False).delete() 

662 else: 

663 raise DomainError([AppError("VALIDATION", f"Cannot delete from section '{section}'.")]) 

664 

665 return _pos_my(request, Loan.objects.get(pk=loan.pk)) 

666 

667 

668# --- submitting --------------------------------------------------------------------- 

669@router.post("/submit", response=PosMyOut) 

670def submit(request): 

671 loan = _require_loan(request) 

672 if loan.stage != LoanStage.APPLICATION_INTAKE.value: 

673 return _pos_my(request, loan) # already submitted: nothing changes 

674 

675 features = _pos_config_out(_get_company())["features"] 

676 primary = _primary(loan) 

677 if not (primary and primary.credit_auth_ind) and not features[ 

678 "allow_submit_without_credit_consent" 

679 ]: 

680 raise DomainError( 

681 [ 

682 AppError( 

683 "POS_CREDIT_CONSENT", 

684 "Credit authorization is required to submit your application.", 

685 ) 

686 ] 

687 ) 

688 transition = assert_transition( 

689 LoanStage.APPLICATION_INTAKE, LoanStage.PREAPPROVED, frozenset() 

690 ) 

691 if not transition.ok: 691 ↛ 692line 691 didn't jump to line 692 because the condition on line 691 was never true

692 raise DomainError(transition.errors) 

693 

694 with transaction.atomic(): 

695 updated = Loan.objects.filter(