Coverage for benefits / enrollment_switchio / models.py: 100%
59 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 15:39 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 15:39 +0000
1import logging
3from django.core.exceptions import ValidationError
4from django.db import models
6from benefits.core.models import EnrollmentGroup, Environment, PemData, SecretNameField, TransitProcessorConfig
7from benefits.secrets import get_secret_by_name
9logger = logging.getLogger(__name__)
12class SwitchioConfig(TransitProcessorConfig):
13 """Configuration for connecting to Switchio, an entity that applies transit agency fare rules to rider transactions."""
15 tokenization_api_key = models.TextField(
16 help_text="The API key used to access the Switchio API for tokenization.", default="", blank=True
17 )
18 tokenization_api_secret_name = SecretNameField(
19 help_text="The name of the secret containing the api_secret value used to access the Switchio API for tokenization.", # noqa: E501
20 default="",
21 blank=True,
22 )
23 enrollment_api_authorization_header = models.TextField(
24 help_text="The value to use for the 'Authorization' header when accessing the Switchio API for enrollment.",
25 default="",
26 blank=True,
27 )
28 pto_id = models.PositiveIntegerField(
29 help_text="The Public Transport Operator ID to use with the Switchio API for enrollment.",
30 default=0,
31 blank=True,
32 )
33 client_certificate = models.ForeignKey(
34 PemData,
35 related_name="+",
36 on_delete=models.PROTECT,
37 help_text="The client certificate for accessing the Switchio API.",
38 null=True,
39 blank=True,
40 default=None,
41 )
42 ca_certificate = models.ForeignKey(
43 PemData,
44 related_name="+",
45 on_delete=models.PROTECT,
46 help_text="The CA certificate chain for accessing the Switchio API.",
47 null=True,
48 blank=True,
49 default=None,
50 )
51 private_key = models.ForeignKey(
52 PemData,
53 related_name="+",
54 on_delete=models.PROTECT,
55 help_text="The private key for accessing the Switchio API.",
56 null=True,
57 blank=True,
58 default=None,
59 )
61 @property
62 def tokenization_api_base_url(self):
63 if self.environment == Environment.DEV.value:
64 return get_secret_by_name("switchio-int-tokenization-api-base-url")
65 if self.environment == Environment.TEST.value:
66 return get_secret_by_name("switchio-acc-tokenization-api-base-url")
67 elif self.environment == Environment.PROD.value:
68 return get_secret_by_name("switchio-prod-tokenization-api-base-url")
69 else:
70 raise ValueError(f"Unexpected value for environment: {self.environment}")
72 @property
73 def enrollment_api_base_url(self):
74 if self.environment == Environment.DEV.value:
75 return get_secret_by_name("switchio-int-enrollment-api-base-url")
76 if self.environment == Environment.TEST.value:
77 return get_secret_by_name("switchio-acc-enrollment-api-base-url")
78 elif self.environment == Environment.PROD.value:
79 return get_secret_by_name("switchio-prod-enrollment-api-base-url")
80 else:
81 raise ValueError(f"Unexpected value for environment: {self.environment}")
83 @property
84 def tokenization_api_secret(self):
85 secret_field = self._meta.get_field("tokenization_api_secret_name")
86 return secret_field.secret_value(self)
88 @property
89 def client_certificate_data(self):
90 """This SwitchioConfig's client certificate as a string."""
91 return self.client_certificate.data
93 @property
94 def ca_certificate_data(self):
95 """This SwitchioConfig's CA certificate as a string."""
96 return self.ca_certificate.data
98 @property
99 def private_key_data(self):
100 """This SwitchioConfig's private key as a string."""
101 return self.private_key.data
103 def clean(self):
104 field_errors = {}
106 if self.pk and self.transitagency_set and any([agency.active for agency in self.transitagency_set.all()]):
107 message = "This field is required when this configuration is referenced by an active transit agency."
108 needed = dict(
109 tokenization_api_key=self.tokenization_api_key,
110 tokenization_api_secret_name=self.tokenization_api_secret_name,
111 enrollment_api_authorization_header=self.enrollment_api_authorization_header,
112 pto_id=self.pto_id,
113 client_certificate=self.client_certificate,
114 ca_certificate=self.ca_certificate,
115 private_key=self.private_key,
116 )
117 field_errors.update({k: ValidationError(message) for k, v in needed.items() if not v})
119 if field_errors:
120 raise ValidationError(field_errors)
123class SwitchioGroup(EnrollmentGroup):
124 group_id = models.TextField(default=None, blank=True, help_text="The ID of the Switchio group for user enrollment.")
126 @staticmethod
127 def by_id(id):
128 """Get a SwitchioGroup instance by its ID."""
129 logger.debug(f"Get {SwitchioGroup.__name__} by id: {id}")
130 return SwitchioGroup.objects.get(id=id)