authentik.events.tests.test_transports

transport tests

  1"""transport tests"""
  2
  3from unittest.mock import PropertyMock, patch
  4
  5from django.core import mail
  6from django.core.mail.backends.locmem import EmailBackend
  7from django.test import TestCase
  8from django.urls import reverse
  9from requests_mock import Mocker
 10
 11from authentik import authentik_full_version
 12from authentik.core.tests.utils import create_test_admin_user
 13from authentik.crypto.models import CertificateKeyPair
 14from authentik.events.api.notification_transports import NotificationTransportSerializer
 15from authentik.events.models import (
 16    Event,
 17    Notification,
 18    NotificationSeverity,
 19    NotificationTransport,
 20    NotificationWebhookMapping,
 21    TransportMode,
 22)
 23from authentik.lib.generators import generate_id
 24from authentik.stages.email.models import get_template_choices
 25
 26
 27class TestEventTransports(TestCase):
 28    """Test Event Transports"""
 29
 30    def setUp(self) -> None:
 31        self.user = create_test_admin_user()
 32        self.event = Event.new("foo", "testing", foo="bar,").set_user(self.user)
 33        self.event.save()
 34        self.notification = Notification.objects.create(
 35            severity=NotificationSeverity.ALERT,
 36            body="foo",
 37            event=self.event,
 38            user=self.user,
 39        )
 40
 41    def test_transport_webhook(self):
 42        """Test webhook transport"""
 43        transport: NotificationTransport = NotificationTransport.objects.create(
 44            name=generate_id(),
 45            mode=TransportMode.WEBHOOK,
 46            webhook_url="http://localhost:1234/test",
 47        )
 48        with Mocker() as mocker:
 49            mocker.post("http://localhost:1234/test")
 50            transport.send(self.notification)
 51            self.assertEqual(mocker.call_count, 1)
 52            self.assertEqual(mocker.request_history[0].method, "POST")
 53            self.assertJSONEqual(
 54                mocker.request_history[0].body.decode(),
 55                {
 56                    "body": "foo",
 57                    "severity": "alert",
 58                    "user_email": self.user.email,
 59                    "user_username": self.user.username,
 60                    "event_user_email": self.user.email,
 61                    "event_user_username": self.user.username,
 62                },
 63            )
 64
 65    def test_transport_webhook_ca_invalid_unset(self):
 66        """Test webhook transport"""
 67        transport: NotificationTransport = NotificationTransport.objects.create(
 68            name=generate_id(),
 69            mode=TransportMode.WEBHOOK,
 70            webhook_url="https://localhost:1234/test",
 71        )
 72        with Mocker() as mocker:
 73            mocker.post("https://localhost:1234/test")
 74            transport.send(self.notification)
 75            self.assertEqual(mocker.call_count, 1)
 76            self.assertTrue(mocker.request_history[0].verify)
 77
 78    def test_transport_webhook_ca(self):
 79        """Test webhook transport"""
 80        kp = CertificateKeyPair.objects.create(
 81            name=generate_id(),
 82            certificate_data="foo",
 83        )
 84        transport: NotificationTransport = NotificationTransport.objects.create(
 85            name=generate_id(),
 86            mode=TransportMode.WEBHOOK,
 87            webhook_url="https://localhost:1234/test",
 88            webhook_ca=kp,
 89        )
 90        with Mocker() as mocker:
 91            mocker.post("https://localhost:1234/test")
 92            transport.send(self.notification)
 93            self.assertEqual(mocker.call_count, 1)
 94            self.assertIsNotNone(mocker.request_history[0].verify)
 95
 96    def test_transport_webhook_mapping(self):
 97        """Test webhook transport with custom mapping"""
 98        mapping_body = NotificationWebhookMapping.objects.create(
 99            name=generate_id(), expression="return request.user"
100        )
101        mapping_headers = NotificationWebhookMapping.objects.create(
102            name=generate_id(), expression="""return {"foo": "bar"}"""
103        )
104        transport: NotificationTransport = NotificationTransport.objects.create(
105            name=generate_id(),
106            mode=TransportMode.WEBHOOK,
107            webhook_url="http://localhost:1234/test",
108            webhook_mapping_body=mapping_body,
109            webhook_mapping_headers=mapping_headers,
110        )
111        with Mocker() as mocker:
112            mocker.post("http://localhost:1234/test")
113            transport.send(self.notification)
114            self.assertEqual(mocker.call_count, 1)
115            self.assertEqual(mocker.request_history[0].method, "POST")
116            self.assertEqual(mocker.request_history[0].headers["foo"], "bar")
117            self.assertJSONEqual(
118                mocker.request_history[0].body.decode(),
119                {"email": self.user.email, "pk": self.user.pk, "username": self.user.username},
120            )
121
122    def test_transport_webhook_slack(self):
123        """Test webhook transport (slack)"""
124        transport: NotificationTransport = NotificationTransport.objects.create(
125            name=generate_id(),
126            mode=TransportMode.WEBHOOK_SLACK,
127            webhook_url="http://localhost:1234/test",
128        )
129        with Mocker() as mocker:
130            mocker.post("http://localhost:1234/test")
131            transport.send(self.notification)
132            self.assertEqual(mocker.call_count, 1)
133            self.assertEqual(mocker.request_history[0].method, "POST")
134            self.assertJSONEqual(
135                mocker.request_history[0].body.decode(),
136                {
137                    "username": "authentik",
138                    "icon_url": "https://goauthentik.io/img/icon.png",
139                    "attachments": [
140                        {
141                            "author_name": "authentik",
142                            "author_link": "https://goauthentik.io",
143                            "author_icon": "https://goauthentik.io/img/icon.png",
144                            "title": "custom_foo",
145                            "color": "#fd4b2d",
146                            "fields": [
147                                {"title": "Severity", "value": "alert", "short": True},
148                                {
149                                    "title": "Dispatched for user",
150                                    "value": self.user.username,
151                                    "short": True,
152                                },
153                                {"short": True, "title": "Event user", "value": self.user.username},
154                                {"title": "foo", "value": "bar,"},
155                            ],
156                            "footer": f"authentik {authentik_full_version()}",
157                        }
158                    ],
159                },
160            )
161
162    def test_transport_email(self):
163        """Test email transport"""
164        transport: NotificationTransport = NotificationTransport.objects.create(
165            name=generate_id(),
166            mode=TransportMode.EMAIL,
167        )
168        with patch(
169            "authentik.stages.email.models.EmailStage.backend_class",
170            PropertyMock(return_value=EmailBackend),
171        ):
172            transport.send(self.notification)
173            self.assertEqual(len(mail.outbox), 1)
174            self.assertEqual(mail.outbox[0].subject, "authentik Notification: custom_foo")
175            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])
176
177    def test_transport_email_custom_template(self):
178        """Test email transport with custom template"""
179        transport: NotificationTransport = NotificationTransport.objects.create(
180            name=generate_id(),
181            mode=TransportMode.EMAIL,
182            email_template="email/event_notification.html",
183        )
184        with patch(
185            "authentik.stages.email.models.EmailStage.backend_class",
186            PropertyMock(return_value=EmailBackend),
187        ):
188            transport.send(self.notification)
189            self.assertEqual(len(mail.outbox), 1)
190            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])
191
192    def test_transport_email_custom_subject_prefix(self):
193        """Test email transport with custom subject prefix"""
194        transport: NotificationTransport = NotificationTransport.objects.create(
195            name=generate_id(),
196            mode=TransportMode.EMAIL,
197            email_subject_prefix="[CUSTOM] ",
198        )
199        with patch(
200            "authentik.stages.email.models.EmailStage.backend_class",
201            PropertyMock(return_value=EmailBackend),
202        ):
203            transport.send(self.notification)
204            self.assertEqual(len(mail.outbox), 1)
205            self.assertEqual(mail.outbox[0].subject, "[CUSTOM] custom_foo")
206
207    def test_transport_email_validation(self):
208        """Test email transport template validation"""
209
210        # Test valid template
211        serializer = NotificationTransportSerializer(
212            data={
213                "name": generate_id(),
214                "mode": TransportMode.EMAIL,
215                "email_template": "email/event_notification.html",
216            }
217        )
218        self.assertTrue(serializer.is_valid())
219
220        # Test invalid template - should fail due to choices validation
221        serializer = NotificationTransportSerializer(
222            data={
223                "name": generate_id(),
224                "mode": TransportMode.EMAIL,
225                "email_template": "invalid/template.html",
226            }
227        )
228        self.assertFalse(serializer.is_valid())
229        self.assertIn("email_template", serializer.errors)
230
231    def test_templates_api_endpoint(self):
232        """Test templates API endpoint returns valid templates"""
233        self.client.force_login(self.user)
234        response = self.client.get(reverse("authentik_api:emailstage-templates"))
235        self.assertEqual(response.status_code, 200)
236
237        data = response.json()
238        self.assertIsInstance(data, list)
239
240        # Check that we have at least the default templates
241        template_names = [item["name"] for item in data]
242        self.assertIn("email/event_notification.html", template_names)
243
244        # Verify all templates are valid choices
245        valid_choices = dict(get_template_choices())
246        for template in data:
247            self.assertIn(template["name"], valid_choices)
248            self.assertEqual(template["description"], valid_choices[template["name"]])
class TestEventTransports(django.test.testcases.TestCase): 
 28class TestEventTransports(TestCase):
 29    """Test Event Transports"""
 30
 31    def setUp(self) -> None:
 32        self.user = create_test_admin_user()
 33        self.event = Event.new("foo", "testing", foo="bar,").set_user(self.user)
 34        self.event.save()
 35        self.notification = Notification.objects.create(
 36            severity=NotificationSeverity.ALERT,
 37            body="foo",
 38            event=self.event,
 39            user=self.user,
 40        )
 41
 42    def test_transport_webhook(self):
 43        """Test webhook transport"""
 44        transport: NotificationTransport = NotificationTransport.objects.create(
 45            name=generate_id(),
 46            mode=TransportMode.WEBHOOK,
 47            webhook_url="http://localhost:1234/test",
 48        )
 49        with Mocker() as mocker:
 50            mocker.post("http://localhost:1234/test")
 51            transport.send(self.notification)
 52            self.assertEqual(mocker.call_count, 1)
 53            self.assertEqual(mocker.request_history[0].method, "POST")
 54            self.assertJSONEqual(
 55                mocker.request_history[0].body.decode(),
 56                {
 57                    "body": "foo",
 58                    "severity": "alert",
 59                    "user_email": self.user.email,
 60                    "user_username": self.user.username,
 61                    "event_user_email": self.user.email,
 62                    "event_user_username": self.user.username,
 63                },
 64            )
 65
 66    def test_transport_webhook_ca_invalid_unset(self):
 67        """Test webhook transport"""
 68        transport: NotificationTransport = NotificationTransport.objects.create(
 69            name=generate_id(),
 70            mode=TransportMode.WEBHOOK,
 71            webhook_url="https://localhost:1234/test",
 72        )
 73        with Mocker() as mocker:
 74            mocker.post("https://localhost:1234/test")
 75            transport.send(self.notification)
 76            self.assertEqual(mocker.call_count, 1)
 77            self.assertTrue(mocker.request_history[0].verify)
 78
 79    def test_transport_webhook_ca(self):
 80        """Test webhook transport"""
 81        kp = CertificateKeyPair.objects.create(
 82            name=generate_id(),
 83            certificate_data="foo",
 84        )
 85        transport: NotificationTransport = NotificationTransport.objects.create(
 86            name=generate_id(),
 87            mode=TransportMode.WEBHOOK,
 88            webhook_url="https://localhost:1234/test",
 89            webhook_ca=kp,
 90        )
 91        with Mocker() as mocker:
 92            mocker.post("https://localhost:1234/test")
 93            transport.send(self.notification)
 94            self.assertEqual(mocker.call_count, 1)
 95            self.assertIsNotNone(mocker.request_history[0].verify)
 96
 97    def test_transport_webhook_mapping(self):
 98        """Test webhook transport with custom mapping"""
 99        mapping_body = NotificationWebhookMapping.objects.create(
100            name=generate_id(), expression="return request.user"
101        )
102        mapping_headers = NotificationWebhookMapping.objects.create(
103            name=generate_id(), expression="""return {"foo": "bar"}"""
104        )
105        transport: NotificationTransport = NotificationTransport.objects.create(
106            name=generate_id(),
107            mode=TransportMode.WEBHOOK,
108            webhook_url="http://localhost:1234/test",
109            webhook_mapping_body=mapping_body,
110            webhook_mapping_headers=mapping_headers,
111        )
112        with Mocker() as mocker:
113            mocker.post("http://localhost:1234/test")
114            transport.send(self.notification)
115            self.assertEqual(mocker.call_count, 1)
116            self.assertEqual(mocker.request_history[0].method, "POST")
117            self.assertEqual(mocker.request_history[0].headers["foo"], "bar")
118            self.assertJSONEqual(
119                mocker.request_history[0].body.decode(),
120                {"email": self.user.email, "pk": self.user.pk, "username": self.user.username},
121            )
122
123    def test_transport_webhook_slack(self):
124        """Test webhook transport (slack)"""
125        transport: NotificationTransport = NotificationTransport.objects.create(
126            name=generate_id(),
127            mode=TransportMode.WEBHOOK_SLACK,
128            webhook_url="http://localhost:1234/test",
129        )
130        with Mocker() as mocker:
131            mocker.post("http://localhost:1234/test")
132            transport.send(self.notification)
133            self.assertEqual(mocker.call_count, 1)
134            self.assertEqual(mocker.request_history[0].method, "POST")
135            self.assertJSONEqual(
136                mocker.request_history[0].body.decode(),
137                {
138                    "username": "authentik",
139                    "icon_url": "https://goauthentik.io/img/icon.png",
140                    "attachments": [
141                        {
142                            "author_name": "authentik",
143                            "author_link": "https://goauthentik.io",
144                            "author_icon": "https://goauthentik.io/img/icon.png",
145                            "title": "custom_foo",
146                            "color": "#fd4b2d",
147                            "fields": [
148                                {"title": "Severity", "value": "alert", "short": True},
149                                {
150                                    "title": "Dispatched for user",
151                                    "value": self.user.username,
152                                    "short": True,
153                                },
154                                {"short": True, "title": "Event user", "value": self.user.username},
155                                {"title": "foo", "value": "bar,"},
156                            ],
157                            "footer": f"authentik {authentik_full_version()}",
158                        }
159                    ],
160                },
161            )
162
163    def test_transport_email(self):
164        """Test email transport"""
165        transport: NotificationTransport = NotificationTransport.objects.create(
166            name=generate_id(),
167            mode=TransportMode.EMAIL,
168        )
169        with patch(
170            "authentik.stages.email.models.EmailStage.backend_class",
171            PropertyMock(return_value=EmailBackend),
172        ):
173            transport.send(self.notification)
174            self.assertEqual(len(mail.outbox), 1)
175            self.assertEqual(mail.outbox[0].subject, "authentik Notification: custom_foo")
176            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])
177
178    def test_transport_email_custom_template(self):
179        """Test email transport with custom template"""
180        transport: NotificationTransport = NotificationTransport.objects.create(
181            name=generate_id(),
182            mode=TransportMode.EMAIL,
183            email_template="email/event_notification.html",
184        )
185        with patch(
186            "authentik.stages.email.models.EmailStage.backend_class",
187            PropertyMock(return_value=EmailBackend),
188        ):
189            transport.send(self.notification)
190            self.assertEqual(len(mail.outbox), 1)
191            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])
192
193    def test_transport_email_custom_subject_prefix(self):
194        """Test email transport with custom subject prefix"""
195        transport: NotificationTransport = NotificationTransport.objects.create(
196            name=generate_id(),
197            mode=TransportMode.EMAIL,
198            email_subject_prefix="[CUSTOM] ",
199        )
200        with patch(
201            "authentik.stages.email.models.EmailStage.backend_class",
202            PropertyMock(return_value=EmailBackend),
203        ):
204            transport.send(self.notification)
205            self.assertEqual(len(mail.outbox), 1)
206            self.assertEqual(mail.outbox[0].subject, "[CUSTOM] custom_foo")
207
208    def test_transport_email_validation(self):
209        """Test email transport template validation"""
210
211        # Test valid template
212        serializer = NotificationTransportSerializer(
213            data={
214                "name": generate_id(),
215                "mode": TransportMode.EMAIL,
216                "email_template": "email/event_notification.html",
217            }
218        )
219        self.assertTrue(serializer.is_valid())
220
221        # Test invalid template - should fail due to choices validation
222        serializer = NotificationTransportSerializer(
223            data={
224                "name": generate_id(),
225                "mode": TransportMode.EMAIL,
226                "email_template": "invalid/template.html",
227            }
228        )
229        self.assertFalse(serializer.is_valid())
230        self.assertIn("email_template", serializer.errors)
231
232    def test_templates_api_endpoint(self):
233        """Test templates API endpoint returns valid templates"""
234        self.client.force_login(self.user)
235        response = self.client.get(reverse("authentik_api:emailstage-templates"))
236        self.assertEqual(response.status_code, 200)
237
238        data = response.json()
239        self.assertIsInstance(data, list)
240
241        # Check that we have at least the default templates
242        template_names = [item["name"] for item in data]
243        self.assertIn("email/event_notification.html", template_names)
244
245        # Verify all templates are valid choices
246        valid_choices = dict(get_template_choices())
247        for template in data:
248            self.assertIn(template["name"], valid_choices)
249            self.assertEqual(template["description"], valid_choices[template["name"]])

