stages: remove more global state #18641

BeryJu · 2025-12-06T01:17:59Z

in preparation for the ability to run multiple flows in parallel

remove more global session state and move it into flow plan

authenticator_duo
authenticator_sms
authenticator_email
consent

remaining:

authentik/core/middleware.py:
  17  
  18: SESSION_KEY_IMPERSONATE_USER = "authentik/impersonate/user"
  19: SESSION_KEY_IMPERSONATE_ORIGINAL_USER = "authentik/impersonate/original_user"
  20  RESPONSE_HEADER_ID = "X-authentik-id"

authentik/core/sources/flow_manager.py:
  49  PLAN_CONTEXT_SOURCE_GROUPS = "source_groups"
  50: SESSION_KEY_SOURCE_FLOW_STAGES = "authentik/flows/source_flow_stages"
  51: SESSION_KEY_SOURCE_FLOW_CONTEXT = "authentik/flows/source_flow_context"
  52: SESSION_KEY_OVERRIDE_FLOW_TOKEN = "authentik/flows/source_override_flow_token"  # nosec
  53  

authentik/flows/views/executor.py:
  64  NEXT_ARG_NAME = "next"
  65: SESSION_KEY_PLAN = "authentik/flows/plan"
  66: SESSION_KEY_GET = "authentik/flows/get"
  67: SESSION_KEY_POST = "authentik/flows/post"
  68: SESSION_KEY_HISTORY = "authentik/flows/history"
  69  QS_KEY_TOKEN = "flow_token"  # nosec

authentik/policies/views.py:
  39  QS_SKIP_BUFFER = "skip_buffer"
  40: SESSION_KEY_BUFFER = "authentik/policies/pav_buffer/%s"
  41  

authentik/providers/oauth2/views/authorize.py:
  75  PLAN_CONTEXT_PARAMS = "goauthentik.io/providers/oauth2/params"
  76: SESSION_KEY_LAST_LOGIN_UID = "authentik/providers/oauth2/last_login_uid"
  77  

authentik/sources/oauth/clients/oauth2.py:
  22  LOGGER = get_logger()
  23: SESSION_KEY_OAUTH_PKCE = "authentik/sources/oauth/pkce"
  24  

authentik/sources/saml/processors/request.py:
  22  
  23: SESSION_KEY_REQUEST_ID = "authentik/sources/saml/request_id"
  24  

authentik/stages/password/stage.py:
  33  PLAN_CONTEXT_METHOD_ARGS = "auth_method_args"
  34: SESSION_KEY_INVALID_TRIES = "authentik/stages/password/user_invalid_tries"
  35  

authentik/stages/user_login/middleware.py:
  15  
  16: SESSION_KEY_BINDING_NET = "authentik/stages/user_login/binding/net"
  17: SESSION_KEY_BINDING_GEO = "authentik/stages/user_login/binding/geo"
  18  LOGGER = get_logger()

Signed-off-by: Jens Langhammer <[email protected]>

netlify · 2025-12-06T01:18:03Z

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	`0668d8f`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-docs/deploys/693448a3894d070008c324cd

netlify · 2025-12-06T01:18:04Z

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	`0668d8f`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/693448a3e12287000712499d

netlify · 2025-12-06T01:18:04Z

✅ Deploy Preview for authentik-integrations ready!

Name	Link
🔨 Latest commit	`0668d8f`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/693448a30349b40008bd4ec0
😎 Deploy Preview	https://deploy-preview-18641--authentik-integrations.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-12-06T01:23:50Z

❌ 3 Tests Failed:

Tests completed	Failed	Passed	Skipped
2821	3	2818	2

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

tests.e2e.test_flows_recovery.TestFlowsRecovery::test_recover_email

Stack Traces | 133s run time

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
selector = 'ak-stage-prompt'
container = <selenium.webdriver.remote.shadowroot.ShadowRoot (session="2c7a0b8334ef4681968ed252989158ec", element="f.1BA4E3F0348D313D873570CD765C7CCD.d.7CC17F9C75A93D18CF2990220CD45122.e.8")>
timeout = 10

    def get_shadow_root(
        self, selector: str, container: WebElement | WebDriver | None = None, timeout: float = 10
    ) -> WebElement:
        """Get the shadow root of a web component specified by `selector`."""
        if not container:
            container = self.driver
        wait = WebDriverWait(container, timeout)
        host: WebElement | None = None
    
        try:
>           host = wait.until(lambda c: c.find_element(By.CSS_SELECTOR, selector))

