authentik.stages.email.tests.test_sending

email tests

View Source

  1"""email tests"""
  2
  3from smtplib import SMTPException
  4from unittest.mock import MagicMock, PropertyMock, patch
  5
  6from django.core import mail
  7from django.core.mail.backends.locmem import EmailBackend
  8from django.urls import reverse
  9
 10from authentik.core.models import User
 11from authentik.core.tests.utils import create_test_admin_user, create_test_flow, create_test_user
 12from authentik.events.models import Event, EventAction
 13from authentik.flows.markers import StageMarker
 14from authentik.flows.models import FlowDesignation, FlowStageBinding
 15from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 16from authentik.flows.tests import FlowTestCase
 17from authentik.flows.views.executor import SESSION_KEY_PLAN
 18from authentik.lib.generators import generate_id
 19from authentik.stages.email.models import EmailStage
 20
 21
 22class TestEmailStageSending(FlowTestCase):
 23    """Email tests"""
 24
 25    def setUp(self):
 26        super().setUp()
 27        self.user = create_test_admin_user()
 28
 29        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
 30        self.stage = EmailStage.objects.create(
 31            name="email",
 32        )
 33        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 34
 35    def test_pending_user(self):
 36        """Test with pending user"""
 37        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 38        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 39        session = self.client.session
 40        session[SESSION_KEY_PLAN] = plan
 41        session.save()
 42        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
 43
 44        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 45        with patch(
 46            "authentik.stages.email.models.EmailStage.backend_class",
 47            PropertyMock(return_value=EmailBackend),
 48        ):
 49            response = self.client.post(url)
 50            self.assertEqual(response.status_code, 200)
 51            self.assertStageResponse(
 52                response,
 53                self.flow,
 54                response_errors={
 55                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
 56                },
 57            )
 58            self.assertEqual(len(mail.outbox), 1)
 59            self.assertEqual(mail.outbox[0].subject, "authentik")
 60            events = Event.objects.filter(action=EventAction.EMAIL_SENT)
 61            self.assertEqual(len(events), 1)
 62            event = events.first()
 63            self.assertEqual(
 64                event.context["message"], f"Email to {self.user.name} <{self.user.email}> sent"
 65            )
 66            self.assertEqual(event.context["subject"], "authentik")
 67            self.assertEqual(event.context["to_email"], [f"{self.user.name} <{self.user.email}>"])
 68            self.assertEqual(event.context["from_email"], "system@authentik.local")
 69
 70    def test_newlines_long_name(self):
 71        """Test with pending user"""
 72        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 73        long_user = create_test_user()
 74        long_user.name = "Test User\r\n Many Words\r\n"
 75        long_user.save()
 76        plan.context[PLAN_CONTEXT_PENDING_USER] = long_user
 77        session = self.client.session
 78        session[SESSION_KEY_PLAN] = plan
 79        session.save()
 80        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
 81
 82        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 83        with patch(
 84            "authentik.stages.email.models.EmailStage.backend_class",
 85            PropertyMock(return_value=EmailBackend),
 86        ):
 87            response = self.client.post(url)
 88            self.assertEqual(response.status_code, 200)
 89            self.assertStageResponse(
 90                response,
 91                self.flow,
 92                response_errors={
 93                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
 94                },
 95            )
 96            self.assertEqual(len(mail.outbox), 1)
 97            self.assertEqual(mail.outbox[0].subject, "authentik")
 98            self.assertEqual(mail.outbox[0].to, [f"Test User   Many Words   <{long_user.email}>"])
 99
