Coverage for apps/fees_trid/api.py: 90%

1"""Fees router: fee set/list, Loan Estimate bucket summary, Change-of-Circumstance tolerance 

2cure, and the QM points-and-fees / APR-APOR test (mounted under /api/loans). 

3 

4Thin ORM<->domain adapter over ``domain/fees.py``. Follows the credit/AUS/pricing shape: 

5``_scoped_loan`` + ``ensure_staff_scope`` + ``requires(...)`` guards. 

6""" 

7 

8from __future__ import annotations 

9 

10from decimal import Decimal 

11from uuid import UUID 

12 

13from ninja import Router 

14 

15from apps.accounts.auth import ensure_staff_scope, requires 

16from apps.core_los.models import Loan, LoanTransaction 

17from apps.disclosures.services import loan_is_disclosed 

18from apps.fees_trid.models import Fee 

19from apps.fees_trid.schemas import ( 

20 FeeIn, 

21 FeeOut, 

22 FeeSetIn, 

23 FeeSummaryOut, 

24 QmResultOut, 

25 ToleranceOut, 

26) 

27from apps.fees_trid.services import replace_fees 

28from domain.enums import PaidByType, ToleranceBucket 

29from domain.fees import ( 

30 FeeInput, 

31 bucket_of, 

32 compute_tolerance_cure, 

33 qm_points_and_fees_test, 

34 round2, 

35) 

36from domain.result import AppError, DomainError 

37 

38router = Router(tags=["fees"]) 

39 

40ZERO = Decimal("0") 

41_BUCKET_ORDER = (ToleranceBucket.ZERO, ToleranceBucket.TEN_PERCENT, ToleranceBucket.UNLIMITED) 

42 

43 

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

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

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

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

48 ensure_staff_scope(request.principal, loan) 

49 return loan 

50 

51 

52def _txn(loan: Loan) -> LoanTransaction | None: 

53 try: 

54 return loan.transaction 

55 except LoanTransaction.DoesNotExist: 

56 return None 

57 

58 

59@router.get("/{guid}/fees", response=list[FeeOut], auth=requires("LOAN_VIEW")) 

60def list_fees(request, guid: UUID): 

61 loan = _scoped_loan(request, guid) 

62 return list(loan.fees.all()) 

63 

64 

65@router.put("/{guid}/fees", response=list[FeeOut], auth=requires("LOAN_EDIT")) 

66def set_fees(request, guid: UUID, payload: FeeSetIn): 

67 loan = _scoped_loan(request, guid) 

68 if loan_is_disclosed(loan): 

69 raise DomainError( 

70 [AppError( 

71 "COC_REQUIRED", 

72 "This loan is disclosed — fee changes require a Change of Circumstance.", 

73 )] 

74 ) 

75 return replace_fees(loan, payload.fees) 

76 

77 

78@router.post("/{guid}/fees", response={201: FeeOut}, auth=requires("LOAN_EDIT")) 

79def add_fee(request, guid: UUID, payload: FeeIn): 

80 loan = _scoped_loan(request, guid) 

81 if loan_is_disclosed(loan): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 raise DomainError( 

83 [AppError( 

84 "COC_REQUIRED", 

85 "This loan is disclosed — fee changes require a Change of Circumstance.", 

86 )] 

87 ) 

88 fee = Fee.objects.create(loan=loan, **payload.model_dump()) 

89 return 201, fee 

90 

91 

92@router.get("/{guid}/fees/summary", response=FeeSummaryOut, auth=requires("LOAN_VIEW")) 

93def fee_summary(request, guid: UUID): 

94 loan = _scoped_loan(request, guid) 

95 by_bucket: dict[str, Decimal] = {} 

96 by_section: dict[str, Decimal] = {} 

97 grand = ZERO 

98 poc_total = ZERO 

99 for f in loan.fees.all(): 

100 amt = f.total_amt or ZERO 

101 if f.poc_ind: 

102 poc_total += amt 

103 continue 

104 bucket = bucket_of(f).value 

