*: lazy task imports by BeryJu · Pull Request #17725 · goauthentik/authentik

BeryJu · 2025-10-27T14:49:00Z

reduce memory usage in worker by only importing task-related things in the task worker

netlify · 2025-10-27T14:49:06Z

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	`e46b70a`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/68ff865f7471b000095db667

netlify · 2025-10-27T14:49:06Z

✅ Deploy Preview for authentik-integrations canceled.

Name	Link
🔨 Latest commit	`e46b70a`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/68ff865f144bbf0008ec7993

netlify · 2025-10-27T14:49:26Z

✅ Deploy Preview for authentik-docs ready!

Name	Link
🔨 Latest commit	`e46b70a`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-docs/deploys/68ff865f18d91200082dd603
😎 Deploy Preview	https://deploy-preview-17725--authentik-docs.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

To edit notification comments on pull requests, go to your Netlify project configuration.

codecov · 2025-10-27T14:54:35Z

❌ 10 Tests Failed:

Tests completed	Failed	Passed	Skipped
3127	10	3117	2

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

authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests::test_validate_challenge_unrestricted

Stack Traces | 0.372s run time

self = <unittest.case._Outcome object at 0x7f4491dd3150>
test_case = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_unrestricted>
subTest = False

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_unrestricted>
result = <TestCaseFunction test_validate_challenge_unrestricted>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_unrestricted>
method = <bound method AuthenticatorValidateStageWebAuthnTests.test_validate_challenge_unrestricted of <authentik.stages.authen..._validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_unrestricted>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_unrestricted>

    def test_validate_challenge_unrestricted(self):
        """Test webauthn authentication (unrestricted webauthn device)"""
>       webauthn_mds_import(force=True)

.../authenticator_validate/tests/test_webauthn.py:361: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Actor(<function webauthn_mds_import at 0x7f44b4f73a00>, queue_name='default', actor_name='authentik.stages.authenticator_webauthn.tasks.webauthn_mds_import')
args = (), kwargs = {'force': True}, start = 493.376240624
delta = 9.18699998919692e-06

    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.14.../site-packages/dramatiq/actor.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

force = True

    @actor(description=_("Background task to import FIDO Alliance MDS blob and AAGUIDs into database."))
    def webauthn_mds_import(force=False):
        """Background task to import FIDO Alliance MDS blob and AAGUIDs into database"""
>       self = CurrentTask.get_task()
               ^^^^^^^^^^^^^^^^^^^^^^

.../stages/authenticator_webauthn/tasks.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> Task:
>       return cast(Task, super().get_task())
                          ^^^^^^^^^^^^^^^^^^

authentik/tasks/middleware.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> TaskBase:
        task = cls._TASKS.get()
        if not task:
>           raise CurrentTaskNotFound()
E           django_dramatiq_postgres.middleware.CurrentTaskNotFound

.../django-dramatiq-postgres/django_dramatiq_postgres/middleware.py:147: CurrentTaskNotFound

authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests::test_device_challenge_webauthn_restricted

Stack Traces | 0.374s run time

self = <unittest.case._Outcome object at 0x7f4492ac5fd0>
test_case = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_device_challenge_webauthn_restricted>
subTest = False

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_device_challenge_webauthn_restricted>
result = <TestCaseFunction test_device_challenge_webauthn_restricted>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_device_challenge_webauthn_restricted>
method = <bound method AuthenticatorValidateStageWebAuthnTests.test_device_challenge_webauthn_restricted of <authentik.stages.a...date.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_device_challenge_webauthn_restricted>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_device_challenge_webauthn_restricted>

    def test_device_challenge_webauthn_restricted(self):
        """Test webauthn (getting device challenges with a webauthn
        device that is not allowed due to aaguid restrictions)"""
>       webauthn_mds_import(force=True)

.../authenticator_validate/tests/test_webauthn.py:136: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Actor(<function webauthn_mds_import at 0x7f44b4f73a00>, queue_name='default', actor_name='authentik.stages.authenticator_webauthn.tasks.webauthn_mds_import')
args = (), kwargs = {'force': True}, start = 494.899245381
delta = 7.573999994292535e-06

    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.14.../site-packages/dramatiq/actor.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