100    def test_utf8_name(self):
101        """Test with pending user"""
102        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
103        utf8_user = create_test_user()
104        utf8_user.name = "Cirilo ЉМНЊ el cirilico И̂ӢЙӤ "
105        utf8_user.email = "cyrillic@authentik.local"
106        utf8_user.save()
107        plan.context[PLAN_CONTEXT_PENDING_USER] = utf8_user
108        session = self.client.session
109        session[SESSION_KEY_PLAN] = plan
110        session.save()
111        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
112
113        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
114        with patch(
115            "authentik.stages.email.models.EmailStage.backend_class",
116            PropertyMock(return_value=EmailBackend),
117        ):
118            response = self.client.post(url)
119            self.assertEqual(response.status_code, 200)
120            self.assertStageResponse(
121                response,
122                self.flow,
123                response_errors={
124                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
125                },
126            )
127            self.assertEqual(len(mail.outbox), 1)
128            self.assertEqual(mail.outbox[0].subject, "authentik")
129            self.assertEqual(mail.outbox[0].to, [f"{utf8_user.email}"])
130
131    def test_pending_fake_user(self):
132        """Test with pending (fake) user"""
133        self.flow.designation = FlowDesignation.RECOVERY
134        self.flow.save()
135        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
136        plan.context[PLAN_CONTEXT_PENDING_USER] = User(username=generate_id())
137        session = self.client.session
138        session[SESSION_KEY_PLAN] = plan
139        session.save()
140
141        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
142        with patch(
143            "authentik.stages.email.models.EmailStage.backend_class",
144            PropertyMock(return_value=EmailBackend),
145        ):
146            response = self.client.post(url)
147            self.assertEqual(response.status_code, 200)
148            self.assertStageResponse(
149                response,
150                self.flow,
151                response_errors={
152                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
153                },
154            )
155            self.assertEqual(len(mail.outbox), 0)
156
157    def test_send_error(self):
158        """Test error during sending (sending will be retried)"""
159        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
160        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
161        session = self.client.session
162        session[SESSION_KEY_PLAN] = plan
163        session.save()
164
165        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
166        with patch(
167            "authentik.stages.email.models.EmailStage.backend_class",
168            PropertyMock(return_value=EmailBackend),
169        ):
170            with patch(
171                "django.core.mail.backends.locmem.EmailBackend.send_messages",
172                MagicMock(side_effect=[SMTPException, EmailBackend.send_messages]),
173            ):
174                response = self.client.post(url)
175            response = self.client.post(url)
176            self.assertEqual(response.status_code, 200)
177            self.assertGreaterEqual(len(mail.outbox), 1)
178            self.assertEqual(mail.outbox[0].subject, "authentik")

class TestEmailStageSending(authentik.flows.tests.FlowTestCase): View Source

 23class TestEmailStageSending(FlowTestCase):
 24    """Email tests"""
 25
 26    def setUp(self):
 27        super().setUp()
 28        self.user = create_test_admin_user()
 29
 30        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
 31        self.stage = EmailStage.objects.create(
 32            name="email",
 33        )
 34        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)
 35
 36    def test_pending_user(self):
 37        """Test with pending user"""
 38        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 39        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
 40        session = self.client.session
 41        session[SESSION_KEY_PLAN] = plan
 42        session.save()
 43        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
 44
 45        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 46        with patch(
 47            "authentik.stages.email.models.EmailStage.backend_class",
 48            PropertyMock(return_value=EmailBackend),
 49        ):
 50            response = self.client.post(url)
 51            self.assertEqual(response.status_code, 200)
 52            self.assertStageResponse(
 53                response,
 54                self.flow,
 55                response_errors={
 56                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
 57                },
 58            )
 59            self.assertEqual(len(mail.outbox), 1)
 60            self.assertEqual(mail.outbox[0].subject, "authentik")
 61            events = Event.objects.filter(action=EventAction.EMAIL_SENT)
 62            self.assertEqual(len(events), 1)
 63            event = events.first()
 64            self.assertEqual(
 65                event.context["message"], f"Email to {self.user.name} <{self.user.email}> sent"
 66            )
 67            self.assertEqual(event.context["subject"], "authentik")
 68            self.assertEqual(event.context["to_email"], [f"{self.user.name} <{self.user.email}>"])
 69            self.assertEqual(event.context["from_email"], "system@authentik.local")
 70
 71    def test_newlines_long_name(self):
 72        """Test with pending user"""
 73        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
 74        long_user = create_test_user()
 75        long_user.name = "Test User\r\n Many Words\r\n"
 76        long_user.save()
 77        plan.context[PLAN_CONTEXT_PENDING_USER] = long_user
 78        session = self.client.session
 79        session[SESSION_KEY_PLAN] = plan
 80        session.save()
 81        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
 82
 83        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
 84        with patch(
 85            "authentik.stages.email.models.EmailStage.backend_class",
 86            PropertyMock(return_value=EmailBackend),
 87        ):
 88            response = self.client.post(url)
 89            self.assertEqual(response.status_code, 200)
 90            self.assertStageResponse(
 91                response,
 92                self.flow,
 93                response_errors={
 94                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
 95                },
 96            )
 97            self.assertEqual(len(mail.outbox), 1)
 98            self.assertEqual(mail.outbox[0].subject, "authentik")
 99            self.assertEqual(mail.outbox[0].to, [f"Test User   Many Words   <{long_user.email}>"])