Test Event Transports

def setUp(self) -> None: 
31    def setUp(self) -> None:
32        self.user = create_test_admin_user()
33        self.event = Event.new("foo", "testing", foo="bar,").set_user(self.user)
34        self.event.save()
35        self.notification = Notification.objects.create(
36            severity=NotificationSeverity.ALERT,
37            body="foo",
38            event=self.event,
39            user=self.user,
40        )

Hook method for setting up the test fixture before exercising it.

def test_transport_webhook(self): 
42    def test_transport_webhook(self):
43        """Test webhook transport"""
44        transport: NotificationTransport = NotificationTransport.objects.create(
45            name=generate_id(),
46            mode=TransportMode.WEBHOOK,
47            webhook_url="http://localhost:1234/test",
48        )
49        with Mocker() as mocker:
50            mocker.post("http://localhost:1234/test")
51            transport.send(self.notification)
52            self.assertEqual(mocker.call_count, 1)
53            self.assertEqual(mocker.request_history[0].method, "POST")
54            self.assertJSONEqual(
55                mocker.request_history[0].body.decode(),
56                {
57                    "body": "foo",
58                    "severity": "alert",
59                    "user_email": self.user.email,
60                    "user_username": self.user.username,
61                    "event_user_email": self.user.email,
62                    "event_user_username": self.user.username,
63                },
64            )

