Coverage for benefits/core/models/transit.py: 100%
146 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-31 18:44 +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 TransitAgency(models.Model):
65 """An agency offering transit service."""
67 class Meta:
68 verbose_name_plural = "transit agencies"
70 id = models.AutoField(primary_key=True)
71 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
72 slug = models.SlugField(
73 choices=core_context.AgencySlug,
74 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
75 )
76 short_name = models.TextField(
77 default="", blank=True, help_text="The user-facing short name for this agency. Often an uppercase acronym."
78 )
79 long_name = models.TextField(
80 default="",
81 blank=True,
82 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
83 )
84 info_url = models.URLField(
85 default="",
86 blank=True,
87 help_text="URL of a website/page with more information about the agency's discounts",
88 )
89 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
90 supported_card_schemes = MultiSelectField(
91 choices=CardSchemes.CHOICES,
92 min_choices=1,
93 max_choices=len(CardSchemes.CHOICES),
94 default=[CardSchemes.VISA, CardSchemes.MASTERCARD],
95 help_text="The contactless card schemes this agency supports.",
96 )
97 eligibility_api_id = models.TextField(
98 help_text="The identifier for this agency used in Eligibility API calls.",
99 blank=True,
100 default="",
101 )
102 eligibility_api_private_key = models.ForeignKey(
103 PemData,
104 related_name="+",
105 on_delete=models.PROTECT,
106 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
107 null=True,
108 blank=True,
109 default=None,
110 )
111 eligibility_api_public_key = models.ForeignKey(
112 PemData,
113 related_name="+",
114 on_delete=models.PROTECT,
115 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
116 null=True,
117 blank=True,
118 default=None,
119 )
120 staff_group = models.OneToOneField(
121 Group,
122 on_delete=models.PROTECT,
123 null=True,
124 blank=True,
125 default=None,
126 help_text="The group of users associated with this TransitAgency.",
127 related_name="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="+",
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 return self.eligibility_api_private_key.data
172 @property
173 def eligibility_api_public_key_data(self):
174 """This Agency's public key as a string."""
175 return self.eligibility_api_public_key.data
177 @property
178 def littlepay_config(self):
179 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "littlepayconfig"):
180 return self.transitprocessorconfig.littlepayconfig
181 else:
182 return None
184 @property
185 def switchio_config(self):
186 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "switchioconfig"):
187 return self.transitprocessorconfig.switchioconfig
188 else:
189 return None
191 @property
192 def transit_processor(self):
193 if self.littlepay_config:
194 return "littlepay"
195 elif self.switchio_config:
196 return "switchio"
197 else:
198 return None
200 @property
201 def in_person_enrollment_index_route(self):
202 """This Agency's in-person enrollment index route, based on its configured transit processor."""
203 if self.littlepay_config:
204 return routes.IN_PERSON_ENROLLMENT_LITTLEPAY_INDEX
205 elif self.switchio_config:
206 return routes.IN_PERSON_ENROLLMENT_SWITCHIO_INDEX
207 else:
208 raise ValueError(
209 (
210 "TransitAgency must have either a LittlepayConfig or SwitchioConfig "
211 "in order to show in-person enrollment index."
212 )
213 )
215 @property
216 def enrollment_index_route(self):
217 """This Agency's enrollment index route, based on its configured transit processor."""
218 if self.littlepay_config:
219 return routes.ENROLLMENT_LITTLEPAY_INDEX
220 elif self.switchio_config:
221 return routes.ENROLLMENT_SWITCHIO_INDEX
222 else:
223 raise ValueError(
224 "TransitAgency must have either a LittlepayConfig or SwitchioConfig in order to show enrollment index."
225 )
227 @property
228 def enrollment_flows(self):
229 return self.enrollmentflow_set
231 def clean(self):
232 field_errors = {}
233 non_field_errors = []
235 if self.active:
236 message = "This field is required for active transit agencies."
237 needed = dict(
238 short_name=self.short_name,
239 long_name=self.long_name,
240 phone=self.phone,
241 info_url=self.info_url,
242 logo=self.logo,
243 )
244 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
246 if self.littlepay_config is None and self.switchio_config is None:
247 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
248 else:
249 if self.littlepay_config:
250 try:
251 self.littlepay_config.clean()
252 except ValidationError as e:
253 message = "Littlepay configuration is missing fields that are required when this agency is active."
254 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
255 non_field_errors.append(ValidationError(message))
257 if self.switchio_config:
258 try:
259 self.switchio_config.clean()
260 except ValidationError as e:
261 message = "Switchio 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 all_errors = {}
266 if field_errors:
267 all_errors.update(field_errors)
268 if non_field_errors:
269 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
270 if all_errors:
271 raise ValidationError(all_errors)
273 @staticmethod
274 def by_id(id):
275 """Get a TransitAgency instance by its ID."""
276 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
277 return TransitAgency.objects.get(id=id)
279 @staticmethod
280 def by_slug(slug):
281 """Get a TransitAgency instance by its slug."""
282 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
283 return TransitAgency.objects.filter(slug=slug).first()
285 @staticmethod
286 def all_active():
287 """Get all TransitAgency instances marked active."""
288 logger.debug(f"Get all active {TransitAgency.__name__}")
289 return TransitAgency.objects.filter(active=True).order_by("long_name")
291 @staticmethod
292 def for_user(user: User):
293 for group in user.groups.all():
294 if hasattr(group, "transit_agency"):
295 return group.transit_agency # this is looking at the TransitAgency's staff_group
297 # the loop above returns the first match found. Return None if no match was found.
298 return None