Coverage for benefits / enrollment / enrollment.py: 98%

1from datetime import datetime, timedelta 

2from enum import Enum 

3 

4import sentry_sdk 

5from django.shortcuts import redirect 

6from django.utils import timezone 

7 

8from benefits.core import models, session 

9from benefits.routes import routes 

10 

11from . import analytics 

12 

13 

14class Status(Enum): 

15 # SUCCESS means the enrollment went through successfully 

16 SUCCESS = 1 

17 

18 # SYSTEM_ERROR means the enrollment system encountered an internal error (returned a 500 HTTP status) 

19 SYSTEM_ERROR = 2 

20 

21 # EXCEPTION means the enrollment system is working, but something unexpected happened 

22 # because of a misconfiguration or invalid request from our side 

23 EXCEPTION = 3 

24 

25 # REENROLLMENT_ERROR means that the user tried to re-enroll but is not within the reenrollment window 

26 REENROLLMENT_ERROR = 4 

27 

28 

29def _is_expired(expiry_date: datetime): 

30 """Returns whether the passed in datetime is expired or not.""" 

31 return expiry_date <= timezone.now() 

32 

33 

34def _is_within_reenrollment_window(expiry_date: datetime, enrollment_reenrollment_date: datetime): 

35 """Returns if we are currently within the reenrollment window.""" 

36 return enrollment_reenrollment_date <= timezone.now() < expiry_date 

37 

38 

39def _calculate_expiry(expiration_days: int): 

40 """Returns the expiry datetime, which should be midnight in our configured timezone of the (N + 1)th day from now, 

41 where N is expiration_days.""" 

42 default_time_zone = timezone.get_default_timezone() 

43 expiry_date = timezone.localtime(timezone=default_time_zone) + timedelta(days=expiration_days + 1) 

44 expiry_datetime = expiry_date.replace(hour=0, minute=0, second=0, microsecond=0) 

45 

46 return expiry_datetime 

47 

48 

49def handle_enrollment_results( 

50 request, 

51 status: Status, 

52 verified_by: str, 

53 exception: Exception = None, 

54 enrollment_method: str = models.EnrollmentMethods.DIGITAL, 

55 route_reenrollment_error=routes.ENROLLMENT_REENROLLMENT_ERROR, 

56 route_success=routes.ENROLLMENT_SUCCESS, 

57 route_system_error=routes.ENROLLMENT_SYSTEM_ERROR, 

58 card_category: str = None, 

59 card_scheme: str = None, 

60): 

61 flow = session.flow(request) 

62 agency = session.agency(request) 

63 group_id = str(session.group(request).group_id) # needs to be a string for the API call 

64 match (status): 

65 case Status.SUCCESS: 

66 expiry = session.enrollment_expiry(request) 

67 oauth_extra_claims = session.oauth_extra_claims(request) 

68 # EnrollmentEvent expects a string value for extra_claims 

69 if oauth_extra_claims: 

70 str_extra_claims = ", ".join(oauth_extra_claims) 

71 else: 

72 str_extra_claims = "" 

73 

74 agencies_to_report = [agency] 

75 agencies_to_report.extend(agency.group_agencies()) 

76 

77 for agency in agencies_to_report: 

78 event = models.EnrollmentEvent.objects.create( 

79 transit_agency=agency, 

80 enrollment_flow=flow, 

81 enrollment_method=enrollment_method, 

82 verified_by=verified_by, 

83 expiration_datetime=expiry, 

84 extra_claims=str_extra_claims, 

85 ) 

86 event.save() 

87 

88 analytics.returned_success( 

89 request, 

90 agency=agency, 

91 enrollment_group=group_id, 

92 transit_processor=agency.transit_processor, 

93 enrollment_method=enrollment_method, 

94 extra_claims=oauth_extra_claims, 

95 card_scheme=card_scheme, 

96 card_category=card_category, 

97 ) 

98 

99 return redirect(route_success) 

100 

101 case Status.SYSTEM_ERROR: 

102 analytics.returned_error( 

103 request, 

104 str(exception), 

105 agency=agency, 

106 enrollment_group=group_id, 

107 transit_processor=agency.transit_processor, 

108 enrollment_method=enrollment_method, 

109 ) 

110 sentry_sdk.capture_exception(exception) 

111 return redirect(route_system_error) 

112 

113 case Status.EXCEPTION: 

114 analytics.returned_error( 

115 request, 

116 str(exception), 

117 agency=agency, 

118 enrollment_group=group_id, 

119 transit_processor=agency.transit_processor, 

120 enrollment_method=enrollment_method, 

121 ) 

122 raise exception 

123 

124 case Status.REENROLLMENT_ERROR: 124 ↛ exitline 124 didn't return from function 'handle_enrollment_results' because the pattern on line 124 always matched

125 analytics.returned_error( 

126 request, 

127 "Re-enrollment error.", 

128 agency=agency, 

129 enrollment_group=group_id, 

130 transit_processor=agency.transit_processor, 

131 enrollment_method=enrollment_method, 

132 ) 

133 return redirect(route_reenrollment_error)