outpost/proxy: rewrite in rust by rissson · Pull Request #21856 · goauthentik/authentik

rissson · 2026-04-27T13:37:04Z

Needs:

netlify · 2026-04-27T13:41:24Z

✅ Deploy Preview for authentik-docs ready!

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

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

codecov · 2026-04-27T13:46:57Z

❌ 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
3828	6	3822	1

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

tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_caddy

Stack Traces | 43.5s run time

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="f7c253af525ca8490da6c82145801007")>
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/selenium.py:244: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="f7c253af525ca8490da6c82145801007")>
method = <function SeleniumTestMixin.get_shadow_root.<locals>.<lambda> at 0x7f94a1272a30>
message = ''

    def until(self, method: Callable[[D], 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``.
    
        Args:
            method: A callable object that takes a WebDriver instance as an
                argument.
            message: Optional message for TimeoutException.
    
        Returns:
            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.wait 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 0x557db41d1df2 <unknown>
E       #1 0x557db3b51398 <unknown>
E       #2 0x557db3ba60f9 <unknown>
E       #3 0x557db3ba64a5 <unknown>
E       #4 0x557db3bf6d88 <unknown>
E       #5 0x557db3bcf321 <unknown>
E       #6 0x557db3bf43c1 <unknown>
E       #7 0x557db3bcf097 <unknown>
E       #8 0x557db3b9928e <unknown>
E       #9 0x557db3b9a125 <unknown>
E       #10 0x557db41960a9 <unknown>
E       #11 0x557db4194a2c <unknown>
E       #12 0x557db419441e <unknown>
E       #13 0x557db4194ea9 <unknown>
E       #14 0x557db4182ea9 <unknown>
E       #15 0x557db419525c <unknown>
E       #16 0x557db416d429 <unknown>
E       #17 0x557db41be279 <unknown>
E       #18 0x557db41be454 <unknown>
E       #19 0x557db41d06ce <unknown>
E       #20 0x7f4c50aeb289 <unknown>

.venv/lib/python3.14.../webdriver/support/wait.py:121: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f949b952e90>
test_case = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
result = <TestCaseFunction test_caddy>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
method = <bound method TestProviderProxyForward.test_caddy of <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>

    @retry()
    def test_caddy(self):
        """Test caddy"""
        local_config_path = (
            Path(__file__).parent / "proxy_forward_auth" / "caddy_single" / "Caddyfile"
        )
        self.run_container(
            image="docker.io/library/caddy:2.8",
            ports={
                "80": "80",
            },
            volumes={
                local_config_path: {
                    "bind": ".../etc/caddy/Caddyfile",
                }
            },
        )
    
        self.prepare()
    
        self.driver.get("http:.../localhost/api")
>       self.login()

tests/e2e/test_provider_proxy_forward.py:221: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
shadow_dom = True, skip_stages = []

    def login(self, shadow_dom=True, skip_stages: list[str] | None = None):
        """Perform the entire authentik login flow."""
        skip_stages = skip_stages or []
    
        if "ak-stage-identification" not in skip_stages:
            if shadow_dom:
>               flow_executor = self.get_shadow_root("ak-flow-executor")
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/selenium.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="f7c253af525ca8490da6c82145801007")>
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/selenium.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_caddy>
msg = 'Timed out waiting for shadow host ak-flow-executor 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-flow-executor to appear

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

tests.e2e.test_provider_proxy.TestProviderProxy::test_proxy_simple

Stack Traces | 47.3s run time

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="d6daba74f8b0d7ff5999fcfb64a9e429")>
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/selenium.py:244: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="d6daba74f8b0d7ff5999fcfb64a9e429")>
method = <function SeleniumTestMixin.get_shadow_root.<locals>.<lambda> at 0x7f94a0de3270>
message = ''

    def until(self, method: Callable[[D], 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``.
    
        Args:
            method: A callable object that takes a WebDriver instance as an
                argument.
            message: Optional message for TimeoutException.
    
        Returns:
            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.wait 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 0x5646e2ad5df2 <unknown>
E       #1 0x5646e2455398 <unknown>
E       #2 0x5646e24aa0f9 <unknown>
E       #3 0x5646e24aa4a5 <unknown>
E       #4 0x5646e24fad88 <unknown>
E       #5 0x5646e24d3321 <unknown>
E       #6 0x5646e24f83c1 <unknown>
E       #7 0x5646e24d3097 <unknown>
E       #8 0x5646e249d28e <unknown>
E       #9 0x5646e249e125 <unknown>
E       #10 0x5646e2a9a0a9 <unknown>
E       #11 0x5646e2a98a2c <unknown>
E       #12 0x5646e2a9841e <unknown>
E       #13 0x5646e2a98ea9 <unknown>
E       #14 0x5646e2a86ea9 <unknown>
E       #15 0x5646e2a9925c <unknown>
E       #16 0x5646e2a71429 <unknown>
E       #17 0x5646e2ac2279 <unknown>
E       #18 0x5646e2ac2454 <unknown>
E       #19 0x5646e2ad46ce <unknown>
E       #20 0x7f2387ed6289 <unknown>

.venv/lib/python3.14.../webdriver/support/wait.py:121: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f949bec9350>
test_case = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
result = <TestCaseFunction test_proxy_simple>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
method = <bound method TestProviderProxy.test_proxy_simple of <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>,)
kwargs = {}

    @wraps(func)
    def wrapper(*args, **kwargs):
        config = apps.get_app_config(app_name)
        if isinstance(config, ManagedAppConfig):
            config._on_startup_callback(None)
>       return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "default/flow-default-provider-authorization-implicit-consent.yaml",
        "default/flow-default-provider-invalidation.yaml",
    )
    @apply_blueprint(
        "system/providers-oauth2.yaml",
        "system/providers-proxy.yaml",
    )
    @reconcile_app("authentik_crypto")
    def test_proxy_simple(self):
        """Test simple outpost setup with single provider"""
        # set additionalHeaders to test later
        self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"}
        self.user.save()
    
        proxy: ProxyProvider = ProxyProvider.objects.create(
            name=generate_id(),
            authorization_flow=Flow.objects.get(
                slug="default-provider-authorization-implicit-consent"
            ),
            invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"),
            internal_host=f"http://{self.host}",
            external_host="http://localhost:9000",
        )
        # Ensure OAuth2 Params are set
        proxy.set_oauth_defaults()
        proxy.save()
        # we need to create an application to actually access the proxy
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy)
        outpost: Outpost = Outpost.objects.create(
            name=generate_id(),
            type=OutpostType.PROXY,
        )
        outpost.providers.add(proxy)
        outpost.build_user_permissions(outpost.user)
    
        self.start_proxy(outpost)
    
        sleep(5)
    
        self.driver.get("http://localhost:9000/api")
