*: replace Celery with DramatiQ

[rissson]

Details

TODO:

• fix the authentik_events.view_systemtask permission check for sync_status endpoints, and associated code
• modify schedules reconciliation to atomically create/delete/update schedules on startup, instead of hoping outdated schedules get deleted
• broker fixes, see todos
• tasks fixes, see todos
• figure out what to do about messages
• ak worker embedded in ak server (?)
• django.setup startup code refactor
• UI

---

Checklist

• Local tests pass (ak test authentik/)
• The code has been formatted (make lint-fix)

If an API change has been made

• The API schema has been updated (make gen-build)

If changes to the frontend have been made

• The code has been formatted (make web)

If applicable

• The documentation has been updated
• The documentation has been formatted (make website)

[netlify]

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	c9f1e34
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-storybook/deploys/67eadf1fd7e8370008584ad4

[netlify]

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	c9f1e34
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/67eadf1fc59a770008f372e9

[rissson]

Superseeds #13436 #9607 #6845

[codecov]

❌ 11 Tests Failed:

Tests completed	Failed	Passed	Skipped
71	11	60	2

View the top 3 failed test(s) by shortest run time

::authentik.tasks.schedules.models

Stack Traces | 0s run time

cls = <class '_pytest.runner.CallInfo'>
func = <function pytest_make_collect_report.<locals>.collect at 0x7f440cce3600>
when = 'collect', reraise = (<class 'KeyboardInterrupt'>, <class 'SystemExit'>)

    @classmethod
    def from_call(
        cls,
        func: Callable[[], TResult],
        when: Literal["collect", "setup", "call", "teardown"],
        reraise: type[BaseException] | tuple[type[BaseException], ...] | None = None,
    ) -> CallInfo[TResult]:
        """Call func, wrapping the result in a CallInfo.
    
        :param func:
            The function to call. Called without arguments.
        :type func: Callable[[], _pytest.runner.TResult]
        :param when:
            The phase in which the function is called.
        :param reraise:
            Exception or exceptions that shall propagate if raised by the
            function, instead of being wrapped in the CallInfo.
        """
        excinfo = None
        start = timing.time()
        precise_start = timing.perf_counter()
        try:
>           result: TResult | None = func()

.venv/lib/python3.12....../site-packages/_pytest/runner.py:341: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    def collect() -> list[Item | Collector]:
        # Before collecting, if this is a Directory, load the conftests.
        # If a conftest import fails to load, it is considered a collection
        # error of the Directory collector. This is why it's done inside of the
        # CallInfo wrapper.
        #
        # Note: initial conftests are loaded early, not here.
        if isinstance(collector, Directory):
            collector.config.pluginmanager._loadconftestmodules(
                collector.path,
                collector.config.getoption("importmode"),
                rootpath=collector.config.rootpath,
                consider_namespace_packages=collector.config.getini(
                    "consider_namespace_packages"
                ),
            )
    
>       return list(collector.collect())

.venv/lib/python3.12....../site-packages/_pytest/runner.py:389: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DoctestModule models.py>

    def collect(self) -> Iterable[DoctestItem]:
        import doctest
    
        class MockAwareDocTestFinder(doctest.DocTestFinder):
            py_ver_info_minor = sys.version_info[:2]
            is_find_lineno_broken = (
                py_ver_info_minor < (3, 11)
                or (py_ver_info_minor == (3, 11) and sys.version_info.micro < 9)
                or (py_ver_info_minor == (3, 12) and sys.version_info.micro < 3)
            )
            if is_find_lineno_broken:
    
                def _find_lineno(self, obj, source_lines):
                    """On older Pythons, doctest code does not take into account
                    `@property`. https://github..../cpython/issues/61648
    
                    Moreover, wrapped Doctests need to be unwrapped so the correct
                    line number is returned. #8796
                    """
                    if isinstance(obj, property):
                        obj = getattr(obj, "fget", obj)
    
                    if hasattr(obj, "__wrapped__"):
                        # Get the main obj in case of it being wrapped
                        obj = inspect.unwrap(obj)
    
                    # Type ignored because this is a private function.
                    return super()._find_lineno(  # type:ignore[misc]
                        obj,
                        source_lines,
                    )
    
            if sys.version_info < (3, 10):
    
                def _find(
                    self, tests, obj, name, module, source_lines, globs, seen
                ) -> None:
                    """Override _find to work around issue in stdlib.
    
                    https://github..../pytest/issues/3456
                    https://github..../cpython/issues/69718
                    """
                    if _is_mocked(obj):
                        return  # pragma: no cover
                    with _patch_unwrap_mock_aware():
                        # Type ignored because this is a private function.
                        super()._find(  # type:ignore[misc]
                            tests, obj, name, module, source_lines, globs, seen
                        )
    
            if sys.version_info < (3, 13):
    
                def _from_module(self, module, object):
                    """`cached_property` objects are never considered a part
                    of the 'current module'. As such they are skipped by doctest.
                    Here we override `_from_module` to check the underlying
                    function instead. https://github..../cpython/issues/107995
                    """
                    if isinstance(object, functools.cached_property):
                        object = object.func
    
                    # Type ignored because this is a private function.
                    return super()._from_module(module, object)  # type: ignore[misc]
    
        try:
