Coverage for benefits/core/models/enrollment.py: 97%
155 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-17 19:57 +0000
1import logging
2import uuid
4from cdt_identity.models import IdentityGatewayConfig, ClaimsVerificationRequest
5from django.core.exceptions import ValidationError
6from django.db import models
7from django.utils import timezone
8from multiselectfield import MultiSelectField
10from .common import PemData, SecretNameField, template_path
11from .transit import TransitAgency
12from benefits.core import context as core_context
13from benefits.eligibility import context as eligibility_context
14from benefits.enrollment import context as enrollment_context
15from benefits.in_person import context as in_person_context
17logger = logging.getLogger(__name__)
20class EnrollmentMethods:
21 DIGITAL = "digital"
22 IN_PERSON = "in_person"
25SUPPORTED_METHODS = (
26 (EnrollmentMethods.DIGITAL, EnrollmentMethods.DIGITAL.capitalize()),
27 (EnrollmentMethods.IN_PERSON, EnrollmentMethods.IN_PERSON.replace("_", "-").capitalize()),
28)
31class EnrollmentFlow(models.Model):
32 """Represents a user journey through the Benefits app for a single eligibility type."""
34 id = models.AutoField(primary_key=True)
35 system_name = models.SlugField(
36 choices=core_context.SystemName,
37 help_text="Primary internal system name for this EnrollmentFlow instance, e.g. in analytics and Eligibility API requests.", # noqa: 501
38 )
39 label = models.TextField(
40 blank=True,
41 default="",
42 help_text="A human readable label, used as the display text in Admin.",
43 )
44 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT, null=True, blank=True)
45 supported_enrollment_methods = MultiSelectField(
46 choices=SUPPORTED_METHODS,
47 max_choices=2,
48 max_length=50,
49 default=[EnrollmentMethods.DIGITAL, EnrollmentMethods.IN_PERSON],
50 help_text="If the flow is supported by digital enrollment, in-person enrollment, or both",
51 )
52 sign_out_button_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out button")
53 sign_out_link_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out link")
54 oauth_config = models.ForeignKey(
55 IdentityGatewayConfig,
56 on_delete=models.PROTECT,
57 null=True,
58 blank=True,
59 help_text="The IdG connection details for this flow.",
60 )
61 claims_request = models.ForeignKey(
62 ClaimsVerificationRequest,
63 on_delete=models.PROTECT,
64 null=True,
65 blank=True,
66 help_text="The claims request details for this flow.",
67 )
68 eligibility_api_url = models.TextField(
69 blank=True, default="", help_text="Fully qualified URL for an Eligibility API server used by this flow."
70 )
71 eligibility_api_auth_header = models.TextField(
72 blank=True,
73 default="",
74 help_text="The auth header to send in Eligibility API requests for this flow.",
75 )
76 eligibility_api_auth_key_secret_name = SecretNameField(
77 blank=True,
78 default="",
79 help_text="The name of a secret containing the value of the auth header to send in Eligibility API requests for this flow.", # noqa: 501
80 )
81 eligibility_api_public_key = models.ForeignKey(
82 PemData,
83 related_name="+",
84 on_delete=models.PROTECT,
85 null=True,
86 blank=True,
87 help_text="The public key used to encrypt Eligibility API requests and to verify signed Eligibility API responses for this flow.", # noqa: E501
88 )
89 eligibility_api_jwe_cek_enc = models.TextField(
90 blank=True,
91 default="",
92 help_text="The JWE-compatible Content Encryption Key (CEK) key-length and mode to use in Eligibility API requests for this flow.", # noqa: E501
93 )
94 eligibility_api_jwe_encryption_alg = models.TextField(
95 blank=True,
96 default="",
97 help_text="The JWE-compatible encryption algorithm to use in Eligibility API requests for this flow.",
98 )
99 eligibility_api_jws_signing_alg = models.TextField(
100 blank=True,
101 default="",
102 help_text="The JWS-compatible signing algorithm to use in Eligibility API requests for this flow.",
103 )
104 selection_label_template_override = models.TextField(
105 blank=True,
106 default="",
107 help_text="Override the default template that defines the end-user UI for selecting this flow among other options.",
108 )
109 supports_expiration = models.BooleanField(
110 default=False, help_text="Indicates if the enrollment expires or does not expire"
111 )
112 expiration_days = models.PositiveSmallIntegerField(
113 null=True, blank=True, help_text="If the enrollment supports expiration, number of days before the eligibility expires"
114 )
115 expiration_reenrollment_days = models.PositiveSmallIntegerField(
116 null=True,
117 blank=True,
118 help_text="If the enrollment supports expiration, number of days preceding the expiration date during which a user can re-enroll in the eligibilty", # noqa: E501
119 )
120 display_order = models.PositiveSmallIntegerField(default=0, blank=False, null=False)
122 class Meta:
123 ordering = ["display_order"]
125 def __str__(self):
126 agency_slug = self.transit_agency.slug if self.transit_agency else "no agency"
127 return f"{self.label} ({agency_slug})"
129 @property
130 def group_id(self):
131 if hasattr(self, "enrollmentgroup"):
132 enrollment_group = self.enrollmentgroup
134 # these are the class names for models in enrollment_littlepay and enrollment_switchio
135 if hasattr(enrollment_group, "littlepaygroup"):
136 return str(enrollment_group.littlepaygroup.group_id)
137 elif hasattr(enrollment_group, "switchiogroup"): 137 ↛ 140line 137 didn't jump to line 140 because the condition on line 137 was always true
138 return enrollment_group.switchiogroup.group_id
139 else:
140 return None
141 else:
142 return None
144 @property
145 def agency_card_name(self):
146 if self.uses_api_verification:
147 return f"{self.transit_agency.slug}-agency-card"
148 else:
149 return ""
151 @property
152 def eligibility_api_auth_key(self):
153 if self.eligibility_api_auth_key_secret_name is not None: 153 ↛ 157line 153 didn't jump to line 157 because the condition on line 153 was always true
154 secret_field = self._meta.get_field("eligibility_api_auth_key_secret_name")
155 return secret_field.secret_value(self)
156 else:
157 return None
159 @property
160 def eligibility_api_public_key_data(self):
161 """This flow's Eligibility API public key as a string."""
162 return self.eligibility_api_public_key.data
164 @property
165 def selection_label_template(self):
166 prefix = "eligibility/includes/selection-label"
167 if self.uses_api_verification:
168 return self.selection_label_template_override or f"{prefix}--{self.agency_card_name}.html"
169 else:
170 return self.selection_label_template_override or f"{prefix}--{self.system_name}.html"
172 @property
173 def eligibility_start_context(self):
174 return eligibility_context.eligibility_start[self.system_name].dict()
176 @property
177 def eligibility_unverified_context(self):
178 ctx = eligibility_context.eligibility_unverified.get(self.system_name)
179 return ctx.dict() if ctx else {}
181 @property
182 def uses_claims_verification(self):
183 """True if this flow verifies via the Identity Gateway and has a scope and claim. False otherwise."""
184 return (
185 self.oauth_config is not None and bool(self.claims_request.scopes) and bool(self.claims_request.eligibility_claim)
186 )
188 @property
189 def uses_api_verification(self):
190 """True if this flow verifies via the Eligibility API. False otherwise."""
191 return bool(self.eligibility_api_url)
193 @property
194 def claims_scheme(self):
195 return self.claims_request.scheme or self.oauth_config.scheme
197 @property
198 def eligibility_verifier(self):
199 """A str representing the entity that verifies eligibility for this flow.
201 Either the client name of the flow's claims provider, or the URL to the eligibility API.
202 """
203 if self.uses_claims_verification:
204 return self.oauth_config.client_name
205 else:
206 return self.eligibility_api_url
208 @property
209 def enrollment_index_context(self):
210 ctx = enrollment_context.enrollment_index.get(self.system_name, enrollment_context.DefaultEnrollmentIndex())
211 return ctx.dict()
213 @property
214 def enrollment_success_context(self):
215 if self.uses_api_verification:
216 return enrollment_context.enrollment_success[self.system_name].dict()
217 else:
218 return enrollment_context.enrollment_success[self.transit_agency.slug].dict()
220 @property
221 def in_person_eligibility_context(self):
222 return in_person_context.eligibility_index[self.system_name].dict()
224 @property
225 def help_context(self):
226 ctx = core_context.flows_help.get(self.system_name)
227 return [c.dict() for c in ctx] if ctx else []
229 @property
230 def supports_sign_out(self):
231 return bool(self.sign_out_button_template) or bool(self.sign_out_link_template)
233 def clean(self):
234 errors = []
236 if self.transit_agency:
237 templates = [
238 self.selection_label_template,
239 ]
241 # since templates are calculated from the pattern or the override field
242 # we can't add a field-level validation error
243 # so just create directly for a missing template
244 for t in templates:
245 if not template_path(t):
246 errors.append(ValidationError(f"Template not found: {t}"))
248 if EnrollmentMethods.IN_PERSON in self.supported_enrollment_methods: 248 ↛ 259line 248 didn't jump to line 259 because the condition on line 248 was always true
249 try:
250 in_person_eligibility_context = self.in_person_eligibility_context
251 except KeyError:
252 in_person_eligibility_context = None
254 if not in_person_eligibility_context:
255 errors.append(
256 ValidationError(f"{self.system_name} not configured for In-person. Please uncheck to continue.")
257 )
259 if self.transit_agency.active and self.group_id is None:
260 errors.append(
261 ValidationError(f"{self.system_name} needs either a LittlepayGroup or SwitchioGroup linked to it.")
262 )
264 if errors:
265 raise ValidationError(errors)
267 @staticmethod
268 def by_id(id):
269 """Get an EnrollmentFlow instance by its ID."""
270 logger.debug(f"Get {EnrollmentFlow.__name__} by id: {id}")
271 return EnrollmentFlow.objects.get(id=id)
274class EnrollmentGroup(models.Model):
275 id = models.AutoField(primary_key=True)
276 enrollment_flow = models.OneToOneField(
277 EnrollmentFlow,
278 on_delete=models.PROTECT,
279 help_text="The enrollment flow that this group is for.",
280 )
282 def __str__(self):
283 return str(self.enrollment_flow)
286class EnrollmentEvent(models.Model):
287 """A record of a successful enrollment."""
289 id = models.UUIDField(primary_key=True, default=uuid.uuid4)
290 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT)
291 enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
292 enrollment_method = models.TextField(
293 choices={
294 EnrollmentMethods.DIGITAL: EnrollmentMethods.DIGITAL,
295 EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
296 }
297 )
298 verified_by = models.TextField()
299 enrollment_datetime = models.DateTimeField(default=timezone.now)
300 expiration_datetime = models.DateTimeField(blank=True, null=True)
301 extra_claims = models.TextField(blank=True, default="")
303 def __str__(self):
304 dt = timezone.localtime(self.enrollment_datetime)
305 ts = dt.strftime("%b %d, %Y, %I:%M %p")
306 return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"