>       self.login()

tests/e2e/test_provider_proxy.py:92: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
shadow_dom = True, skip_stages = []

    def login(self, shadow_dom=True, skip_stages: list[str] | None = None):
        """Perform the entire authentik login flow."""
        skip_stages = skip_stages or []
    
        if "ak-stage-identification" not in skip_stages:
            if shadow_dom:
>               flow_executor = self.get_shadow_root("ak-flow-executor")
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/selenium.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="d6daba74f8b0d7ff5999fcfb64a9e429")>
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/selenium.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_simple>
msg = 'Timed out waiting for shadow host ak-flow-executor 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-flow-executor to appear

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

tests.e2e.test_provider_proxy.TestProviderProxy::test_proxy_basic_auth

Stack Traces | 47.6s run time

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="1518c22d6f3ca8a5ce99e116f32bf9af")>
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/selenium.py:244: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="1518c22d6f3ca8a5ce99e116f32bf9af")>
method = <function SeleniumTestMixin.get_shadow_root.<locals>.<lambda> at 0x7f94a119b8a0>
message = ''

    def until(self, method: Callable[[D], 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``.
    
        Args:
            method: A callable object that takes a WebDriver instance as an
                argument.
            message: Optional message for TimeoutException.
    
        Returns:
            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.wait 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 0x560091352df2 <unknown>
E       #1 0x560090cd2398 <unknown>
E       #2 0x560090d270f9 <unknown>
E       #3 0x560090d274a5 <unknown>
E       #4 0x560090d77d88 <unknown>
E       #5 0x560090d50321 <unknown>
E       #6 0x560090d753c1 <unknown>
E       #7 0x560090d50097 <unknown>
E       #8 0x560090d1a28e <unknown>
E       #9 0x560090d1b125 <unknown>
E       #10 0x5600913170a9 <unknown>
E       #11 0x560091315a2c <unknown>
E       #12 0x56009131541e <unknown>
E       #13 0x560091315ea9 <unknown>
E       #14 0x560091303ea9 <unknown>
E       #15 0x56009131625c <unknown>
E       #16 0x5600912ee429 <unknown>
E       #17 0x56009133f279 <unknown>
E       #18 0x56009133f454 <unknown>
E       #19 0x5600913516ce <unknown>
E       #20 0x7ff9e44f5289 <unknown>

.venv/lib/python3.14.../webdriver/support/wait.py:121: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f949bec82b0>
test_case = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
result = <TestCaseFunction test_proxy_basic_auth>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
method = <bound method TestProviderProxy.test_proxy_basic_auth of <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>,)
kwargs = {}

    @wraps(func)
    def wrapper(*args, **kwargs):
        config = apps.get_app_config(app_name)
        if isinstance(config, ManagedAppConfig):
            config._on_startup_callback(None)
>       return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^

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

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "default/flow-default-provider-authorization-implicit-consent.yaml",
        "default/flow-default-provider-invalidation.yaml",
    )
    @apply_blueprint(
        "system/providers-oauth2.yaml",
        "system/providers-proxy.yaml",
    )
    @reconcile_app("authentik_crypto")
    def test_proxy_basic_auth(self):
        """Test simple outpost setup with single provider"""
        cred = generate_id()
        attr = "basic-password"  # nosec
        self.user.attributes["basic-username"] = cred
        self.user.attributes[attr] = cred
        self.user.save()
    
        proxy: ProxyProvider = ProxyProvider.objects.create(
            name=generate_id(),
            authorization_flow=Flow.objects.get(
                slug="default-provider-authorization-implicit-consent"
            ),
            invalidation_flow=Flow.objects.get(slug="default-provider-invalidation-flow"),
            internal_host=f"http://{self.host}",
            external_host="http://localhost:9000",
            basic_auth_enabled=True,
            basic_auth_user_attribute="basic-username",
            basic_auth_password_attribute=attr,
        )
        # Ensure OAuth2 Params are set
        proxy.set_oauth_defaults()
        proxy.save()
        # we need to create an application to actually access the proxy
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy)
        outpost: Outpost = Outpost.objects.create(
            name=generate_id(),
            type=OutpostType.PROXY,
        )
        outpost.providers.add(proxy)
        outpost.build_user_permissions(outpost.user)
    
        self.start_proxy(outpost)
    
        sleep(5)
    
        self.driver.get("http://localhost:9000/api")
