Coverage for benefits/core/models/enrollment.py: 98%
149 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-22 21:13 +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 return ctx.dict()
209 @property
210 def enrollment_success_context(self):
211 if self.uses_api_verification:
212 return enrollment_context.enrollment_success[self.system_name].dict()
213 else:
214 return enrollment_context.enrollment_success[self.transit_agency.slug].dict()
216 @property
217 def in_person_eligibility_context(self):
218 return in_person_context.eligibility_index[self.system_name].dict()
220 @property
221 def help_context(self):
222 ctx = core_context.flows_help.get(self.system_name)
223 return [c.dict() for c in ctx] if ctx else []
225 @property
226 def supports_sign_out(self):
227 return bool(self.sign_out_button_template) or bool(self.sign_out_link_template)
229 def clean(self):
230 errors = []
232 if self.transit_agency:
233 templates = [
234 self.selection_label_template,
235 ]
236 if self.supports_expiration:
237 templates.append(self.reenrollment_error_template)
239 # since templates are calculated from the pattern or the override field
240 # we can't add a field-level validation error
241 # so just create directly for a missing template
242 for t in templates:
243 if not template_path(t):
244 errors.append(ValidationError(f"Template not found: {t}"))
246 if EnrollmentMethods.IN_PERSON in self.supported_enrollment_methods: 246 ↛ 257line 246 didn't jump to line 257 because the condition on line 246 was always true
247 try:
248 in_person_eligibility_context = self.in_person_eligibility_context
249 except KeyError:
250 in_person_eligibility_context = None
252 if not in_person_eligibility_context:
253 errors.append(
254 ValidationError(f"{self.system_name} not configured for In-person. Please uncheck to continue.")
255 )
257 if errors:
258 raise ValidationError(errors)
260 def eligibility_form_instance(self, *args, **kwargs):
261 """Return an instance of this flow's EligibilityForm, or None."""
262 if not self.uses_api_verification:
263 return None
265 # inspired by https://stackoverflow.com/a/30941292
266 module_name, class_name = self.eligibility_form_class.rsplit(".", 1)
267 FormClass = getattr(importlib.import_module(module_name), class_name)
269 return FormClass(*args, **kwargs)
271 @staticmethod
272 def by_id(id):
273 """Get an EnrollmentFlow instance by its ID."""
274 logger.debug(f"Get {EnrollmentFlow.__name__} by id: {id}")
275 return EnrollmentFlow.objects.get(id=id)
278class EnrollmentEvent(models.Model):
279 """A record of a successful enrollment."""
281 id = models.UUIDField(primary_key=True, default=uuid.uuid4)
282 transit_agency = models.ForeignKey(TransitAgency, on_delete=models.PROTECT)
283 enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
284 enrollment_method = models.TextField(
285 choices={
286 EnrollmentMethods.DIGITAL: EnrollmentMethods.DIGITAL,
287 EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
288 }
289 )
290 verified_by = models.TextField()
291 enrollment_datetime = models.DateTimeField(default=timezone.now)
292 expiration_datetime = models.DateTimeField(blank=True, null=True)
293 extra_claims = models.TextField(blank=True, default="")
295 def __str__(self):
296 dt = timezone.localtime(self.enrollment_datetime)
297 ts = dt.strftime("%b %d, %Y, %I:%M %p")
298 return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"