tests/e2e/utils.py:321: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <[AttributeError("'ShadowRoot' object has no attribute 'session_id'") raised in repr()] WebDriverWait object at 0x7f630c045e10>
method = <function SeleniumTestCase.get_shadow_root.<locals>.<lambda> at 0x7f630afdb9c0>
message = ''

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: 
E       Stacktrace:
E       #0 0x55fabe6ec052 <unknown>
E       #1 0x55fabe1512db <unknown>
E       #2 0x55fabe1a202b <unknown>
E       #3 0x55fabe1a2235 <unknown>
E       #4 0x55fabe19641a <unknown>
E       #5 0x55fabe1c6d91 <unknown>
E       #6 0x55fabe1962f1 <unknown>
E       #7 0x55fabe1c6f52 <unknown>
E       #8 0x55fabe1e8d76 <unknown>
E       #9 0x55fabe1c6b17 <unknown>
E       #10 0x55fabe194781 <unknown>
E       #11 0x55fabe195565 <unknown>
E       #12 0x55fabe6b5904 <unknown>
E       #13 0x55fabe6b8daf <unknown>
E       #14 0x55fabe6b884c <unknown>
E       #15 0x55fabe6b9259 <unknown>
E       #16 0x55fabe69f31b <unknown>
E       #17 0x55fabe6b95e4 <unknown>
E       #18 0x55fabe688b0d <unknown>
E       #19 0x55fabe6d8989 <unknown>
E       #20 0x55fabe6d8b7f <unknown>
E       #21 0x55fabe6eaa09 <unknown>
E       #22 0x7f702e8d3aa4 <unknown>
E       #23 0x7f702e960a64 __clone

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f630c1b0830>
test_case = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
subTest = False

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

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
result = <TestCaseFunction test_recover_email>

    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.13.9.............../x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
method = <bound method TestFlowsRecovery.test_recover_email of <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>>

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

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>,)
kwargs = {}, file = 'example/flows-recovery-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Recovery with...    model: authentik_policies.policybinding\n    attrs:\n      negate: false\n      enabled: true\n      timeout: 30\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-recovery-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_recover_email(self):
        """Test recovery with Email verification"""
        # Attach recovery flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.recovery_flow = Flow.objects.filter(slug="default-recovery-flow").first()
        ident_stage.save()
    
        user = create_test_admin_user()
    
        self.driver.get(self.live_server_url)
        self.initial_stages(user)
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        self.driver.find_element(By.ID, "confirm").click()
        self.driver.close()
        self.driver.switch_to.window(self.driver.window_handles[0])
    
        sleep(2)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
        # We can now enter the new password
        flow_executor = self.get_shadow_root("ak-flow-executor")
>       prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor)

tests/e2e/test_flows_recovery.py:97: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
selector = 'ak-stage-prompt'
container = <selenium.webdriver.remote.shadowroot.ShadowRoot (session="2c7a0b8334ef4681968ed252989158ec", element="f.1BA4E3F0348D313D873570CD765C7CCD.d.7CC17F9C75A93D18CF2990220CD45122.e.8")>
timeout = 10

    def get_shadow_root(
        self, selector: str, container: WebElement | WebDriver | None = None, timeout: float = 10
    ) -> WebElement:
        """Get the shadow root of a web component specified by `selector`."""
        if not container:
            container = self.driver
        wait = WebDriverWait(container, timeout)
        host: WebElement | None = None
    
        try:
            host = wait.until(lambda c: c.find_element(By.CSS_SELECTOR, selector))
        except TimeoutException:
>           self.fail(f"Timed out waiting for shadow host {selector} to appear")

tests/e2e/utils.py:323: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_recovery.TestFlowsRecovery testMethod=test_recover_email>
msg = 'Timed out waiting for shadow host ak-stage-prompt to appear'

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: Timed out waiting for shadow host ak-stage-prompt to appear

.../hostedtoolcache/Python/3.13.9.............../x64/lib/python3.13/unittest/case.py:732: AssertionError

tests.e2e.test_flows_enroll.TestFlowsEnroll::test_enroll_email_pretend_email_scanner