Test webhook transport

def test_transport_webhook_ca_invalid_unset(self): 
66    def test_transport_webhook_ca_invalid_unset(self):
67        """Test webhook transport"""
68        transport: NotificationTransport = NotificationTransport.objects.create(
69            name=generate_id(),
70            mode=TransportMode.WEBHOOK,
71            webhook_url="https://localhost:1234/test",
72        )
73        with Mocker() as mocker:
74            mocker.post("https://localhost:1234/test")
75            transport.send(self.notification)
76            self.assertEqual(mocker.call_count, 1)
77            self.assertTrue(mocker.request_history[0].verify)

Test webhook transport

def test_transport_webhook_ca(self): 
79    def test_transport_webhook_ca(self):
80        """Test webhook transport"""
81        kp = CertificateKeyPair.objects.create(
82            name=generate_id(),
83            certificate_data="foo",
84        )
85        transport: NotificationTransport = NotificationTransport.objects.create(
86            name=generate_id(),
87            mode=TransportMode.WEBHOOK,
88            webhook_url="https://localhost:1234/test",
89            webhook_ca=kp,
90        )
91        with Mocker() as mocker:
92            mocker.post("https://localhost:1234/test")
93            transport.send(self.notification)
94            self.assertEqual(mocker.call_count, 1)
95            self.assertIsNotNone(mocker.request_history[0].verify)

