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