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

1import os 

2import logging 

3 

4from django.core.exceptions import ValidationError 

5from django.contrib.auth.models import Group, User 

6from django.db import models 

7from django.urls import reverse 

8 

9from benefits.routes import routes 

10from .common import PemData, SecretNameField, template_path 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15def _agency_logo(instance, filename, size): 

16 base, ext = os.path.splitext(filename) 

17 return f"agencies/{instance.slug}-{size}" + ext 

18 

19 

20def agency_logo_small(instance, filename): 

21 return _agency_logo(instance, filename, "sm") 

22 

23 

24def agency_logo_large(instance, filename): 

25 return _agency_logo(instance, filename, "lg") 

26 

27 

28class TransitProcessor(models.Model): 

29 """An entity that applies transit agency fare rules to rider transactions.""" 

30 

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 ) 

46 

47 def __str__(self): 

48 return self.name 

49 

50 

51class TransitAgency(models.Model): 

52 """An agency offering transit service.""" 

53 

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 ) 

158 

159 def __str__(self): 

160 return self.long_name 

161 

162 @property 

163 def index_template(self): 

164 return self.index_template_override or f"core/index--{self.slug}.html" 

165 

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]) 

170 

171 @property 

172 def eligibility_index_template(self): 

173 return self.eligibility_index_template_override or f"eligibility/index--{self.slug}.html" 

174 

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]) 

179 

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 

184 

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 

189 

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) 

194 

195 @property 

196 def enrollment_flows(self): 

197 return self.enrollmentflow_set 

198 

199 def clean(self): 

200 field_errors = {} 

201 template_errors = [] 

202 

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}") 

209 

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}) 

228 

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}")) 

235 

236 if field_errors: 

237 raise ValidationError(field_errors) 

238 if template_errors: 

239 raise ValidationError(template_errors) 

240 

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) 

246 

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() 

252 

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) 

258 

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 

264 

265 # the loop above returns the first match found. Return None if no match was found. 

266 return None