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