Coverage for benefits/core/models/enrollment.py: 96%
135 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 00:01 +0000
1import logging
2import uuid
4from cdt_identity.models import ClaimsVerificationRequest, IdentityGatewayConfig
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
12logger = logging.getLogger(__name__)
15class EnrollmentMethods:
16 SELF_SERVICE = "self_service"
17 IN_PERSON = "in_person"
20SUPPORTED_METHODS = (
21 (EnrollmentMethods.SELF_SERVICE, EnrollmentMethods.SELF_SERVICE.replace("_", "-").capitalize()),
22 (EnrollmentMethods.IN_PERSON, EnrollmentMethods.IN_PERSON.replace("_", "-").capitalize()),
23)
26class SystemName(models.TextChoices):
27 CALFRESH = "calfresh"
28 COURTESY_CARD = "courtesy_card"
29 MEDICARE = "medicare"
30 OLDER_ADULT = "senior"
31 REDUCED_FARE_MOBILITY_ID = "mobility_pass"
32 VETERAN = "veteran"
33 GCTD_CARD = "gctd_card"
36class EligibilityApiVerificationRequest(models.Model):
37 """Represents configuration for eligibility verification via Eligibility API calls."""
39 id = models.AutoField(primary_key=True)
40 label = models.SlugField(
41 help_text="A human readable label, used as the display text in Admin.",
42 )
43 api_url = models.URLField(help_text="Fully qualified URL for an Eligibility API server.")
44 api_auth_header = models.CharField(
45 help_text="The auth header to send in Eligibility API requests.",
46 max_length=50,
47 )
48 api_auth_key_secret_name = SecretNameField(
49 help_text="The name of a secret containing the value of the auth header to send in Eligibility API requests.",
50 )
51 client_private_key = models.ForeignKey(
52 PemData,
53 related_name="+",
54 on_delete=models.PROTECT,
55 default=None,
56 null=True,
57 help_text="Private key used to sign Eligibility API tokens created on behalf of the Benefits client.",
58 )
59 client_public_key = models.ForeignKey(
60 PemData,
61 related_name="+",
62 on_delete=models.PROTECT,
63 default=None,
64 null=True,
65 help_text="Public key corresponding to the Benefits client's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
66 )
67 api_public_key = models.ForeignKey(
68 PemData,
69 related_name="+",
70 on_delete=models.PROTECT,
71 help_text="The public key used to encrypt Eligibility API requests and to verify signed Eligibility API responses.",
72 )
73 api_jwe_cek_enc = models.CharField(
74 help_text="The JWE-compatible Content Encryption Key (CEK) key-length and mode to use in Eligibility API requests.",
75 max_length=50,
76 )
77 api_jwe_encryption_alg = models.CharField(
78 help_text="The JWE-compatible encryption algorithm to use in Eligibility API requests.",
79 max_length=50,
80 )
81 api_jws_signing_alg = models.CharField(
82 help_text="The JWS-compatible signing algorithm to use in Eligibility API requests.",
83 max_length=50,
84 )
86 def __str__(self):
87 return self.label
89 @property
90 def api_auth_key(self):
91 """The Eligibility API auth key as a string."""
92 secret_field = self._meta.get_field("api_auth_key_secret_name")
93 return secret_field.secret_value(self)
95 @property
96 def client_private_key_data(self):
97 """The private key used to sign Eligibility API tokens created by the Benefits client as a string."""
98 return self.client_private_key.data
100 @property
101 def client_public_key_data(self):
102 """The public key corresponding to the Benefits client's private key as a string."""
103 return self.client_public_key.data
105 @property
106 def api_public_key_data(self):
107 """The Eligibility API public key as a string."""
108 return self.api_public_key.data
111SUPPORTED_IN_PERSON_FLOWS = (
112 SystemName.COURTESY_CARD,
113 SystemName.MEDICARE,
114 SystemName.OLDER_ADULT,
115 SystemName.REDUCED_FARE_MOBILITY_ID,
116 SystemName.GCTD_CARD,
117)
120class EnrollmentFlow(models.Model):
121 """Represents a user journey through the Benefits app for a single eligibility type."""
123 id = models.AutoField(primary_key=True)
124 system_name = models.SlugField(
125 choices=SystemName,
126 help_text="Primary internal system name for this EnrollmentFlow instance, e.g. in analytics and Eligibility API requests.", # noqa: 501
127 )
128 label = models.TextField(
129 blank=True,
130 default="",
131 help_text="A human readable label, used as the display text in Admin.",
132 )
133 supported_enrollment_methods = MultiSelectField(
134 choices=SUPPORTED_METHODS,
135 max_choices=2,
136 max_length=50,
137 default=[EnrollmentMethods.SELF_SERVICE, EnrollmentMethods.IN_PERSON],
138 help_text="If the flow is supported by self-service enrollment, in-person enrollment, or both",
139 )
140 sign_out_button_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out button")
141 sign_out_link_template = models.TextField(default="", blank=True, help_text="Template that renders sign-out link")
142 oauth_config = models.ForeignKey(
143 IdentityGatewayConfig,
144 on_delete=models.PROTECT,
145 null=True,
146 blank=True,
147 help_text="The IdG connection details for this flow.",
148 )
149 claims_request = models.ForeignKey(
150 ClaimsVerificationRequest,
151 on_delete=models.PROTECT,
152 null=True,
153 blank=True,
154 help_text="The claims request details for this flow.",
155 )
156 api_request = models.ForeignKey(
157 EligibilityApiVerificationRequest,
158 on_delete=models.PROTECT,
159 null=True,
160 blank=True,
161 help_text="The Eligibility API request details for this flow.",
162 )
163 supports_expiration = models.BooleanField(
164 default=False, help_text="Indicates if the enrollment expires or does not expire"
165 )
166 expiration_days = models.PositiveSmallIntegerField(
167 null=True, blank=True, help_text="If the enrollment supports expiration, number of days before the eligibility expires"
168 )
169 expiration_reenrollment_days = models.PositiveSmallIntegerField(
170 null=True,
171 blank=True,
172 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
173 )
174 display_order = models.PositiveSmallIntegerField(default=0, blank=False, null=False)
176 class Meta:
177 ordering = ["display_order"]
179 def __str__(self):
180 return self.label
182 @property
183 def eligibility_api_auth_key(self):
184 if self.uses_api_verification: 184 ↛ 187line 184 didn't jump to line 187 because the condition on line 184 was always true
185 return self.api_request.api_auth_key
186 else:
187 return None
189 @property
190 def eligibility_api_public_key_data(self):
191 """This flow's Eligibility API public key as a string."""
192 if self.uses_api_verification: 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was always true
193 return self.api_request.api_public_key_data
194 else:
195 return None
197 @property
198 def selection_label_template(self):
199 return f"eligibility/includes/selection-label--{self.system_name}.html"
201 @property
202 def uses_claims_verification(self):
203 """True if this flow verifies via the Identity Gateway and has a scope and claim. False otherwise."""
204 return (
205 self.oauth_config is not None and bool(self.claims_request.scopes) and bool(self.claims_request.eligibility_claim)
206 )
208 @property
209 def uses_api_verification(self):
210 """True if this flow verifies via the Eligibility API. False otherwise."""
211 return self.api_request is not None
213 @property
214 def claims_scheme(self):
215 if self.uses_claims_verification: 215 ↛ 218line 215 didn't jump to line 218 because the condition on line 215 was always true
216 return self.claims_request.scheme or self.oauth_config.scheme
217 else:
218 return None
220 @property
221 def eligibility_verifier(self):
222 """A str representing the entity that verifies eligibility for this flow.
224 Either the client name of the flow's claims provider, or the URL to the eligibility API.
225 """
226 if self.uses_claims_verification:
227 return self.oauth_config.client_name
228 elif self.uses_api_verification:
229 return self.api_request.api_url
230 else:
231 return "undefined"
233 @property
234 def supports_sign_out(self):
235 return bool(self.sign_out_button_template) or bool(self.sign_out_link_template)
237 # until we can make time to consolidate, additional validation logic can be found in EnrollmentFlow.clean()
238 # see https://cal-itp.slack.com/archives/C037Y3UE71P/p1779234784673319?thread_ts=1779231543.096499&cid=C037Y3UE71P
239 def clean(self):
240 errors = []
242 supports_self_service = EnrollmentMethods.SELF_SERVICE in self.supported_enrollment_methods
243 supports_in_person = EnrollmentMethods.IN_PERSON in self.supported_enrollment_methods
244 t = self.selection_label_template
246 if supports_self_service and not template_path(t):
247 # we can't add a field-level validation error
248 # because the actual template for the self-service flow is derived from a pattern
249 errors.append(ValidationError(f"Template not found: {t}"))
251 if supports_in_person and self.system_name not in SUPPORTED_IN_PERSON_FLOWS:
252 errors.append(
253 ValidationError(f"{self.system_name} not configured for in-person enrollment. Please uncheck to continue.")
254 )
256 if errors:
257 raise ValidationError(errors)
259 @staticmethod
260 def by_id(id):
261 """Get an EnrollmentFlow instance by its ID."""
262 logger.debug(f"Get {EnrollmentFlow.__name__} by id: {id}")
263 return EnrollmentFlow.objects.get(id=id)
266class EnrollmentGroup(models.Model):
267 id = models.AutoField(primary_key=True)
268 transit_agency = models.ForeignKey(
269 "core.TransitAgency",
270 on_delete=models.PROTECT,
271 help_text="The transit agency that this group is for.",
272 )
273 enrollment_flow = models.ForeignKey(
274 EnrollmentFlow,
275 on_delete=models.PROTECT,
276 help_text="The enrollment flow that this group is for.",
277 )
279 def __str__(self):
280 return f"{self.enrollment_flow} ({self.transit_agency.slug})"
283class EnrollmentEvent(models.Model):
284 """A record of a successful enrollment."""
286 id = models.UUIDField(primary_key=True, default=uuid.uuid4)
287 transit_agency = models.ForeignKey("core.TransitAgency", on_delete=models.PROTECT)
288 enrollment_flow = models.ForeignKey(EnrollmentFlow, on_delete=models.PROTECT)
289 enrollment_method = models.TextField(
290 choices={
291 EnrollmentMethods.SELF_SERVICE: EnrollmentMethods.SELF_SERVICE,
292 EnrollmentMethods.IN_PERSON: EnrollmentMethods.IN_PERSON,
293 }
294 )
295 verified_by = models.TextField()
296 enrollment_datetime = models.DateTimeField(default=timezone.now)
297 expiration_datetime = models.DateTimeField(blank=True, null=True)
298 extra_claims = models.TextField(blank=True, default="")
300 def __str__(self):
301 dt = timezone.localtime(self.enrollment_datetime)
302 ts = dt.strftime("%b %d, %Y, %I:%M %p")
303 return f"{ts}, {self.transit_agency}, {self.enrollment_flow}"