Test webhook transport

def test_transport_webhook_mapping(self): 
 97    def test_transport_webhook_mapping(self):
 98        """Test webhook transport with custom mapping"""
 99        mapping_body = NotificationWebhookMapping.objects.create(
100            name=generate_id(), expression="return request.user"
101        )
102        mapping_headers = NotificationWebhookMapping.objects.create(
103            name=generate_id(), expression="""return {"foo": "bar"}"""
104        )
105        transport: NotificationTransport = NotificationTransport.objects.create(
106            name=generate_id(),
107            mode=TransportMode.WEBHOOK,
108            webhook_url="http://localhost:1234/test",
109            webhook_mapping_body=mapping_body,
110            webhook_mapping_headers=mapping_headers,
111        )
112        with Mocker() as mocker:
113            mocker.post("http://localhost:1234/test")
114            transport.send(self.notification)
115            self.assertEqual(mocker.call_count, 1)
116            self.assertEqual(mocker.request_history[0].method, "POST")
117            self.assertEqual(mocker.request_history[0].headers["foo"], "bar")
118            self.assertJSONEqual(
119                mocker.request_history[0].body.decode(),
120                {"email": self.user.email, "pk": self.user.pk, "username": self.user.username},
121            )

Test webhook transport with custom mapping

def test_transport_webhook_slack(self): 
123    def test_transport_webhook_slack(self):
124        """Test webhook transport (slack)"""
125        transport: NotificationTransport = NotificationTransport.objects.create(
126            name=generate_id(),
127            mode=TransportMode.WEBHOOK_SLACK,
128            webhook_url="http://localhost:1234/test",
129        )
130        with Mocker() as mocker:
131            mocker.post("http://localhost:1234/test")
132            transport.send(self.notification)
133            self.assertEqual(mocker.call_count, 1)
134            self.assertEqual(mocker.request_history[0].method, "POST")
135            self.assertJSONEqual(
136                mocker.request_history[0].body.decode(),
137                {
138                    "username": "authentik",
139                    "icon_url": "https://goauthentik.io/img/icon.png",
140                    "attachments": [
141                        {
142                            "author_name": "authentik",
143                            "author_link": "https://goauthentik.io",
144                            "author_icon": "https://goauthentik.io/img/icon.png",
145                            "title": "custom_foo",
146                            "color": "#fd4b2d",
147                            "fields": [
148                                {"title": "Severity", "value": "alert", "short": True},
149                                {
150                                    "title": "Dispatched for user",
151                                    "value": self.user.username,
152                                    "short": True,
153                                },
154                                {"short": True, "title": "Event user", "value": self.user.username},
155                                {"title": "foo", "value": "bar,"},
156                            ],
157                            "footer": f"authentik {authentik_full_version()}",
158                        }
159                    ],
160                },
161            )