force = True

    @actor(description=_("Background task to import FIDO Alliance MDS blob and AAGUIDs into database."))
    def webauthn_mds_import(force=False):
        """Background task to import FIDO Alliance MDS blob and AAGUIDs into database"""
>       self = CurrentTask.get_task()
               ^^^^^^^^^^^^^^^^^^^^^^

.../stages/authenticator_webauthn/tasks.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> Task:
>       return cast(Task, super().get_task())
                          ^^^^^^^^^^^^^^^^^^

authentik/tasks/middleware.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> TaskBase:
        task = cls._TASKS.get()
        if not task:
>           raise CurrentTaskNotFound()
E           django_dramatiq_postgres.middleware.CurrentTaskNotFound

.../django-dramatiq-postgres/django_dramatiq_postgres/middleware.py:147: CurrentTaskNotFound

authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests::test_validate_challenge_restricted

Stack Traces | 0.375s run time

self = <unittest.case._Outcome object at 0x7f4491dd11d0>
test_case = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_restricted>
subTest = False

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_restricted>
result = <TestCaseFunction test_validate_challenge_restricted>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_restricted>
method = <bound method AuthenticatorValidateStageWebAuthnTests.test_validate_challenge_restricted of <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_restricted>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_validate.tests.test_webauthn.AuthenticatorValidateStageWebAuthnTests testMethod=test_validate_challenge_restricted>

    def test_validate_challenge_restricted(self):
        """Test webauthn authentication (restricted device type, failure)"""
>       webauthn_mds_import(force=True)

.../authenticator_validate/tests/test_webauthn.py:435: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Actor(<function webauthn_mds_import at 0x7f44b4f73a00>, queue_name='default', actor_name='authentik.stages.authenticator_webauthn.tasks.webauthn_mds_import')
args = (), kwargs = {'force': True}, start = 490.658125913
delta = 0.00010362300002952907

    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.14.../site-packages/dramatiq/actor.py:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

force = True

    @actor(description=_("Background task to import FIDO Alliance MDS blob and AAGUIDs into database."))
    def webauthn_mds_import(force=False):
        """Background task to import FIDO Alliance MDS blob and AAGUIDs into database"""
>       self = CurrentTask.get_task()
               ^^^^^^^^^^^^^^^^^^^^^^

.../stages/authenticator_webauthn/tasks.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> Task:
>       return cast(Task, super().get_task())
                          ^^^^^^^^^^^^^^^^^^

authentik/tasks/middleware.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @classmethod
    def get_task(cls) -> TaskBase:
        task = cls._TASKS.get()
        if not task:
>           raise CurrentTaskNotFound()
E           django_dramatiq_postgres.middleware.CurrentTaskNotFound

.../django-dramatiq-postgres/django_dramatiq_postgres/middleware.py:147: CurrentTaskNotFound

authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage::test_register_restricted_device_type_deny

Stack Traces | 0.578s run time

self = <unittest.case._Outcome object at 0x7fd69ff8fee0>
test_case = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
subTest = False

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