>       self.login()

tests/e2e/test_provider_proxy.py:180: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
shadow_dom = True, skip_stages = []

    def login(self, shadow_dom=True, skip_stages: list[str] | None = None):
        """Perform the entire authentik login flow."""
        skip_stages = skip_stages or []
    
        if "ak-stage-identification" not in skip_stages:
            if shadow_dom:
>               flow_executor = self.get_shadow_root("ak-flow-executor")
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/selenium.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="1518c22d6f3ca8a5ce99e116f32bf9af")>
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/selenium.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxy testMethod=test_proxy_basic_auth>
msg = 'Timed out waiting for shadow host ak-flow-executor 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-flow-executor to appear

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

tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_envoy

Stack Traces | 47.9s run time

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="30a8324a8123bbe01eab396ba0775bce")>
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/selenium.py:244: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="30a8324a8123bbe01eab396ba0775bce")>
method = <function SeleniumTestMixin.get_shadow_root.<locals>.<lambda> at 0x7f94a16d4ca0>
message = ''

    def until(self, method: Callable[[D], 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``.
    
        Args:
            method: A callable object that takes a WebDriver instance as an
                argument.
            message: Optional message for TimeoutException.
    
        Returns:
            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.wait 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 0x556584af2df2 <unknown>
E       #1 0x556584472398 <unknown>
E       #2 0x5565844c70f9 <unknown>
E       #3 0x5565844c74a5 <unknown>
E       #4 0x556584517d88 <unknown>
E       #5 0x5565844f0321 <unknown>
E       #6 0x5565845153c1 <unknown>
E       #7 0x5565844f0097 <unknown>
E       #8 0x5565844ba28e <unknown>
E       #9 0x5565844bb125 <unknown>
E       #10 0x556584ab70a9 <unknown>
E       #11 0x556584ab5a2c <unknown>
E       #12 0x556584ab541e <unknown>
E       #13 0x556584ab5ea9 <unknown>
E       #14 0x556584aa3ea9 <unknown>
E       #15 0x556584ab625c <unknown>
E       #16 0x556584a8e429 <unknown>
E       #17 0x556584adf279 <unknown>
E       #18 0x556584adf454 <unknown>
E       #19 0x556584af16ce <unknown>
E       #20 0x7ff3d4ce5289 <unknown>

.venv/lib/python3.14.../webdriver/support/wait.py:121: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f949b950190>
test_case = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
result = <TestCaseFunction test_envoy>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
method = <bound method TestProviderProxyForward.test_envoy of <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>

    @retry()
    def test_envoy(self):
        """Test envoy"""
        self.run_container(
            image="docker.io/envoyproxy/envoy:v1.25-latest",
            ports={
                "10000": "80",
            },
            volumes={
                f"{Path(__file__).parent / "proxy_forward_auth" / "envoy_single" / "envoy.yaml"}": {
                    "bind": ".../etc/envoy/envoy.yaml",
                }
            },
        )
    
        self.prepare()
    
        self.driver.get("http:.../localhost/api")
>       self.login()

tests/e2e/test_provider_proxy_forward.py:180: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
shadow_dom = True, skip_stages = []

    def login(self, shadow_dom=True, skip_stages: list[str] | None = None):
        """Perform the entire authentik login flow."""
        skip_stages = skip_stages or []
    
        if "ak-stage-identification" not in skip_stages:
            if shadow_dom:
>               flow_executor = self.get_shadow_root("ak-flow-executor")
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/selenium.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="30a8324a8123bbe01eab396ba0775bce")>
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/selenium.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_envoy>
msg = 'Timed out waiting for shadow host ak-flow-executor 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-flow-executor to appear

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

tests.e2e.test_provider_proxy_forward.TestProviderProxyForward::test_traefik

Stack Traces | 130s run time

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="d441dc192b16ede5f35025d81a32743c")>
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/selenium.py:244: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.support.wait.WebDriverWait (session="d441dc192b16ede5f35025d81a32743c")>
method = <function SeleniumTestMixin.get_shadow_root.<locals>.<lambda> at 0x7f94a0de3b60>
message = ''

    def until(self, method: Callable[[D], 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``.
    
        Args:
            method: A callable object that takes a WebDriver instance as an
                argument.
            message: Optional message for TimeoutException.
    
        Returns:
            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.wait 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 0x561582652df2 <unknown>
E       #1 0x561581fd2398 <unknown>
E       #2 0x5615820270f9 <unknown>
E       #3 0x5615820274a5 <unknown>
E       #4 0x561582077d88 <unknown>
E       #5 0x561582050321 <unknown>
E       #6 0x5615820753c1 <unknown>
E       #7 0x561582050097 <unknown>
E       #8 0x56158201a28e <unknown>
E       #9 0x56158201b125 <unknown>
E       #10 0x5615826170a9 <unknown>
E       #11 0x561582615a2c <unknown>
E       #12 0x56158261541e <unknown>
E       #13 0x561582615ea9 <unknown>
E       #14 0x561582603ea9 <unknown>
E       #15 0x56158261625c <unknown>
E       #16 0x5615825ee429 <unknown>
E       #17 0x56158263f279 <unknown>
E       #18 0x56158263f454 <unknown>
E       #19 0x5615826516ce <unknown>
E       #20 0x7fc25ea6f289 <unknown>

.venv/lib/python3.14.../webdriver/support/wait.py:121: TimeoutException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f94a19fe3c0>
test_case = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
result = <TestCaseFunction test_traefik>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
method = <bound method TestProviderProxyForward.test_traefik of <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>>

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

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>

    @retry()
    def test_traefik(self):
        """Test traefik"""
        local_config_path = (
            Path(__file__).parent / "proxy_forward_auth" / "traefik_single" / "config-static.yaml"
        )
        self.run_container(
            image="docker.io/library/traefik:3.1",
            ports={
                "80": "80",
            },
            volumes={
                local_config_path: {
                    "bind": ".../etc/traefik/traefik.yml",
                }
            },
        )
    
        self.prepare()
    
        self.driver.get("http:.../localhost/api")
>       self.login()

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

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
shadow_dom = True, skip_stages = []

    def login(self, shadow_dom=True, skip_stages: list[str] | None = None):
        """Perform the entire authentik login flow."""
        skip_stages = skip_stages or []
    
        if "ak-stage-identification" not in skip_stages:
            if shadow_dom:
>               flow_executor = self.get_shadow_root("ak-flow-executor")
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/selenium.py:293: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
selector = 'ak-flow-executor'
container = <selenium.webdriver.remote.webdriver.WebDriver (session="d441dc192b16ede5f35025d81a32743c")>
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/selenium.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy_forward.TestProviderProxyForward testMethod=test_traefik>
msg = 'Timed out waiting for shadow host ak-flow-executor 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-flow-executor to appear

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

tests.e2e.test_provider_proxy.TestProviderProxyConnect::test_proxy_connectivity

Stack Traces | 156s run time

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @wraps(func)
    def wrapper(*args, **kwargs):
        config = apps.get_app_config(app_name)
        if isinstance(config, ManagedAppConfig):
            config._on_startup_callback(None)
>       return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>

    @retry(exceptions=[AssertionError])
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "default/flow-default-provider-authorization-implicit-consent.yaml",
    )
    @reconcile_app("authentik_crypto")
    def test_proxy_connectivity(self):
        """Test proxy connectivity over websocket"""
        proxy: ProxyProvider = ProxyProvider.objects.create(
            name=generate_id(),
            authorization_flow=Flow.objects.get(
                slug="default-provider-authorization-implicit-consent"
            ),
            internal_host="http://localhost",
            external_host="http://localhost:9000",
        )
        # Ensure OAuth2 Params are set
        proxy.set_oauth_defaults()
        proxy.save()
        # we need to create an application to actually access the proxy
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy)
    
        outpost: Outpost = Outpost.objects.create(
            name=generate_id(),
            type=OutpostType.PROXY,
            _config=asdict(OutpostConfig(authentik_host=self.live_server_url, log_level="debug")),
        )
        outpost.providers.add(proxy)
        outpost.build_user_permissions(outpost.user)
    
