Coverage for apps/core_los/leads_api.py: 86%

1"""Leads (CRM) router: GET/POST/DELETE /api/leads, stats, convert (mounted at "").""" 

2 

3from ninja import Router 

4 

5from apps.accounts.auth import requires 

6from apps.core_los.api import _build_loan 

7from apps.core_los.models import Lead 

8from apps.core_los.schemas import ( 

9 LeadConversionOut, 

10 LeadOut, 

11 LeadStatsOut, 

12 NewLeadInput, 

13) 

14from domain.result import AppError, DomainError 

15 

16router = Router(tags=["leads"]) 

17 

18 

19@router.get("/leads", response=list[LeadOut], auth=requires("LEAD_VIEW")) 

20def list_leads(request): 

21 return list(Lead.objects.select_related("converted_loan").order_by("-created_at")) 

22 

23 

24@router.get("/leads/stats", response=LeadStatsOut, auth=requires("LEAD_VIEW")) 

25def lead_stats(request): 

26 return { 

27 "total": Lead.objects.count(), 

28 "new": Lead.objects.filter(status="New").count(), 

29 "qualified": Lead.objects.filter(status="Qualified").count(), 

30 "converted": Lead.objects.filter(status="Converted").count(), 

31 } 

32 

33 

34@router.post("/leads", response={201: LeadOut}, auth=requires("LEAD_EDIT")) 

35def create_lead(request, payload: NewLeadInput): 

36 if not payload.first_name.strip() or not payload.last_name.strip(): 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 raise DomainError([AppError("VALIDATION", "First and last name are required")]) 

38 lead = Lead.objects.create( 

39 first_name=payload.first_name, 

40 last_name=payload.last_name, 

41 email=payload.email, 

42 phone=payload.phone, 

43 source=payload.source, 

44 purpose=payload.purpose, 

45 est_loan_amount=payload.est_loan_amount, 

46 ) 

47 return 201, lead 

48 

49 

50@router.delete("/leads/{lead_id}", response={204: None}, auth=requires("LEAD_EDIT")) 

51def delete_lead(request, lead_id: int): 

52 deleted, _ = Lead.objects.filter(pk=lead_id).delete() 

53 if deleted == 0: 

54 raise DomainError([AppError("NOT_FOUND", "Lead not found")]) 

55 return 204, None 

56 

57 

58@router.post( 

59 "/leads/{lead_id}/convert", response={201: LeadConversionOut}, auth=requires("LOAN_CREATE") 

60) 

61def convert_lead(request, lead_id: int): 

62 if "LEAD_EDIT" not in request.principal.permissions: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 raise DomainError([AppError("FORBIDDEN", "LEAD_EDIT permission required")]) 

64 

65 lead = Lead.objects.filter(pk=lead_id).first() 

66 if lead is None: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise DomainError([AppError("NOT_FOUND", "Lead not found")]) 

68 if lead.status == "Converted": 

69 raise DomainError([AppError("VALIDATION", "Lead is already converted")]) 

70 

71 loan = _build_loan( 

72 loan_officer_id=request.principal.user_id, 

73 state_code="FL", 

74 occupancy="PrimaryResidence", 

75 purpose=lead.purpose or "Purchase", 

76 first_name=lead.first_name, 

77 last_name=lead.last_name, 

78 email=lead.email or "", 

79 ) 

80 lead.status = "Converted" 

81 lead.converted_loan = loan 

82 lead.save(update_fields=["status", "converted_loan"]) 

83 return 201, {"loan_guid": str(loan.guid), "loan_number": loan.arive_loan_no}