105 by_bucket[bucket] = by_bucket.get(bucket, ZERO) + amt 

106 by_section[f.section] = by_section.get(f.section, ZERO) + amt 

107 grand += amt 

108 return { 

109 "by_bucket": [{"bucket": b, "total": round2(t)} for b, t in by_bucket.items()], 

110 "by_section": [{"section": s, "total": round2(t)} for s, t in by_section.items()], 

111 "grand_total": round2(grand), 

112 "poc_total": round2(poc_total), 

113 } 

114 

115 

116@router.post("/{guid}/fees/tolerance", response=ToleranceOut, auth=requires("LOAN_VIEW")) 

117def fee_tolerance(request, guid: UUID, payload: FeeSetIn): 

118 loan = _scoped_loan(request, guid) 

119 baseline = [ 

120 FeeInput( 

121 section=f.section, 

122 total_amt=f.total_amt, 

123 tolerance_bucket=f.tolerance_bucket, 

124 poc_ind=f.poc_ind, 

125 name=f.name, 

126 paid_by=f.paid_by, 

127 paid_to=f.paid_to, 

128 ) 

129 for f in payload.fees 

130 ] 

131 result = compute_tolerance_cure(baseline, list(loan.fees.all())) 

132 return { 

133 "by_bucket": [ 

134 { 

135 "bucket": b.value, 

136 "baseline": result.by_bucket[b].baseline, 

137 "revised": result.by_bucket[b].revised, 

138 "variance": result.by_bucket[b].variance, 

139 "cure": result.by_bucket[b].cure, 

140 } 

141 for b in _BUCKET_ORDER 

142 ], 

143 "total_cure": result.total_cure, 

144 } 

145 

146 

147@router.get("/{guid}/fees/qm", response=QmResultOut, auth=requires("LOAN_VIEW")) 

148def fee_qm( 

149 request, 

150 guid: UUID, 

151 loan_amount: Decimal | None = None, 

152 points_and_fees: Decimal | None = None, 

153 apr: Decimal | None = None, 

154 apor: Decimal | None = None, 

155): 

156 loan = _scoped_loan(request, guid) 

157 txn = _txn(loan) 

158 amt = loan_amount if loan_amount is not None else (txn.base_loan_amt if txn else ZERO) 

159 

160 if points_and_fees is not None: 

161 pf = points_and_fees 

162 source = "supplied" 

163 else: 

164 pf = sum( 

165 ( 

166 (f.total_amt or ZERO) 

167 for f in loan.fees.all() 

168 if not f.poc_ind and f.paid_by == PaidByType.BORROWER.value 

169 ), 

170 ZERO, 

171 ) 

172 source = "auto: borrower-paid non-POC fees" 

173 

174 result = qm_points_and_fees_test(amt, pf, apr, apor) 

175 return { 

176 "loan_amount": result.loan_amount, 

177 "points_and_fees": result.points_and_fees, 

178 "threshold": result.threshold, 

179 "pct": result.pct, 

180 "passed": result.passed, 

181 "qm_type": result.qm_type.value, 

182 "apr_spread": result.apr_spread, 

183 "findings": result.findings, 

184 "points_and_fees_source": source, 

185 } 

186 

187 

188# NOTE: /{guid}/fees/{fee_id} is registered AFTER literal routes like /summary and /tolerance 

189@router.delete("/{guid}/fees/{fee_id}", response={204: None}, auth=requires("LOAN_EDIT")) 

190def delete_fee(request, guid: UUID, fee_id: int): 

191 loan = _scoped_loan(request, guid) 

192 if loan_is_disclosed(loan): 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 raise DomainError( 

194 [AppError( 

195 "COC_REQUIRED", 

196 "This loan is disclosed — fee changes require a Change of Circumstance.", 

197 )] 

198 ) 

199 fee = loan.fees.filter(pk=fee_id).first() 

200 if fee is None: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true

201 raise DomainError([AppError("NOT_FOUND", "Fee not found on this loan")]) 

202 fee.delete() 

203 return 204, None 

204