>       self.run_container(
            image=self.get_container_image("ghcr.io/goauthentik/dev-proxy"),
            ports={
                "9000": "9000",
            },
            environment={
                "AUTHENTIK_TOKEN": outpost.token.key,
            },
        )

tests/e2e/test_provider_proxy.py:252: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
specs = {'image': 'ghcr.io/goauthentik/dev-proxy:gh-rust-proxy', 'ports': {'9000': '9000'}, 'environment': {'AUTHENTIK_TOKEN':...AUTHENTIK_HOST': 'http://10.1.0.167:37377'}, 'network': 'authentik-test-nhgpMp2CsSit3OCG64bj5H1vzIdy4Y5AuQTGew0E', ...}

    def run_container(self, **specs: Any) -> Container:
        if "network_mode" not in specs:
            specs["network"] = self.__network.name
        specs["labels"] = self.docker_labels
        specs["detach"] = True
        specs.setdefault("environment", {})
        specs["environment"]["AUTHENTIK_LOG_LEVEL"] = "debug"
        if hasattr(self, "live_server_url"):
            specs["environment"]["AUTHENTIK_HOST"] = self.live_server_url
        container: Container = self.docker_client.containers.run(**specs)
        container.reload()
        state = container.attrs.get("State", {})
        if "Health" not in state:
            return container
>       self.wait_for_container(container)

tests/docker.py:74: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
container = <Container: 6b4895aabe26>

    def wait_for_container(self, container: Container) -> Container:
        """Check that container is health"""
        attempt = 0
        while True:
            container.reload()
            status = container.attrs.get("State", {}).get("Health", {}).get("Status")
            if status == "healthy":
                return container
            sleep(1)
            attempt += 1
            if attempt >= self.max_healthcheck_attempts:
                self.output_container_logs(container)
>               raise self.failureException("Container failed to start")
E               AssertionError: Container failed to start

tests/docker.py:54: AssertionError

During handling of the above exception, another exception occurred:

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @wraps(func)
    def wrapper(*args, **kwargs):
        config = apps.get_app_config(app_name)
        if isinstance(config, ManagedAppConfig):
            config._on_startup_callback(None)
>       return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>

    @retry(exceptions=[AssertionError])
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "default/flow-default-provider-authorization-implicit-consent.yaml",
    )
    @reconcile_app("authentik_crypto")
    def test_proxy_connectivity(self):
        """Test proxy connectivity over websocket"""
        proxy: ProxyProvider = ProxyProvider.objects.create(
            name=generate_id(),
            authorization_flow=Flow.objects.get(
                slug="default-provider-authorization-implicit-consent"
            ),
            internal_host="http://localhost",
            external_host="http://localhost:9000",
        )
        # Ensure OAuth2 Params are set
        proxy.set_oauth_defaults()
        proxy.save()
        # we need to create an application to actually access the proxy
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy)
    
        outpost: Outpost = Outpost.objects.create(
            name=generate_id(),
            type=OutpostType.PROXY,
            _config=asdict(OutpostConfig(authentik_host=self.live_server_url, log_level="debug")),
        )
        outpost.providers.add(proxy)
        outpost.build_user_permissions(outpost.user)
    