100
101    def test_utf8_name(self):
102        """Test with pending user"""
103        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
104        utf8_user = create_test_user()
105        utf8_user.name = "Cirilo ЉМНЊ el cirilico И̂ӢЙӤ "
106        utf8_user.email = "cyrillic@authentik.local"
107        utf8_user.save()
108        plan.context[PLAN_CONTEXT_PENDING_USER] = utf8_user
109        session = self.client.session
110        session[SESSION_KEY_PLAN] = plan
111        session.save()
112        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
113
114        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
115        with patch(
116            "authentik.stages.email.models.EmailStage.backend_class",
117            PropertyMock(return_value=EmailBackend),
118        ):
119            response = self.client.post(url)
120            self.assertEqual(response.status_code, 200)
121            self.assertStageResponse(
122                response,
123                self.flow,
124                response_errors={
125                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
126                },
127            )
128            self.assertEqual(len(mail.outbox), 1)
129            self.assertEqual(mail.outbox[0].subject, "authentik")
130            self.assertEqual(mail.outbox[0].to, [f"{utf8_user.email}"])
131
132    def test_pending_fake_user(self):
133        """Test with pending (fake) user"""
134        self.flow.designation = FlowDesignation.RECOVERY
135        self.flow.save()
136        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
137        plan.context[PLAN_CONTEXT_PENDING_USER] = User(username=generate_id())
138        session = self.client.session
139        session[SESSION_KEY_PLAN] = plan
140        session.save()
141
142        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
143        with patch(
144            "authentik.stages.email.models.EmailStage.backend_class",
145            PropertyMock(return_value=EmailBackend),
146        ):
147            response = self.client.post(url)
148            self.assertEqual(response.status_code, 200)
149            self.assertStageResponse(
150                response,
151                self.flow,
152                response_errors={
153                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
154                },
155            )
156            self.assertEqual(len(mail.outbox), 0)
157
158    def test_send_error(self):
159        """Test error during sending (sending will be retried)"""
160        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
161        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
162        session = self.client.session
163        session[SESSION_KEY_PLAN] = plan
164        session.save()
165
166        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
167        with patch(
168            "authentik.stages.email.models.EmailStage.backend_class",
169            PropertyMock(return_value=EmailBackend),
170        ):
171            with patch(
172                "django.core.mail.backends.locmem.EmailBackend.send_messages",
173                MagicMock(side_effect=[SMTPException, EmailBackend.send_messages]),
174            ):
175                response = self.client.post(url)
176            response = self.client.post(url)
177            self.assertEqual(response.status_code, 200)
178            self.assertGreaterEqual(len(mail.outbox), 1)
179            self.assertEqual(mail.outbox[0].subject, "authentik")

Email tests

def setUp(self): View Source

26    def setUp(self):
27        super().setUp()
28        self.user = create_test_admin_user()
29
30        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
31        self.stage = EmailStage.objects.create(
32            name="email",
33        )
34        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=2)

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

def test_pending_user(self): View Source

36    def test_pending_user(self):
37        """Test with pending user"""
38        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
39        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
40        session = self.client.session
41        session[SESSION_KEY_PLAN] = plan
42        session.save()
43        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
44
45        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
46        with patch(
47            "authentik.stages.email.models.EmailStage.backend_class",
48            PropertyMock(return_value=EmailBackend),
49        ):
50            response = self.client.post(url)
51            self.assertEqual(response.status_code, 200)
52            self.assertStageResponse(
53                response,
54                self.flow,
55                response_errors={
56                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
57                },
58            )
59            self.assertEqual(len(mail.outbox), 1)
60            self.assertEqual(mail.outbox[0].subject, "authentik")
61            events = Event.objects.filter(action=EventAction.EMAIL_SENT)
62            self.assertEqual(len(events), 1)
63            event = events.first()
64            self.assertEqual(
65                event.context["message"], f"Email to {self.user.name} <{self.user.email}> sent"
66            )
67            self.assertEqual(event.context["subject"], "authentik")
68            self.assertEqual(event.context["to_email"], [f"{self.user.name} <{self.user.email}>"])
69            self.assertEqual(event.context["from_email"], "system@authentik.local")

