Coverage for benefits/core/models/transit.py: 99%
114 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-29 21:21 +0000
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-29 21:21 +0000
1import os
2import logging
4from django.core.exceptions import ValidationError
5from django.contrib.auth.models import Group, User
6from django.db import models
7from django.urls import reverse
9from benefits.routes import routes
10from .common import PemData, SecretNameField, template_path
12logger = logging.getLogger(__name__)
15def _agency_logo(instance, filename, size):
16 base, ext = os.path.splitext(filename)
17 return f"agencies/{instance.slug}-{size}" + ext
20def agency_logo_small(instance, filename):
21 return _agency_logo(instance, filename, "sm")
24def agency_logo_large(instance, filename):
25 return _agency_logo(instance, filename, "lg")
28class TransitProcessor(models.Model):
29 """An entity that applies transit agency fare rules to rider transactions."""
31 id = models.AutoField(primary_key=True)
32 name = models.TextField(help_text="Primary internal display name for this TransitProcessor instance, e.g. in the Admin.")
33 api_base_url = models.TextField(help_text="The absolute base URL for the TransitProcessor's API, including https://.")
34 card_tokenize_url = models.TextField(
35 help_text="The absolute URL for the client-side card tokenization library provided by the TransitProcessor."
36 )
37 card_tokenize_func = models.TextField(
38 help_text="The function from the card tokenization library to call on the client to initiate the process."
39 )
40 card_tokenize_env = models.TextField(help_text="The environment in which card tokenization is occurring.")
41 portal_url = models.TextField(
42 default="",
43 blank=True,
44 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
45 )
47 def __str__(self):
48 return self.name
51class TransitAgency(models.Model):
52 """An agency offering transit service."""
54 id = models.AutoField(primary_key=True)
55 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
56 slug = models.SlugField(help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}")
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 index_template_override = models.TextField(
72 help_text="Override the default template used for this agency's landing page",
73 blank=True,
74 default="",
75 )
76 eligibility_index_template_override = models.TextField(
77 help_text="Override the default template used for this agency's eligibility landing page",
78 blank=True,
79 default="",
80 )
81 eligibility_api_id = models.TextField(
82 help_text="The identifier for this agency used in Eligibility API calls.",
83 blank=True,
84 default="",
85 )
86 eligibility_api_private_key = models.ForeignKey(
87 PemData,
88 related_name="+",
89 on_delete=models.PROTECT,
90 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
91 null=True,
92 blank=True,
93 default=None,
94 )
95 eligibility_api_public_key = models.ForeignKey(
96 PemData,
97 related_name="+",
98 on_delete=models.PROTECT,
99 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
100 null=True,
101 blank=True,
102 default=None,
103 )
104 transit_processor = models.ForeignKey(
105 TransitProcessor,
106 on_delete=models.PROTECT,
107 null=True,
108 blank=True,
109 default=None,
110 help_text="This agency's TransitProcessor.",
111 )
112 transit_processor_audience = models.TextField(
113 help_text="This agency's audience value used to access the TransitProcessor's API.", default="", blank=True
114 )
115 transit_processor_client_id = models.TextField(
116 help_text="This agency's client_id value used to access the TransitProcessor's API.", default="", blank=True
117 )
118 transit_processor_client_secret_name = SecretNameField(
119 help_text="The name of the secret containing this agency's client_secret value used to access the TransitProcessor's API.", # noqa: E501
120 default="",
121 blank=True,
122 )
123 staff_group = models.OneToOneField(
124 Group,
125 on_delete=models.PROTECT,
126 null=True,
127 blank=True,
128 default=None,
129 help_text="The group of users associated with this TransitAgency.",
130 related_name="transit_agency",
131 )
132 sso_domain = models.TextField(
133 blank=True,
134 default="",
135 help_text="The email domain of users to automatically add to this agency's staff group upon login.",
136 )
137 customer_service_group = models.OneToOneField(
138 Group,
139 on_delete=models.PROTECT,
140 null=True,
141 blank=True,
142 default=None,
143 help_text="The group of users who are allowed to do in-person eligibility verification and enrollment.",
144 related_name="+",
145 )
146 logo_large = models.ImageField(
147 default="",
148 blank=True,
149 upload_to=agency_logo_large,
150 help_text="The large version of the transit agency's logo.",
151 )
152 logo_small = models.ImageField(
153 default="",
154 blank=True,
155 upload_to=agency_logo_small,
156 help_text="The small version of the transit agency's logo.",
157 )
159 def __str__(self):
160 return self.long_name
162 @property
163 def index_template(self):
164 return self.index_template_override or f"core/index--{self.slug}.html"
166 @property
167 def index_url(self):
168 """Public-facing URL to the TransitAgency's landing page."""
169 return reverse(routes.AGENCY_INDEX, args=[self.slug])
171 @property
172 def eligibility_index_template(self):
173 return self.eligibility_index_template_override or f"eligibility/index--{self.slug}.html"
175 @property
176 def eligibility_index_url(self):
177 """Public facing URL to the TransitAgency's eligibility page."""
178 return reverse(routes.AGENCY_ELIGIBILITY_INDEX, args=[self.slug])
180 @property
181 def eligibility_api_private_key_data(self):
182 """This Agency's private key as a string."""
183 return self.eligibility_api_private_key.data
185 @property
186 def eligibility_api_public_key_data(self):
187 """This Agency's public key as a string."""
188 return self.eligibility_api_public_key.data
190 @property
191 def transit_processor_client_secret(self):
192 secret_field = self._meta.get_field("transit_processor_client_secret_name")
193 return secret_field.secret_value(self)
195 @property
196 def enrollment_flows(self):
197 return self.enrollmentflow_set
199 def clean(self):
200 field_errors = {}
201 template_errors = []
203 if self.active:
204 for flow in self.enrollment_flows.all():
205 try:
206 flow.clean()
207 except ValidationError:
208 raise ValidationError(f"Invalid EnrollmentFlow: {flow.label}")
210 message = "This field is required for active transit agencies."
211 needed = dict(
212 short_name=self.short_name,
213 long_name=self.long_name,
214 phone=self.phone,
215 info_url=self.info_url,
216 logo_large=self.logo_large,
217 logo_small=self.logo_small,
218 )
219 if self.transit_processor: 219 ↛ 227line 219 didn't jump to line 227 because the condition on line 219 was always true
220 needed.update(
221 dict(
222 transit_processor_audience=self.transit_processor_audience,
223 transit_processor_client_id=self.transit_processor_client_id,
224 transit_processor_client_secret_name=self.transit_processor_client_secret_name,
225 )
226 )
227 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
229 # since templates are calculated from the pattern or the override field
230 # we can't add a field-level validation error
231 # so just create directly for a missing template
232 for t in [self.index_template, self.eligibility_index_template]:
233 if not template_path(t):
234 template_errors.append(ValidationError(f"Template not found: {t}"))
236 if field_errors:
237 raise ValidationError(field_errors)
238 if template_errors:
239 raise ValidationError(template_errors)
241 @staticmethod
242 def by_id(id):
243 """Get a TransitAgency instance by its ID."""
244 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
245 return TransitAgency.objects.get(id=id)
247 @staticmethod
248 def by_slug(slug):
249 """Get a TransitAgency instance by its slug."""
250 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
251 return TransitAgency.objects.filter(slug=slug).first()
253 @staticmethod
254 def all_active():
255 """Get all TransitAgency instances marked active."""
256 logger.debug(f"Get all active {TransitAgency.__name__}")
257 return TransitAgency.objects.filter(active=True)
259 @staticmethod
260 def for_user(user: User):
261 for group in user.groups.all():
262 if hasattr(group, "transit_agency"):
263 return group.transit_agency # this is looking at the TransitAgency's staff_group
265 # the loop above returns the first match found. Return None if no match was found.
266 return None