>       self.run_container(
            image=self.get_container_image("ghcr.io/goauthentik/dev-proxy"),
            ports={
                "9000": "9000",
            },
            environment={
                "AUTHENTIK_TOKEN": outpost.token.key,
            },
        )

tests/e2e/test_provider_proxy.py:252: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
specs = {'image': 'ghcr.io/goauthentik/dev-proxy:gh-rust-proxy', 'ports': {'9000': '9000'}, 'environment': {'AUTHENTIK_TOKEN':...AUTHENTIK_HOST': 'http://10.1.0.167:37377'}, 'network': 'authentik-test-604EYk09sfm6RuMHgJT0eYjm9Fjdsb34VboscxPT', ...}

    def run_container(self, **specs: Any) -> Container:
        if "network_mode" not in specs:
            specs["network"] = self.__network.name
        specs["labels"] = self.docker_labels
        specs["detach"] = True
        specs.setdefault("environment", {})
        specs["environment"]["AUTHENTIK_LOG_LEVEL"] = "debug"
        if hasattr(self, "live_server_url"):
            specs["environment"]["AUTHENTIK_HOST"] = self.live_server_url
        container: Container = self.docker_client.containers.run(**specs)
        container.reload()
        state = container.attrs.get("State", {})
        if "Health" not in state:
            return container
>       self.wait_for_container(container)

tests/docker.py:74: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
container = <Container: 8e02efd087bc>

    def wait_for_container(self, container: Container) -> Container:
        """Check that container is health"""
        attempt = 0
        while True:
            container.reload()
            status = container.attrs.get("State", {}).get("Health", {}).get("Status")
            if status == "healthy":
                return container
            sleep(1)
            attempt += 1
            if attempt >= self.max_healthcheck_attempts:
                self.output_container_logs(container)
>               raise self.failureException("Container failed to start")
E               AssertionError: Container failed to start

tests/docker.py:54: AssertionError

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f94a0b58710>
test_case = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
subTest = False

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
result = <TestCaseFunction test_proxy_connectivity>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
method = <bound method TestProviderProxyConnect.test_proxy_connectivity of <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>>

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:67: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
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/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @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_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}

    @wraps(func)
    def wrapper(*args, **kwargs):
        config = apps.get_app_config(app_name)
        if isinstance(config, ManagedAppConfig):
            config._on_startup_callback(None)
>       return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>

    @retry(exceptions=[AssertionError])
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @apply_blueprint(
        "default/flow-default-provider-authorization-implicit-consent.yaml",
    )
    @reconcile_app("authentik_crypto")
    def test_proxy_connectivity(self):
        """Test proxy connectivity over websocket"""
        proxy: ProxyProvider = ProxyProvider.objects.create(
            name=generate_id(),
            authorization_flow=Flow.objects.get(
                slug="default-provider-authorization-implicit-consent"
            ),
            internal_host="http://localhost",
            external_host="http://localhost:9000",
        )
        # Ensure OAuth2 Params are set
        proxy.set_oauth_defaults()
        proxy.save()
        # we need to create an application to actually access the proxy
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=proxy)
    
        outpost: Outpost = Outpost.objects.create(
            name=generate_id(),
            type=OutpostType.PROXY,
            _config=asdict(OutpostConfig(authentik_host=self.live_server_url, log_level="debug")),
        )
        outpost.providers.add(proxy)
        outpost.build_user_permissions(outpost.user)
    
