Coverage for benefits / core / models / enrollment.py: 94%
168 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-22 19:08 +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 EligibilityApiVerificationRequest(models.Model):
32 """Represents configuration for eligibility verification via Eligibility API calls."""
34 id = models.AutoField(primary_key=True)
35 label = models.SlugField(
36 help_text="A human readable label, used as the display text in Admin.",
37 )
38 api_url = models.URLField(help_text="Fully qualified URL for an Eligibility API server.")
39 api_auth_header = models.CharField(
40 help_text="The auth header to send in Eligibility API requests.",
41 max_length=50,
42 )
43 api_auth_key_secret_name = SecretNameField(
44 help_text="The name of a secret containing the value of the auth header to send in Eligibility API requests.",
45 )
46 api_public_key = models.ForeignKey(
47 PemData,
48 related_name="+",
49 on_delete=models.PROTECT,
50 help_text="The public key used to encrypt Eligibility API requests and to verify signed Eligibility API responses.",
51 )
52 api_jwe_cek_enc = models.CharField(
53 help_text="The JWE-compatible Content Encryption Key (CEK) key-length and mode to use in Eligibility API requests.",
54 max_length=50,
55 )
56 api_jwe_encryption_alg = models.CharField(
57 help_text="The JWE-compatible encryption algorithm to use in Eligibility API requests.",
58 max_length=50,
59 )
60 api_jws_signing_alg = models.CharField(
61 help_text="The JWS-compatible signing algorithm to use in Eligibility API requests.",
62 max_length=50,
63 )
65 def __str__(self):
66 return self.label
68 @property
69 def api_auth_key(self):
70 """The Eligibility API auth key as a string."""
71 secret_field = self._meta.get_field("api_auth_key_secret_name")
72 return secret_field.secret_value(self)
74 @property
75 def api_public_key_data(self):
76 """The Eligibility API public key as a string."""
77 return self.api_public_key.data
80class EnrollmentFlow(models.Model):
81 """Represents a user journey through the Benefits app for a single eligibility type."""
83 id = models.AutoField(primary_key=True)
84 system_name = models.SlugField(
85 choices=core_context.SystemName,
86 help_text="Primary internal system name for this EnrollmentFlow instance, e.g. in analytics and Eligibility API requests.", # noqa: 501
87 )
88 label = models.TextField(
89 blank=True,
90 default="",
91 help_text="A human readable label, used as the display text in Admin.",
92 )
93 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT, null=True, blank=True)
94 supported_enrollment_methods = MultiSelectField(
95 choices=SUPPORTED_METHODS,
96 max_choices=2,
97 max_length=50,
98 default=[EnrollmentMethods.DIGITAL, EnrollmentMethods.IN_PERSON],
99 help_text="If the flow is supported by digital enrollment, in-person enrollment, or both",
100 )
101 sign_out_button_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out button")
102 sign_out_link_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out link")
103 oauth_config = models.ForeignKey(
104 IdentityGatewayConfig,
105 on_delete=models.PROTECT,
106 null=True,
107 blank=True,
108 help_text="The IdG connection details for this flow.",
109 )
110 claims_request = models.ForeignKey(
111 ClaimsVerificationRequest,
112 on_delete=models.PROTECT,
113 null=True,
114 blank=True,
115 help_text="The claims request details for this flow.",
116 )
117 api_request = models.ForeignKey(
118 EligibilityApiVerificationRequest,
119 on_delete=models.PROTECT,
120 null=True,
121 blank=True,
122 help_text="The Eligibility API request details for this flow.",
123 )
124 selection_label_template_override = models.TextField(
125 blank=True,
126 default="",
127 help_text="Override the default template that defines the end-user UI for selecting this flow among other options.",
128 )
129 supports_expiration = models.BooleanField(
130 default=False, help_text="Indicates if the enrollment expires or does not expire"
131 )
132 expiration_days = models.PositiveSmallIntegerField(
133 null=True, blank=True, help_text="If the enrollment supports expiration, number of days before the eligibility expires"
134 )
135 expiration_reenrollment_days = models.PositiveSmallIntegerField(
136 null=True,
137 blank=True,
138 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
139 )
140 display_order = models.PositiveSmallIntegerField(default=0, blank=False, null=False)
142 class Meta:
143 ordering = ["display_order"]
145 def __str__(self):
146 agency_slug = self.transit_agency.slug if self.transit_agency else "no agency"
147 return f"{self.label} ({agency_slug})"
149 @property
150 def group_id(self):
151 if hasattr(self, "enrollmentgroup"):
152 enrollment_group = self.enrollmentgroup
154 # these are the class names for models in enrollment_littlepay and enrollment_switchio
155 if hasattr(enrollment_group, "littlepaygroup"):
156 return str(enrollment_group.littlepaygroup.group_id)
157 elif hasattr(enrollment_group, "switchiogroup"): 157 ↛ 160line 157 didn't jump to line 160 because the condition on line 157 was always true
158 return enrollment_group.switchiogroup.group_id
159 else:
160 return None
161 else:
162 return None
164 @property
165 def agency_card_name(self):
166 if self.uses_api_verification:
167 return f"{self.transit_agency.slug}-agency-card"
168 else:
169 return ""
171 @property
172 def eligibility_api_auth_key(self):
173 if self.uses_api_verification: 173 ↛ 176line 173 didn't jump to line 176 because the condition on line 173 was always true
174 return self.api_request.api_auth_key
175 else:
176 return None
178 @property
179 def eligibility_api_public_key_data(self):
180 """This flow's Eligibility API public key as a string."""
181 if self.uses_api_verification: 181 ↛ 184line 181 didn't jump to line 184 because the condition on line 181 was always true
182 return self.api_request.api_public_key_data
183 else:
184 return None
186 @property
187 def selection_label_template(self):
188 prefix = "eligibility/includes/selection-label"
189 if self.uses_api_verification:
190 return self.selection_label_template_override or f"{prefix}--{self.agency_card_name}.html"
191 else:
192 return self.selection_label_template_override or f"{prefix}--{self.system_name}.html"
194 @property
195 def eligibility_start_context(self):
196 return eligibility_context.eligibility_start[self.system_name].dict()
198 @property
199 def eligibility_unverified_context(self):
200 ctx = eligibility_context.eligibility_unverified.get(self.system_name)
201 return ctx.dict() if ctx else {}
203 @property
204 def uses_claims_verification(self):
205 """True if this flow verifies via the Identity Gateway and has a scope and claim. False otherwise."""
206 return (
207 self.oauth_config is not None and bool(self.claims_request.scopes) and bool(self.claims_request.eligibility_claim)
208 )
210 @property
211 def uses_api_verification(self):
212 """True if this flow verifies via the Eligibility API. False otherwise."""
213 return self.api_request is not None
215 @property
216 def claims_scheme(self):
217 if self.uses_claims_verification: 217 ↛ 220line 217 didn't jump to line 220 because the condition on line 217 was always true
218 return self.claims_request.scheme or self.oauth_config.scheme
219 else:
220 return None
222 @property
223 def eligibility_verifier(self):
224 """A str representing the entity that verifies eligibility for this flow.
226 Either the client name of the flow's claims provider, or the URL to the eligibility API.
227 """
228 if self.uses_claims_verification:
229 return self.oauth_config.client_name
230 elif self.uses_api_verification:
231 return self.api_request.api_url
232 else:
233 return "undefined"
235 @property
236 def enrollment_index_context(self):
237 ctx = enrollment_context.enrollment_index.get(self.system_name, enrollment_context.DefaultEnrollmentIndex())
238 return ctx.dict()
240 @property
241 def in_person_eligibility_context(self):
242 return in_person_context.eligibility_index[self.system_name].dict()
244 @property
245 def help_context(self):
246 ctx = core_context.flows_help.get(self.system_name)
247 return [c.dict() for c in ctx] if ctx else []
249 @property
250 def supports_sign_out(self):
251 return bool(self.sign_out_button_template) or bool(self.sign_out_link_template)
253 def clean(self):
254 errors = []
256 if self.transit_agency:
257 templates = [
258 self.selection_label_template,
259 ]
261 # since templates are calculated from the pattern or the override field
262 # we can't add a field-level validation error
263 # so just create directly for a missing template
264 for t in templates:
265 if not template_path(t):
266 errors.append(ValidationError(f"Template not found: {t}"))
268 if EnrollmentMethods.IN_PERSON in self.supported_enrollment_methods: 268 ↛ 279line 268 didn't jump to line 279 because the condition on line 268 was always true
269 try:
270 in_person_eligibility_context = self.in_person_eligibility_context
271 except KeyError:
272 in_person_eligibility_context = None
274 if not in_person_eligibility_context:
275 errors.append(
276 ValidationError(f"{self.system_name} not configured for In-person. Please uncheck to continue.")
277 )
279 if self.transit_agency.active and self.group_id is None: 279 ↛ 284line 279 didn't jump to line 284 because the condition on line 279 was always true
280 errors.append(
281 ValidationError(f"{self.system_name} needs either a LittlepayGroup or SwitchioGroup linked to it.")
282 )
284 if errors:
285 raise ValidationError(errors)
287 @staticmethod
288 def by_id(id):
289 """Get an EnrollmentFlow instance by its ID."""
290 logger.debug(f"Get {EnrollmentFlow.__name__} by id: {id}")
291 return EnrollmentFlow.objects.get(id=id)
294class EnrollmentGroup(models.Model):
295 id = models.AutoField(primary_key=True)
296 enrollment_flow = models.OneToOneField(
297 EnrollmentFlow,
298 on_delete=models.PROTECT,
299 help_text="The enrollment flow that this group is for.",
300 )
302 def __str__(self):
303 return str(self.enrollment_flow)
306class EnrollmentEvent(models.Model):
307 """A record of a successful enrollment."""
309 id = models.UUIDField(primary_key=True, default=uuid.uuid4)
310 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT)
311 enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
312 enrollment_method = models.TextField(
313 choices={
314 EnrollmentMethods.DIGITAL: EnrollmentMethods.DIGITAL,
315 EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
316 }
317 )
318 verified_by = models.TextField()
319 enrollment_datetime = models.DateTimeField(default=timezone.now)
320 expiration_datetime = models.DateTimeField(blank=True, null=True)
321 extra_claims = models.TextField(blank=True, default="")
323 def __str__(self):
324 dt = timezone.localtime(self.enrollment_datetime)
325 ts = dt.strftime("%b %d, %Y, %I:%M %p")
326 return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"