authentik.events.tests.test_notifications

Notification tests

  1"""Notification tests"""
  2
  3from unittest.mock import MagicMock, patch
  4
  5from django.urls import reverse
  6from rest_framework.test import APITestCase
  7
  8from authentik.core.models import Group, User
  9from authentik.core.tests.utils import create_test_user
 10from authentik.events.models import (
 11    Event,
 12    EventAction,
 13    Notification,
 14    NotificationRule,
 15    NotificationSeverity,
 16    NotificationTransport,
 17    NotificationWebhookMapping,
 18    TransportMode,
 19)
 20from authentik.lib.generators import generate_id
 21from authentik.policies.event_matcher.models import EventMatcherPolicy
 22from authentik.policies.exceptions import PolicyException
 23from authentik.policies.models import PolicyBinding
 24
 25
 26class TestEventsNotifications(APITestCase):
 27    """Test Event Notifications"""
 28
 29    def setUp(self) -> None:
 30        self.group = Group.objects.create(name="test-group")
 31        self.user = User.objects.create(name="test-user", username="test")
 32        self.group.users.add(self.user)
 33        self.group.save()
 34
 35    def test_trigger_empty(self):
 36        """Test trigger without any policies attached"""
 37        transport = NotificationTransport.objects.create(name=generate_id())
 38        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
 39        trigger.transports.add(transport)
 40        trigger.save()
 41
 42        execute_mock = MagicMock()
 43        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 44            Event.new(EventAction.CUSTOM_PREFIX).save()
 45        self.assertEqual(execute_mock.call_count, 0)
 46
 47    def test_trigger_single(self):
 48        """Test simple transport triggering"""
 49        transport = NotificationTransport.objects.create(name=generate_id())
 50        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
 51        trigger.transports.add(transport)
 52        trigger.save()
 53        matcher = EventMatcherPolicy.objects.create(
 54            name="matcher", action=EventAction.CUSTOM_PREFIX
 55        )
 56        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 57
 58        execute_mock = MagicMock()
 59        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 60            Event.new(EventAction.CUSTOM_PREFIX).save()
 61        self.assertEqual(execute_mock.call_count, 1)
 62
 63    def test_trigger_event_user(self):
 64        """Test trigger with event user"""
 65        user = create_test_user()
 66        transport = NotificationTransport.objects.create(name=generate_id())
 67        trigger = NotificationRule.objects.create(name=generate_id(), destination_event_user=True)
 68        trigger.transports.add(transport)
 69        trigger.save()
 70        matcher = EventMatcherPolicy.objects.create(
 71            name="matcher", action=EventAction.CUSTOM_PREFIX
 72        )
 73        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 74
 75        execute_mock = MagicMock()
 76        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 77            Event.new(EventAction.CUSTOM_PREFIX).set_user(user).save()
 78        self.assertEqual(execute_mock.call_count, 1)
 79        notification: Notification = execute_mock.call_args[0][0]
 80        self.assertEqual(notification.user, user)
 81
 82    def test_trigger_no_group(self):
 83        """Test trigger without group"""
 84        trigger = NotificationRule.objects.create(name=generate_id())
 85        matcher = EventMatcherPolicy.objects.create(
 86            name="matcher", action=EventAction.CUSTOM_PREFIX
 87        )
 88        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 89
 90        execute_mock = MagicMock()
 91        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 92            Event.new(EventAction.CUSTOM_PREFIX).save()
 93        self.assertEqual(execute_mock.call_count, 0)
 94
 95    def test_policy_error_recursive(self):
 96        """Test Policy error which would cause recursion"""
 97        transport = NotificationTransport.objects.create(name=generate_id())
 98        NotificationRule.objects.filter(name__startswith="default").delete()
 99        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
