core: migrate all sessions to the database

[rissson]

Details

Still a WIP, nothing has been tested

---

Checklist

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

If an API change has been made

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

If changes to the frontend have been made

• The code has been formatted (make web)

If applicable

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

[netlify]

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	b8149c2
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/66449fe5ae1fe900083843ee

[netlify]

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	b8149c2
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-storybook/deploys/66449fe56faef80008e790e0

[netlify]

✅ Deploy Preview for authentik-docs ready!

Name	Link
🔨 Latest commit	71f9ffa
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-docs/deploys/67c0c04c206cb300088b8b0a
😎 Deploy Preview	https://deploy-preview-9736--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.

[netlify]

✅ Deploy Preview for authentik-storybook ready!

Name	Link
🔨 Latest commit	71f9ffa
🔍 Latest deploy log	https://app.netlify.com/sites/authentik-storybook/deploys/67c0c04c10a76d00081f01ad
😎 Deploy Preview	https://deploy-preview-9736--authentik-storybook.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.

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
1717	1	1716	2

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

tests.e2e.test_flows_stage_setup.TestFlowsStageSetup::test_password_change

Stack Traces | 207s run time

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:294: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
kwargs = {}, file = 'default/flow-password-change.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Password change flow\nentries:\n- attrs:\n    check_static_rules: true\n    ...n    stage: !KeyOf default-password-change-write\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_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
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: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>

    @retry()
    @apply_blueprint("default/flow-password-change.yaml")
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_password_change(self):
        """test password change flow"""
        # Ensure that password stage has change_flow set
        flow = Flow.objects.get(
            slug="default-password-change",
            designation=FlowDesignation.STAGE_CONFIGURATION,
        )
    
        stage = PasswordStage.objects.get(name="default-authentication-password")
        stage.configure_flow = flow
        stage.save()
    
        new_password = generate_key()
    
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
        self.login()
        self.wait_for_url(self.if_user_url("/library"))
    
        self.driver.get(
            self.url(
                "authentik_flows:configure",
                stage_uuid=PasswordStage.objects.first().stage_uuid,
            )
        )
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor)
    
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(new_password)
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            new_password
        )
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            Keys.ENTER
        )
    
>       self.wait_for_url(self.if_user_url("/library"))

tests/e2e/test_flows_stage_setup.py:64: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
desired_url = 'http://10.1.0.20:45589/if/user/#/library'

    def wait_for_url(self, desired_url):
        """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}",
        )

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

self = <selenium.webdriver.support.wait.WebDriverWait (session="c7bda2769a05fc384533ebb093898223")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f7b3bc65620>
message = "URL http://10.1.0.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library"

    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.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:294: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
kwargs = {}, file = 'default/flow-password-change.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Password change flow\nentries:\n- attrs:\n    check_static_rules: true\n    ...n    stage: !KeyOf default-password-change-write\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_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
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: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>

    @retry()
    @apply_blueprint("default/flow-password-change.yaml")
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_password_change(self):
        """test password change flow"""
        # Ensure that password stage has change_flow set
        flow = Flow.objects.get(
            slug="default-password-change",
            designation=FlowDesignation.STAGE_CONFIGURATION,
        )
    
        stage = PasswordStage.objects.get(name="default-authentication-password")
        stage.configure_flow = flow
        stage.save()
    
        new_password = generate_key()
    
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
        self.login()
        self.wait_for_url(self.if_user_url("/library"))
    
        self.driver.get(
            self.url(
                "authentik_flows:configure",
                stage_uuid=PasswordStage.objects.first().stage_uuid,
            )
        )
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor)
    
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(new_password)
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            new_password
        )
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            Keys.ENTER
        )
    
>       self.wait_for_url(self.if_user_url("/library"))

tests/e2e/test_flows_stage_setup.py:64: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
desired_url = 'http://10.1.0.20:45589/if/user/#/library'

    def wait_for_url(self, desired_url):
        """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}",
        )

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

self = <selenium.webdriver.support.wait.WebDriverWait (session="69194e458d7f0a1f04fce3b165b4a984")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f7b3bcbd120>
message = "URL http://10.1.0.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library"

    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.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../webdriver/support/wait.py:146: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f7b3827cfb0>
test_case = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
subTest = False

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

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

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
result = <TestCaseFunction test_password_change>

    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 = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
method = <bound method TestFlowsStageSetup.test_password_change of <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>>

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

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

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:307: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:307: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:301: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
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:294: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
kwargs = {}, file = 'default/flow-password-change.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Password change flow\nentries:\n- attrs:\n    check_static_rules: true\n    ...n    stage: !KeyOf default-password-change-write\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_stage_setup.TestFlowsStageSetup testMethod=test_password_change>,)
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: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>

    @retry()
    @apply_blueprint("default/flow-password-change.yaml")
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_password_change(self):
        """test password change flow"""
        # Ensure that password stage has change_flow set
        flow = Flow.objects.get(
            slug="default-password-change",
            designation=FlowDesignation.STAGE_CONFIGURATION,
        )
    
        stage = PasswordStage.objects.get(name="default-authentication-password")
        stage.configure_flow = flow
        stage.save()
    
        new_password = generate_key()
    
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
        self.login()
        self.wait_for_url(self.if_user_url("/library"))
    
        self.driver.get(
            self.url(
                "authentik_flows:configure",
                stage_uuid=PasswordStage.objects.first().stage_uuid,
            )
        )
    
        flow_executor = self.get_shadow_root("ak-flow-executor")
        prompt_stage = self.get_shadow_root("ak-stage-prompt", flow_executor)
    
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(new_password)
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            new_password
        )
        prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
            Keys.ENTER
        )
    
>       self.wait_for_url(self.if_user_url("/library"))

tests/e2e/test_flows_stage_setup.py:64: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_stage_setup.TestFlowsStageSetup testMethod=test_password_change>
desired_url = 'http://10.1.0.20:45589/if/user/#/library'

    def wait_for_url(self, desired_url):
        """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}",
        )

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

self = <selenium.webdriver.support.wait.WebDriverWait (session="b488bd8482efff5c6229b648c26e4d5d")>
method = <function SeleniumTestCase.wait_for_url.<locals>.<lambda> at 0x7f7b2ed144a0>
message = "URL http://10.1.0.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library"

    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.20:.../if/flow/default-password-change/?inspector=available doesn't match expected URL http://10.1.0.20:45589/if/user/#/library

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../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.}

[rissson]

TODO: also delete session?

[rissson]

TODO: custom channels authmiddleware

EDIT: done