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