Stack Traces | 214s run time

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email_pretend_email_scanner(self):
        """Test enroll with Email verification. Open the email link twice to pretend we have an
        email scanner that clicks on links"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        confirmation_link = self.driver.find_element(By.ID, "confirm").get_attribute("href")
    
        main_tab = self.driver.current_window_handle
    
        self.driver.switch_to.new_window("tab")
        confirm_tab = self.driver.current_window_handle
    
        # On the new tab, check that we have the confirmation screen
        self.driver.get(confirmation_link)
        self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-flow-executor")))
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
    
        self.assertEqual(
            "Continue to confirm this email address.",
            consent_stage.find_element(By.CSS_SELECTOR, "[data-test-id='stage-heading']").text,
        )
    
        # Back on the main tab, confirm
        self.driver.switch_to.window(main_tab)
        self.driver.get(confirmation_link)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:210: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="427948d0b329deaf4f5c3b539138c3f9")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f6313276660>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=ZGBZCV7jLIF5JZDAvRZ6ZHwM7X91y9UOVw9P...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=ZGBZCV7jLIF5JZDAvRZ6ZHwM7X91y9UOVw9PLmAU5RHAoYqqIlZLcqIgPapF doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email_pretend_email_scanner(self):
        """Test enroll with Email verification. Open the email link twice to pretend we have an
        email scanner that clicks on links"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        confirmation_link = self.driver.find_element(By.ID, "confirm").get_attribute("href")
    
        main_tab = self.driver.current_window_handle
    
        self.driver.switch_to.new_window("tab")
        confirm_tab = self.driver.current_window_handle
    
        # On the new tab, check that we have the confirmation screen
        self.driver.get(confirmation_link)
        self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-flow-executor")))
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
    
        self.assertEqual(
            "Continue to confirm this email address.",
            consent_stage.find_element(By.CSS_SELECTOR, "[data-test-id='stage-heading']").text,
        )
    
        # Back on the main tab, confirm
        self.driver.switch_to.window(main_tab)
        self.driver.get(confirmation_link)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:210: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="db1638441c32727b0a4b79912c7e1b92")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f6300bf4ea0>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=U57Z5T7u04oHsmoaJGUDGkzxvmJQeXSog0Iu...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=U57Z5T7u04oHsmoaJGUDGkzxvmJQeXSog0IuLa9dyznTiDs58zz02uvlO9nV doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f631351a8b0>
test_case = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
subTest = False

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

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
result = <TestCaseFunction test_enroll_email_pretend_email_scanner>

    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.13.9................../x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
method = <bound method TestFlowsEnroll.test_enroll_email_pretend_email_scanner of <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>>

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

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
                raise exc
            logger.debug("Retrying on error", exc=exc, test=self)
            self.tearDown()
            self._post_teardown()
            self._pre_setup()
            self.setUp()
>           return wrapper(self, *args, **kwargs)

tests/e2e/utils.py:474: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
                raise exc
            logger.debug("Retrying on error", exc=exc, test=self)
            self.tearDown()
            self._post_teardown()
            self._pre_setup()
            self.setUp()
>           return wrapper(self, *args, **kwargs)

tests/e2e/utils.py:474: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
>               raise exc