.../hostedtoolcache/Python/3.14.3.............../x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
result = <TestCaseFunction test_register_restricted_device_type_deny>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3.............../x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
method = <bound method TestAuthenticatorWebAuthnStage.test_register_restricted_device_type_deny of <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3.............../x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>

    def test_register_restricted_device_type_deny(self):
        """Test registration with restricted devices (fail)"""
        webauthn_mds_import.send(force=True)
        self.stage.device_type_restrictions.set(
            WebAuthnDeviceType.objects.filter(description="YubiKey 5 Series")
        )
    
        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
        plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
        plan.context[PLAN_CONTEXT_WEBAUTHN_CHALLENGE] = b64decode(
            b"03Xodi54gKsfnP5I9VFfhaGXVVE2NUyZpBBXns/JI+x6V9RY2Tw2QmxRJkhh7174EkRazUntIwjMVY9bFG60Lw=="
        )
        session = self.client.session
        session[SESSION_KEY_PLAN] = plan
        session.save()
        response = self.client.post(
            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
            data={
                "component": "ak-stage-authenticator-webauthn",
                "response": {
                    "id": "kqnmrVLnDG-OwsSNHkihYZaNz5s",
                    "rawId": "kqnmrVLnDG-OwsSNHkihYZaNz5s",
                    "type": "public-key",
                    "registrationClientExtensions": "{}",
                    "response": {
                        "clientDataJSON": (
                            "eyJ0eXBlIjoid2ViYXV0aG4uY3JlYXRlIiwiY2hhbGxlbmd"
                            "lIjoiMDNYb2RpNTRnS3NmblA1STlWRmZoYUdYVlZFMk5VeV"
                            "pwQkJYbnNfSkkteDZWOVJZMlR3MlFteFJKa2hoNzE3NEVrU"
                            "mF6VW50SXdqTVZZOWJGRzYwTHciLCJvcmlnaW4iOiJodHRw"
                            "Oi8vbG9jYWxob3N0OjkwMDAiLCJjcm9zc09yaWdpbiI6ZmFsc2V9"
                        ),
                        "attestationObject": (
                            "o2NmbXRkbm9uZWdhdHRTdG10oGhhdXRoRGF0YViYSZYN5Yg"
                            "OjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2NdAAAAAPv8MA"
                            "cVTk7MjAtuAgVX170AFJKp5q1S5wxvjsLEjR5IoWGWjc-bp"
                            "QECAyYgASFYIKtcZHPumH37XHs0IM1v3pUBRIqHVV_SE-Lq"
                            "2zpJAOVXIlgg74Fg_WdB0kuLYqCKbxogkEPaVtR_iR3IyQFIJAXBzds"
                        ),
                    },
                },
            },
            SERVER_NAME="localhost",
            SERVER_PORT="9000",
        )
        self.assertEqual(response.status_code, 200)
>       self.assertStageResponse(
            response,
            flow=self.flow,
            component="ak-stage-authenticator-webauthn",
            response_errors={
                "response": [
                    {
                        "string": (
                            "Invalid device type. Contact your authentik administrator for help."
                        ),
                        "code": "invalid",
                    }
                ]
            },
        )

.../stages/authenticator_webauthn/tests.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
response = <HttpChallengeResponse status_code=200, "application/json">
flow = <Flow: Flow AezTxiCfHB (aeztxicfhb)>, user = None
kwargs = {'component': 'ak-stage-authenticator-webauthn', 'response_errors': {'response': [{'code': 'invalid', 'string': 'Invalid device type. Contact your authentik administrator for help.'}]}}
raw_response = {'component': 'xak-flow-redirect', 'to': '/'}

    def assertStageResponse(
        self,
        response: HttpResponse,
        flow: Flow | None = None,
        user: User | None = None,
        **kwargs,
    ) -> dict[str, Any]:
        """Assert various attributes of a stage response"""
        self.assertEqual(response.status_code, 200)
        raw_response = loads(response.content.decode())
        self.assertIsNotNone(raw_response["component"])
        if flow:
>           self.assertIn("flow_info", raw_response)

.../flows/tests/__init__.py:34: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
member = 'flow_info', container = {'component': 'xak-flow-redirect', 'to': '/'}
msg = None

    def assertIn(self, member, container, msg=None):
        """Just like self.assertTrue(a in b), but with a nicer default message."""
        if member not in container:
            standardMsg = '%s not found in %s' % (safe_repr(member),
                                                  safe_repr(container))
>           self.fail(self._formatMessage(msg, standardMsg))

.../hostedtoolcache/Python/3.14.3.............../x64/lib/python3.14/unittest/case.py:1192: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_webauthn.tests.TestAuthenticatorWebAuthnStage testMethod=test_register_restricted_device_type_deny>
msg = "'flow_info' not found in {'component': 'xak-flow-redirect', 'to': '/'}"

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: 'flow_info' not found in {'component': 'xak-flow-redirect', 'to': '/'}

.../hostedtoolcache/Python/3.14.3.............../x64/lib/python3.14/unittest/case.py:750: AssertionError

tests.integration.test_proxy_kubernetes.TestProxyKubernetes::test_kubernetes_controller_ingress

Stack Traces | 0.678s run time