Test webhook transport (slack)

def test_transport_email(self): 
163    def test_transport_email(self):
164        """Test email transport"""
165        transport: NotificationTransport = NotificationTransport.objects.create(
166            name=generate_id(),
167            mode=TransportMode.EMAIL,
168        )
169        with patch(
170            "authentik.stages.email.models.EmailStage.backend_class",
171            PropertyMock(return_value=EmailBackend),
172        ):
173            transport.send(self.notification)
174            self.assertEqual(len(mail.outbox), 1)
175            self.assertEqual(mail.outbox[0].subject, "authentik Notification: custom_foo")
176            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])

Test email transport

def test_transport_email_custom_template(self): 
178    def test_transport_email_custom_template(self):
179        """Test email transport with custom template"""
180        transport: NotificationTransport = NotificationTransport.objects.create(
181            name=generate_id(),
182            mode=TransportMode.EMAIL,
183            email_template="email/event_notification.html",
184        )
185        with patch(
186            "authentik.stages.email.models.EmailStage.backend_class",
187            PropertyMock(return_value=EmailBackend),
188        ):
189            transport.send(self.notification)
190            self.assertEqual(len(mail.outbox), 1)
191            self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])

Test email transport with custom template

def test_transport_email_custom_subject_prefix(self): 
193    def test_transport_email_custom_subject_prefix(self):
194        """Test email transport with custom subject prefix"""
195        transport: NotificationTransport = NotificationTransport.objects.create(
196            name=generate_id(),
197            mode=TransportMode.EMAIL,
198            email_subject_prefix="[CUSTOM] ",
199        )
200        with patch(
201            "authentik.stages.email.models.EmailStage.backend_class",
202            PropertyMock(return_value=EmailBackend),
203        ):
204            transport.send(self.notification)
205            self.assertEqual(len(mail.outbox), 1)
206            self.assertEqual(mail.outbox[0].subject, "[CUSTOM] custom_foo")