>           module = self.obj

.venv/lib/python3.12.../site-packages/_pytest/doctest.py:566: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DoctestModule models.py>

    @property
    def obj(self):
        """Underlying Python object."""
        obj = getattr(self, "_obj", None)
        if obj is None:
>           self._obj = obj = self._getobj()

.venv/lib/python3.12........./site-packages/_pytest/python.py:284: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DoctestModule models.py>

    def _getobj(self):
>       return importtestmodule(self.path, self.config)

.venv/lib/python3.12........./site-packages/_pytest/python.py:546: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

path = PosixPath('.../tasks/schedules/models.py')
config = <_pytest.config.Config object at 0x7f44121db2c0>

    def importtestmodule(
        path: Path,
        config: Config,
    ):
        # We assume we are only called once per module.
        importmode = config.getoption("--import-mode")
        try:
>           mod = import_path(
                path,
                mode=importmode,
                root=config.rootpath,
                consider_namespace_packages=config.getini("consider_namespace_packages"),
            )

.venv/lib/python3.12........./site-packages/_pytest/python.py:493: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

path = PosixPath('.../tasks/schedules/models.py')

    def import_path(
        path: str | os.PathLike[str],
        *,
        mode: str | ImportMode = ImportMode.prepend,
        root: Path,
        consider_namespace_packages: bool,
    ) -> ModuleType:
        """
        Import and return a module from the given path, which can be a file (a module) or
        a directory (a package).
    
        :param path:
            Path to the file to import.
    
        :param mode:
            Controls the underlying import mechanism that will be used:
    
            * ImportMode.prepend: the directory containing the module (or package, taking
              `__init__.py` files into account) will be put at the *start* of `sys.path` before
              being imported with `importlib.import_module`.
    
            * ImportMode.append: same as `prepend`, but the directory will be appended
              to the end of `sys.path`, if not already in `sys.path`.
    
            * ImportMode.importlib: uses more fine control mechanisms provided by `importlib`
              to import the module, which avoids having to muck with `sys.path` at all. It effectively
              allows having same-named test modules in different places.
    
        :param root:
            Used as an anchor when mode == ImportMode.importlib to obtain
            a unique name for the module being imported so it can safely be stored
            into ``sys.modules``.
    
        :param consider_namespace_packages:
            If True, consider namespace packages when resolving module names.
    
        :raises ImportPathMismatchError:
            If after importing the given `path` and the module `__file__`
            are different. Only raised in `prepend` and `append` modes.
        """
        path = Path(path)
        mode = ImportMode(mode)
    
        if not path.exists():
            raise ImportError(path)
    
        if mode is ImportMode.importlib:
            # Try to import this module using the standard import mechanisms, but
            # without touching sys.path.
            try:
                pkg_root, module_name = resolve_pkg_root_and_module_name(
                    path, consider_namespace_packages=consider_namespace_packages
                )
            except CouldNotResolvePathError:
                pass
            else:
                # If the given module name is already in sys.modules, do not import it again.
                with contextlib.suppress(KeyError):
                    return sys.modules[module_name]
    
