Coverage for benefits/core/models/transit.py: 99%
103 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-13 23:09 +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.core import context as core_context
10from benefits.eligibility import context as eligibility_context
11from benefits.routes import routes
12from .common import PemData, SecretNameField
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 card_tokenize_url = models.TextField(
37 help_text="The absolute URL for the client-side card tokenization library provided by the TransitProcessor."
38 )
39 card_tokenize_func = models.TextField(
40 help_text="The function from the card tokenization library to call on the client to initiate the process."
41 )
42 card_tokenize_env = models.TextField(help_text="The environment in which card tokenization is occurring.")
43 portal_url = models.TextField(
44 default="",
45 blank=True,
46 help_text="The absolute base URL for the TransitProcessor's control portal, including https://.",
47 )
49 def __str__(self):
50 return self.name
53class TransitAgency(models.Model):
54 """An agency offering transit service."""
56 id = models.AutoField(primary_key=True)
57 active = models.BooleanField(default=False, help_text="Determines if this Agency is enabled for users")
58 slug = models.SlugField(
59 choices=core_context.AgencySlug,
60 help_text="Used for URL navigation for this agency, e.g. the agency homepage url is /{slug}",
61 )
62 short_name = models.TextField(
63 default="", blank=True, help_text="The user-facing short name for this agency. Often an uppercase acronym."
64 )
65 long_name = models.TextField(
66 default="",
67 blank=True,
68 help_text="The user-facing long name for this agency. Often the short_name acronym, spelled out.",
69 )
70 info_url = models.URLField(
71 default="",
72 blank=True,
73 help_text="URL of a website/page with more information about the agency's discounts",
74 )
75 phone = models.TextField(default="", blank=True, help_text="Agency customer support phone number")
76 eligibility_api_id = models.TextField(
77 help_text="The identifier for this agency used in Eligibility API calls.",
78 blank=True,
79 default="",
80 )
81 eligibility_api_private_key = models.ForeignKey(
82 PemData,
83 related_name="+",
84 on_delete=models.PROTECT,
85 help_text="Private key used to sign Eligibility API tokens created on behalf of this Agency.",
86 null=True,
87 blank=True,
88 default=None,
89 )
90 eligibility_api_public_key = models.ForeignKey(
91 PemData,
92 related_name="+",
93 on_delete=models.PROTECT,
94 help_text="Public key corresponding to the agency's private key, used by Eligibility Verification servers to encrypt responses.", # noqa: E501
95 null=True,
96 blank=True,
97 default=None,
98 )
99 transit_processor = models.ForeignKey(
100 TransitProcessor,
101 on_delete=models.PROTECT,
102 null=True,
103 blank=True,
104 default=None,
105 help_text="This agency's TransitProcessor.",
106 )
107 transit_processor_audience = models.TextField(
108 help_text="This agency's audience value used to access the TransitProcessor's API.", default="", blank=True
109 )
110 transit_processor_client_id = models.TextField(
111 help_text="This agency's client_id value used to access the TransitProcessor's API.", default="", blank=True
112 )
113 transit_processor_client_secret_name = SecretNameField(
114 help_text="The name of the secret containing this agency's client_secret value used to access the TransitProcessor's API.", # noqa: E501
115 default="",
116 blank=True,
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 transit_processor_client_secret(self):
187 secret_field = self._meta.get_field("transit_processor_client_secret_name")
188 return secret_field.secret_value(self)
190 @property
191 def enrollment_flows(self):
192 return self.enrollmentflow_set
194 def clean(self):
195 field_errors = {}
197 if self.active:
198 message = "This field is required for active transit agencies."
199 needed = dict(
200 short_name=self.short_name,
201 long_name=self.long_name,
202 phone=self.phone,
203 info_url=self.info_url,
204 logo_large=self.logo_large,
205 logo_small=self.logo_small,
206 )
207 if self.transit_processor: 207 ↛ 215line 207 didn't jump to line 215 because the condition on line 207 was always true
208 needed.update(
209 dict(
210 transit_processor_audience=self.transit_processor_audience,
211 transit_processor_client_id=self.transit_processor_client_id,
212 transit_processor_client_secret_name=self.transit_processor_client_secret_name,
213 )
214 )
215 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
217 if field_errors:
218 raise ValidationError(field_errors)
220 @staticmethod
221 def by_id(id):
222 """Get a TransitAgency instance by its ID."""
223 logger.debug(f"Get {TransitAgency.__name__} by id: {id}")
224 return TransitAgency.objects.get(id=id)
226 @staticmethod
227 def by_slug(slug):
228 """Get a TransitAgency instance by its slug."""
229 logger.debug(f"Get {TransitAgency.__name__} by slug: {slug}")
230 return TransitAgency.objects.filter(slug=slug).first()
232 @staticmethod
233 def all_active():
234 """Get all TransitAgency instances marked active."""
235 logger.debug(f"Get all active {TransitAgency.__name__}")
236 return TransitAgency.objects.filter(active=True)
238 @staticmethod
239 def for_user(user: User):
240 for group in user.groups.all():
241 if hasattr(group, "transit_agency"):
242 return group.transit_agency # this is looking at the TransitAgency's staff_group
244 # the loop above returns the first match found. Return None if no match was found.
245 return None