self = <unittest.case._Outcome object at 0x7fb754e83d50>
test_case = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_ingress>
subTest = False

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_ingress>
result = <TestCaseFunction test_kubernetes_controller_ingress>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_ingress>
method = <bound method TestProxyKubernetes.test_kubernetes_controller_ingress of <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_ingress>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_ingress>

    @pytest.mark.timeout(120, func_only=True)
    def test_kubernetes_controller_ingress(self):
        """Test Kubernetes Controller's Ingress"""
        provider: ProxyProvider = ProxyProvider.objects.create(
            name="test",
            internal_host="http://localhost",
            external_host="https://localhost",
            authorization_flow=create_test_flow(),
        )
        provider2: ProxyProvider = ProxyProvider.objects.create(
            name="test2",
            internal_host="http://otherhost",
            external_host="https://otherhost",
            mode=ProxyMode.FORWARD_SINGLE,
            authorization_flow=create_test_flow(),
        )
    
        service_connection = KubernetesServiceConnection.objects.first()
        outpost: Outpost = Outpost.objects.create(
            name="test",
            type=OutpostType.PROXY,
            service_connection=service_connection,
        )
        outpost.providers.add(provider)
    
>       self.controller = ProxyKubernetesController(outpost, service_connection)
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/integration/test_proxy_kubernetes.py:76: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb756c13410>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb756c13410>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb756c13770>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

tests.integration.test_proxy_kubernetes.TestProxyKubernetes::test_kubernetes_controller_static

Stack Traces | 0.795s run time

self = <unittest.case._Outcome object at 0x7fb754e83950>
test_case = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_static>
subTest = False

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_static>
result = <TestCaseFunction test_kubernetes_controller_static>

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_static>
method = <bound method TestProxyKubernetes.test_kubernetes_controller_static of <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_static>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_kubernetes.TestProxyKubernetes testMethod=test_kubernetes_controller_static>

    @pytest.mark.timeout(120, func_only=True)
    def test_kubernetes_controller_static(self):
        """Test Kubernetes Controller"""
        provider: ProxyProvider = ProxyProvider.objects.create(
            name="test",
            internal_host="http://localhost",
            external_host="http://localhost",
            authorization_flow=create_test_flow(),
        )
        service_connection = KubernetesServiceConnection.objects.first()
        outpost: Outpost = Outpost.objects.create(
            name="test",
            type=OutpostType.PROXY,
            service_connection=service_connection,
        )
        outpost.providers.add(provider)
        outpost.save()
    
>       self.controller = ProxyKubernetesController(outpost, service_connection)
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/integration/test_proxy_kubernetes.py:47: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb7579b16e0>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb7579b16e0>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb7579b03e0>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_controller_rename

Stack Traces | 0.877s run time

self = <unittest.case._Outcome object at 0x7fb76cdd2780>
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.14.3........./x64/lib/python3.14/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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>
method = <bound method OutpostKubernetesTests.test_controller_rename of <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_rename>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @pytest.mark.timeout(120, func_only=True)
    def test_controller_rename(self):
        """test that objects get deleted and re-created with new names"""
>       controller = ProxyKubernetesController(self.outpost, self.service_connection)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb753b2c2d0>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb753b2c2d0>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb75403d950>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_controller_full_update

Stack Traces | 0.879s run time

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

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_full_update>
method = <bound method OutpostKubernetesTests.test_controller_full_update of <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_controller_full_update>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @pytest.mark.timeout(120, func_only=True)
    def test_controller_full_update(self):
        """Test an update that triggers all objects"""
>       controller = ProxyKubernetesController(self.outpost, self.service_connection)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb76cdd09d0>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb76cdd09d0>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb76cdd3950>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_deployment_reconciler

Stack Traces | 0.897s run time

self = <unittest.case._Outcome object at 0x7fb750713890>
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.14.3........./x64/lib/python3.14/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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>
method = <bound method OutpostKubernetesTests.test_deployment_reconciler of <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_deployment_reconciler>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @pytest.mark.timeout(120, func_only=True)
    def test_deployment_reconciler(self):
        """test that deployment requires update"""
>       controller = ProxyKubernetesController(self.outpost, self.service_connection)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb757aae900>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb757aae900>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb757aacec0>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

tests.integration.test_outpost_kubernetes.OutpostKubernetesTests::test_service_reconciler

