authentik.events.tests.test_api

  1"""Event API tests"""
  2
  3from datetime import timedelta
  4from json import loads
  5
  6from django.urls import reverse
  7from django.utils.datastructures import MultiValueDict
  8from django.utils.http import urlencode
  9from django.utils.timezone import now
 10from rest_framework.test import APITestCase
 11
 12from authentik.core.tests.utils import create_test_admin_user
 13from authentik.events.models import (
 14    Event,
 15    EventAction,
 16    Notification,
 17    NotificationSeverity,
 18    TransportMode,
 19)
 20from authentik.events.utils import model_to_dict
 21from authentik.lib.generators import generate_id
 22from authentik.providers.oauth2.models import OAuth2Provider
 23
 24
 25class TestEventsAPI(APITestCase):
 26    """Test Event API"""
 27
 28    def setUp(self) -> None:
 29        self.user = create_test_admin_user()
 30        self.client.force_login(self.user)
 31
 32    def test_filter_model_pk_int(self):
 33        """Test event list with context_model_pk and integer PKs"""
 34        provider = OAuth2Provider.objects.create(
 35            name=generate_id(),
 36        )
 37        event = Event.new(EventAction.MODEL_CREATED, model=model_to_dict(provider))
 38        event.save()
 39        response = self.client.get(
 40            reverse("authentik_api:event-list"),
 41            data={
 42                "context_model_pk": provider.pk,
 43                "context_model_app": "authentik_providers_oauth2",
 44                "context_model_name": "oauth2provider",
 45            },
 46        )
 47        self.assertEqual(response.status_code, 200)
 48        body = loads(response.content)
 49        self.assertEqual(body["pagination"]["count"], 1)
 50
 51    def test_top_n(self):
 52        """Test top_per_user"""
 53        event = Event.new(EventAction.AUTHORIZE_APPLICATION)
 54        event.save()  # We save to ensure nothing is un-saveable
 55        response = self.client.get(
 56            reverse("authentik_api:event-top-per-user"),
 57            data={"filter_action": EventAction.AUTHORIZE_APPLICATION},
 58        )
 59        self.assertEqual(response.status_code, 200)
 60
 61    def test_actions(self):
 62        """Test actions"""
 63        response = self.client.get(
 64            reverse("authentik_api:event-actions"),
 65        )
 66        self.assertEqual(response.status_code, 200)
 67
 68    def test_notifications(self):
 69        """Test notifications"""
 70        notification = Notification.objects.create(
 71            user=self.user, severity=NotificationSeverity.ALERT, body="", seen=False
 72        )
 73        self.client.post(
 74            reverse("authentik_api:notification-mark-all-seen"),
 75        )
 76        notification.refresh_from_db()
 77        self.assertTrue(notification.seen)
 78
 79    def test_transport(self):
 80        """Test transport API"""
 81        response = self.client.post(
 82            reverse("authentik_api:notificationtransport-list"),
 83            data={
 84                "name": "foo-with",
 85                "mode": TransportMode.WEBHOOK,
 86                "webhook_url": "http://foo.com",
 87            },
 88        )
 89        self.assertEqual(response.status_code, 201)
 90        response = self.client.post(
 91            reverse("authentik_api:notificationtransport-list"),
 92            data={
 93                "name": "foo-without",
 94                "mode": TransportMode.WEBHOOK,
 95            },
 96        )
 97        self.assertEqual(response.status_code, 400)
 98
 99    def test_volume(self):
100        Event.objects.all().delete()
101        Event.new(EventAction.LOGIN).set_user(self.user).save()
102        evt = Event.new(EventAction.LOGIN).set_user(self.user)
103        evt.created = now() - timedelta(days=6)
104        evt.save()
105        res = self.client.get(
106            reverse("authentik_api:event-volume")
107            + "?"
108            + urlencode(
109                {
110                    "action": EventAction.LOGIN,
111                }
112            )
113        )
114        self.assertEqual(res.status_code, 200)
115        data = loads(res.content)
116        self.assertEqual(len(data), 1)
117
118    def test_stats(self):
119        Event.objects.all().delete()
120        Event.new(EventAction.LOGIN).set_user(self.user).save()
121        evt = Event.new(EventAction.LOGIN).set_user(self.user)
122        evt.created = now() - timedelta(days=6)
123        evt.save()
124        res = self.client.get(
125            reverse("authentik_api:event-stats")
126            + "?"
127            + urlencode(
128                MultiValueDict({"count_steps": ["hours=24", "days=7", "days=240"]}), doseq=True
129            )
130        )
131        self.assertEqual(res.status_code, 200, res.content)
132        self.assertJSONEqual(
133            res.content, {"unique_users": 1, "count_step": {"hours24": 2, "days7": 2, "days240": 2}}
134        )
135
136    def test_stats_invalid(self):
137        res = self.client.get(
138            reverse("authentik_api:event-stats")
139            + "?"
140            + urlencode({"count_steps": "24d"}, doseq=True)
141        )
142        self.assertEqual(res.status_code, 400)
143        self.assertJSONEqual(
144            res.content,
145            {"count_steps": {"0": ["24d is not in the correct format of 'hours=3;minutes=1'."]}},
146        )