Coverage for benefits/core/models/transit.py: 99%
117 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-06 20:07 +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_flows(self):
187 return self.enrollmentflow_set
189 def clean(self):
190 field_errors = {}
191 non_field_errors = []
193 if self.active:
194 message = "This field is required for active transit agencies."
195 needed = dict(
196 short_name=self.short_name,
197 long_name=self.long_name,
198 phone=self.phone,
199 info_url=self.info_url,
200 logo_large=self.logo_large,
201 logo_small=self.logo_small,
202 )
203 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
205 if self.transit_processor: 205 ↛ 209line 205 didn't jump to line 209 because the condition on line 205 was always true
206 if self.littlepay_config is None and self.switchio_config is None:
207 non_field_errors.append(ValidationError("Must fill out configuration for either Littlepay or Switchio."))
209 if self.littlepay_config:
210 try:
211 self.littlepay_config.clean()
212 except ValidationError as e:
213 message = "Littlepay configuration is missing fields that are required when this agency is active."
214 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
215 non_field_errors.append(ValidationError(message))
217 if self.switchio_config:
218 try:
219 self.switchio_config.clean(agency=self)
220 except ValidationError as e:
221 message = "Switchio configuration is missing fields that are required when this agency is active."
222 message += f" Missing fields: {', '.join(e.error_dict.keys())}"
223 non_field_errors.append(ValidationError(message))
225 all_errors = {}
226 if field_errors:
227 all_errors.update(field_errors)
228 if non_field_errors:
229 all_errors.update({NON_FIELD_ERRORS: value for value in non_field_errors})
230 if all_errors:
231 raise ValidationError(all_errors)
233 @staticmethod
234 def by_id(id):
235 """Get a TransitAgency instance by its ID."""
236 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
237 return TransitAgency.objects.get(id=id)
239 @staticmethod
240 def by_slug(slug):
241 """Get a TransitAgency instance by its slug."""
242 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
243 return TransitAgency.objects.filter(slug=slug).first()
245 @staticmethod
246 def all_active():
247 """Get all TransitAgency instances marked active."""
248 logger.debug(f"Get all active {TransitAgency.__name__}")
249 return TransitAgency.objects.filter(active=True)
251 @staticmethod
252 def for_user(user: User):
253 for group in user.groups.all():
254 if hasattr(group, "transit_agency"):
255 return group.transit_agency # this is looking at the TransitAgency's staff_group
257 # the loop above returns the first match found. Return None if no match was found.
258 return None