Stack Traces | 0.904s run time

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

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

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_service_reconciler>
result = <TestCaseFunction test_service_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()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:669: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_service_reconciler>
method = <bound method OutpostKubernetesTests.test_service_reconciler of <tests.integration.test_outpost_kubernetes.OutpostKubernetesTests testMethod=test_service_reconciler>>

    def _callTestMethod(self, method):
>       result = method()
                 ^^^^^^^^

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:615: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    @pytest.mark.timeout(120, func_only=True)
    def test_service_reconciler(self):
        """test that service requires update"""
>       controller = ProxyKubernetesController(self.outpost, self.service_connection)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

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

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb75634d1d0>
outpost = <Outpost: Outpost test>, connection = None

    def __init__(self, outpost: Outpost, connection: KubernetesServiceConnection):
>       super().__init__(outpost, connection)

.../proxy/controllers/kubernetes.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.kubernetes.ProxyKubernetesController object at 0x7fb75634d1d0>
outpost = <Outpost: Outpost test>, connection = None, client = None

    def __init__(
        self,
        outpost: Outpost,
        connection: KubernetesServiceConnection,
        client: KubernetesClient | None = None,
    ) -> None:
        super().__init__(outpost, connection)
>       self.client = client if client else KubernetesClient(connection)
                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../outposts/controllers/kubernetes.py:72: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.kubernetes.KubernetesClient object at 0x7fb754c42350>
connection = None

    def __init__(self, connection: KubernetesServiceConnection):
        config = Configuration()
        try:
>           if connection.local:
               ^^^^^^^^^^^^^^^^
E           AttributeError: 'NoneType' object has no attribute 'local'

.../outposts/controllers/kubernetes.py:37: AttributeError

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

Signed-off-by: Jens Langhammer <jens@goauthentik.io> # Conflicts: # authentik/tasks/middleware.py

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

BeryJu assigned rissson Oct 27, 2025

BeryJu requested a review from a team as a code owner October 27, 2025 14:49

BeryJu added the area:backend label Oct 27, 2025

BeryJu added this to authentik Core Oct 27, 2025

github-project-automation Bot moved this to Todo in authentik Core Oct 27, 2025

rissson mentioned this pull request Nov 3, 2025

How to lower idle RAM usage for small instances #17869

Open

*: lazy task imports

adc017a

Signed-off-by: Jens Langhammer <jens@goauthentik.io> # Conflicts: # authentik/tasks/middleware.py

BeryJu force-pushed the tasks/lazy-task-imports branch from e46b70a to adc017a Compare March 29, 2026 13:56

BeryJu added 6 commits March 29, 2026 16:04

format, don't use global

3d79aff

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

typo

6a6ac25

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

cleanup

242ee0c

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

remove setproctitle, it uses like 35 MB of ram for some reason

719edac

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

fix duplicate import

c521e9a

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

test mds directly

9cd9ff1

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

Navigation Menu

Uh oh!

Uh oh!

*: lazy task imports - #17725

*: lazy task imports#17725
BeryJu wants to merge 7 commits into
mainfrom
tasks/lazy-task-imports

BeryJu commented Oct 27, 2025 •

edited by rissson

Loading

Uh oh!

netlify Bot commented Oct 27, 2025 •

edited

Loading

Uh oh!

netlify Bot commented Oct 27, 2025 •

edited

Loading

Uh oh!

netlify Bot commented Oct 27, 2025 •

edited

Loading

Uh oh!

codecov Bot commented Oct 27, 2025 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Uh oh!

Conversation

BeryJu commented Oct 27, 2025 • edited by rissson Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

netlify Bot commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-storybook canceled.

Uh oh!

netlify Bot commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-integrations canceled.

Uh oh!

netlify Bot commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-docs ready!

Uh oh!

codecov Bot commented Oct 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 10 Tests Failed:

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

BeryJu commented Oct 27, 2025 •

edited by rissson

Loading

netlify Bot commented Oct 27, 2025 •

edited

Loading

netlify Bot commented Oct 27, 2025 •

edited

Loading

netlify Bot commented Oct 27, 2025 •

edited

Loading

codecov Bot commented Oct 27, 2025 •

edited

Loading