Coverage for benefits/core/models/transit.py: 100%
144 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-17 22:53 +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, size):
34 base, ext = os.path.splitext(filename)
35 return f"agencies/{instance.slug}-{size}" + ext
38def agency_logo_small(instance, filename):
39 return _agency_logo(instance, filename, "sm")
42def agency_logo_large(instance, filename):
43 return _agency_logo(instance, filename, "lg")
46class TransitProcessorConfig(models.Model):
47 id = models.AutoField(primary_key=True)
48 environment = models.TextField(
49 choices=Environment,
50 help_text="A label to indicate which environment this configuration is for.",
51 )
52 transit_agency = models.OneToOneField(
53 "TransitAgency",
54 on_delete=models.PROTECT,
55 null=True,
56 blank=True,
57 default=None,
58 help_text="The transit agency that uses this configuration.",
59 )
60 portal_url = models.TextField(
61 default="",
62 blank=True,
63 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
64 )
66 def __str__(self):
67 environment_label = Environment(self.environment).label if self.environment else "unknown"
68 agency_slug = self.transit_agency.slug if self.transit_agency else "(no agency)"
69 return f"({environment_label}) {agency_slug}"
72class TransitAgency(models.Model):
73 """An agency offering transit service."""
75 class Meta:
76 verbose_name_plural = "transit agencies"
78 id = models.AutoField(primary_key=True)
79 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
80 slug = models.SlugField(
81 choices=core_context.AgencySlug,
82 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
83 )
84 short_name = models.TextField(
85 default="", blank=True, help_text="The user-facing short name for this agency. Often an uppercase acronym."
86 )
87 long_name = models.TextField(
88 default="",
89 blank=True,
90 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
91 )
92 info_url = models.URLField(
93 default="",
94 blank=True,
95 help_text="URL of a website/page with more information about the agency's discounts",
96 )
97 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
98 supported_card_schemes = MultiSelectField(
99 choices=CardSchemes.CHOICES,
100 min_choices=1,
101 max_choices=len(CardSchemes.CHOICES),
102 default=[CardSchemes.VISA, CardSchemes.MASTERCARD],
103 help_text="The contactless card schemes this agency supports.",
104 )
105 eligibility_api_id = models.TextField(
106 help_text="The identifier for this agency used in Eligibility API calls.",
107 blank=True,
108 default="",
109 )
110 eligibility_api_private_key = models.ForeignKey(
111 PemData,
112 related_name="+",
113 on_delete=models.PROTECT,
114 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
115 null=True,
116 blank=True,
117 default=None,
118 )
119 eligibility_api_public_key = models.ForeignKey(
120 PemData,
121 related_name="+",
122 on_delete=models.PROTECT,
123 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
124 null=True,
125 blank=True,
126 default=None,
127 )
128 staff_group = models.OneToOneField(
129 Group,
130 on_delete=models.PROTECT,
131 null=True,
132 blank=True,
133 default=None,
134 help_text="The group of users associated with this TransitAgency.",
135 related_name="transit_agency",
136 )
137 sso_domain = models.TextField(
138 blank=True,
139 default="",
140 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
141 )
142 customer_service_group = models.OneToOneField(
143 Group,
144 on_delete=models.PROTECT,
145 null=True,
146 blank=True,
147 default=None,
148 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
149 related_name="+",
150 )
151 logo_large = models.ImageField(
152 default="",
153 blank=True,
154 upload_to=agency_logo_large,
155 help_text="The large version of the transit agency's logo.",
156 )
157 logo_small = models.ImageField(
158 default="",
159 blank=True,
160 upload_to=agency_logo_small,
161 help_text="The small version of the transit agency's logo.",
162 )
164 def __str__(self):
165 return self.long_name
167 @property
168 def index_context(self):
169 return core_context.agency_index[self.slug].dict()
171 @property
172 def index_url(self):
173 """Public-facing URL to the TransitAgency's landing page."""
174 return reverse(routes.AGENCY_INDEX, args=[self.slug])
176 @property
177 def eligibility_index_url(self):
178 """Public facing URL to the TransitAgency's eligibility page."""
179 return reverse(routes.AGENCY_ELIGIBILITY_INDEX, args=[self.slug])
181 @property
182 def eligibility_api_private_key_data(self):
183 """This Agency's private key as a string."""
184 return self.eligibility_api_private_key.data
186 @property
187 def eligibility_api_public_key_data(self):
188 """This Agency's public key as a string."""
189 return self.eligibility_api_public_key.data
191 @property
192 def littlepay_config(self):
193 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "littlepayconfig"):
194 return self.transitprocessorconfig.littlepayconfig
195 else:
196 return None
198 @property
199 def switchio_config(self):
200 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "switchioconfig"):
201 return self.transitprocessorconfig.switchioconfig
202 else:
203 return None
205 @property
206 def in_person_enrollment_index_route(self):
207 """This Agency's in-person enrollment index route, based on its configured transit processor."""
208 if self.littlepay_config:
209 return routes.IN_PERSON_ENROLLMENT_LITTLEPAY_INDEX
210 elif self.switchio_config:
211 return routes.IN_PERSON_ENROLLMENT_SWITCHIO_INDEX
212 else:
213 raise ValueError(
214 (
215 "TransitAgency must have either a LittlepayConfig or SwitchioConfig "
216 "in order to show in-person enrollment index."
217 )
218 )
220 @property
221 def enrollment_index_route(self):
222 """This Agency's enrollment index route, based on its configured transit processor."""
223 if self.littlepay_config:
224 return routes.ENROLLMENT_LITTLEPAY_INDEX
225 elif self.switchio_config:
226 return routes.ENROLLMENT_SWITCHIO_INDEX
227 else:
228 raise ValueError(
229 "TransitAgency must have either a LittlepayConfig or SwitchioConfig in order to show enrollment index."
230 )
232 @property
233 def enrollment_flows(self):
234 return self.enrollmentflow_set
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 short_name=self.short_name,
244 long_name=self.long_name,
245 phone=self.phone,
246 info_url=self.info_url,
247 logo_large=self.logo_large,
248 logo_small=self.logo_small,
249 )
250 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
252 if self.littlepay_config is None and self.switchio_config is None:
253 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
254 else:
255 if self.littlepay_config:
256 try:
257 self.littlepay_config.clean()
258 except ValidationError as e:
259 message = "Littlepay configuration is missing fields that are required when this agency is active."
260 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
261 non_field_errors.append(ValidationError(message))
263 if self.switchio_config:
264 try:
265 self.switchio_config.clean()
266 except ValidationError as e:
267 message = "Switchio configuration is missing fields that are required when this agency is active."
268 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
269 non_field_errors.append(ValidationError(message))
271 all_errors = {}
272 if field_errors:
273 all_errors.update(field_errors)
274 if non_field_errors:
275 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
276 if all_errors:
277 raise ValidationError(all_errors)
279 @staticmethod
280 def by_id(id):
281 """Get a TransitAgency instance by its ID."""
282 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
283 return TransitAgency.objects.get(id=id)
285 @staticmethod
286 def by_slug(slug):
287 """Get a TransitAgency instance by its slug."""
288 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
289 return TransitAgency.objects.filter(slug=slug).first()
291 @staticmethod
292 def all_active():
293 """Get all TransitAgency instances marked active."""
294 logger.debug(f"Get all active {TransitAgency.__name__}")
295 return TransitAgency.objects.filter(active=True)
297 @staticmethod
298 def for_user(user: User):
299 for group in user.groups.all():
300 if hasattr(group, "transit_agency"):
301 return group.transit_agency # this is looking at the TransitAgency's staff_group
303 # the loop above returns the first match found. Return None if no match was found.
304 return None