wip

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

authentik/admin/tasks.py

@@ -3,6 +3,7 @@
 from django.core.cache import cache
 from django.db import DatabaseError, InternalError, ProgrammingError
 from django.utils.translation import gettext_lazy as _
+from dramatiq import actor
 from packaging.version import parse
 from requests import RequestException
 from structlog.stdlib import get_logger
@@ -33,9 +34,7 @@ def _set_prom_info():
     )
 
 
-@CELERY_APP.task(
-    throws=(DatabaseError, ProgrammingError, InternalError),
-)
+@actor(throws=(DatabaseError, ProgrammingError, InternalError))
 def clear_update_notifications():
     """Clear update notifications on startup if the notification was for the version
     we're running now."""

authentik/admin/tests/test_tasks.py

@@ -11,6 +11,7 @@ from authentik.admin.tasks import (
 )
 from authentik.events.models import Event, EventAction
 from authentik.lib.config import CONFIG
+from authentik.tasks.tests import TaskTestCase
 
 RESPONSE_VALID = {
     "$schema": "https://version.goauthentik.io/schema.json",
@@ -23,7 +24,7 @@ RESPONSE_VALID = {
 }
 
 
-class TestAdminTasks(TestCase):
+class TestAdminTasks(TaskTestCase):
     """test admin tasks"""
 
     def test_version_valid_response(self):

authentik/root/celery.py

@@ -98,11 +98,11 @@ def _get_startup_tasks_default_tenant() -> list[Callable]:
 
 def _get_startup_tasks_all_tenants() -> list[Callable]:
     """Get all tasks to be run on startup for all tenants"""
-    from authentik.admin.tasks import clear_update_notifications
+    # from authentik.admin.tasks import clear_update_notifications
     from authentik.providers.proxy.tasks import proxy_set_defaults
 
     return [
-        clear_update_notifications,
+        # clear_update_notifications,
         proxy_set_defaults,
     ]
 

authentik/tasks/models.py

@@ -1,3 +1,4 @@
+from django.utils.translation import gettext_lazy as _
 from enum import StrEnum, auto
 from uuid import uuid4
 import pgtrigger
@@ -11,6 +12,35 @@ from authentik.tenants.models import Tenant
 CHANNEL_PREFIX = "authentik.tasks"
 
 
+# class Schedule(SerializerModel):
+#     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
+#
+#     name = models.TextField(editable=False)
+#     func = models.TextField(editable=False)
+#     args = models.TextField(editable=False)
+#     kwargs = models.TextField(editable=False)
+#
+#     class Meta:
+#         verbose_name = _("Schedule")
+#         verbose_name_plural = _("Schedules")
+#         indexes = (
+#             models.Index(
+#                 fields=(
+#                     "name",
+#                     "func",
+#                 ),
+#             ),
+#         )
+#
+#     def __str__(self):
+#         return self.name
+#
+#     @property
+#     def serializer(self):
+#         # TODO: fixme
+#         pass
+
+
 class ChannelIdentifier(StrEnum):
     ENQUEUE = auto()
     LOCK = auto()
@@ -39,6 +69,8 @@ class Task(SerializerModel):
     messages = models.JSONField(blank=True, null=True, editable=False)
 
     class Meta:
+        verbose_name = _("Task")
+        verbose_name_plural = _("Tasks")
         indexes = (models.Index(fields=("state", "mtime")),)
         triggers = (
             pgtrigger.Trigger(

authentik/tasks/tests.py

@@ -0,0 +1,18 @@
+from dramatiq import Worker, get_broker
+from django.test import TransactionTestCase
+
+
+class TaskTestCase(TransactionTestCase):
+    def _pre_setup(self):
+        super()._pre_setup()
+
+        self.broker = get_broker()
+        self.broker.flush_all()
+
+        self.worker = Worker(self.broker, worker_timeout=100)
+        self.worker.start()
+
+    def _post_teardown(self):
+        self.worker.stop()
+
+        super()._post_teardown()