>               mod = _import_module_using_spec(
                    module_name, path, pkg_root, insert_modules=False
                )

.venv/lib/python3.12....../site-packages/_pytest/pathlib.py:549: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

module_name = 'schedules.models'
module_path = PosixPath('.../tasks/schedules/models.py')
module_location = PosixPath('.../authentik/authentik/tasks')

    def _import_module_using_spec(
        module_name: str, module_path: Path, module_location: Path, *, insert_modules: bool
    ) -> ModuleType | None:
        """
        Tries to import a module by its canonical name, path, and its parent location.
    
        :param module_name:
            The expected module name, will become the key of `sys.modules`.
    
        :param module_path:
            The file path of the module, for example `.../foo/bar/test_demo.py`.
            If module is a package, pass the path to the  `__init__.py` of the package.
            If module is a namespace package, pass directory path.
    
        :param module_location:
            The parent location of the module.
            If module is a package, pass the directory containing the `__init__.py` file.
    
        :param insert_modules:
            If True, will call `insert_missing_modules` to create empty intermediate modules
            with made-up module names (when importing test files not reachable from `sys.path`).
    
        Example 1 of parent_module_*:
    
            module_name:        "a.b.c.demo"
            module_path:        Path(".../b/c/demo.py")
            module_location:    Path("a/b/c/")
            if "a.b.c" is package (".../b/c/__init__.py" exists), then
                parent_module_name:         "a.b.c"
                parent_module_path:         Path(".../b/c/__init__.py")
                parent_module_location:     Path("a/b/c/")
            else:
                parent_module_name:         "a.b.c"
                parent_module_path:         Path("a/b/c")
                parent_module_location:     Path("a/b/")
    
        Example 2 of parent_module_*:
    
            module_name:        "a.b.c"
            module_path:        Path(".../b/c/__init__.py")
            module_location:    Path("a/b/c/")
            if  "a.b" is package ("a/b/__init__.py" exists), then
                parent_module_name:         "a.b"
                parent_module_path:         Path("a/b/__init__.py")
                parent_module_location:     Path("a/b/")
            else:
                parent_module_name:         "a.b"
                parent_module_path:         Path("a/b/")
                parent_module_location:     Path("a/")
        """
        # Attempt to import the parent module, seems is our responsibility:
        # https://github..../Lib/importlib/_bootstrap.py#L1308-L1311
        parent_module_name, _, name = module_name.rpartition(".")
        parent_module: ModuleType | None = None
        if parent_module_name:
            parent_module = sys.modules.get(parent_module_name)
            # If the parent_module lacks the `__path__` attribute, AttributeError when finding a submodule's spec,
            # requiring re-import according to the path.
            need_reimport = not hasattr(parent_module, "__path__")
            if parent_module is None or need_reimport:
                # Get parent_location based on location, get parent_path based on path.
                if module_path.name == "__init__.py":
                    # If the current module is in a package,
                    # need to leave the package first and then enter the parent module.
                    parent_module_path = module_path.parent.parent
                else:
                    parent_module_path = module_path.parent
    
                if (parent_module_path / "__init__.py").is_file():
                    # If the parent module is a package, loading by  __init__.py file.
                    parent_module_path = parent_module_path / "__init__.py"
    
                parent_module = _import_module_using_spec(
                    parent_module_name,
                    parent_module_path,
                    parent_module_path.parent,
                    insert_modules=insert_modules,
                )
    
        # Checking with sys.meta_path first in case one of its hooks can import this module,
        # such as our own assertion-rewrite hook.
        for meta_importer in sys.meta_path:
            module_name_of_meta = getattr(meta_importer.__class__, "__module__", "")
            if module_name_of_meta == "_pytest.assertion.rewrite" and module_path.is_file():
                # Import modules in subdirectories by module_path
                # to ensure assertion rewrites are not missed (#12659).
                find_spec_path = [str(module_location), str(module_path)]
            else:
                find_spec_path = [str(module_location)]
    
            spec = meta_importer.find_spec(module_name, find_spec_path)
    
            if spec_matches_module_path(spec, module_path):
                break
        else:
            loader = None
            if module_path.is_dir():
                # The `spec_from_file_location` matches a loader based on the file extension by default.
                # For a namespace package, need to manually specify a loader.
                loader = NamespaceLoader(name, module_path, PathFinder())
    
            spec = importlib.util.spec_from_file_location(
                module_name, str(module_path), loader=loader
            )
    
        if spec_matches_module_path(spec, module_path):
            assert spec is not None
            # Find spec and import this module.
            mod = importlib.util.module_from_spec(spec)
            sys.modules[module_name] = mod
