Coverage for benefits/core/models/transit.py: 99%
136 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-08 16:26 +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
9from benefits.core import context as core_context
10from benefits.routes import routes
11from .common import Environment, PemData
13logger = logging.getLogger(__name__)
16def _agency_logo(instance, filename, size):
17 base, ext = os.path.splitext(filename)
18 return f"agencies/{instance.slug}-{size}" + ext
21def agency_logo_small(instance, filename):
22 return _agency_logo(instance, filename, "sm")
25def agency_logo_large(instance, filename):
26 return _agency_logo(instance, filename, "lg")
29class TransitProcessor(models.Model):
30 """An entity that applies transit agency fare rules to rider transactions."""
32 id = models.AutoField(primary_key=True)
33 name = models.TextField(help_text="Primary internal display name for this TransitProcessor instance, e.g. in the Admin.")
34 portal_url = models.TextField(
35 default="",
36 blank=True,
37 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
38 )
40 def __str__(self):
41 return self.name
44class TransitProcessorConfig(models.Model):
45 id = models.AutoField(primary_key=True)
46 environment = models.TextField(
47 choices=Environment,
48 help_text="A label to indicate which environment this configuration is for.",
49 )
50 transit_agency = models.OneToOneField(
51 "TransitAgency",
52 on_delete=models.PROTECT,
53 null=True,
54 blank=True,
55 default=None,
56 help_text="The transit agency that uses this configuration.",
57 )
59 def __str__(self):
60 environment_label = Environment(self.environment).label if self.environment else "unknown"
61 agency_slug = self.transit_agency.slug if self.transit_agency else "(no agency)"
62 return f"({environment_label}) {agency_slug}"
65class TransitAgency(models.Model):
66 """An agency offering transit service."""
68 class Meta:
69 verbose_name_plural = "transit agencies"
71 id = models.AutoField(primary_key=True)
72 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
73 slug = models.SlugField(
74 choices=core_context.AgencySlug,
75 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
76 )
77 short_name = models.TextField(
78 default="", blank=True, help_text="The user-facing short name for this agency. Often an uppercase acronym."
79 )
80 long_name = models.TextField(
81 default="",
82 blank=True,
83 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
84 )
85 info_url = models.URLField(
86 default="",
87 blank=True,
88 help_text="URL of a website/page with more information about the agency's discounts",
89 )
90 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
91 eligibility_api_id = models.TextField(
92 help_text="The identifier for this agency used in Eligibility API calls.",
93 blank=True,
94 default="",
95 )
96 eligibility_api_private_key = models.ForeignKey(
97 PemData,
98 related_name="+",
99 on_delete=models.PROTECT,
100 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
101 null=True,
102 blank=True,
103 default=None,
104 )
105 eligibility_api_public_key = models.ForeignKey(
106 PemData,
107 related_name="+",
108 on_delete=models.PROTECT,
109 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
110 null=True,
111 blank=True,
112 default=None,
113 )
114 transit_processor = models.ForeignKey(
115 TransitProcessor,
116 on_delete=models.PROTECT,
117 null=True,
118 blank=True,
119 default=None,
120 help_text="This agency's TransitProcessor.",
121 )
122 staff_group = models.OneToOneField(
123 Group,
124 on_delete=models.PROTECT,
125 null=True,
126 blank=True,
127 default=None,
128 help_text="The group of users associated with this TransitAgency.",
129 related_name="transit_agency",
130 )
131 sso_domain = models.TextField(
132 blank=True,
133 default="",
134 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
135 )
136 customer_service_group = models.OneToOneField(
137 Group,
138 on_delete=models.PROTECT,
139 null=True,
140 blank=True,
141 default=None,
142 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
143 related_name="+",
144 )
145 logo_large = models.ImageField(
146 default="",
147 blank=True,
148 upload_to=agency_logo_large,
149 help_text="The large version of the transit agency's logo.",
150 )
151 logo_small = models.ImageField(
152 default="",
153 blank=True,
154 upload_to=agency_logo_small,
155 help_text="The small version of the transit agency's logo.",
156 )
158 def __str__(self):
159 return self.long_name
161 @property
162 def index_context(self):
163 return core_context.agency_index[self.slug].dict()
165 @property
166 def index_url(self):
167 """Public-facing URL to the TransitAgency's landing page."""
168 return reverse(routes.AGENCY_INDEX, args=[self.slug])
170 @property
171 def eligibility_index_url(self):
172 """Public facing URL to the TransitAgency's eligibility page."""
173 return reverse(routes.AGENCY_ELIGIBILITY_INDEX, args=[self.slug])
175 @property
176 def eligibility_api_private_key_data(self):
177 """This Agency's private key as a string."""
178 return self.eligibility_api_private_key.data
180 @property
181 def eligibility_api_public_key_data(self):
182 """This Agency's public key as a string."""
183 return self.eligibility_api_public_key.data
185 @property
186 def littlepay_config(self):
187 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "littlepayconfig"):
188 return self.transitprocessorconfig.littlepayconfig
189 else:
190 return None
192 @property
193 def switchio_config(self):
194 if hasattr(self, "transitprocessorconfig") and hasattr(self.transitprocessorconfig, "switchioconfig"):
195 return self.transitprocessorconfig.switchioconfig
196 else:
197 return None
199 @property
200 def enrollment_index_route(self):
201 """This Agency's enrollment index route, based on its configured transit processor."""
202 if self.littlepay_config:
203 return routes.ENROLLMENT_LITTLEPAY_INDEX
204 elif self.switchio_config:
205 return routes.ENROLLMENT_SWITCHIO_INDEX
206 else:
207 raise ValueError(
208 "TransitAgency must have either a LittlepayConfig or SwitchioConfig in order to show enrollment index."
209 )
211 @property
212 def enrollment_flows(self):
213 return self.enrollmentflow_set
215 def clean(self):
216 field_errors = {}
217 non_field_errors = []
219 if self.active:
220 message = "This field is required for active transit agencies."
221 needed = dict(
222 short_name=self.short_name,
223 long_name=self.long_name,
224 phone=self.phone,
225 info_url=self.info_url,
226 logo_large=self.logo_large,
227 logo_small=self.logo_small,
228 )
229 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
231 if self.transit_processor: 231 ↛ 235line 231 didn't jump to line 235 because the condition on line 231 was always true
232 if self.littlepay_config is None and self.switchio_config is None:
233 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
235 if self.littlepay_config:
236 try:
237 self.littlepay_config.clean()
238 except ValidationError as e:
239 message = "Littlepay configuration is missing fields that are required when this agency is active."
240 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
241 non_field_errors.append(ValidationError(message))
243 if self.switchio_config:
244 try:
245 self.switchio_config.clean()
246 except ValidationError as e:
247 message = "Switchio configuration is missing fields that are required when this agency is active."
248 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
249 non_field_errors.append(ValidationError(message))
251 all_errors = {}
252 if field_errors:
253 all_errors.update(field_errors)
254 if non_field_errors:
255 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
256 if all_errors:
257 raise ValidationError(all_errors)
259 @staticmethod
260 def by_id(id):
261 """Get a TransitAgency instance by its ID."""
262 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
263 return TransitAgency.objects.get(id=id)
265 @staticmethod
266 def by_slug(slug):
267 """Get a TransitAgency instance by its slug."""
268 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
269 return TransitAgency.objects.filter(slug=slug).first()
271 @staticmethod
272 def all_active():
273 """Get all TransitAgency instances marked active."""
274 logger.debug(f"Get all active {TransitAgency.__name__}")
275 return TransitAgency.objects.filter(active=True)
277 @staticmethod
278 def for_user(user: User):
279 for group in user.groups.all():
280 if hasattr(group, "transit_agency"):
281 return group.transit_agency # this is looking at the TransitAgency's staff_group
283 # the loop above returns the first match found. Return None if no match was found.
284 return None