Coverage for benefits/core/models/transit.py: 94%
134 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-14 01:41 +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 api_base_url = models.TextField(help_text="The absolute base URL for the TransitProcessor's API, including https://.")
36 portal_url = models.TextField(
37 default="",
38 blank=True,
39 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
40 )
42 def __str__(self):
43 return self.name
46class TransitAgency(models.Model):
47 """An agency offering transit service."""
49 class Meta:
50 verbose_name_plural = "transit agencies"
52 id = models.AutoField(primary_key=True)
53 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
54 slug = models.SlugField(
55 choices=core_context.AgencySlug,
56 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
57 )
58 short_name = models.TextField(
59 default="", blank=True, help_text="The user-facing short name for this agency. Often an uppercase acronym."
60 )
61 long_name = models.TextField(
62 default="",
63 blank=True,
64 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
65 )
66 info_url = models.URLField(
67 default="",
68 blank=True,
69 help_text="URL of a website/page with more information about the agency's discounts",
70 )
71 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
72 eligibility_api_id = models.TextField(
73 help_text="The identifier for this agency used in Eligibility API calls.",
74 blank=True,
75 default="",
76 )
77 eligibility_api_private_key = models.ForeignKey(
78 PemData,
79 related_name="+",
80 on_delete=models.PROTECT,
81 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
82 null=True,
83 blank=True,
84 default=None,
85 )
86 eligibility_api_public_key = models.ForeignKey(
87 PemData,
88 related_name="+",
89 on_delete=models.PROTECT,
90 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
91 null=True,
92 blank=True,
93 default=None,
94 )
95 transit_processor = models.ForeignKey(
96 TransitProcessor,
97 on_delete=models.PROTECT,
98 null=True,
99 blank=True,
100 default=None,
101 help_text="This agency's TransitProcessor.",
102 )
103 littlepay_config = models.OneToOneField(
104 "enrollment_littlepay.LittlepayConfig",
105 on_delete=models.PROTECT,
106 null=True,
107 blank=True,
108 default=None,
109 help_text="The Littlepay configuration used by this agency for enrollment.",
110 )
111 switchio_config = models.ForeignKey(
112 "enrollment_switchio.SwitchioConfig",
113 on_delete=models.PROTECT,
114 null=True,
115 blank=True,
116 default=None,
117 help_text="The Switchio configuration used by this agency for enrollment.",
118 )
119 staff_group = models.OneToOneField(
120 Group,
121 on_delete=models.PROTECT,
122 null=True,
123 blank=True,
124 default=None,
125 help_text="The group of users associated with this TransitAgency.",
126 related_name="transit_agency",
127 )
128 sso_domain = models.TextField(
129 blank=True,
130 default="",
131 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
132 )
133 customer_service_group = models.OneToOneField(
134 Group,
135 on_delete=models.PROTECT,
136 null=True,
137 blank=True,
138 default=None,
139 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
140 related_name="+",
141 )
142 logo_large = models.ImageField(
143 default="",
144 blank=True,
145 upload_to=agency_logo_large,
146 help_text="The large version of the transit agency's logo.",
147 )
148 logo_small = models.ImageField(
149 default="",
150 blank=True,
151 upload_to=agency_logo_small,
152 help_text="The small version of the transit agency's logo.",
153 )
155 def __str__(self):
156 return self.long_name
158 @property
159 def index_context(self):
160 return core_context.agency_index[self.slug].dict()
162 @property
163 def index_url(self):
164 """Public-facing URL to the TransitAgency's landing page."""
165 return reverse(routes.AGENCY_INDEX, args=[self.slug])
167 @property
168 def eligibility_index_context(self):
169 return eligibility_context.eligibility_index[self.slug].dict()
171 @property
172 def eligibility_index_url(self):
173 """Public facing URL to the TransitAgency's eligibility page."""
174 return reverse(routes.AGENCY_ELIGIBILITY_INDEX, args=[self.slug])
176 @property
177 def eligibility_api_private_key_data(self):
178 """This Agency's private key as a string."""
179 return self.eligibility_api_private_key.data
181 @property
182 def eligibility_api_public_key_data(self):
183 """This Agency's public key as a string."""
184 return self.eligibility_api_public_key.data
186 @property
187 def enrollment_index_template(self):
188 if self.littlepay_config:
189 template = self.littlepay_config.enrollment_index_template
190 elif self.switchio_config: 190 ↛ 193line 190 didn't jump to line 193 because the condition on line 190 was always true
191 template = self.switchio_config.enrollment_index_template
192 else:
193 raise ValueError("Transit agency does not have a Littlepay or Switchio config")
195 return template
197 @property
198 def enrollment_flows(self):
199 return self.enrollmentflow_set
201 @property
202 def transit_processor_context(self):
203 if self.littlepay_config: 203 ↛ 205line 203 didn't jump to line 205 because the condition on line 203 was always true
204 context = self.littlepay_config.transit_processor_context
205 elif self.switchio_config:
206 context = self.switchio_config.transit_processor_context
207 else:
208 raise ValueError("Transit agency does not have a Littlepay or Switchio config")
210 return context
212 def clean(self):
213 field_errors = {}
214 non_field_errors = []
216 if self.active:
217 message = "This field is required for active transit agencies."
218 needed = dict(
219 short_name=self.short_name,
220 long_name=self.long_name,
221 phone=self.phone,
222 info_url=self.info_url,
223 logo_large=self.logo_large,
224 logo_small=self.logo_small,
225 )
226 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
228 if self.transit_processor: 228 ↛ 232line 228 didn't jump to line 232 because the condition on line 228 was always true
229 if self.littlepay_config is None and self.switchio_config is None:
230 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
232 if self.littlepay_config:
233 try:
234 self.littlepay_config.clean()
235 except ValidationError as e:
236 message = "Littlepay configuration is missing fields that are required when this agency is active."
237 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
238 non_field_errors.append(ValidationError(message))
240 if self.switchio_config:
241 try:
242 self.switchio_config.clean(agency=self)
243 except ValidationError as e:
244 message = "Switchio configuration is missing fields that are required when this agency is active."
245 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
246 non_field_errors.append(ValidationError(message))
248 all_errors = {}
249 if field_errors:
250 all_errors.update(field_errors)
251 if non_field_errors:
252 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
253 if all_errors:
254 raise ValidationError(all_errors)
256 @staticmethod
257 def by_id(id):
258 """Get a TransitAgency instance by its ID."""
259 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
260 return TransitAgency.objects.get(id=id)
262 @staticmethod
263 def by_slug(slug):
264 """Get a TransitAgency instance by its slug."""
265 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
266 return TransitAgency.objects.filter(slug=slug).first()
268 @staticmethod
269 def all_active():
270 """Get all TransitAgency instances marked active."""
271 logger.debug(f"Get all active {TransitAgency.__name__}")
272 return TransitAgency.objects.filter(active=True)
274 @staticmethod
275 def for_user(user: User):
276 for group in user.groups.all():
277 if hasattr(group, "transit_agency"):
278 return group.transit_agency # this is looking at the TransitAgency's staff_group
280 # the loop above returns the first match found. Return None if no match was found.
281 return None