Coverage for apps/underwriting/condition_api.py: 89%

1"""Underwriting condition router (mounted under /api/loans).""" 

2 

3from uuid import UUID 

4 

5from django.core.exceptions import FieldDoesNotExist 

6from django.db import models 

7from ninja import Router 

8 

9from apps.accounts.auth import ensure_staff_scope, requires 

10from apps.client_needs.models import Document 

11from apps.core_los.models import Loan 

12from apps.underwriting.models import Condition 

13from apps.underwriting.schemas import ( 

14 ConditionCreate, 

15 ConditionOut, 

16 ConditionPatch, 

17 ConditionStatusPatch, 

18) 

19from domain.conditions import assert_condition_transition 

20from domain.enums import ConditionStatus 

21from domain.result import AppError, DomainError 

22 

23router = Router(tags=["conditions"]) 

24 

25 

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

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

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

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

30 ensure_staff_scope(request.principal, loan) 

31 return loan 

32 

33 

34def _scoped_condition(loan: Loan, condition_id: int) -> Condition: 

35 condition = Condition.objects.filter(pk=condition_id, loan=loan).first() 

36 if condition is None: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 raise DomainError([AppError("NOT_FOUND", "Condition not found on this loan")]) 

38 return condition 

39 

40 

41def _coerce_required_text_nulls(model: type[models.Model], data: dict) -> dict: 

42 """A PATCH schema types every text field `str | None` so a field can be omitted, but a 

43 CharField/TextField column without `null=True` rejects an explicit `null` the client 

44 sends to clear it (Django convention: text columns use blank='', never NULL). Treat 

45 `null` for one of those the same as "clear this field" instead of letting it reach the 

46 DB and raise an IntegrityError.""" 

47 for field, value in data.items(): 

48 if value is None: 48 ↛ 47line 48 didn't jump to line 47 because the condition on line 48 was always true

49 try: 

50 f = model._meta.get_field(field) 

51 except FieldDoesNotExist: 

52 continue 

53 if isinstance(f, (models.CharField, models.TextField)) and not f.null: 53 ↛ 47line 53 didn't jump to line 47 because the condition on line 53 was always true

54 data[field] = "" 

55 return data 

56 

57 

58@router.get("/{guid}/conditions", response=list[ConditionOut], auth=requires("LOAN_VIEW")) 

59def list_conditions(request, guid: UUID): 

60 loan = _scoped_loan(request, guid) 

61 return list(loan.conditions.all().order_by("-created_at")) 

62 

63 

64@router.post("/{guid}/conditions", response={201: ConditionOut}, auth=requires("LOAN_EDIT")) 

65def create_condition(request, guid: UUID, payload: ConditionCreate): 

66 loan = _scoped_loan(request, guid) 

67 condition = Condition.objects.create(loan=loan, **payload.model_dump()) 

68 return 201, condition 

69 

70 

71@router.patch( 

72 "/{guid}/conditions/{condition_id}", response=ConditionOut, auth=requires("LOAN_EDIT") 

73) 

74def update_condition(request, guid: UUID, condition_id: int, payload: ConditionPatch): 

75 loan = _scoped_loan(request, guid) 

76 condition = _scoped_condition(loan, condition_id) 

77 data = payload.model_dump(exclude_unset=True) 

78 

79 document_ids = data.pop("document_ids", None) 

80 if document_ids is not None: 

81 documents = list(Document.objects.filter(pk__in=document_ids, loan=loan)) 

82 if len(documents) != len(set(document_ids)): 

83 raise DomainError( 

84 [ 

85 AppError( 

86 "VALIDATION", "One or more documents not found on this loan", "document_ids" 

87 ) 

88 ] 

89 ) 

90 

91 for field, value in _coerce_required_text_nulls(Condition, data).items(): 

92 setattr(condition, field, value) 

93 condition.save() 

94 

95 if document_ids is not None: 

96 condition.documents.set(documents) 

97 

98 return condition 

99 

100 

101@router.patch( 

102 "/{guid}/conditions/{condition_id}/status", 

103 response=ConditionOut, 

104 auth=requires("LOAN_EDIT"), 

105) 

106def transition_condition_status( 

107 request, guid: UUID, condition_id: int, payload: ConditionStatusPatch 

108): 

109 loan = _scoped_loan(request, guid) 

110 condition = _scoped_condition(loan, condition_id) 

111 result = assert_condition_transition( 

112 ConditionStatus(condition.status), ConditionStatus(payload.status) 

113 ) 

114 if not result.ok: 

115 raise DomainError(result.errors) 

116 condition.status = payload.status 

117 condition.save(update_fields=["status", "updated_at"]) 

118 return condition