Coverage for apps/borrower_portal/invite_api.py: 98%

1"""Staff bringing borrowers into the borrower portal — mounted under /api/loans 

2(add-borrower-portal-api spec borrower-invitations; design.md D11).""" 

3 

4from __future__ import annotations 

5 

6import secrets 

7import string 

8from uuid import UUID 

9 

10from django.conf import settings 

11from django.db import transaction 

12from django.utils import timezone 

13from ninja import Router 

14 

15from apps.accounts.auth import ensure_staff_scope, requires 

16from apps.application_1003.models import Borrower 

17from apps.borrower_portal.models import BorrowerAccount 

18from apps.borrower_portal.schemas import BorrowerPosStatusOut, PosInviteInput, PosInviteOut 

19from apps.borrower_portal.services import company_name, portal_url, send_borrower_message 

20from apps.communications import emails 

21from apps.communications.delivery import EmailSendError 

22from apps.communications.models import Communication 

23from apps.core_los.models import Loan 

24from domain.pos import EMAIL_MSG, MAX_TEXT, blank, is_email 

25from domain.result import AppError, DomainError 

26 

27router = Router(tags=["borrower-invitations"]) 

28 

29 

30def _scoped_loan(request, guid: UUID) -> Loan: 

31 loan = Loan.objects.filter(guid=guid).first() 

32 if loan is None: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 raise DomainError([AppError("NOT_FOUND", "Loan not found")]) 

34 ensure_staff_scope(request.principal, loan) 

35 return loan 

36 

37 

38def _temporary_password() -> str: 

39 """12 characters with at least one letter and one digit (meets the portal's password rule).""" 

40 alphabet = string.ascii_letters + string.digits 

41 while True: 

42 candidate = "".join(secrets.choice(alphabet) for _ in range(12)) 

43 if any(c.isalpha() for c in candidate) and any(c.isdigit() for c in candidate): 

44 return candidate 

45 

46 

47@router.post( 

48 "/{guid}/pos-invite", 

49 response={201: PosInviteOut}, 

50 auth=requires("LOAN_EDIT"), 

51 exclude_none=True, 

52) 

53def invite_borrower(request, guid: UUID, payload: PosInviteInput): 

54 loan = _scoped_loan(request, guid) 

55 name, email = payload.name.strip(), payload.email.strip().lower() 

56 if blank(name) or len(name) > MAX_TEXT: 

57 raise DomainError([AppError("VALIDATION", "Enter the borrower's name.", "name")]) 

58 if not is_email(email): 

59 raise DomainError([AppError("VALIDATION", EMAIL_MSG, "email")]) 

60 if BorrowerAccount.objects.filter(email=email).exists(): 

61 raise DomainError( 

62 [AppError("VALIDATION", "An account with that email already exists.", "email")] 

63 ) 

64 

65 temp_password = _temporary_password() 

66 with transaction.atomic(): 

67 account = BorrowerAccount(email=email, name=name, invited_at=timezone.now()) 

68 account.set_password(temp_password) 

69 account.save() 

70 if not loan.borrowers.filter(email__iexact=email).exists(): 

71 parts = name.split() 

72 Borrower.objects.create( 

73 loan=loan, 

74 is_primary=False, 

75 first_name=parts[0], 

76 last_name=" ".join(parts[1:]) or parts[0], 

77 email=email, 

78 ) 

79 lo = loan.loan_officer 

80 invited_by = f"{lo.first_name} {lo.last_name}".strip() if lo else "Your loan officer" 

81 try: 

82 # The account exists only if its email went out: a failed send rolls it back. 

83 subject, content = emails.borrower_invite( 

84 company=company_name(), 

85 name=name, 

86 invited_by=invited_by, 

87 loan_number=loan.arive_loan_no, 

88 login_url=portal_url("/login"), 

89 email=email, 

90 temp_password=temp_password, 

91 ) 

92 send_borrower_message( 

93 channel="email", 

94 to=email, 

95 subject=subject, 

96 content=content, 

97 trigger="pos:invite", 

98 loan=loan, 

99 raise_on_failure=True, 

100 ) 

101 except EmailSendError as exc: 

102 raise DomainError( 

103 [AppError("EMAIL_SEND_FAILED", "We couldn't send the invite email. Try again.")] 

104 ) from exc 

105 # The password is only in the email; development also returns it for local testing. 

106 result = {"user_id": account.id, "email_sent": True} 

107 if settings.DEBUG: 

108 result["devTempPassword"] = temp_password 

109 return 201, result 

110 

111 

112@router.get( 

113 "/{guid}/borrower-pos-status", 

114 response=list[BorrowerPosStatusOut], 

115 auth=requires("LOAN_VIEW"), 

116) 

117def borrower_pos_status(request, guid: UUID): 

118 loan = _scoped_loan(request, guid) 

119 borrowers = list(loan.borrowers.order_by("-is_primary", "pk")) 

120 emails = {b.email.strip().lower() for b in borrowers if b.email} 

121 accounts = {a.email: a for a in BorrowerAccount.objects.filter(email__in=emails)} 

122 # Latest invite email per address (newest first, so the first one seen wins). 

123 invite_status: dict[str, str] = {} 

124 for to, status in ( 

125 Communication.objects.filter(loan=loan, trigger="pos:invite") 

126 .order_by("-created_at", "-pk") 

127 .values_list("to", "status") 

128 ): 

129 invite_status.setdefault(to.strip().lower(), status) 

130 rows = [] 

131 for b in borrowers: 

132 account = accounts.get((b.email or "").strip().lower()) 

133 rows.append( 

134 { 

135 "borrower_id": b.id, 

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

137 "email": b.email, 

138 "invited": bool(account and account.invited_at), 

139 "started": bool(account and account.last_login_at), 

140 "has_account": account is not None, 

141 "invite_email_status": invite_status.get((b.email or "").strip().lower()), 

142 } 

143 ) 

144 return rows