Coverage for benefits / core / models / transit.py: 99%
155 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 15:39 +0000
1import logging
2import os
4from django.contrib.auth.models import Group, User
5from django.core.exceptions import NON_FIELD_ERRORS, ValidationError
6from django.db import models
7from django.urls import reverse
8from multiselectfield import MultiSelectField
10from benefits.routes import routes
12from .common import Environment
13from .enrollment import EnrollmentFlow
15logger = logging.getLogger(__name__)
18class CardSchemes:
19 VISA = "visa"
20 MASTERCARD = "mastercard"
21 DISCOVER = "discover"
22 AMEX = "amex"
24 CHOICES = dict(
25 [
26 (VISA, "Visa"),
27 (MASTERCARD, "Mastercard"),
28 (DISCOVER, "Discover"),
29 (AMEX, "American Express"),
30 ]
31 )
34def agency_logo(instance, filename):
35 base, ext = os.path.splitext(filename)
36 return f"agencies/{instance.slug}" + ext
39class TransitProcessorConfig(models.Model):
40 id = models.AutoField(primary_key=True)
41 environment = models.TextField(
42 choices=Environment,
43 help_text="A label to indicate which environment this configuration is for.",
44 )
45 label = models.TextField(
46 default="",
47 blank=True,
48 help_text="A label for internal use.",
49 )
50 portal_url = models.TextField(
51 default="",
52 blank=True,
53 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
54 )
56 def __str__(self):
57 environment_label = Environment(self.environment).label if self.environment else "unknown"
58 return f"({environment_label}) {self.label}"
61class TransitAgency(models.Model):
62 """An agency offering transit service."""
64 class Meta:
65 verbose_name_plural = "transit agencies"
67 id = models.AutoField(primary_key=True)
68 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
69 slug = models.SlugField(
70 unique=True,
71 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
72 )
73 short_name = models.TextField(
74 default="", help_text="The user-facing short name for this agency. Often an uppercase acronym."
75 )
76 long_name = models.TextField(
77 default="",
78 blank=True,
79 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
80 )
81 info_url = models.URLField(
82 default="",
83 blank=True,
84 help_text="URL of a website/page with more information about the agency's discounts",
85 )
86 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
87 enrollment_flows = models.ManyToManyField(
88 EnrollmentFlow,
89 help_text="Select the enrollment flows this agency supports.",
90 )
91 supported_card_schemes = MultiSelectField(
92 choices=CardSchemes.CHOICES,
93 min_choices=1,
94 max_choices=len(CardSchemes.CHOICES),
95 default=[CardSchemes.VISA, CardSchemes.MASTERCARD],
96 help_text="The contactless card schemes this agency supports.",
97 )
98 sso_domain = models.TextField(
99 blank=True,
100 default="",
101 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
102 )
103 customer_service_group = models.OneToOneField(
104 Group,
105 on_delete=models.PROTECT,
106 null=True,
107 blank=True,
108 default=None,
109 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
110 related_name="transit_agency",
111 )
112 logo = models.ImageField(
113 default="",
114 blank=True,
115 upload_to=agency_logo,
116 help_text="The transit agency's logo.",
117 )
118 transit_processor_config = models.ForeignKey(
119 TransitProcessorConfig,
120 on_delete=models.PROTECT,
121 null=True,
122 blank=True,
123 default=None,
124 help_text="The transit processor configuration to use for enrollment.",
125 )
127 def __str__(self):
128 if self.long_name:
129 return self.long_name
130 return self.short_name
132 @property
133 def index_url(self):
134 """Public-facing URL to the TransitAgency's landing page."""
135 return reverse(routes.AGENCY_INDEX, args=[self.slug])
137 @property
138 def entrypoint_url(self):
139 """For grouped agencies, we display an interstitial view prior to commencing the eligibility check."""
140 if self.group_agencies():
141 return reverse(routes.ADDITIONAL_AGENCIES)
143 return reverse(routes.ELIGIBILITY_INDEX)
145 @property
146 def littlepay_config(self):
147 if self.transit_processor_config and hasattr(self.transit_processor_config, "littlepayconfig"):
148 return self.transit_processor_config.littlepayconfig
149 else:
150 return None
152 @property
153 def switchio_config(self):
154 if hasattr(self, "transit_processor_config") and hasattr(self.transit_processor_config, "switchioconfig"):
155 return self.transit_processor_config.switchioconfig
156 else:
157 return None
159 @property
160 def transit_processor(self):
161 if self.littlepay_config:
162 return "littlepay"
163 elif self.switchio_config:
164 return "switchio"
165 else:
166 return None
168 @property
169 def in_person_enrollment_index_route(self):
170 """This Agency's in-person enrollment index route, based on its configured transit processor."""
171 if self.littlepay_config:
172 return routes.IN_PERSON_ENROLLMENT_LITTLEPAY_INDEX
173 elif self.switchio_config:
174 return routes.IN_PERSON_ENROLLMENT_SWITCHIO_INDEX
175 else:
176 raise ValueError(
177 (
178 "TransitAgency must have either a LittlepayConfig or SwitchioConfig "
179 "in order to show in-person enrollment index."
180 )
181 )
183 @property
184 def enrollment_index_route(self):
185 """This Agency's enrollment index route, based on its configured transit processor."""
186 if self.littlepay_config:
187 return routes.ENROLLMENT_LITTLEPAY_INDEX
188 elif self.switchio_config:
189 return routes.ENROLLMENT_SWITCHIO_INDEX
190 else:
191 raise ValueError(
192 "TransitAgency must have either a LittlepayConfig or SwitchioConfig in order to show enrollment index."
193 )
195 @property
196 def customer_service_group_name(self):
197 """Returns the standardized name for this Agency's customer service group."""
198 return f"{self.short_name} Customer Service"
200 def group_agencies(self):
201 """The set of all agencies in all groups associated with this agency, excluding itself.
203 If an agency is not associated with any other agencies via TransitAgencyGroup,
204 this returns an empty list.
205 """
206 return list(
207 TransitAgency.objects.filter(transitagencygroup__in=list(self.transitagencygroup_set.all()))
208 .distinct()
209 .exclude(active=False)
210 .exclude(pk=self.pk)
211 .order_by("short_name")
212 )
214 def group_agency_short_names(self):
215 """A list of agency short names for this agency and any agencies it shares a group with.
217 The list begins with the current agency and the rest follow in alphabetical order.
218 If an agency is not associated with any other agencies via TransitAgencyGroup,
219 this returns an empty list.
220 """
221 agencies = [self] + self.group_agencies()
223 if len(agencies) > 1:
224 return [agency.short_name for agency in agencies]
225 else:
226 return []
228 def clean(self):
229 field_errors = {}
230 non_field_errors = []
232 if self.active:
233 message = "This field is required for active transit agencies."
234 needed = dict(
235 long_name=self.long_name,
236 phone=self.phone,
237 info_url=self.info_url,
238 logo=self.logo,
239 )
240 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
242 if self.littlepay_config is None and self.switchio_config is None:
243 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
244 else:
245 if self.littlepay_config:
246 try:
247 self.littlepay_config.clean()
248 except ValidationError as e:
249 message = "Littlepay configuration is missing fields that are required when this agency is active."
250 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
251 non_field_errors.append(ValidationError(message))
253 if self.switchio_config:
254 try:
255 self.switchio_config.clean()
256 except ValidationError as e:
257 message = "Switchio configuration is missing fields that are required when this agency is active."
258 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
259 non_field_errors.append(ValidationError(message))
261 if self.pk: # prohibit updating short_name with blank customer_service_group 261 ↛ 272line 261 didn't jump to line 272 because the condition on line 261 was always true
262 original_obj = TransitAgency.objects.get(pk=self.pk)
263 if self.short_name != original_obj.short_name and not self.customer_service_group:
264 field_errors.update(
265 {
266 "customer_service_group": ValidationError(
267 "Blank not allowed. Set to its original value if changing the Short Name."
268 )
269 }
270 )
272 all_errors = {}
273 if field_errors:
274 all_errors.update(field_errors)
275 if non_field_errors:
276 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
277 if all_errors:
278 raise ValidationError(all_errors)
280 @staticmethod
281 def by_id(id):
282 """Get a TransitAgency instance by its ID."""
283 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
284 return TransitAgency.objects.get(id=id)
286 @staticmethod
287 def by_slug(slug):
288 """Get a TransitAgency instance by its slug."""
289 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
290 return TransitAgency.objects.filter(slug=slug).first()
292 @staticmethod
293 def all_active():
294 """Get all TransitAgency instances marked active."""
295 logger.debug(f"Get all active {TransitAgency.__name__}")
296 return TransitAgency.objects.filter(active=True).order_by("long_name")
298 @staticmethod
299 def for_user(user: User):
300 for group in user.groups.all():
301 if hasattr(group, "transit_agency"):
302 return group.transit_agency # this is looking at the TransitAgency's customer_service_group
304 # the loop above returns the first match found. Return None if no match was found.
305 return None
308class TransitAgencyGroup(models.Model):
309 id = models.AutoField(primary_key=True)
310 label = models.TextField(
311 help_text="A human readable label, used as the display text in Admin.",
312 )
313 transit_agencies = models.ManyToManyField(
314 TransitAgency,
315 help_text="Select the agencies that belong to this group.",
316 )
318 def __str__(self):
319 return self.label