Test email transport with custom subject prefix

def test_transport_email_validation(self): 
208    def test_transport_email_validation(self):
209        """Test email transport template validation"""
210
211        # Test valid template
212        serializer = NotificationTransportSerializer(
213            data={
214                "name": generate_id(),
215                "mode": TransportMode.EMAIL,
216                "email_template": "email/event_notification.html",
217            }
218        )
219        self.assertTrue(serializer.is_valid())
220
221        # Test invalid template - should fail due to choices validation
222        serializer = NotificationTransportSerializer(
223            data={
224                "name": generate_id(),
225                "mode": TransportMode.EMAIL,
226                "email_template": "invalid/template.html",
227            }
228        )
229        self.assertFalse(serializer.is_valid())
230        self.assertIn("email_template", serializer.errors)

Test email transport template validation

def test_templates_api_endpoint(self): 
232    def test_templates_api_endpoint(self):
233        """Test templates API endpoint returns valid templates"""
234        self.client.force_login(self.user)
235        response = self.client.get(reverse("authentik_api:emailstage-templates"))
236        self.assertEqual(response.status_code, 200)
237
238        data = response.json()
239        self.assertIsInstance(data, list)
240
241        # Check that we have at least the default templates
242        template_names = [item["name"] for item in data]
243        self.assertIn("email/event_notification.html", template_names)
244
245        # Verify all templates are valid choices
246        valid_choices = dict(get_template_choices())
247        for template in data:
248            self.assertIn(template["name"], valid_choices)
249            self.assertEqual(template["description"], valid_choices[template["name"]])

Test templates API endpoint returns valid templates