Coverage for benefits / core / models / transit.py: 97%
161 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-22 19:08 +0000
1import os
2import logging
4from django.core.exceptions import ValidationError, NON_FIELD_ERRORS
5from django.contrib.auth.models import Group, User
6from django.db import models
7from django.urls import reverse
8from multiselectfield import MultiSelectField
10from benefits.core import context as core_context
11from benefits.routes import routes
12from .common import Environment, PemData
14logger = logging.getLogger(__name__)
17class CardSchemes:
18 VISA = "visa"
19 MASTERCARD = "mastercard"
20 DISCOVER = "discover"
21 AMEX = "amex"
23 CHOICES = dict(
24 [
25 (VISA, "Visa"),
26 (MASTERCARD, "Mastercard"),
27 (DISCOVER, "Discover"),
28 (AMEX, "American Express"),
29 ]
30 )
33def agency_logo(instance, filename):
34 base, ext = os.path.splitext(filename)
35 return f"agencies/{instance.slug}" + ext
38class TransitProcessorConfig(models.Model):
39 id = models.AutoField(primary_key=True)
40 environment = models.TextField(
41 choices=Environment,
42 help_text="A label to indicate which environment this configuration is for.",
43 )
44 transit_agency = models.OneToOneField(
45 "TransitAgency",
46 on_delete=models.PROTECT,
47 null=True,
48 blank=True,
49 default=None,
50 help_text="The transit agency that uses this configuration.",
51 )
52 portal_url = models.TextField(
53 default="",
54 blank=True,
55 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
56 )
58 def __str__(self):
59 environment_label = Environment(self.environment).label if self.environment else "unknown"
60 agency_slug = self.transit_agency.slug if self.transit_agency else "(no agency)"
61 return f"({environment_label}) {agency_slug}"
64class EligibilityApiConfig(models.Model):
65 """Per-agency configuration for Eligibility Server integrations via the Eligibility API."""
67 id = models.AutoField(primary_key=True)
68 api_id = models.SlugField(
69 help_text="The identifier for this agency used in Eligibility API calls.",
70 )
71 api_private_key = models.ForeignKey(
72 PemData,
73 related_name="+",
74 on_delete=models.PROTECT,
75 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
76 )
77 api_public_key = models.ForeignKey(
78 PemData,
79 related_name="+",
80 on_delete=models.PROTECT,
81 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
82 )
84 def __str__(self):
85 return self.api_id
88class TransitAgency(models.Model):
89 """An agency offering transit service."""
91 class Meta:
92 verbose_name_plural = "transit agencies"
94 id = models.AutoField(primary_key=True)
95 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
96 slug = models.SlugField(
97 choices=core_context.AgencySlug,
98 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
99 )
100 short_name = models.TextField(
101 default="", help_text="The user-facing short name for this agency. Often an uppercase acronym."
102 )
103 long_name = models.TextField(
104 default="",
105 blank=True,
106 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
107 )
108 info_url = models.URLField(
109 default="",
110 blank=True,
111 help_text="URL of a website/page with more information about the agency's discounts",
112 )
113 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
114 supported_card_schemes = MultiSelectField(
115 choices=CardSchemes.CHOICES,
116 min_choices=1,
117 max_choices=len(CardSchemes.CHOICES),
118 default=[CardSchemes.VISA, CardSchemes.MASTERCARD],
119 help_text="The contactless card schemes this agency supports.",
120 )
121 eligibility_api_config = models.ForeignKey(
122 EligibilityApiConfig,
123 on_delete=models.PROTECT,
124 null=True,
125 blank=True,
126 default=None,
127 help_text="The Eligibility API configuration for this transit agency.",
128 )
129 sso_domain = models.TextField(
130 blank=True,
131 default="",
132 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
133 )
134 customer_service_group = models.OneToOneField(
135 Group,
136 on_delete=models.PROTECT,
137 null=True,
138 blank=True,
139 default=None,
140 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
141 related_name="transit_agency",
142 )
143 logo = models.ImageField(
144 default="",
145 blank=True,
146 upload_to=agency_logo,
147 help_text="The transit agency's logo.",
148 )
150 def __str__(self):
151 return self.long_name
153 @property
154 def index_context(self):
155 return core_context.agency_index[self.slug].dict()
157 @property
158 def index_url(self):
159 """Public-facing URL to the TransitAgency's landing page."""
160 return reverse(routes.AGENCY_INDEX, args=[self.slug])
162 @property
163 def eligibility_index_url(self):
164 """Public facing URL to the TransitAgency's eligibility page."""
165 return reverse(routes.AGENCY_ELIGIBILITY_INDEX, args=[self.slug])
167 @property
168 def eligibility_api_private_key_data(self):
169 """This Agency's private key as a string."""
170 if self.eligibility_api_config: 170 ↛ 172line 170 didn't jump to line 172 because the condition on line 170 was always true
171 return self.eligibility_api_config.api_private_key.data
172 return None
174 @property
175 def eligibility_api_public_key_data(self):
176 """This Agency's public key as a string."""
177 if self.eligibility_api_config: 177 ↛ 179line 177 didn't jump to line 179 because the condition on line 177 was always true
178 return self.eligibility_api_config.api_public_key.data
179 return None
181 @property
182 def littlepay_config(self):
183 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "littlepayconfig"):
184 return self.transitprocessorconfig.littlepayconfig
185 else:
186 return None
188 @property
189 def switchio_config(self):
190 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "switchioconfig"):
191 return self.transitprocessorconfig.switchioconfig
192 else:
193 return None
195 @property
196 def transit_processor(self):
197 if self.littlepay_config:
198 return "littlepay"
199 elif self.switchio_config:
200 return "switchio"
201 else:
202 return None
204 @property
205 def in_person_enrollment_index_route(self):
206 """This Agency's in-person enrollment index route, based on its configured transit processor."""
207 if self.littlepay_config:
208 return routes.IN_PERSON_ENROLLMENT_LITTLEPAY_INDEX
209 elif self.switchio_config:
210 return routes.IN_PERSON_ENROLLMENT_SWITCHIO_INDEX
211 else:
212 raise ValueError(
213 (
214 "TransitAgency must have either a LittlepayConfig or SwitchioConfig "
215 "in order to show in-person enrollment index."
216 )
217 )
219 @property
220 def enrollment_index_route(self):
221 """This Agency's enrollment index route, based on its configured transit processor."""
222 if self.littlepay_config:
223 return routes.ENROLLMENT_LITTLEPAY_INDEX
224 elif self.switchio_config:
225 return routes.ENROLLMENT_SWITCHIO_INDEX
226 else:
227 raise ValueError(
228 "TransitAgency must have either a LittlepayConfig or SwitchioConfig in order to show enrollment index."
229 )
231 @property
232 def enrollment_flows(self):
233 return self.enrollmentflow_set
235 @property
236 def customer_service_group_name(self):
237 """Returns the standardized name for this Agency's customer service group."""
238 return f"{self.short_name} Customer Service"
240 def clean(self):
241 field_errors = {}
242 non_field_errors = []
244 if self.active:
245 message = "This field is required for active transit agencies."
246 needed = dict(
247 long_name=self.long_name,
248 phone=self.phone,
249 info_url=self.info_url,
250 logo=self.logo,
251 )
252 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
254 if self.littlepay_config is None and self.switchio_config is None:
255 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
256 else:
257 if self.littlepay_config:
258 try:
259 self.littlepay_config.clean()
260 except ValidationError as e:
261 message = "Littlepay configuration is missing fields that are required when this agency is active."
262 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
263 non_field_errors.append(ValidationError(message))
265 if self.switchio_config:
266 try:
267 self.switchio_config.clean()
268 except ValidationError as e:
269 message = "Switchio configuration is missing fields that are required when this agency is active."
270 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
271 non_field_errors.append(ValidationError(message))
273 if self.pk: # prohibit updating short_name with blank customer_service_group 273 ↛ 284line 273 didn't jump to line 284 because the condition on line 273 was always true
274 original_obj = TransitAgency.objects.get(pk=self.pk)
275 if self.short_name != original_obj.short_name and not self.customer_service_group:
276 field_errors.update(
277 {
278 "customer_service_group": ValidationError(
279 "Blank not allowed. Set to its original value if changing the Short Name."
280 )
281 }
282 )
284 all_errors = {}
285 if field_errors:
286 all_errors.update(field_errors)
287 if non_field_errors:
288 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
289 if all_errors:
290 raise ValidationError(all_errors)
292 @staticmethod
293 def by_id(id):
294 """Get a TransitAgency instance by its ID."""
295 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
296 return TransitAgency.objects.get(id=id)
298 @staticmethod
299 def by_slug(slug):
300 """Get a TransitAgency instance by its slug."""
301 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
302 return TransitAgency.objects.filter(slug=slug).first()
304 @staticmethod
305 def all_active():
306 """Get all TransitAgency instances marked active."""
307 logger.debug(f"Get all active {TransitAgency.__name__}")
308 return TransitAgency.objects.filter(active=True).order_by("long_name")
310 @staticmethod
311 def for_user(user: User):
312 for group in user.groups.all():
313 if hasattr(group, "transit_agency"):
314 return group.transit_agency # this is looking at the TransitAgency's customer_service_group
316 # the loop above returns the first match found. Return None if no match was found.
317 return None