lib: add tests for ak_create_event (#5710)

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
# Conflicts:
#	locale/en/LC_MESSAGES/django.po

authentik/events/models.py

@@ -7,7 +7,6 @@ from smtplib import SMTPException
 from typing import TYPE_CHECKING, Optional
 from uuid import uuid4
 
-from django.conf import settings
 from django.db import models
 from django.db.models import Count, ExpressionWrapper, F
 from django.db.models.fields import DurationField
@@ -207,9 +206,7 @@ class Event(SerializerModel, ExpiringModel):
         self.user = get_user(user)
         return self
 
-    def from_http(
-        self, request: HttpRequest, user: Optional[settings.AUTH_USER_MODEL] = None
-    ) -> "Event":
+    def from_http(self, request: HttpRequest, user: Optional[User] = None) -> "Event":
         """Add data from a Django-HttpRequest, allowing the creation of
         Events independently from requests.
         `user` arguments optionally overrides user from requests."""

authentik/lib/expression/evaluator.py

@@ -140,17 +140,19 @@ class BaseEvaluator:
     def expr_event_create(self, action: str, **kwargs):
         """Create event with supplied data and try to extract as much relevant data
         from the context"""
+        context = self._context.copy()
         # If the result was a complex variable, we don't want to re-use it
-        self._context.pop("result", None)
-        self._context.pop("handler", None)
-        kwargs["context"] = self._context
+        context.pop("result", None)
+        context.pop("handler", None)
+        event_kwargs = context
+        event_kwargs.update(kwargs)
         event = Event.new(
             action,
             app=self._filename,
-            **kwargs,
+            **event_kwargs,
         )
-        if "request" in self._context and isinstance(self._context["request"], PolicyRequest):
-            policy_request: PolicyRequest = self._context["request"]
+        if "request" in context and isinstance(context["request"], PolicyRequest):
+            policy_request: PolicyRequest = context["request"]
             if policy_request.http_request:
                 event.from_http(policy_request)
                 return

authentik/lib/tests/test_evaluator.py

@@ -2,28 +2,41 @@
 from django.test import TestCase
 
 from authentik.core.tests.utils import create_test_admin_user
+from authentik.events.models import Event
 from authentik.lib.expression.evaluator import BaseEvaluator
+from authentik.lib.generators import generate_id
 
 
 class TestEvaluator(TestCase):
     """Test Evaluator base functions"""
 
-    def test_regex_match(self):
+    def test_expr_regex_match(self):
         """Test expr_regex_match"""
         self.assertFalse(BaseEvaluator.expr_regex_match("foo", "bar"))
         self.assertTrue(BaseEvaluator.expr_regex_match("foo", "foo"))
 
-    def test_regex_replace(self):
+    def test_expr_regex_replace(self):
         """Test expr_regex_replace"""
         self.assertEqual(BaseEvaluator.expr_regex_replace("foo", "o", "a"), "faa")
 
-    def test_user_by(self):
+    def test_expr_user_by(self):
         """Test expr_user_by"""
         user = create_test_admin_user()
         self.assertIsNotNone(BaseEvaluator.expr_user_by(username=user.username))
         self.assertIsNone(BaseEvaluator.expr_user_by(username="bar"))
         self.assertIsNone(BaseEvaluator.expr_user_by(foo="bar"))
 
-    def test_is_group_member(self):
+    def test_expr_is_group_member(self):
         """Test expr_is_group_member"""
         self.assertFalse(BaseEvaluator.expr_is_group_member(create_test_admin_user(), name="test"))
+
+    def test_expr_event_create(self):
+        """Test expr_event_create"""
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {
+            "foo": "bar",
+        }
+        evaluator.evaluate("ak_create_event('foo', bar='baz')")
+        event = Event.objects.filter(action="custom_foo").first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.context, {"bar": "baz", "foo": "bar"})