100        trigger.transports.add(transport)
101        trigger.save()
102        matcher = EventMatcherPolicy.objects.create(
103            name="matcher", action=EventAction.CUSTOM_PREFIX
104        )
105        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
106
107        execute_mock = MagicMock()
108        passes = MagicMock(side_effect=PolicyException)
109        with patch("authentik.policies.event_matcher.models.EventMatcherPolicy.passes", passes):
110            with patch("authentik.events.models.NotificationTransport.send", execute_mock):
111                Event.new(EventAction.CUSTOM_PREFIX).save()
112        self.assertEqual(passes.call_count, 1)
113
114    def test_transport_once(self):
115        """Test transport's send_once"""
116        user2 = User.objects.create(name="test2-user", username="test2")
117        self.group.users.add(user2)
118        self.group.save()
119
120        transport = NotificationTransport.objects.create(name=generate_id(), send_once=True)
121        NotificationRule.objects.filter(name__startswith="default").delete()
122        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
123        trigger.transports.add(transport)
124        trigger.save()
125        matcher = EventMatcherPolicy.objects.create(
126            name="matcher", action=EventAction.CUSTOM_PREFIX
127        )
128        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
129
130        execute_mock = MagicMock()
131        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
132            Event.new(EventAction.CUSTOM_PREFIX).save()
133        self.assertEqual(execute_mock.call_count, 1)
134
135    def test_transport_mapping(self):
136        """Test transport mapping"""
137        mapping = NotificationWebhookMapping.objects.create(
138            name=generate_id(),
139            expression="""notification.body = 'foo'""",
140        )
141
142        transport = NotificationTransport.objects.create(
143            name=generate_id(), webhook_mapping_body=mapping, mode=TransportMode.LOCAL
144        )
145        NotificationRule.objects.filter(name__startswith="default").delete()
146        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
147        trigger.transports.add(transport)
148        matcher = EventMatcherPolicy.objects.create(
149            name="matcher", action=EventAction.CUSTOM_PREFIX
150        )
151        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
152
153        Notification.objects.all().delete()
154        Event.new(EventAction.CUSTOM_PREFIX).save()
155        self.assertEqual(Notification.objects.first().body, "foo")
156
157    def test_api_mark_all_seen(self):
158        """Test mark_all_seen"""
159        self.client.force_login(self.user)
160
161        Notification.objects.create(
162            severity=NotificationSeverity.NOTICE, body="foo", user=self.user, seen=False
163        )
164
165        response = self.client.post(reverse("authentik_api:notification-mark-all-seen"))
166        self.assertEqual(response.status_code, 204)
167        self.assertFalse(Notification.objects.filter(body="foo", seen=False).exists())
class TestEventsNotifications(rest_framework.test.APITestCase): 
 27class TestEventsNotifications(APITestCase):
 28    """Test Event Notifications"""
 29
 30    def setUp(self) -> None:
 31        self.group = Group.objects.create(name="test-group")
 32        self.user = User.objects.create(name="test-user", username="test")
 33        self.group.users.add(self.user)
 34        self.group.save()
 35
 36    def test_trigger_empty(self):
 37        """Test trigger without any policies attached"""
 38        transport = NotificationTransport.objects.create(name=generate_id())
 39        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
 40        trigger.transports.add(transport)
 41        trigger.save()
 42
 43        execute_mock = MagicMock()
 44        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 45            Event.new(EventAction.CUSTOM_PREFIX).save()
 46        self.assertEqual(execute_mock.call_count, 0)
 47
 48    def test_trigger_single(self):
 49        """Test simple transport triggering"""
 50        transport = NotificationTransport.objects.create(name=generate_id())
 51        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
 52        trigger.transports.add(transport)
 53        trigger.save()
 54        matcher = EventMatcherPolicy.objects.create(
 55            name="matcher", action=EventAction.CUSTOM_PREFIX
 56        )
 57        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 58
 59        execute_mock = MagicMock()
 60        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 61            Event.new(EventAction.CUSTOM_PREFIX).save()
 62        self.assertEqual(execute_mock.call_count, 1)
 63
 64    def test_trigger_event_user(self):
 65        """Test trigger with event user"""
 66        user = create_test_user()
 67        transport = NotificationTransport.objects.create(name=generate_id())
 68        trigger = NotificationRule.objects.create(name=generate_id(), destination_event_user=True)
 69        trigger.transports.add(transport)
 70        trigger.save()
 71        matcher = EventMatcherPolicy.objects.create(
 72            name="matcher", action=EventAction.CUSTOM_PREFIX
 73        )
 74        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 75
 76        execute_mock = MagicMock()
 77        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 78            Event.new(EventAction.CUSTOM_PREFIX).set_user(user).save()
 79        self.assertEqual(execute_mock.call_count, 1)
 80        notification: Notification = execute_mock.call_args[0][0]
 81        self.assertEqual(notification.user, user)
 82
 83    def test_trigger_no_group(self):
 84        """Test trigger without group"""
 85        trigger = NotificationRule.objects.create(name=generate_id())
 86        matcher = EventMatcherPolicy.objects.create(
 87            name="matcher", action=EventAction.CUSTOM_PREFIX
 88        )
 89        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
 90
 91        execute_mock = MagicMock()
 92        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
 93            Event.new(EventAction.CUSTOM_PREFIX).save()
 94        self.assertEqual(execute_mock.call_count, 0)
 95
 96    def test_policy_error_recursive(self):
 97        """Test Policy error which would cause recursion"""
 98        transport = NotificationTransport.objects.create(name=generate_id())
 99        NotificationRule.objects.filter(name__startswith="default").delete()