tests/e2e/utils.py:468: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email_pretend_email_scanner(self):
        """Test enroll with Email verification. Open the email link twice to pretend we have an
        email scanner that clicks on links"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        confirmation_link = self.driver.find_element(By.ID, "confirm").get_attribute("href")
    
        main_tab = self.driver.current_window_handle
    
        self.driver.switch_to.new_window("tab")
        confirm_tab = self.driver.current_window_handle
    
        # On the new tab, check that we have the confirmation screen
        self.driver.get(confirmation_link)
        self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-flow-executor")))
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
    
        self.assertEqual(
            "Continue to confirm this email address.",
            consent_stage.find_element(By.CSS_SELECTOR, "[data-test-id='stage-heading']").text,
        )
    
        # Back on the main tab, confirm
        self.driver.switch_to.window(main_tab)
        self.driver.get(confirmation_link)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:210: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email_pretend_email_scanner>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="427f3cffd6d85036f9c8f3efe65b28c8")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f63083c0180>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=6ZYG7MtbSPAPBEjgZiMC8wcJhJixqQv0NuTs...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=6ZYG7MtbSPAPBEjgZiMC8wcJhJixqQv0NuTsfgu5KQ2Ii4cf9Z7rTdICfDs3 doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

tests.e2e.test_flows_enroll.TestFlowsEnroll::test_enroll_email

Stack Traces | 216s run time

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email(self):
        """Test enroll with Email verification"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        self.driver.find_element(By.ID, "confirm").click()
        self.driver.close()
        self.driver.switch_to.window(self.driver.window_handles[0])
    
        sleep(2)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:102: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="7c39b3eed6e03e7855ecfd003c97a328")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f6300bf7100>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=WQwXMQvYviBe2DB8TnTPwo9ex7nmqVbsKchu...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=WQwXMQvYviBe2DB8TnTPwo9ex7nmqVbsKchu3lDX6BruwAux3Myp4ReSYUid doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email(self):
        """Test enroll with Email verification"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        self.driver.find_element(By.ID, "confirm").click()
        self.driver.close()
        self.driver.switch_to.window(self.driver.window_handles[0])
    
        sleep(2)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:102: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="6fca72f3fdbb4ffb0f1b06d49d2c3539")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f6300b70900>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=luYYuGhtEGjMv25W7WNxqz1H7fstLtF7z3W4...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=luYYuGhtEGjMv25W7WNxqz1H7fstLtF7z3W4yjL9tj7M3a7gfGvBoGyFq3RC doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f630befac10>
test_case = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
subTest = False

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

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
result = <TestCaseFunction test_enroll_email>

    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.13.9................../x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
method = <bound method TestFlowsEnroll.test_enroll_email of <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>>

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

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
                raise exc
            logger.debug("Retrying on error", exc=exc, test=self)
            self.tearDown()
            self._post_teardown()
            self._pre_setup()
            self.setUp()
>           return wrapper(self, *args, **kwargs)

tests/e2e/utils.py:474: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
                raise exc
            logger.debug("Retrying on error", exc=exc, test=self)
            self.tearDown()
            self._post_teardown()
            self._pre_setup()
            self.setUp()
>           return wrapper(self, *args, **kwargs)

tests/e2e/utils.py:474: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
            return func(self, *args, **kwargs)
    
        except tuple(exceptions) as exc:
            count += 1
            if count > max_retires:
                logger.debug("Exceeded retry count", exc=exc, test=self)
    
>               raise exc

tests/e2e/utils.py:468: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
args = (), kwargs = {}

    @wraps(func)
    def wrapper(self: TransactionTestCase, *args, **kwargs):
        """Run test again if we're below max_retries, including tearDown and
        setUp. Otherwise raise the error"""
        nonlocal count
        try:
>           return func(self, *args, **kwargs)

tests/e2e/utils.py:461: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'default/flow-default-invalidation-flow.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Invalidation flow\nentries:\n- attrs:\n    designation: invalidation\n    na...0\n    stage: !KeyOf default-invalidation-logout\n    target: !KeyOf flow\n  model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwargs = {}, file = 'example/flows-enrollment-email-verification.yaml'
content = 'version: 1\nmetadata:\n  labels:\n    blueprints.goauthentik.io/instantiate: "false"\n  name: Example - Enrollment wi...ow\n      stage: !KeyOf default-enrollment-user-login\n      order: 100\n    model: authentik_flows.flowstagebinding\n'

    @wraps(func)
    def wrapper(*args, **kwargs):
        for file in files:
            content = BlueprintInstance(path=file).retrieve()
            Importer.from_string(content).apply()
>       return func(*args, **kwargs)

.../blueprints/tests/__init__.py:25: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>,)
kwds = {}

    @wraps(func)
    def inner(*args, **kwds):
        with self._recreate_cm():
>           return func(*args, **kwds)

.../hostedtoolcache/Python/3.13.9................../x64/lib/python3.13/contextlib.py:85: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "example/flows-enrollment-email-verification.yaml",
    )
    @CONFIG.patch("email.port", 1025)
    def test_enroll_email(self):
        """Test enroll with Email verification"""
        # Attach enrollment flow to identification stage
        ident_stage: IdentificationStage = IdentificationStage.objects.get(
            name="default-authentication-identification"
        )
        ident_stage.enrollment_flow = Flow.objects.get(slug="default-enrollment-flow")
        ident_stage.save()
    
        self.driver.get(self.live_server_url)
        self.initial_stages()
    
        # Email stage
        flow_executor = self.get_shadow_root("ak-flow-executor")
        email_stage = self.get_shadow_root("ak-stage-email", flow_executor)
    
        wait = WebDriverWait(email_stage, self.wait_timeout)
    
        # Wait for the success message so we know the email is sent
        wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, ".pf-c-form p")))
    
        # Open Mailpit
        self.driver.get("http://localhost:8025")
    
        # Click on first message
        self.wait.until(ec.presence_of_element_located((By.CLASS_NAME, "message")))
        self.driver.find_element(By.CLASS_NAME, "message").click()
        self.driver.switch_to.frame(self.driver.find_element(By.ID, "preview-html"))
        self.driver.find_element(By.ID, "confirm").click()
        self.driver.close()
        self.driver.switch_to.window(self.driver.window_handles[0])
    
        sleep(2)
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
        consent_stage.find_element(
            By.CSS_SELECTOR,
            "[type=submit]",
        ).click()
    
>       self.wait_for_url(self.if_user_url())

tests/e2e/test_flows_enroll.py:102: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_enroll.TestFlowsEnroll testMethod=test_enroll_email>
desired_url = 'http://10.1.0.190:53995/if/user/'

    def wait_for_url(self, desired_url: str):
        """Wait until URL is `desired_url`."""
    
>       self.wait.until(
            lambda driver: driver.current_url == desired_url,
            f"URL {self.driver.current_url} doesn't match expected URL {desired_url}. "
            f"HTML: {self.driver.page_source[:1000]}",
        )

tests/e2e/utils.py:220: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="4b0a0c5cb336c89ea1dcb0452997aa14")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f630b96e7a0>
message = 'URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=e3NW5TUaitWFkkB9712nNH0zeETYp2kt6Lp7...ry_dsn\\u0022: \\u0022https://[email protected]\\u002Dreporting.a7k.io/4504163677503489'

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
                value = method(self._driver)
                if value:
                    return value
            except self._ignored_exceptions as exc:
                screen = getattr(exc, "screen", None)
                stacktrace = getattr(exc, "stacktrace", None)
            if time.monotonic() > end_time:
                break
            time.sleep(self._poll)
>       raise TimeoutException(message, screen, stacktrace)
E       selenium.common.exceptions.TimeoutException: Message: URL http://10.1.0.190:.../if/flow/default-enrollment-flow/?next=%2F&flow_token=e3NW5TUaitWFkkB9712nNH0zeETYp2kt6Lp74Rq8c0CuynQ44Z7xBhtxCFnO doesn't match expected URL http://10.1.0.190:.../if/user/. HTML: <html lang="en" data-theme="light" data-theme-choice="auto"><head><style>body {transition: opacity ease-in 0.2s; } 
E       body[unresolved] {opacity: 0; display: block; overflow: hidden; position: relative; } 
E       </style>
E               <meta charset="UTF-8">
E               <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1">
E               
E               <meta name="darkreader-lock">
E               <title>Welcome to authentik! - authentik</title>
E               <link rel="icon" href=".../assets/icons/icon.png">
E               <link rel="shortcut icon" href=".../assets/icons/icon.png">
E       
E               
E       
E               
E       <link rel="prefetch" href=".../assets/images/flow_background.jpg">
E       
E       
E       
E       
E       <script data-id="authentik-config">
E           "use strict";
E       
E           window.authentik = {
E               locale: "en",
E               config: JSON.parse('{\u0022error_reporting\u0022: {\u0022enabled\u0022: false, \u0022sentry_dsn\u0022: \u0022https://[email protected]\u002Dreporting.a7k.io/4504163677503489

.venv/lib/python3.13.../webdriver/support/wait.py:146: TimeoutException

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 <[email protected]>

BeryJu added 5 commits December 6, 2025 01:49

add auth for active flow

7b2b768

Signed-off-by: Jens Langhammer <[email protected]>

migrate duo

c8bde57

Signed-off-by: Jens Langhammer <[email protected]>

migrate sms

c2ae9d8

Signed-off-by: Jens Langhammer <[email protected]>

migrate consent

742d587

Signed-off-by: Jens Langhammer <[email protected]>

migrate email and fix broken tests

d0904ab

Signed-off-by: Jens Langhammer <[email protected]>

BeryJu requested a review from a team as a code owner December 6, 2025 01:18

fallback

0668d8f

Signed-off-by: Jens Langhammer <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

stages: remove more global state #18641

stages: remove more global state #18641

Uh oh!

BeryJu commented Dec 6, 2025 •

edited

Loading

Uh oh!

netlify bot commented Dec 6, 2025 •

edited

Loading

Uh oh!

netlify bot commented Dec 6, 2025 •

edited

Loading

Uh oh!

netlify bot commented Dec 6, 2025 •

edited

Loading

Uh oh!

codecov bot commented Dec 6, 2025 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

stages: remove more global state #18641

Are you sure you want to change the base?

stages: remove more global state #18641

Uh oh!

Conversation

BeryJu commented Dec 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

netlify bot commented Dec 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-docs canceled.

Uh oh!

netlify bot commented Dec 6, 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 Dec 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-integrations ready!

Uh oh!

codecov bot commented Dec 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 3 Tests Failed:

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

BeryJu commented Dec 6, 2025 •

edited

Loading

netlify bot commented Dec 6, 2025 •

edited

Loading

netlify bot commented Dec 6, 2025 •

edited

Loading

netlify bot commented Dec 6, 2025 •

edited

Loading

codecov bot commented Dec 6, 2025 •

edited

Loading