Test with pending user

def test_newlines_long_name(self): View Source

71    def test_newlines_long_name(self):
72        """Test with pending user"""
73        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
74        long_user = create_test_user()
75        long_user.name = "Test User\r\n Many Words\r\n"
76        long_user.save()
77        plan.context[PLAN_CONTEXT_PENDING_USER] = long_user
78        session = self.client.session
79        session[SESSION_KEY_PLAN] = plan
80        session.save()
81        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
82
83        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
84        with patch(
85            "authentik.stages.email.models.EmailStage.backend_class",
86            PropertyMock(return_value=EmailBackend),
87        ):
88            response = self.client.post(url)
89            self.assertEqual(response.status_code, 200)
90            self.assertStageResponse(
91                response,
92                self.flow,
93                response_errors={
94                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
95                },
96            )
97            self.assertEqual(len(mail.outbox), 1)
98            self.assertEqual(mail.outbox[0].subject, "authentik")
99            self.assertEqual(mail.outbox[0].to, [f"Test User   Many Words   <{long_user.email}>"])

Test with pending user

def test_utf8_name(self): View Source

101    def test_utf8_name(self):
102        """Test with pending user"""
103        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
104        utf8_user = create_test_user()
105        utf8_user.name = "Cirilo ЉМНЊ el cirilico И̂ӢЙӤ "
106        utf8_user.email = "cyrillic@authentik.local"
107        utf8_user.save()
108        plan.context[PLAN_CONTEXT_PENDING_USER] = utf8_user
109        session = self.client.session
110        session[SESSION_KEY_PLAN] = plan
111        session.save()
112        Event.objects.filter(action=EventAction.EMAIL_SENT).delete()
113
114        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
115        with patch(
116            "authentik.stages.email.models.EmailStage.backend_class",
117            PropertyMock(return_value=EmailBackend),
118        ):
119            response = self.client.post(url)
120            self.assertEqual(response.status_code, 200)
121            self.assertStageResponse(
122                response,
123                self.flow,
124                response_errors={
125                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
126                },
127            )
128            self.assertEqual(len(mail.outbox), 1)
129            self.assertEqual(mail.outbox[0].subject, "authentik")
130            self.assertEqual(mail.outbox[0].to, [f"{utf8_user.email}"])

Test with pending user

def test_pending_fake_user(self): View Source

132    def test_pending_fake_user(self):
133        """Test with pending (fake) user"""
134        self.flow.designation = FlowDesignation.RECOVERY
135        self.flow.save()
136        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
137        plan.context[PLAN_CONTEXT_PENDING_USER] = User(username=generate_id())
138        session = self.client.session
139        session[SESSION_KEY_PLAN] = plan
140        session.save()
141
142        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
143        with patch(
144            "authentik.stages.email.models.EmailStage.backend_class",
145            PropertyMock(return_value=EmailBackend),
146        ):
147            response = self.client.post(url)
148            self.assertEqual(response.status_code, 200)
149            self.assertStageResponse(
150                response,
151                self.flow,
152                response_errors={
153                    "non_field_errors": [{"string": "email-sent", "code": "email-sent"}]
154                },
155            )
156            self.assertEqual(len(mail.outbox), 0)

Test with pending (fake) user

def test_send_error(self): View Source

158    def test_send_error(self):
159        """Test error during sending (sending will be retried)"""
160        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
161        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
162        session = self.client.session
163        session[SESSION_KEY_PLAN] = plan
164        session.save()
165
166        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
167        with patch(
168            "authentik.stages.email.models.EmailStage.backend_class",
169            PropertyMock(return_value=EmailBackend),
170        ):
171            with patch(
172                "django.core.mail.backends.locmem.EmailBackend.send_messages",
173                MagicMock(side_effect=[SMTPException, EmailBackend.send_messages]),
174            ):
175                response = self.client.post(url)
176            response = self.client.post(url)
177            self.assertEqual(response.status_code, 200)
178            self.assertGreaterEqual(len(mail.outbox), 1)
179            self.assertEqual(mail.outbox[0].subject, "authentik")

Test error during sending (sending will be retried)

Inherited Members

authentik.flows.tests.FlowTestCase: assertStageResponse; get_flow_plan; set_flow_plan; assertStageRedirects; assertFlowFinishes