100        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
101        trigger.transports.add(transport)
102        trigger.save()
103        matcher = EventMatcherPolicy.objects.create(
104            name="matcher", action=EventAction.CUSTOM_PREFIX
105        )
106        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
107
108        execute_mock = MagicMock()
109        passes = MagicMock(side_effect=PolicyException)
110        with patch("authentik.policies.event_matcher.models.EventMatcherPolicy.passes", passes):
111            with patch("authentik.events.models.NotificationTransport.send", execute_mock):
112                Event.new(EventAction.CUSTOM_PREFIX).save()
113        self.assertEqual(passes.call_count, 1)
114
115    def test_transport_once(self):
116        """Test transport's send_once"""
117        user2 = User.objects.create(name="test2-user", username="test2")
118        self.group.users.add(user2)
119        self.group.save()
120
121        transport = NotificationTransport.objects.create(name=generate_id(), send_once=True)
122        NotificationRule.objects.filter(name__startswith="default").delete()
123        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
124        trigger.transports.add(transport)
125        trigger.save()
126        matcher = EventMatcherPolicy.objects.create(
127            name="matcher", action=EventAction.CUSTOM_PREFIX
128        )
129        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
130
131        execute_mock = MagicMock()
132        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
133            Event.new(EventAction.CUSTOM_PREFIX).save()
134        self.assertEqual(execute_mock.call_count, 1)
135
136    def test_transport_mapping(self):
137        """Test transport mapping"""
138        mapping = NotificationWebhookMapping.objects.create(
139            name=generate_id(),
140            expression="""notification.body = 'foo'""",
141        )
142
143        transport = NotificationTransport.objects.create(
144            name=generate_id(), webhook_mapping_body=mapping, mode=TransportMode.LOCAL
145        )
146        NotificationRule.objects.filter(name__startswith="default").delete()
147        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
148        trigger.transports.add(transport)
149        matcher = EventMatcherPolicy.objects.create(
150            name="matcher", action=EventAction.CUSTOM_PREFIX
151        )
152        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
153
154        Notification.objects.all().delete()
155        Event.new(EventAction.CUSTOM_PREFIX).save()
156        self.assertEqual(Notification.objects.first().body, "foo")
157
158    def test_api_mark_all_seen(self):
159        """Test mark_all_seen"""
160        self.client.force_login(self.user)
161
162        Notification.objects.create(
163            severity=NotificationSeverity.NOTICE, body="foo", user=self.user, seen=False
164        )
165
166        response = self.client.post(reverse("authentik_api:notification-mark-all-seen"))
167        self.assertEqual(response.status_code, 204)
168        self.assertFalse(Notification.objects.filter(body="foo", seen=False).exists())

Test Event Notifications

def setUp(self) -> None: 
30    def setUp(self) -> None:
31        self.group = Group.objects.create(name="test-group")
32        self.user = User.objects.create(name="test-user", username="test")
33        self.group.users.add(self.user)
34        self.group.save()

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

def test_trigger_empty(self): 
36    def test_trigger_empty(self):
37        """Test trigger without any policies attached"""
38        transport = NotificationTransport.objects.create(name=generate_id())
39        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
40        trigger.transports.add(transport)
41        trigger.save()
42
43        execute_mock = MagicMock()
44        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
45            Event.new(EventAction.CUSTOM_PREFIX).save()
46        self.assertEqual(execute_mock.call_count, 0)

Test trigger without any policies attached

def test_trigger_single(self): 
48    def test_trigger_single(self):
49        """Test simple transport triggering"""
50        transport = NotificationTransport.objects.create(name=generate_id())
51        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
52        trigger.transports.add(transport)
53        trigger.save()
54        matcher = EventMatcherPolicy.objects.create(
55            name="matcher", action=EventAction.CUSTOM_PREFIX
56        )
57        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
58
59        execute_mock = MagicMock()
60        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
61            Event.new(EventAction.CUSTOM_PREFIX).save()
62        self.assertEqual(execute_mock.call_count, 1)

Test simple transport triggering

