Coverage for benefits/core/models/transit.py: 99%
158 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 00:01 +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, only_active=True):
201 """The set of agencies in all groups associated with this agency, excluding itself.
202 If only_active is True, only active agencies are returned. If only_active is False,
203 all agencies are returned.
205 If an agency is not associated with any other agencies via TransitAgencyGroup,
206 this returns an empty list.
207 """
209 agencies_in_group = (
210 TransitAgency.objects.filter(transitagencygroup__in=list(self.transitagencygroup_set.all()))
211 .distinct()
212 .exclude(pk=self.pk)
213 )
215 if only_active:
216 agencies_in_group = agencies_in_group.exclude(active=False)
218 return list(agencies_in_group.order_by("short_name"))
220 def group_agency_short_names(self, only_active=True):
221 """A list of agency short names for this agency and any agencies it shares a group with.
222 If only_active is True, only active short names are returned. If only_active is False,
223 all short names are returned.
225 The list begins with the current agency and the rest follow in alphabetical order.
226 If an agency is not associated with any other agencies via TransitAgencyGroup,
227 this returns an empty list.
228 """
229 agencies = [self] + self.group_agencies(only_active=only_active)
231 if len(agencies) > 1:
232 return [agency.short_name for agency in agencies]
233 else:
234 return []
236 def clean(self):
237 field_errors = {}
238 non_field_errors = []
240 if self.active:
241 message = "This field is required for active transit agencies."
242 needed = dict(
243 long_name=self.long_name,
244 phone=self.phone,
245 info_url=self.info_url,
246 logo=self.logo,
247 )
248 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
250 if self.littlepay_config is None and self.switchio_config is None:
251 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
252 else:
253 if self.littlepay_config:
254 try:
255 self.littlepay_config.clean()
256 except ValidationError as e:
257 message = "Littlepay 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.switchio_config:
262 try:
263 self.switchio_config.clean()
264 except ValidationError as e:
265 message = "Switchio configuration is missing fields that are required when this agency is active."
266 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
267 non_field_errors.append(ValidationError(message))
269 if self.pk: # prohibit updating short_name with blank customer_service_group 269 ↛ 280line 269 didn't jump to line 280 because the condition on line 269 was always true
270 original_obj = TransitAgency.objects.get(pk=self.pk)
271 if self.short_name != original_obj.short_name and not self.customer_service_group:
272 field_errors.update(
273 {
274 "customer_service_group": ValidationError(
275 "Blank not allowed. Set to its original value if changing the Short Name."
276 )
277 }
278 )
280 all_errors = {}
281 if field_errors:
282 all_errors.update(field_errors)
283 if non_field_errors:
284 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
285 if all_errors:
286 raise ValidationError(all_errors)
288 @staticmethod
289 def by_id(id):
290 """Get a TransitAgency instance by its ID."""
291 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
292 return TransitAgency.objects.get(id=id)
294 @staticmethod
295 def by_slug(slug):
296 """Get a TransitAgency instance by its slug."""
297 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
298 return TransitAgency.objects.filter(slug=slug).first()
300 @staticmethod
301 def all_active():
302 """Get all TransitAgency instances marked active."""
303 logger.debug(f"Get all active {TransitAgency.__name__}")
304 return TransitAgency.objects.filter(active=True).order_by("long_name")
306 @staticmethod
307 def for_user(user: User):
308 for group in user.groups.all():
309 if hasattr(group, "transit_agency"):
310 return group.transit_agency # this is looking at the TransitAgency's customer_service_group
312 # the loop above returns the first match found. Return None if no match was found.
313 return None
316class TransitAgencyGroup(models.Model):
317 id = models.AutoField(primary_key=True)
318 label = models.TextField(
319 help_text="A human readable label, used as the display text in Admin.",
320 )
321 transit_agencies = models.ManyToManyField(
322 TransitAgency,
323 help_text="Select the agencies that belong to this group.",
324 )
326 def __str__(self):
327 return self.label