>       self.run_container(
            image=self.get_container_image("ghcr.io/goauthentik/dev-proxy"),
            ports={
                "9000": "9000",
            },
            environment={
                "AUTHENTIK_TOKEN": outpost.token.key,
            },
        )

tests/e2e/test_provider_proxy.py:252: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
specs = {'image': 'ghcr.io/goauthentik/dev-proxy:gh-rust-proxy', 'ports': {'9000': '9000'}, 'environment': {'AUTHENTIK_TOKEN':...AUTHENTIK_HOST': 'http://10.1.0.167:37377'}, 'network': 'authentik-test-4rva7hfjH4uwG1aVMVtI40bq2bs5oDIHnLAtrdL1', ...}

    def run_container(self, **specs: Any) -> Container:
        if "network_mode" not in specs:
            specs["network"] = self.__network.name
        specs["labels"] = self.docker_labels
        specs["detach"] = True
        specs.setdefault("environment", {})
        specs["environment"]["AUTHENTIK_LOG_LEVEL"] = "debug"
        if hasattr(self, "live_server_url"):
            specs["environment"]["AUTHENTIK_HOST"] = self.live_server_url
        container: Container = self.docker_client.containers.run(**specs)
        container.reload()
        state = container.attrs.get("State", {})
        if "Health" not in state:
            return container
>       self.wait_for_container(container)

tests/docker.py:74: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
container = <Container: c0dabb4d47d8>

    def wait_for_container(self, container: Container) -> Container:
        """Check that container is health"""
        attempt = 0
        while True:
            container.reload()
            status = container.attrs.get("State", {}).get("Health", {}).get("Status")
            if status == "healthy":
                return container
            sleep(1)
            attempt += 1
            if attempt >= self.max_healthcheck_attempts:
                self.output_container_logs(container)
>               raise self.failureException("Container failed to start")
E               AssertionError: Container failed to start

tests/docker.py:54: AssertionError

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

netlify · 2026-04-27T16:05:26Z

✅ Deploy Preview for authentik-docs ready!

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

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

netlify · 2026-04-30T11:35:42Z

✅ Deploy Preview for authentik-integrations ready!

Name	Link
🔨 Latest commit	`fa223dc`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/6a0f1f4835adb20008693854
😎 Deploy Preview	https://deploy-preview-21856--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.

netlify · 2026-04-30T11:36:19Z

✅ Deploy Preview for authentik-storybook ready!

Name	Link
🔨 Latest commit	`1aa98c8`
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/6a5a4438eb63a60008e6d90d
😎 Deploy Preview	https://deploy-preview-21856--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 project configuration.

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

github-actions · 2026-07-17T17:57:16Z

authentik PR Installation instructions

Instructions for docker-compose

Add the following block to your .env file:

AUTHENTIK_IMAGE=ghcr.io/goauthentik/dev-server
AUTHENTIK_TAG=gh-f64a736f85fb49eeca6126e9ab9ee7feebb2d8b1
AUTHENTIK_OUTPOSTS__CONTAINER_IMAGE_BASE=ghcr.io/goauthentik/dev-%(type)s:gh-%(build_hash)s

Afterwards, run the upgrade commands from the latest release notes.

Instructions for Kubernetes

Add the following block to your values.yml file:

authentik:
    outposts:
        container_image_base: ghcr.io/goauthentik/dev-%(type)s:gh-%(build_hash)s
global:
    image:
        repository: ghcr.io/goauthentik/dev-server
        tag: gh-f64a736f85fb49eeca6126e9ab9ee7feebb2d8b1

Afterwards, run the upgrade commands from the latest release notes.

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

rissson · 2026-07-20T13:17:30Z

    needs:
      - build-compute-tags
      - build
-      - ci-core-mark


NOTE TO SELF: revert this

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

rissson self-assigned this Apr 27, 2026

rissson added the area:backend label Apr 27, 2026

rissson added this to authentik Core Apr 27, 2026

github-project-automation Bot moved this to Todo in authentik Core Apr 27, 2026

rissson force-pushed the rust-proxy branch from 04669c9 to b11d8b6 Compare April 27, 2026 13:37

rissson force-pushed the rust-proxy branch from 0018bc4 to 94a6646 Compare April 27, 2026 16:03

rissson force-pushed the rust-proxy branch 4 times, most recently from 8e041fc to 440a7ac Compare April 29, 2026 12:28

rissson moved this from Todo to Needs review in authentik Core Apr 30, 2026

rissson moved this from Needs review to Todo in authentik Core Apr 30, 2026

rissson moved this from Todo to In Progress in authentik Core Apr 30, 2026

rissson force-pushed the rust-proxy branch 4 times, most recently from c54653c to 5cb0c4f Compare May 7, 2026 13:42

rissson mentioned this pull request May 11, 2026

feat(outpost): add configurable TLS ciphers and minimum version #22203

Closed

rissson force-pushed the rust-proxy branch 4 times, most recently from fbcabd2 to 4d141dd Compare May 13, 2026 13:14

rissson force-pushed the rust-proxy branch 3 times, most recently from f6e7000 to fa223dc Compare May 21, 2026 15:05

rissson force-pushed the rust-proxy branch from a910bb8 to ea970dd Compare June 30, 2026 13:21

rissson added 16 commits June 30, 2026 15:22

Merge branch 'main' into rust-proxy

c67c850

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

revert

e67fe8a

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

lint

e619ccb

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

temporarly disable health check

d67d979

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

remove all underscore headers

422e9f0

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

add headers to configuration error report

a24cf58

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

no auth if session fails to write to backend

d6fada5

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

spellcheck

ab5923a

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

fix envoy url allowlisting

d393392

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

Merge branch 'main' into rust-proxy

d940a9a

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

fix build

634cbfb

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

reenable healthcheck

a56c350

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

Merge branch 'main' into rust-proxy

45997b7

Merge branch 'main' into rust-proxy

a78b6b4

spell

75d76cf

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

turn on debug logging

fbe9f86

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

rissson mentioned this pull request Jul 9, 2026

2026.2.x Elevated Postgres Connections (1.5-3x from 2025.12.4) #20714

Open

rissson added 5 commits July 9, 2026 17:53

Merge branch 'main' into rust-proxy

97bb97f

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

don't build web, we don't need it here

f7ab236

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

Merge branch 'main' into rust-proxy

1aa98c8

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

fixup

a8b3957

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

REVERT ME: don't wait for tests to push

be435db

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

rissson added 4 commits July 20, 2026 14:28

Merge branch 'main' into rust-proxy

46b338a

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

spell

21657a0

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

packages/ak-common/tracing: make console subscriber optional

ef16a90

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

wip

44dbc1a

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

rissson commented Jul 20, 2026

View reviewed changes

spell

f64a736

Signed-off-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

outpost/proxy: rewrite in rust#21856

outpost/proxy: rewrite in rust#21856
rissson wants to merge 100 commits into
mainfrom
rust-proxy

rissson commented Apr 27, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Apr 27, 2026

Uh oh!

codecov Bot commented Apr 27, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Apr 27, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Apr 30, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Apr 30, 2026 •

edited

Loading

Uh oh!

github-actions Bot commented Jul 17, 2026 •

edited

Loading

Uh oh!

rissson Jul 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Uh oh!

Conversation

rissson commented Apr 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

netlify Bot commented Apr 27, 2026

✅ Deploy Preview for authentik-docs ready!

Uh oh!

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

Uh oh!

❌ 6 Tests Failed:

Uh oh!

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

Uh oh!

✅ Deploy Preview for authentik-docs ready!

Uh oh!

netlify Bot commented Apr 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-integrations ready!

Uh oh!

netlify Bot commented Apr 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for authentik-storybook ready!

Uh oh!

github-actions Bot commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rissson Jul 20, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

rissson commented Apr 27, 2026 •

edited

Loading

codecov Bot commented Apr 27, 2026 •

edited

Loading

netlify Bot commented Apr 27, 2026 •

edited

Loading

netlify Bot commented Apr 30, 2026 •

edited

Loading

netlify Bot commented Apr 30, 2026 •

edited

Loading

github-actions Bot commented Jul 17, 2026 •

edited

Loading