def test_trigger_event_user(self): 
64    def test_trigger_event_user(self):
65        """Test trigger with event user"""
66        user = create_test_user()
67        transport = NotificationTransport.objects.create(name=generate_id())
68        trigger = NotificationRule.objects.create(name=generate_id(), destination_event_user=True)
69        trigger.transports.add(transport)
70        trigger.save()
71        matcher = EventMatcherPolicy.objects.create(
72            name="matcher", action=EventAction.CUSTOM_PREFIX
73        )
74        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
75
76        execute_mock = MagicMock()
77        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
78            Event.new(EventAction.CUSTOM_PREFIX).set_user(user).save()
79        self.assertEqual(execute_mock.call_count, 1)
80        notification: Notification = execute_mock.call_args[0][0]
81        self.assertEqual(notification.user, user)

Test trigger with event user

def test_trigger_no_group(self): 
83    def test_trigger_no_group(self):
84        """Test trigger without group"""
85        trigger = NotificationRule.objects.create(name=generate_id())
86        matcher = EventMatcherPolicy.objects.create(
87            name="matcher", action=EventAction.CUSTOM_PREFIX
88        )
89        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
90
91        execute_mock = MagicMock()
92        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
93            Event.new(EventAction.CUSTOM_PREFIX).save()
94        self.assertEqual(execute_mock.call_count, 0)

Test trigger without group

def test_policy_error_recursive(self): 
 96    def test_policy_error_recursive(self):
 97        """Test Policy error which would cause recursion"""
 98        transport = NotificationTransport.objects.create(name=generate_id())
 99        NotificationRule.objects.filter(name__startswith="default").delete()
100        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
101        trigger.transports.add(transport)
102        trigger.save()
103        matcher = EventMatcherPolicy.objects.create(
104            name="matcher", action=EventAction.CUSTOM_PREFIX
105        )
106        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
107
108        execute_mock = MagicMock()
109        passes = MagicMock(side_effect=PolicyException)
110        with patch("authentik.policies.event_matcher.models.EventMatcherPolicy.passes", passes):
111            with patch("authentik.events.models.NotificationTransport.send", execute_mock):
112                Event.new(EventAction.CUSTOM_PREFIX).save()
113        self.assertEqual(passes.call_count, 1)

Test Policy error which would cause recursion

def test_transport_once(self): 
115    def test_transport_once(self):
116        """Test transport's send_once"""
117        user2 = User.objects.create(name="test2-user", username="test2")
118        self.group.users.add(user2)
119        self.group.save()
120
121        transport = NotificationTransport.objects.create(name=generate_id(), send_once=True)
122        NotificationRule.objects.filter(name__startswith="default").delete()
123        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
124        trigger.transports.add(transport)
125        trigger.save()
126        matcher = EventMatcherPolicy.objects.create(
127            name="matcher", action=EventAction.CUSTOM_PREFIX
128        )
129        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
130
131        execute_mock = MagicMock()
132        with patch("authentik.events.models.NotificationTransport.send", execute_mock):
133            Event.new(EventAction.CUSTOM_PREFIX).save()
134        self.assertEqual(execute_mock.call_count, 1)

Test transport's send_once

def test_transport_mapping(self): 
136    def test_transport_mapping(self):
137        """Test transport mapping"""
138        mapping = NotificationWebhookMapping.objects.create(
139            name=generate_id(),
140            expression="""notification.body = 'foo'""",
141        )
142
143        transport = NotificationTransport.objects.create(
144            name=generate_id(), webhook_mapping_body=mapping, mode=TransportMode.LOCAL
145        )
146        NotificationRule.objects.filter(name__startswith="default").delete()
147        trigger = NotificationRule.objects.create(name=generate_id(), destination_group=self.group)
148        trigger.transports.add(transport)
149        matcher = EventMatcherPolicy.objects.create(
150            name="matcher", action=EventAction.CUSTOM_PREFIX
151        )
152        PolicyBinding.objects.create(target=trigger, policy=matcher, order=0)
153
154        Notification.objects.all().delete()
155        Event.new(EventAction.CUSTOM_PREFIX).save()
156        self.assertEqual(Notification.objects.first().body, "foo")

Test transport mapping

def test_api_mark_all_seen(self): 
158    def test_api_mark_all_seen(self):
159        """Test mark_all_seen"""
160        self.client.force_login(self.user)
161
162        Notification.objects.create(
163            severity=NotificationSeverity.NOTICE, body="foo", user=self.user, seen=False
164        )
165
166        response = self.client.post(reverse("authentik_api:notification-mark-all-seen"))
167        self.assertEqual(response.status_code, 204)
168        self.assertFalse(Notification.objects.filter(body="foo", seen=False).exists())

Test mark_all_seen