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