>           spec.loader.exec_module(mod)  # type: ignore[union-attr]

.venv/lib/python3.12....../site-packages/_pytest/pathlib.py:725: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_frozen_importlib_external.SourceFileLoader object at 0x7f440ba5b050>
module = <module 'schedules.models' from '.../tasks/schedules/models.py'>

>   ???

<frozen importlib._bootstrap_external>:999: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

f = <built-in function exec>
args = (<code object <module> at 0x7f440be61a70, file ".../authentik/authentik/tasks/schedules/models...ntentType': <class 'django.contrib.contenttypes.models.ContentType'>, 'Cron': <class 'cron_converter.cron.Cron'>, ...})
kwds = {}

>   ???

<frozen importlib._bootstrap>:488: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    import pickle  # nosec
    from uuid import uuid4
    
    from cron_converter import Cron
    from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
    from django.contrib.contenttypes.models import ContentType
    from django.core.exceptions import ValidationError
    from django.db import models
    from django.utils.timezone import datetime
    from django.utils.translation import gettext_lazy as _
    from dramatiq.actor import Actor
    from dramatiq.broker import Broker, get_broker
    from dramatiq.message import Message
    
    from authentik.lib.models import SerializerModel
    from authentik.tasks.schedules.lib import ScheduleSpec
    
    
    def validate_crontab(value):
        try:
            Cron(value)
        except ValueError as exc:
            raise ValidationError(
                _("%(value)s is not a valid crontab"),
                params={"value": value},
            ) from exc
    
    
>   class Schedule(SerializerModel):

.../tasks/schedules/models.py:29: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'django.db.models.base.ModelBase'>, name = 'Schedule'
bases = (<class 'authentik.lib.models.SerializerModel'>,)
attrs = {'__qualname__': 'Schedule', '__str__': <function Schedule.__str__ at 0x7f440ba384a0>, 'actor_name': <django.db.models.fields.TextField>, 'args': <django.db.models.fields.BinaryField>, ...}
kwargs = {}
super_new = <built-in method __new__ of type object at 0x7f441dd31260>
parents = [<class 'authentik.lib.models.SerializerModel'>]
module = 'schedules.models'
new_attrs = {'__module__': 'schedules.models', '__qualname__': 'Schedule', '__str__': <function Schedule.__str__ at 0x7f440ba384a0>, 'calculate_next_run': <function Schedule.calculate_next_run at 0x7f440baa0900>, ...}
classcell = None, attr_meta = <class 'schedules.models.Schedule.Meta'>

    def __new__(cls, name, bases, attrs, **kwargs):
        super_new = super().__new__
    
        # Also ensure initialization is only performed for subclasses of Model
        # (excluding Model class itself).
        parents = [b for b in bases if isinstance(b, ModelBase)]
        if not parents:
            return super_new(cls, name, bases, attrs)
    
        # Create the class.
        module = attrs.pop("__module__")
        new_attrs = {"__module__": module}
        classcell = attrs.pop("__classcell__", None)
        if classcell is not None:
            new_attrs["__classcell__"] = classcell
        attr_meta = attrs.pop("Meta", None)
        # Pass all attrs without a (Django-specific) contribute_to_class()
        # method to type.__new__() so that they're properly initialized
        # (i.e. __set_name__()).
        contributable_attrs = {}
        for obj_name, obj in attrs.items():
            if _has_contribute_to_class(obj):
                contributable_attrs[obj_name] = obj
            else:
                new_attrs[obj_name] = obj
        new_class = super_new(cls, name, bases, new_attrs, **kwargs)
    
        abstract = getattr(attr_meta, "abstract", False)
        meta = attr_meta or getattr(new_class, "Meta", None)
        base_meta = getattr(new_class, "_meta", None)
    
        app_label = None
    
        # Look for an application configuration to attach the model to.
        app_config = apps.get_containing_app_config(module)
    
        if getattr(meta, "app_label", None) is None:
            if app_config is None:
                if not abstract:
>                   raise RuntimeError(
                        "Model class %s.%s doesn't declare an explicit "
                        "app_label and isn't in an application in "
                        "INSTALLED_APPS." % (module, name)
                    )
E                   RuntimeError: Model class schedules.models.Schedule doesn't declare an explicit app_label and isn't in an application in INSTALLED_APPS.

.venv/lib/python3.12.../db/models/base.py:134: RuntimeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_deployment_reconciler

Stack Traces | 0.006s run time

self = <unittest.case._Outcome object at 0x7f3c308b0c50>
test_case = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>
result = <TestCaseFunction test_deployment_reconciler>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:630: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:586: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>

    def setUp(self):
        super().setUp()
        # Ensure that local connection have been created
>       outpost_connection_discovery()

tests/integration/test_outpost_kubernetes.py:27: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Actor(<function outpost_connection_discovery at 0x7f3c34ad32e0>, queue_name='default', actor_name='authentik.outposts.tasks.outpost_connection_discovery')
args = (), kwargs = {}, start = 526.227761991, delta = 1.357500002541201e-05

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Any | R | Awaitable[R]:
        """Synchronously call this actor.
    
        Parameters:
          *args: Positional arguments to send to the actor.
          **kwargs: Keyword arguments to send to the actor.
    
        Returns:
          Whatever the underlying function backing this actor returns.
        """
        try:
            self.logger.debug("Received args=%r kwargs=%r.", args, kwargs)
            start = time.perf_counter()
>           return self.fn(*args, **kwargs)

.venv/lib/python3.12.../site-packages/dramatiq/actor.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    @actor
    def outpost_connection_discovery():
        """Checks the local environment and create Service connections."""
>       self: Task = CurrentTask.get_task()

authentik/outposts/tasks.py:223: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'authentik.tasks.middleware.CurrentTask'>

    @classmethod
    def get_task(cls) -> Task:
        task = cls._TASK.get()
        if task is None:
>           raise RuntimeError("CurrentTask.get_task() should only be called in a running task")
E           RuntimeError: CurrentTask.get_task() should only be called in a running task

authentik/tasks/middleware.py:27: RuntimeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_controller_rename

Stack Traces | 0.007s run time

self = <unittest.case._Outcome object at 0x7f3c30dbf500>
test_case = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>
result = <TestCaseFunction test_controller_rename>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:630: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.12.9........./x64/lib/python3.12/unittest/case.py:586: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>

    def setUp(self):
        super().setUp()
        # Ensure that local connection have been created
>       outpost_connection_discovery()

tests/integration/test_outpost_kubernetes.py:27: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Actor(<function outpost_connection_discovery at 0x7f3c34ad32e0>, queue_name='default', actor_name='authentik.outposts.tasks.outpost_connection_discovery')
args = (), kwargs = {}, start = 526.334270707, delta = 1.3274000025376154e-05

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Any | R | Awaitable[R]:
        """Synchronously call this actor.
    
        Parameters:
          *args: Positional arguments to send to the actor.
          **kwargs: Keyword arguments to send to the actor.
    
        Returns:
          Whatever the underlying function backing this actor returns.
        """
        try:
            self.logger.debug("Received args=%r kwargs=%r.", args, kwargs)
            start = time.perf_counter()
>           return self.fn(*args, **kwargs)

.venv/lib/python3.12.../site-packages/dramatiq/actor.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    @actor
    def outpost_connection_discovery():
        """Checks the local environment and create Service connections."""
>       self: Task = CurrentTask.get_task()

authentik/outposts/tasks.py:223: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'authentik.tasks.middleware.CurrentTask'>

    @classmethod
    def get_task(cls) -> Task:
        task = cls._TASK.get()
        if task is None:
>           raise RuntimeError("CurrentTask.get_task() should only be called in a running task")
E           RuntimeError: CurrentTask.get_task() should only be called in a running task

authentik/tasks/middleware.py:27: RuntimeError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}