web/user: fix race condition in user settings flow executor (cherry-pick #13163)

[gcp-cherry-pick-bot]

Cherry-picked web/user: fix race condition in user settings flow executor (#13163)

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

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
1712	1	1711	2

View the full list of 1 ❄️ flaky tests

authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage::test_session_management

Flake rate in main: 13.46% (Passed 90 times, Failed 14 times)

Stack Traces | 0.948s run time

self = <unittest.case._Outcome object at 0x7f224c0daff0>
test_case = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>
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 = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>
result = <TestCaseFunction test_session_management>

    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.12.9.............../x64/lib/python3.12/unittest/case.py:634: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>
method = <bound method TestAuthenticatorEmailStage.test_session_management of <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>>

    def _callTestMethod(self, method):
>       if method() is not None:

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

self = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>

    def test_session_management(self):
        """Test session device management"""
        # Test device creation in session
        with patch(
            "authentik.stages.authenticator_email.models.AuthenticatorEmailStage.backend_class",
            PropertyMock(return_value=EmailBackend),
        ):
            # Delete any existing devices for this test
            EmailDevice.objects.filter(user=self.user).delete()
    
            response = self.client.get(
                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
            )
            self.assertIn(SESSION_KEY_EMAIL_DEVICE, self.client.session)
            device = self.client.session[SESSION_KEY_EMAIL_DEVICE]
            self.assertIsInstance(device, EmailDevice)
            self.assertFalse(device.confirmed)
            self.assertEqual(device.user, self.user)
    
            # Test device confirmation and cleanup
            device.confirmed = True
            device.email = "new_test@authentik.local"  # Use a different email
            self.client.session[SESSION_KEY_EMAIL_DEVICE] = device
            self.client.session.save()
            response = self.client.post(
                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
                data={"component": "ak-stage-authenticator-email", "code": device.token},
            )
            self.assertEqual(response.status_code, 200)
            self.assertTrue(device.confirmed)
            # Session key should be removed after device is saved
            device.save()
>           self.assertNotIn(SESSION_KEY_EMAIL_DEVICE, self.client.session)

.../stages/authenticator_email/tests.py:305: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>
member = '.../stages/authenticator_email/email_device'
container = <django.contrib.sessions.backends.cache.SessionStore object at 0x7f224d0d8080>
msg = None

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

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

self = <authentik.stages.authenticator_email.tests.TestAuthenticatorEmailStage testMethod=test_session_management>
msg = "'.../stages/authenticator_email/email_device' unexpectedly found in <django.contrib.sessions.backends.cache.SessionStore object at 0x7f224d0d8080>"

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: '.../stages/authenticator_email/email_device' unexpectedly found in <django.contrib.sessions.backends.cache.SessionStore object at 0x7f224d0d8080>

.../hostedtoolcache/Python/3.12.9.............../x64/lib/python3.12/unittest/case.py:715: AssertionError

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

[netlify]

✅ Deploy Preview for authentik-docs ready!

Name	Link
🔨 Latest commit	4663dfd
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/67b8d5a5f2d62700084f5647
😎 Deploy Preview	https://deploy-preview-13169--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 site configuration.