Coverage for benefits/core/models/enrollment.py: 98%
151 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-14 01:41 +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 group_id = models.TextField(
54 blank=True, default="", help_text="Reference to the TransitProcessor group for user enrollment"
55 )
56 sign_out_button_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out button")
57 sign_out_link_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out link")
58 oauth_config = models.ForeignKey(
59 IdentityGatewayConfig,
60 on_delete=models.PROTECT,
61 null=True,
62 blank=True,
63 help_text="The IdG connection details for this flow.",
64 )
65 claims_request = models.ForeignKey(
66 ClaimsVerificationRequest,
67 on_delete=models.PROTECT,
68 null=True,
69 blank=True,
70 help_text="The claims request details for this flow.",
71 )
72 eligibility_api_url = models.TextField(
73 blank=True, default="", help_text="Fully qualified URL for an Eligibility API server used by this flow."
74 )
75 eligibility_api_auth_header = models.TextField(
76 blank=True,
77 default="",
78 help_text="The auth header to send in Eligibility API requests for this flow.",
79 )
80 eligibility_api_auth_key_secret_name = SecretNameField(
81 blank=True,
82 default="",
83 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
84 )
85 eligibility_api_public_key = models.ForeignKey(
86 PemData,
87 related_name="+",
88 on_delete=models.PROTECT,
89 null=True,
90 blank=True,
91 help_text="The public key used to encrypt Eligibility API requests and to verify signed Eligibility API responses for this flow.", # noqa: E501
92 )
93 eligibility_api_jwe_cek_enc = models.TextField(
94 blank=True,
95 default="",
96 help_text="The JWE-compatible Content Encryption Key (CEK) key-length and mode to use in Eligibility API requests for this flow.", # noqa: E501
97 )
98 eligibility_api_jwe_encryption_alg = models.TextField(
99 blank=True,
100 default="",
101 help_text="The JWE-compatible encryption algorithm to use in Eligibility API requests for this flow.",
102 )
103 eligibility_api_jws_signing_alg = models.TextField(
104 blank=True,
105 default="",
106 help_text="The JWS-compatible signing algorithm to use in Eligibility API requests for this flow.",
107 )
108 eligibility_form_class = models.TextField(
109 blank=True,
110 default="",
111 help_text="The fully qualified Python path of a form class used by this flow, e.g. benefits.eligibility.forms.FormClass", # noqa: E501
112 )
113 selection_label_template_override = models.TextField(
114 blank=True,
115 default="",
116 help_text="Override the default template that defines the end-user UI for selecting this flow among other options.",
117 )
118 supports_expiration = models.BooleanField(
119 default=False, help_text="Indicates if the enrollment expires or does not expire"
120 )
121 expiration_days = models.PositiveSmallIntegerField(
122 null=True, blank=True, help_text="If the enrollment supports expiration, number of days before the eligibility expires"
123 )
124 expiration_reenrollment_days = models.PositiveSmallIntegerField(
125 null=True,
126 blank=True,
127 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
128 )
129 reenrollment_error_template = models.TextField(
130 blank=True, default="", help_text="Template for a re-enrollment error associated with the enrollment flow"
131 )
132 display_order = models.PositiveSmallIntegerField(default=0, blank=False, null=False)
134 class Meta:
135 ordering = ["display_order"]
137 def __str__(self):
138 return self.label
140 @property
141 def agency_card_name(self):
142 if self.uses_api_verification:
143 return f"{self.transit_agency.slug}-agency-card"
144 else:
145 return ""
147 @property
148 def eligibility_api_auth_key(self):
149 if self.eligibility_api_auth_key_secret_name is not None: 149 ↛ 153line 149 didn't jump to line 153 because the condition on line 149 was always true
150 secret_field = self._meta.get_field("eligibility_api_auth_key_secret_name")
151 return secret_field.secret_value(self)
152 else:
153 return None
155 @property
156 def eligibility_api_public_key_data(self):
157 """This flow's Eligibility API public key as a string."""
158 return self.eligibility_api_public_key.data
160 @property
161 def selection_label_template(self):
162 prefix = "eligibility/includes/selection-label"
163 if self.uses_api_verification:
164 return self.selection_label_template_override or f"{prefix}--{self.agency_card_name}.html"
165 else:
166 return self.selection_label_template_override or f"{prefix}--{self.system_name}.html"
168 @property
169 def eligibility_start_context(self):
170 return eligibility_context.eligibility_start[self.system_name].dict()
172 @property
173 def eligibility_unverified_context(self):
174 ctx = eligibility_context.eligibility_unverified.get(self.system_name)
175 return ctx.dict() if ctx else {}
177 @property
178 def uses_claims_verification(self):
179 """True if this flow verifies via the Identity Gateway and has a scope and claim. False otherwise."""
180 return (
181 self.oauth_config is not None and bool(self.claims_request.scopes) and bool(self.claims_request.eligibility_claim)
182 )
184 @property
185 def uses_api_verification(self):
186 """True if this flow verifies via the Eligibility API. False otherwise."""
187 return bool(self.eligibility_api_url) and bool(self.eligibility_form_class)
189 @property
190 def claims_scheme(self):
191 return self.claims_request.scheme or self.oauth_config.scheme
193 @property
194 def eligibility_verifier(self):
195 """A str representing the entity that verifies eligibility for this flow.
197 Either the client name of the flow's claims provider, or the URL to the eligibility API.
198 """
199 if self.uses_claims_verification:
200 return self.oauth_config.client_name
201 else:
202 return self.eligibility_api_url
204 @property
205 def enrollment_index_context(self):
206 ctx = enrollment_context.enrollment_index.get(self.system_name, enrollment_context.DefaultEnrollmentIndex())
207 ctx_dict = ctx.dict()
208 ctx_dict["transit_processor"] = self.transit_agency.transit_processor_context
210 return ctx_dict
212 @property
213 def enrollment_success_context(self):
214 if self.uses_api_verification:
215 return enrollment_context.enrollment_success[self.system_name].dict()
216 else:
217 return enrollment_context.enrollment_success[self.transit_agency.slug].dict()
219 @property
220 def in_person_eligibility_context(self):
221 return in_person_context.eligibility_index[self.system_name].dict()
223 @property
224 def help_context(self):
225 ctx = core_context.flows_help.get(self.system_name)
226 return [c.dict() for c in ctx] if ctx else []
228 @property
229 def supports_sign_out(self):
230 return bool(self.sign_out_button_template) or bool(self.sign_out_link_template)
232 def clean(self):
233 errors = []
235 if self.transit_agency:
236 templates = [
237 self.selection_label_template,
238 ]
239 if self.supports_expiration:
240 templates.append(self.reenrollment_error_template)
242 # since templates are calculated from the pattern or the override field
243 # we can't add a field-level validation error
244 # so just create directly for a missing template
245 for t in templates:
246 if not template_path(t):
247 errors.append(ValidationError(f"Template not found: {t}"))
249 if EnrollmentMethods.IN_PERSON in self.supported_enrollment_methods: 249 ↛ 260line 249 didn't jump to line 260 because the condition on line 249 was always true
250 try:
251 in_person_eligibility_context = self.in_person_eligibility_context
252 except KeyError:
253 in_person_eligibility_context = None
255 if not in_person_eligibility_context:
256 errors.append(
257 ValidationError(f"{self.system_name} not configured for In-person. Please uncheck to continue.")
258 )
260 if errors:
261 raise ValidationError(errors)
263 def eligibility_form_instance(self, *args, **kwargs):
264 """Return an instance of this flow's EligibilityForm, or None."""
265 if not self.uses_api_verification:
266 return None
268 # inspired by https://stackoverflow.com/a/30941292
269 module_name, class_name = self.eligibility_form_class.rsplit(".", 1)
270 FormClass = getattr(importlib.import_module(module_name), class_name)
272 return FormClass(*args, **kwargs)
274 @staticmethod
275 def by_id(id):
276 """Get an EnrollmentFlow instance by its ID."""
277 logger.debug(f"Get {EnrollmentFlow.__name__} by id: {id}")
278 return EnrollmentFlow.objects.get(id=id)
281class EnrollmentEvent(models.Model):
282 """A record of a successful enrollment."""
284 id = models.UUIDField(primary_key=True, default=uuid.uuid4)
285 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT)
286 enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
287 enrollment_method = models.TextField(
288 choices={
289 EnrollmentMethods.DIGITAL: EnrollmentMethods.DIGITAL,
290 EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
291 }
292 )
293 verified_by = models.TextField()
294 enrollment_datetime = models.DateTimeField(default=timezone.now)
295 expiration_datetime = models.DateTimeField(blank=True, null=True)
296 extra_claims = models.TextField(blank=True, default="")
298 def __str__(self):
299 dt = timezone.localtime(self.enrollment_datetime)
300 ts = dt.strftime("%b %d, %Y, %I:%M %p")
301 return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"