tests.e2e.test_flows_login.TestFlowsLogin::test_login_compatibility_mode

Stack Traces | 9.63s run time

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:334: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>,)
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_login.TestFlowsLogin testMethod=test_login_compatibility_mode>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_login_compatibility_mode(self):
        """test default login flow with compatibility mode enabled"""
        Flow.objects.filter(slug="default-authentication-flow").update(compatibility_mode=True)
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
>       self.login(shadow_dom=False)

tests/e2e/test_flows_login.py:48: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
shadow_dom = False

    def login(self, shadow_dom=True):
        """Do entire login flow"""
    
        if shadow_dom:
            flow_executor = self.get_shadow_root("ak-flow-executor")
            identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        else:
            flow_executor = self.shady_dom()
            identification_stage = self.shady_dom()
    
        wait = WebDriverWait(identification_stage, self.wait_timeout)
>       wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=uidField]")))

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

self = <[AttributeError("'wrapper' object has no attribute 'session_id'") raised in repr()] WebDriverWait object at 0x7f99a47c2c60>
method = <function presence_of_element_located.<locals>._predicate at 0x7f99a37b0ea0>
message = ''

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
>               value = method(self._driver)

.venv/lib/python3.13.../webdriver/support/wait.py:137: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

driver = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c702f0>

    def _predicate(driver: WebDriverOrWebElement):
>       return driver.find_element(*locator)

.venv/lib/python3.13.../webdriver/support/expected_conditions.py:110: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c702f0>
by = 'css selector', selector = 'input[name=uidField]'

    def find_element(self, by: str, selector: str) -> WebElement:
>       return self.container.execute_script(
            "return document.__shady_native_querySelector(arguments[0])", selector
        )

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

self = <selenium.webdriver.remote.webdriver.WebDriver (session="c328ceddb9902e5f6aaaa1a9ab9df2f7")>
script = 'return document.__shady_native_querySelector(arguments[0])'
args = ('input[name=uidField]',), converted_args = ['input[name=uidField]']
command = 'w3cExecuteScript'

    def execute_script(self, script, *args):
        """Synchronously Executes JavaScript in the current window/frame.
    
        Parameters:
        -----------
        script : str
            - The javascript to execute.
    
        *args : tuple
            - Any applicable arguments for your JavaScript.
    
        Example:
        --------
        >>> input_id = "username"
        >>> input_value = "test_user"
        >>> driver.execute_script(
        ...     "document.getElementById(arguments[0]).value = arguments[1];", input_id, input_value
        ... )
        """
        if isinstance(script, ScriptKey):
            try:
                script = self.pinned_scripts[script.id]
            except KeyError:
                raise JavascriptException("Pinned script could not be found")
    
        converted_args = list(args)
        command = Command.W3C_EXECUTE_SCRIPT
    
>       return self.execute(command, {"script": script, "args": converted_args})["value"]

.venv/lib/python3.13.../webdriver/remote/webdriver.py:547: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.webdriver.WebDriver (session="c328ceddb9902e5f6aaaa1a9ab9df2f7")>
driver_command = 'w3cExecuteScript'
params = {'args': ['input[name=uidField]'], 'script': 'return document.__shady_native_querySelector(arguments[0])'}

    def execute(self, driver_command: str, params: dict = None) -> dict:
        """Sends a command to be executed by a command.CommandExecutor.
    
        Parameters:
        -----------
        driver_command : str
            - The name of the command to execute as a string.
    
        params : dict
            - A dictionary of named Parameters to send with the command.
    
        Returns:
        --------
          dict - The command's JSON response loaded into a dictionary object.
        """
        params = self._wrap_value(params)
    
        if self.session_id:
            if not params:
                params = {"sessionId": self.session_id}
            elif "sessionId" not in params:
                params["sessionId"] = self.session_id
    
        response = self.command_executor.execute(driver_command, params)
        if response:
>           self.error_handler.check_response(response)

.venv/lib/python3.13.../webdriver/remote/webdriver.py:448: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.errorhandler.ErrorHandler object at 0x7f99a4390850>
response = {'status': 500, 'value': '{"value":{"error":"javascript error","message":"javascript error: document.__shady_native_qu...unknown>\\n#17 0x55699d141586 \\u003Cunknown>\\n#18 0x7f305c3d1aa4 \\u003Cunknown>\\n#19 0x7f305c45ea34 __clone\\n"}}'}

    def check_response(self, response: Dict[str, Any]) -> None:
        """Checks that a JSON response from the WebDriver does not have an
        error.
    
        :Args:
         - response - The JSON response from the WebDriver server as a dictionary
           object.
    
        :Raises: If the response contains an error message.
        """
        status = response.get("status", None)
        if not status or status == ErrorCode.SUCCESS:
            return
        value = None
        message = response.get("message", "")
        screen: str = response.get("screen", "")
        stacktrace = None
        if isinstance(status, int):
            value_json = response.get("value", None)
            if value_json and isinstance(value_json, str):
                import json
    
                try:
                    value = json.loads(value_json)
                    if len(value) == 1:
                        value = value["value"]
                    status = value.get("error", None)
                    if not status:
                        status = value.get("status", ErrorCode.UNKNOWN_ERROR)
                        message = value.get("value") or value.get("message")
                        if not isinstance(message, str):
                            value = message
                            message = message.get("message")
                    else:
                        message = value.get("message", None)
                except ValueError:
                    pass
    
        exception_class: Type[WebDriverException]
        e = ErrorCode()
        error_codes = [item for item in dir(e) if not item.startswith("__")]
        for error_code in error_codes:
            error_info = getattr(ErrorCode, error_code)
            if isinstance(error_info, list) and status in error_info:
                exception_class = getattr(ExceptionMapping, error_code, WebDriverException)
                break
        else:
            exception_class = WebDriverException
    
        if not value:
            value = response["value"]
        if isinstance(value, str):
            raise exception_class(value)
        if message == "" and "message" in value:
            message = value["message"]
    
        screen = None  # type: ignore[assignment]
        if "screen" in value:
            screen = value["screen"]
    
        stacktrace = None
        st_value = value.get("stackTrace") or value.get("stacktrace")
        if st_value:
            if isinstance(st_value, str):
                stacktrace = st_value.split("\n")
            else:
                stacktrace = []
                try:
                    for frame in st_value:
                        line = frame.get("lineNumber", "")
                        file = frame.get("fileName", "<anonymous>")
                        if line:
                            file = f"{file}:{line}"
                        meth = frame.get("methodName", "<anonymous>")
                        if "className" in frame:
                            meth = f"{frame['className']}.{meth}"
                        msg = "    at %s (%s)"
                        msg = msg % (meth, file)
                        stacktrace.append(msg)
                except TypeError:
                    pass
        if exception_class == UnexpectedAlertPresentException:
            alert_text = None
            if "data" in value:
                alert_text = value["data"].get("text")
            elif "alert" in value:
                alert_text = value["alert"].get("text")
            raise exception_class(message, screen, stacktrace, alert_text)  # type: ignore[call-arg]  # mypy is not smart enough here
>       raise exception_class(message, screen, stacktrace)
E       selenium.common.exceptions.JavascriptException: Message: javascript error: document.__shady_native_querySelector is not a function
E         (Session info: chrome=136.0.7103.113)
E       Stacktrace:
E       #0 0x55699d14271a <unknown>
E       #1 0x55699cbe50a0 <unknown>
E       #2 0x55699cbebcff <unknown>
E       #3 0x55699cbee857 <unknown>
E       #4 0x55699cc832de <unknown>
E       #5 0x55699cc5c3a2 <unknown>
E       #6 0x55699cc822a0 <unknown>
E       #7 0x55699cc5c173 <unknown>
E       #8 0x55699cc28d4b <unknown>
E       #9 0x55699cc299b1 <unknown>
E       #10 0x55699d1078cb <unknown>
E       #11 0x55699d10b7ca <unknown>
E       #12 0x55699d0ef622 <unknown>
E       #13 0x55699d10c354 <unknown>
E       #14 0x55699d0d445f <unknown>
E       #15 0x55699d1304f8 <unknown>
E       #16 0x55699d1306d6 <unknown>
E       #17 0x55699d141586 <unknown>
E       #18 0x7f305c3d1aa4 <unknown>
E       #19 0x7f305c45ea34 __clone

.venv/lib/python3.13.../webdriver/remote/errorhandler.py:232: JavascriptException

During handling of the above exception, another exception occurred:

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:334: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>,)
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_login.TestFlowsLogin testMethod=test_login_compatibility_mode>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_login_compatibility_mode(self):
        """test default login flow with compatibility mode enabled"""
        Flow.objects.filter(slug="default-authentication-flow").update(compatibility_mode=True)
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
>       self.login(shadow_dom=False)

tests/e2e/test_flows_login.py:48: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
shadow_dom = False

    def login(self, shadow_dom=True):
        """Do entire login flow"""
    
        if shadow_dom:
            flow_executor = self.get_shadow_root("ak-flow-executor")
            identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        else:
            flow_executor = self.shady_dom()
            identification_stage = self.shady_dom()
    
        wait = WebDriverWait(identification_stage, self.wait_timeout)
>       wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=uidField]")))

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

self = <[AttributeError("'wrapper' object has no attribute 'session_id'") raised in repr()] WebDriverWait object at 0x7f99a47c50f0>
method = <function presence_of_element_located.<locals>._predicate at 0x7f99a37b1d00>
message = ''

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
>               value = method(self._driver)

.venv/lib/python3.13.../webdriver/support/wait.py:137: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

driver = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c70590>

    def _predicate(driver: WebDriverOrWebElement):
>       return driver.find_element(*locator)

.venv/lib/python3.13.../webdriver/support/expected_conditions.py:110: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c70590>
by = 'css selector', selector = 'input[name=uidField]'

    def find_element(self, by: str, selector: str) -> WebElement:
>       return self.container.execute_script(
            "return document.__shady_native_querySelector(arguments[0])", selector
        )

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

self = <selenium.webdriver.remote.webdriver.WebDriver (session="ee01c9573116f56b8977f35ea5f38002")>
script = 'return document.__shady_native_querySelector(arguments[0])'
args = ('input[name=uidField]',), converted_args = ['input[name=uidField]']
command = 'w3cExecuteScript'

    def execute_script(self, script, *args):
        """Synchronously Executes JavaScript in the current window/frame.
    
        Parameters:
        -----------
        script : str
            - The javascript to execute.
    
        *args : tuple
            - Any applicable arguments for your JavaScript.
    
        Example:
        --------
        >>> input_id = "username"
        >>> input_value = "test_user"
        >>> driver.execute_script(
        ...     "document.getElementById(arguments[0]).value = arguments[1];", input_id, input_value
        ... )
        """
        if isinstance(script, ScriptKey):
            try:
                script = self.pinned_scripts[script.id]
            except KeyError:
                raise JavascriptException("Pinned script could not be found")
    
        converted_args = list(args)
        command = Command.W3C_EXECUTE_SCRIPT
    
>       return self.execute(command, {"script": script, "args": converted_args})["value"]

.venv/lib/python3.13.../webdriver/remote/webdriver.py:547: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.webdriver.WebDriver (session="ee01c9573116f56b8977f35ea5f38002")>
driver_command = 'w3cExecuteScript'
params = {'args': ['input[name=uidField]'], 'script': 'return document.__shady_native_querySelector(arguments[0])'}

    def execute(self, driver_command: str, params: dict = None) -> dict:
        """Sends a command to be executed by a command.CommandExecutor.
    
        Parameters:
        -----------
        driver_command : str
            - The name of the command to execute as a string.
    
        params : dict
            - A dictionary of named Parameters to send with the command.
    
        Returns:
        --------
          dict - The command's JSON response loaded into a dictionary object.
        """
        params = self._wrap_value(params)
    
        if self.session_id:
            if not params:
                params = {"sessionId": self.session_id}
            elif "sessionId" not in params:
                params["sessionId"] = self.session_id
    
        response = self.command_executor.execute(driver_command, params)
        if response:
>           self.error_handler.check_response(response)

.venv/lib/python3.13.../webdriver/remote/webdriver.py:448: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.errorhandler.ErrorHandler object at 0x7f99a47c37a0>
response = {'status': 500, 'value': '{"value":{"error":"javascript error","message":"javascript error: document.__shady_native_qu...unknown>\\n#17 0x5556d2ea8586 \\u003Cunknown>\\n#18 0x7f4b0c684aa4 \\u003Cunknown>\\n#19 0x7f4b0c711a34 __clone\\n"}}'}

    def check_response(self, response: Dict[str, Any]) -> None:
        """Checks that a JSON response from the WebDriver does not have an
        error.
    
        :Args:
         - response - The JSON response from the WebDriver server as a dictionary
           object.
    
        :Raises: If the response contains an error message.
        """
        status = response.get("status", None)
        if not status or status == ErrorCode.SUCCESS:
            return
        value = None
        message = response.get("message", "")
        screen: str = response.get("screen", "")
        stacktrace = None
        if isinstance(status, int):
            value_json = response.get("value", None)
            if value_json and isinstance(value_json, str):
                import json
    
                try:
                    value = json.loads(value_json)
                    if len(value) == 1:
                        value = value["value"]
                    status = value.get("error", None)
                    if not status:
                        status = value.get("status", ErrorCode.UNKNOWN_ERROR)
                        message = value.get("value") or value.get("message")
                        if not isinstance(message, str):
                            value = message
                            message = message.get("message")
                    else:
                        message = value.get("message", None)
                except ValueError:
                    pass
    
        exception_class: Type[WebDriverException]
        e = ErrorCode()
        error_codes = [item for item in dir(e) if not item.startswith("__")]
        for error_code in error_codes:
            error_info = getattr(ErrorCode, error_code)
            if isinstance(error_info, list) and status in error_info:
                exception_class = getattr(ExceptionMapping, error_code, WebDriverException)
                break
        else:
            exception_class = WebDriverException
    
        if not value:
            value = response["value"]
        if isinstance(value, str):
            raise exception_class(value)
        if message == "" and "message" in value:
            message = value["message"]
    
        screen = None  # type: ignore[assignment]
        if "screen" in value:
            screen = value["screen"]
    
        stacktrace = None
        st_value = value.get("stackTrace") or value.get("stacktrace")
        if st_value:
            if isinstance(st_value, str):
                stacktrace = st_value.split("\n")
            else:
                stacktrace = []
                try:
                    for frame in st_value:
                        line = frame.get("lineNumber", "")
                        file = frame.get("fileName", "<anonymous>")
                        if line:
                            file = f"{file}:{line}"
                        meth = frame.get("methodName", "<anonymous>")
                        if "className" in frame:
                            meth = f"{frame['className']}.{meth}"
                        msg = "    at %s (%s)"
                        msg = msg % (meth, file)
                        stacktrace.append(msg)
                except TypeError:
                    pass
        if exception_class == UnexpectedAlertPresentException:
            alert_text = None
            if "data" in value:
                alert_text = value["data"].get("text")
            elif "alert" in value:
                alert_text = value["alert"].get("text")
            raise exception_class(message, screen, stacktrace, alert_text)  # type: ignore[call-arg]  # mypy is not smart enough here
>       raise exception_class(message, screen, stacktrace)
E       selenium.common.exceptions.JavascriptException: Message: javascript error: document.__shady_native_querySelector is not a function
E         (Session info: chrome=136.0.7103.113)
E       Stacktrace:
E       #0 0x5556d2ea971a <unknown>
E       #1 0x5556d294c0a0 <unknown>
E       #2 0x5556d2952cff <unknown>
E       #3 0x5556d2955857 <unknown>
E       #4 0x5556d29ea2de <unknown>
E       #5 0x5556d29c33a2 <unknown>
E       #6 0x5556d29e92a0 <unknown>
E       #7 0x5556d29c3173 <unknown>
E       #8 0x5556d298fd4b <unknown>
E       #9 0x5556d29909b1 <unknown>
E       #10 0x5556d2e6e8cb <unknown>
E       #11 0x5556d2e727ca <unknown>
E       #12 0x5556d2e56622 <unknown>
E       #13 0x5556d2e73354 <unknown>
E       #14 0x5556d2e3b45f <unknown>
E       #15 0x5556d2e974f8 <unknown>
E       #16 0x5556d2e976d6 <unknown>
E       #17 0x5556d2ea8586 <unknown>
E       #18 0x7f4b0c684aa4 <unknown>
E       #19 0x7f4b0c711a34 __clone

.venv/lib/python3.13.../webdriver/remote/errorhandler.py:232: JavascriptException

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7f99a3f21220>
test_case = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
subTest = False

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

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

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
result = <TestCaseFunction test_login_compatibility_mode>

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

.../hostedtoolcache/Python/3.13.3........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
method = <bound method TestFlowsLogin.test_login_compatibility_mode of <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>>

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

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

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:347: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:347: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:341: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
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:334: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>,)
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_login.TestFlowsLogin testMethod=test_login_compatibility_mode>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_login_compatibility_mode(self):
        """test default login flow with compatibility mode enabled"""
        Flow.objects.filter(slug="default-authentication-flow").update(compatibility_mode=True)
        self.driver.get(
            self.url(
                "authentik_core:if-flow",
                flow_slug="default-authentication-flow",
            )
        )
>       self.login(shadow_dom=False)

tests/e2e/test_flows_login.py:48: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_flows_login.TestFlowsLogin testMethod=test_login_compatibility_mode>
shadow_dom = False

    def login(self, shadow_dom=True):
        """Do entire login flow"""
    
        if shadow_dom:
            flow_executor = self.get_shadow_root("ak-flow-executor")
            identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        else:
            flow_executor = self.shady_dom()
            identification_stage = self.shady_dom()
    
        wait = WebDriverWait(identification_stage, self.wait_timeout)
>       wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=uidField]")))

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

self = <[AttributeError("'wrapper' object has no attribute 'session_id'") raised in repr()] WebDriverWait object at 0x7f99a3f17ad0>
method = <function presence_of_element_located.<locals>._predicate at 0x7f99a37fc540>
message = ''

    def until(self, method: Callable[[D], Union[Literal[False], T]], message: str = "") -> T:
        """Wait until the method returns a value that is not False.
    
        Calls the method provided with the driver as an argument until the
        return value does not evaluate to ``False``.
    
        Parameters:
        -----------
        method: callable(WebDriver)
            - A callable object that takes a WebDriver instance as an argument.
    
        message: str
            - Optional message for :exc:`TimeoutException`
    
        Return:
        -------
        object: T
            - The result of the last call to `method`
    
        Raises:
        -------
        TimeoutException
            - If 'method' does not return a truthy value within the WebDriverWait
            object's timeout
    
        Example:
        --------
        >>> from selenium.webdriver.common.by import By
        >>> from selenium.webdriver.support.ui import WebDriverWait
        >>> from selenium.webdriver.support import expected_conditions as EC
    
        # Wait until an element is visible on the page
        >>> wait = WebDriverWait(driver, 10)
        >>> element = wait.until(EC.visibility_of_element_located((By.ID, "exampleId")))
        >>> print(element.text)
        """
        screen = None
        stacktrace = None
    
        end_time = time.monotonic() + self._timeout
        while True:
            try:
>               value = method(self._driver)

.venv/lib/python3.13.../webdriver/support/wait.py:137: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

driver = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c70980>

    def _predicate(driver: WebDriverOrWebElement):
>       return driver.find_element(*locator)

.venv/lib/python3.13.../webdriver/support/expected_conditions.py:110: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.utils.SeleniumTestCase.shady_dom.<locals>.wrapper object at 0x7f99a3c70980>
by = 'css selector', selector = 'input[name=uidField]'

    def find_element(self, by: str, selector: str) -> WebElement:
>       return self.container.execute_script(
            "return document.__shady_native_querySelector(arguments[0])", selector
        )

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

self = <selenium.webdriver.remote.webdriver.WebDriver (session="ee0bf7ea5958624a07cb6661653a692c")>
script = 'return document.__shady_native_querySelector(arguments[0])'
args = ('input[name=uidField]',), converted_args = ['input[name=uidField]']
command = 'w3cExecuteScript'

    def execute_script(self, script, *args):
        """Synchronously Executes JavaScript in the current window/frame.
    
        Parameters:
        -----------
        script : str
            - The javascript to execute.
    
        *args : tuple
            - Any applicable arguments for your JavaScript.
    
        Example:
        --------
        >>> input_id = "username"
        >>> input_value = "test_user"
        >>> driver.execute_script(
        ...     "document.getElementById(arguments[0]).value = arguments[1];", input_id, input_value
        ... )
        """
        if isinstance(script, ScriptKey):
            try:
                script = self.pinned_scripts[script.id]
            except KeyError:
                raise JavascriptException("Pinned script could not be found")
    
        converted_args = list(args)
        command = Command.W3C_EXECUTE_SCRIPT
    
>       return self.execute(command, {"script": script, "args": converted_args})["value"]

.venv/lib/python3.13.../webdriver/remote/webdriver.py:547: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.webdriver.WebDriver (session="ee0bf7ea5958624a07cb6661653a692c")>
driver_command = 'w3cExecuteScript'
params = {'args': ['input[name=uidField]'], 'script': 'return document.__shady_native_querySelector(arguments[0])'}

    def execute(self, driver_command: str, params: dict = None) -> dict:
        """Sends a command to be executed by a command.CommandExecutor.
    
        Parameters:
        -----------
        driver_command : str
            - The name of the command to execute as a string.
    
        params : dict
            - A dictionary of named Parameters to send with the command.
    
        Returns:
        --------
          dict - The command's JSON response loaded into a dictionary object.
        """
        params = self._wrap_value(params)
    
        if self.session_id:
            if not params:
                params = {"sessionId": self.session_id}
            elif "sessionId" not in params:
                params["sessionId"] = self.session_id
    
        response = self.command_executor.execute(driver_command, params)
        if response:
>           self.error_handler.check_response(response)

.venv/lib/python3.13.../webdriver/remote/webdriver.py:448: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <selenium.webdriver.remote.errorhandler.ErrorHandler object at 0x7f99a47c3110>
response = {'status': 500, 'value': '{"value":{"error":"javascript error","message":"javascript error: document.__shady_native_qu...unknown>\\n#17 0x556f65744586 \\u003Cunknown>\\n#18 0x7ff677ddaaa4 \\u003Cunknown>\\n#19 0x7ff677e67a34 __clone\\n"}}'}

    def check_response(self, response: Dict[str, Any]) -> None:
        """Checks that a JSON response from the WebDriver does not have an
        error.
    
        :Args:
         - response - The JSON response from the WebDriver server as a dictionary
           object.
    
        :Raises: If the response contains an error message.
        """
        status = response.get("status", None)
        if not status or status == ErrorCode.SUCCESS:
            return
        value = None
        message = response.get("message", "")
        screen: str = response.get("screen", "")
        stacktrace = None
        if isinstance(status, int):
            value_json = response.get("value", None)
            if value_json and isinstance(value_json, str):
                import json
    
                try:
                    value = json.loads(value_json)
                    if len(value) == 1:
                        value = value["value"]
                    status = value.get("error", None)
                    if not status:
                        status = value.get("status", ErrorCode.UNKNOWN_ERROR)
                        message = value.get("value") or value.get("message")
                        if not isinstance(message, str):
                            value = message
                            message = message.get("message")
                    else:
                        message = value.get("message", None)
                except ValueError:
                    pass
    
        exception_class: Type[WebDriverException]
        e = ErrorCode()
        error_codes = [item for item in dir(e) if not item.startswith("__")]
        for error_code in error_codes:
            error_info = getattr(ErrorCode, error_code)
            if isinstance(error_info, list) and status in error_info:
                exception_class = getattr(ExceptionMapping, error_code, WebDriverException)
                break
        else:
            exception_class = WebDriverException
    
        if not value:
            value = response["value"]
        if isinstance(value, str):
            raise exception_class(value)
        if message == "" and "message" in value:
            message = value["message"]
    
        screen = None  # type: ignore[assignment]
        if "screen" in value:
            screen = value["screen"]
    
        stacktrace = None
        st_value = value.get("stackTrace") or value.get("stacktrace")
        if st_value:
            if isinstance(st_value, str):
                stacktrace = st_value.split("\n")
            else:
                stacktrace = []
                try:
                    for frame in st_value:
                        line = frame.get("lineNumber", "")
                        file = frame.get("fileName", "<anonymous>")
                        if line:
                            file = f"{file}:{line}"
                        meth = frame.get("methodName", "<anonymous>")
                        if "className" in frame:
                            meth = f"{frame['className']}.{meth}"
                        msg = "    at %s (%s)"
                        msg = msg % (meth, file)
                        stacktrace.append(msg)
                except TypeError:
                    pass
        if exception_class == UnexpectedAlertPresentException:
            alert_text = None
            if "data" in value:
                alert_text = value["data"].get("text")
            elif "alert" in value:
                alert_text = value["alert"].get("text")
            raise exception_class(message, screen, stacktrace, alert_text)  # type: ignore[call-arg]  # mypy is not smart enough here
>       raise exception_class(message, screen, stacktrace)
E       selenium.common.exceptions.JavascriptException: Message: javascript error: document.__shady_native_querySelector is not a function
E         (Session info: chrome=136.0.7103.113)
E       Stacktrace:
E       #0 0x556f6574571a <unknown>
E       #1 0x556f651e80a0 <unknown>
E       #2 0x556f651eecff <unknown>
E       #3 0x556f651f1857 <unknown>
E       #4 0x556f652862de <unknown>
E       #5 0x556f6525f3a2 <unknown>
E       #6 0x556f652852a0 <unknown>
E       #7 0x556f6525f173 <unknown>
E       #8 0x556f6522bd4b <unknown>
E       #9 0x556f6522c9b1 <unknown>
E       #10 0x556f6570a8cb <unknown>
E       #11 0x556f6570e7ca <unknown>
E       #12 0x556f656f2622 <unknown>
E       #13 0x556f6570f354 <unknown>
E       #14 0x556f656d745f <unknown>
E       #15 0x556f657334f8 <unknown>
E       #16 0x556f657336d6 <unknown>
E       #17 0x556f65744586 <unknown>
E       #18 0x7ff677ddaaa4 <unknown>
E       #19 0x7ff677e67a34 __clone

.venv/lib/python3.13.../webdriver/remote/errorhandler.py:232: JavascriptException

tests.integration.test_outpost_docker.OutpostDockerTests::test_docker_controller

Stack Traces | 14.1s run time

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50........./containers/ak-outpost-test/json

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>

    def _get_container(self) -> tuple[Container, bool]:
        try:
>           return self.client.containers.get(self.name), False

.../outposts/controllers/docker.py:194: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.containers.ContainerCollection object at 0x7fddeb24db80>
container_id = 'ak-outpost-test'

    def get(self, container_id):
        """
        Get a container by name or ID.
    
        Args:
            container_id (str): Container name or ID.
    
        Returns:
            A :py:class:`Container` object.
    
        Raises:
            :py:class:`docker.errors.NotFound`
                If the container does not exist.
            :py:class:`docker.errors.APIError`
                If the server returns an error.
        """
>       resp = self.client.api.inspect_container(container_id)

.venv/lib/python3.13.../docker/models/containers.py:954: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
resource_id = 'ak-outpost-test', args = (), kwargs = {}

    @functools.wraps(f)
    def wrapped(self, resource_id=None, *args, **kwargs):
        if resource_id is None and kwargs.get(resource_name):
            resource_id = kwargs.pop(resource_name)
        if isinstance(resource_id, dict):
            resource_id = resource_id.get('Id', resource_id.get('ID'))
        if not resource_id:
            raise errors.NullResource(
                'Resource ID was not provided'
            )
>       return f(self, resource_id, *args, **kwargs)

.venv/lib/python3.13.../docker/utils/decorators.py:19: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
container = 'ak-outpost-test'

    @utils.check_resource('container')
    def inspect_container(self, container):
        """
        Identical to the `docker inspect` command, but only for containers.
    
        Args:
            container (str): The container to inspect
    
        Returns:
            (dict): Similar to the output of `docker inspect`, but as a
            single dict
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
        """
>       return self._result(
            self._get(self._url("/containers/{0}/json", container)), True
        )

.venv/lib/python3.13.../docker/api/container.py:793: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>, json = True, binary = False

    def _result(self, response, json=False, binary=False):
        assert not (json and binary)
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/client.py:281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50........./containers/ak-outpost-test/json')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50........./containers/ak-outpost-test/json: Not Found ("No such container: ak-outpost-test")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

During handling of the above exception, another exception occurred:

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>

    def try_pull_image(self):
        """Try to pull the image needed for this outpost based on the CONFIG
        `outposts.container_image_base`, but fall back to known-good images"""
        image = self.get_container_image()
        try:
>           self.client.images.pull(image)

.../outposts/controllers/docker.py:186: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.images.ImageCollection object at 0x7fddec4ab6d0>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-cherry-pick-390f65-version-2025.6', all_tags = False, kwargs = {}
image_tag = 'gh-cherry-pick-390f65-version-2025.6'

    def pull(self, repository, tag=None, all_tags=False, **kwargs):
        """
        Pull an image of the given name and return it. Similar to the
        ``docker pull`` command.
        If ``tag`` is ``None`` or empty, it is set to ``latest``.
        If ``all_tags`` is set, the ``tag`` parameter is ignored and all image
        tags will be pulled.
    
        If you want to get the raw pull output, use the
        :py:meth:`~docker.api.image.ImageApiMixin.pull` method in the
        low-level API.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags
    
        Returns:
            (:py:class:`Image` or list): The image that has been pulled.
                If ``all_tags`` is True, the method will return a list
                of :py:class:`Image` objects belonging to this repository.
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> # Pull the image tagged `latest` in the busybox repo
            >>> image = client.images.pull('busybox')
    
            >>> # Pull all tags in the busybox repo
            >>> images = client.images.pull('busybox', all_tags=True)
        """
        repository, image_tag = parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if 'stream' in kwargs:
            warnings.warn(
                '`stream` is not a valid parameter for this method'
                ' and will be overridden',
                stacklevel=1,
            )
            del kwargs['stream']
    
>       pull_log = self.client.api.pull(
            repository, tag=tag, stream=True, all_tags=all_tags, **kwargs
        )

.venv/lib/python3.13.../docker/models/images.py:464: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-cherry-pick-390f65-version-2025.6', stream = True, auth_config = None
decode = False, platform = None, all_tags = False

    def pull(self, repository, tag=None, stream=False, auth_config=None,
             decode=False, platform=None, all_tags=False):
        """
        Pulls an image. Similar to the ``docker pull`` command.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull. If ``tag`` is ``None`` or empty, it
                is set to ``latest``.
            stream (bool): Stream the output as a generator. Make sure to
                consume the generator, otherwise pull might get cancelled.
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            decode (bool): Decode the JSON data from the server into dicts.
                Only applies with ``stream=True``
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags, the ``tag`` parameter is
                ignored.
    
        Returns:
            (generator or str): The output
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> resp = client.api.pull('busybox', stream=True, decode=True)
            ... for line in resp:
            ...     print(json.dumps(line, indent=4))
            {
                "status": "Pulling image (latest) from busybox",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
            {
                "status": "Pulling image (latest) from busybox, endpoint: ...",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
    
        """
        repository, image_tag = utils.parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if all_tags:
            tag = None
    
        registry, repo_name = auth.resolve_repository_name(repository)
    
        params = {
            'tag': tag,
            'fromImage': repository
        }
        headers = {}
    
        if auth_config is None:
            header = auth.get_config_header(self, registry)
            if header:
                headers['X-Registry-Auth'] = header
        else:
            log.debug('Sending supplied auth config')
            headers['X-Registry-Auth'] = auth.encode_header(auth_config)
    
        if platform is not None:
            if utils.version_lt(self._version, '1.32'):
                raise errors.InvalidVersion(
                    'platform was only introduced in API version 1.32'
                )
            params['platform'] = platform
    
        response = self._post(
            self._url('/images/create'), params=params, headers=headers,
            stream=stream, timeout=None
        )
    
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/image.py:429: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy: Not Found ("manifest unknown")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

During handling of the above exception, another exception occurred:

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>
depth = 1

    def up(self, depth=1):
        if self.outpost.managed == MANAGED_OUTPOST:
            return None
        if depth >= DOCKER_MAX_ATTEMPTS:
            raise ControllerException("Giving up since we exceeded recursion limit.")
        self._migrate_container_name()
        try:
>           container, has_been_created = self._get_container()

.../outposts/controllers/docker.py:238: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>

    def _get_container(self) -> tuple[Container, bool]:
        try:
            return self.client.containers.get(self.name), False
        except NotFound:
            self.logger.info("(Re-)creating container...")
>           image_name = self.try_pull_image()

.../outposts/controllers/docker.py:197: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>

    def try_pull_image(self):
        """Try to pull the image needed for this outpost based on the CONFIG
        `outposts.container_image_base`, but fall back to known-good images"""
        image = self.get_container_image()
        try:
            self.client.images.pull(image)
        except DockerException:  # pragma: no cover
            image = f"ghcr.io/goauthentik/{self.outpost.type}:{__version__}"
>           self.client.images.pull(image)

.../outposts/controllers/docker.py:189: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.images.ImageCollection object at 0x7fddec2c69e0>
repository = 'ghcr.io/goauthentik/proxy', tag = '2025.6.0', all_tags = False
kwargs = {}, image_tag = '2025.6.0'

    def pull(self, repository, tag=None, all_tags=False, **kwargs):
        """
        Pull an image of the given name and return it. Similar to the
        ``docker pull`` command.
        If ``tag`` is ``None`` or empty, it is set to ``latest``.
        If ``all_tags`` is set, the ``tag`` parameter is ignored and all image
        tags will be pulled.
    
        If you want to get the raw pull output, use the
        :py:meth:`~docker.api.image.ImageApiMixin.pull` method in the
        low-level API.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags
    
        Returns:
            (:py:class:`Image` or list): The image that has been pulled.
                If ``all_tags`` is True, the method will return a list
                of :py:class:`Image` objects belonging to this repository.
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> # Pull the image tagged `latest` in the busybox repo
            >>> image = client.images.pull('busybox')
    
            >>> # Pull all tags in the busybox repo
            >>> images = client.images.pull('busybox', all_tags=True)
        """
        repository, image_tag = parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if 'stream' in kwargs:
            warnings.warn(
                '`stream` is not a valid parameter for this method'
                ' and will be overridden',
                stacklevel=1,
            )
            del kwargs['stream']
    
>       pull_log = self.client.api.pull(
            repository, tag=tag, stream=True, all_tags=all_tags, **kwargs
        )

.venv/lib/python3.13.../docker/models/images.py:464: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
repository = 'ghcr.io/goauthentik/proxy', tag = '2025.6.0', stream = True
auth_config = None, decode = False, platform = None, all_tags = False

    def pull(self, repository, tag=None, stream=False, auth_config=None,
             decode=False, platform=None, all_tags=False):
        """
        Pulls an image. Similar to the ``docker pull`` command.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull. If ``tag`` is ``None`` or empty, it
                is set to ``latest``.
            stream (bool): Stream the output as a generator. Make sure to
                consume the generator, otherwise pull might get cancelled.
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            decode (bool): Decode the JSON data from the server into dicts.
                Only applies with ``stream=True``
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags, the ``tag`` parameter is
                ignored.
    
        Returns:
            (generator or str): The output
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> resp = client.api.pull('busybox', stream=True, decode=True)
            ... for line in resp:
            ...     print(json.dumps(line, indent=4))
            {
                "status": "Pulling image (latest) from busybox",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
            {
                "status": "Pulling image (latest) from busybox, endpoint: ...",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
    
        """
        repository, image_tag = utils.parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if all_tags:
            tag = None
    
        registry, repo_name = auth.resolve_repository_name(repository)
    
        params = {
            'tag': tag,
            'fromImage': repository
        }
        headers = {}
    
        if auth_config is None:
            header = auth.get_config_header(self, registry)
            if header:
                headers['X-Registry-Auth'] = header
        else:
            log.debug('Sending supplied auth config')
            headers['X-Registry-Auth'] = auth.encode_header(auth_config)
    
        if platform is not None:
            if utils.version_lt(self._version, '1.32'):
                raise errors.InvalidVersion(
                    'platform was only introduced in API version 1.32'
                )
            params['platform'] = platform
    
        response = self._post(
            self._url('/images/create'), params=params, headers=headers,
            stream=stream, timeout=None
        )
    
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/image.py:429: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddec8ce550>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy: Not Found ("manifest unknown")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

The above exception was the direct cause of the following exception:

self = <unittest.case._Outcome object at 0x7fddec8ce050>
test_case = <tests.integration.test_outpost_docker.OutpostDockerTests testMethod=test_docker_controller>
subTest = False

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

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

self = <tests.integration.test_outpost_docker.OutpostDockerTests testMethod=test_docker_controller>
result = <TestCaseFunction test_docker_controller>

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

.../hostedtoolcache/Python/3.13.3........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_outpost_docker.OutpostDockerTests testMethod=test_docker_controller>
method = <bound method OutpostDockerTests.test_docker_controller of <tests.integration.test_outpost_docker.OutpostDockerTests testMethod=test_docker_controller>>

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

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

self = <tests.integration.test_outpost_docker.OutpostDockerTests testMethod=test_docker_controller>

    @pytest.mark.timeout(120)
    def test_docker_controller(self):
        """test that deployment requires update"""
        controller = DockerController(self.outpost, self.service_connection)
>       controller.up()

tests/integration/test_outpost_docker.py:94: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfce850>
depth = 1

    def up(self, depth=1):
        if self.outpost.managed == MANAGED_OUTPOST:
            return None
        if depth >= DOCKER_MAX_ATTEMPTS:
            raise ControllerException("Giving up since we exceeded recursion limit.")
        self._migrate_container_name()
        try:
            container, has_been_created = self._get_container()
            if has_been_created:
                container.start()
                return None
            # Check if the container is out of date, delete it and retry
            if len(container.image.tags) > 0:
                should_image = self.try_pull_image()
                if should_image not in container.image.tags:  # pragma: no cover
                    self.logger.info(
                        "Container has mismatched image, re-creating...",
                        has=container.image.tags,
                        should=should_image,
                    )
                    self.down()
                    return self.up(depth + 1)
            # Check container's ports
            if self._comp_ports(container):
                self.logger.info("Container has mis-matched ports, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container values match our values
            if self._comp_env(container):
                self.logger.info("Container has outdated config, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container values match our values
            if self._comp_labels(container):
                self.logger.info("Container has outdated labels, re-creating...")
                self.down()
                return self.up(depth + 1)
            if (
                container.attrs.get("HostConfig", {})
                .get("RestartPolicy", {})
                .get("Name", "")
                .lower()
                != "unless-stopped"
            ):
                self.logger.info("Container has mis-matched restart policy, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container is healthy
            if container.status == "running" and container.attrs.get("State", {}).get(
                "Health", {}
            ).get("Status", "") not in ["healthy", "starting"]:
                # At this point we know the config is correct, but the container isn't healthy,
                # so we just restart it with the same config
                if has_been_created:
                    # Since we've just created the container, give it some time to start.
                    # If its still not up by then, restart it
                    self.logger.info("Container is unhealthy and new, giving it time to boot.")
                    sleep(60)
                self.logger.info("Container is unhealthy, restarting...")
                container.restart()
                return None
            # Check that container is running
            if container.status != "running":
                self.logger.info("Container is not running, restarting...")
                container.start()
                return None
            self.logger.info("Container is running")
            return None
        except DockerException as exc:
>           raise ControllerException(str(exc)) from exc
E           authentik.outposts.controllers.base.ControllerException: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy: Not Found ("manifest unknown")

.../outposts/controllers/docker.py:300: ControllerException

tests.integration.test_proxy_docker.TestProxyDocker::test_docker_controller

Stack Traces | 15.1s run time

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50........./containers/ak-outpost-test/json

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>

    def _get_container(self) -> tuple[Container, bool]:
        try:
>           return self.client.containers.get(self.name), False

.../outposts/controllers/docker.py:194: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.containers.ContainerCollection object at 0x7fddeb8d92c0>
container_id = 'ak-outpost-test'

    def get(self, container_id):
        """
        Get a container by name or ID.
    
        Args:
            container_id (str): Container name or ID.
    
        Returns:
            A :py:class:`Container` object.
    
        Raises:
            :py:class:`docker.errors.NotFound`
                If the container does not exist.
            :py:class:`docker.errors.APIError`
                If the server returns an error.
        """
>       resp = self.client.api.inspect_container(container_id)

.venv/lib/python3.13.../docker/models/containers.py:954: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
resource_id = 'ak-outpost-test', args = (), kwargs = {}

    @functools.wraps(f)
    def wrapped(self, resource_id=None, *args, **kwargs):
        if resource_id is None and kwargs.get(resource_name):
            resource_id = kwargs.pop(resource_name)
        if isinstance(resource_id, dict):
            resource_id = resource_id.get('Id', resource_id.get('ID'))
        if not resource_id:
            raise errors.NullResource(
                'Resource ID was not provided'
            )
>       return f(self, resource_id, *args, **kwargs)

.venv/lib/python3.13.../docker/utils/decorators.py:19: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
container = 'ak-outpost-test'

    @utils.check_resource('container')
    def inspect_container(self, container):
        """
        Identical to the `docker inspect` command, but only for containers.
    
        Args:
            container (str): The container to inspect
    
        Returns:
            (dict): Similar to the output of `docker inspect`, but as a
            single dict
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
        """
>       return self._result(
            self._get(self._url("/containers/{0}/json", container)), True
        )

.venv/lib/python3.13.../docker/api/container.py:793: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>, json = True, binary = False

    def _result(self, response, json=False, binary=False):
        assert not (json and binary)
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/client.py:281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50........./containers/ak-outpost-test/json')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50........./containers/ak-outpost-test/json: Not Found ("No such container: ak-outpost-test")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

During handling of the above exception, another exception occurred:

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>

    def try_pull_image(self):
        """Try to pull the image needed for this outpost based on the CONFIG
        `outposts.container_image_base`, but fall back to known-good images"""
        image = self.get_container_image()
        try:
>           self.client.images.pull(image)

.../outposts/controllers/docker.py:186: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.images.ImageCollection object at 0x7fddeb5c57e0>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-cherry-pick-390f65-version-2025.6', all_tags = False, kwargs = {}
image_tag = 'gh-cherry-pick-390f65-version-2025.6'

    def pull(self, repository, tag=None, all_tags=False, **kwargs):
        """
        Pull an image of the given name and return it. Similar to the
        ``docker pull`` command.
        If ``tag`` is ``None`` or empty, it is set to ``latest``.
        If ``all_tags`` is set, the ``tag`` parameter is ignored and all image
        tags will be pulled.
    
        If you want to get the raw pull output, use the
        :py:meth:`~docker.api.image.ImageApiMixin.pull` method in the
        low-level API.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags
    
        Returns:
            (:py:class:`Image` or list): The image that has been pulled.
                If ``all_tags`` is True, the method will return a list
                of :py:class:`Image` objects belonging to this repository.
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> # Pull the image tagged `latest` in the busybox repo
            >>> image = client.images.pull('busybox')
    
            >>> # Pull all tags in the busybox repo
            >>> images = client.images.pull('busybox', all_tags=True)
        """
        repository, image_tag = parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if 'stream' in kwargs:
            warnings.warn(
                '`stream` is not a valid parameter for this method'
                ' and will be overridden',
                stacklevel=1,
            )
            del kwargs['stream']
    
>       pull_log = self.client.api.pull(
            repository, tag=tag, stream=True, all_tags=all_tags, **kwargs
        )

.venv/lib/python3.13.../docker/models/images.py:464: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-cherry-pick-390f65-version-2025.6', stream = True, auth_config = None
decode = False, platform = None, all_tags = False

    def pull(self, repository, tag=None, stream=False, auth_config=None,
             decode=False, platform=None, all_tags=False):
        """
        Pulls an image. Similar to the ``docker pull`` command.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull. If ``tag`` is ``None`` or empty, it
                is set to ``latest``.
            stream (bool): Stream the output as a generator. Make sure to
                consume the generator, otherwise pull might get cancelled.
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            decode (bool): Decode the JSON data from the server into dicts.
                Only applies with ``stream=True``
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags, the ``tag`` parameter is
                ignored.
    
        Returns:
            (generator or str): The output
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> resp = client.api.pull('busybox', stream=True, decode=True)
            ... for line in resp:
            ...     print(json.dumps(line, indent=4))
            {
                "status": "Pulling image (latest) from busybox",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
            {
                "status": "Pulling image (latest) from busybox, endpoint: ...",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
    
        """
        repository, image_tag = utils.parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if all_tags:
            tag = None
    
        registry, repo_name = auth.resolve_repository_name(repository)
    
        params = {
            'tag': tag,
            'fromImage': repository
        }
        headers = {}
    
        if auth_config is None:
            header = auth.get_config_header(self, registry)
            if header:
                headers['X-Registry-Auth'] = header
        else:
            log.debug('Sending supplied auth config')
            headers['X-Registry-Auth'] = auth.encode_header(auth_config)
    
        if platform is not None:
            if utils.version_lt(self._version, '1.32'):
                raise errors.InvalidVersion(
                    'platform was only introduced in API version 1.32'
                )
            params['platform'] = platform
    
        response = self._post(
            self._url('/images/create'), params=params, headers=headers,
            stream=stream, timeout=None
        )
    
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/image.py:429: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=gh-cherry-pick-390f65-version-2025.6&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy: Not Found ("manifest unknown")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

During handling of the above exception, another exception occurred:

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
>           response.raise_for_status()

.venv/lib/python3.13.../docker/api/client.py:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Response [404]>

    def raise_for_status(self):
        """Raises :class:`HTTPError`, if one occurred."""
    
        http_error_msg = ""
        if isinstance(self.reason, bytes):
            # We attempt to decode utf-8 first because some servers
            # choose to localize their reason strings. If the string
            # isn't utf-8, we fall back to iso-8859-1 for all other
            # encodings. (See PR #3538)
            try:
                reason = self.reason.decode("utf-8")
            except UnicodeDecodeError:
                reason = self.reason.decode("iso-8859-1")
        else:
            reason = self.reason
    
        if 400 <= self.status_code < 500:
            http_error_msg = (
                f"{self.status_code} Client Error: {reason} for url: {self.url}"
            )
    
        elif 500 <= self.status_code < 600:
            http_error_msg = (
                f"{self.status_code} Server Error: {reason} for url: {self.url}"
            )
    
        if http_error_msg:
>           raise HTTPError(http_error_msg, response=self)
E           requests.exceptions.HTTPError: 404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy

.venv/lib/python3.13........./site-packages/requests/models.py:1024: HTTPError

The above exception was the direct cause of the following exception:

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>
depth = 1

    def up(self, depth=1):
        if self.outpost.managed == MANAGED_OUTPOST:
            return None
        if depth >= DOCKER_MAX_ATTEMPTS:
            raise ControllerException("Giving up since we exceeded recursion limit.")
        self._migrate_container_name()
        try:
>           container, has_been_created = self._get_container()

.../outposts/controllers/docker.py:238: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>

    def _get_container(self) -> tuple[Container, bool]:
        try:
            return self.client.containers.get(self.name), False
        except NotFound:
            self.logger.info("(Re-)creating container...")
>           image_name = self.try_pull_image()

.../outposts/controllers/docker.py:197: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>

    def try_pull_image(self):
        """Try to pull the image needed for this outpost based on the CONFIG
        `outposts.container_image_base`, but fall back to known-good images"""
        image = self.get_container_image()
        try:
            self.client.images.pull(image)
        except DockerException:  # pragma: no cover
            image = f"ghcr.io/goauthentik/{self.outpost.type}:{__version__}"
>           self.client.images.pull(image)

.../outposts/controllers/docker.py:189: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.images.ImageCollection object at 0x7fddebfc7950>
repository = 'ghcr.io/goauthentik/proxy', tag = '2025.6.0', all_tags = False
kwargs = {}, image_tag = '2025.6.0'

    def pull(self, repository, tag=None, all_tags=False, **kwargs):
        """
        Pull an image of the given name and return it. Similar to the
        ``docker pull`` command.
        If ``tag`` is ``None`` or empty, it is set to ``latest``.
        If ``all_tags`` is set, the ``tag`` parameter is ignored and all image
        tags will be pulled.
    
        If you want to get the raw pull output, use the
        :py:meth:`~docker.api.image.ImageApiMixin.pull` method in the
        low-level API.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags
    
        Returns:
            (:py:class:`Image` or list): The image that has been pulled.
                If ``all_tags`` is True, the method will return a list
                of :py:class:`Image` objects belonging to this repository.
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> # Pull the image tagged `latest` in the busybox repo
            >>> image = client.images.pull('busybox')
    
            >>> # Pull all tags in the busybox repo
            >>> images = client.images.pull('busybox', all_tags=True)
        """
        repository, image_tag = parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if 'stream' in kwargs:
            warnings.warn(
                '`stream` is not a valid parameter for this method'
                ' and will be overridden',
                stacklevel=1,
            )
            del kwargs['stream']
    
>       pull_log = self.client.api.pull(
            repository, tag=tag, stream=True, all_tags=all_tags, **kwargs
        )

.venv/lib/python3.13.../docker/models/images.py:464: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
repository = 'ghcr.io/goauthentik/proxy', tag = '2025.6.0', stream = True
auth_config = None, decode = False, platform = None, all_tags = False

    def pull(self, repository, tag=None, stream=False, auth_config=None,
             decode=False, platform=None, all_tags=False):
        """
        Pulls an image. Similar to the ``docker pull`` command.
    
        Args:
            repository (str): The repository to pull
            tag (str): The tag to pull. If ``tag`` is ``None`` or empty, it
                is set to ``latest``.
            stream (bool): Stream the output as a generator. Make sure to
                consume the generator, otherwise pull might get cancelled.
            auth_config (dict): Override the credentials that are found in the
                config for this request.  ``auth_config`` should contain the
                ``username`` and ``password`` keys to be valid.
            decode (bool): Decode the JSON data from the server into dicts.
                Only applies with ``stream=True``
            platform (str): Platform in the format ``os[/arch[/variant]]``
            all_tags (bool): Pull all image tags, the ``tag`` parameter is
                ignored.
    
        Returns:
            (generator or str): The output
    
        Raises:
            :py:class:`docker.errors.APIError`
                If the server returns an error.
    
        Example:
    
            >>> resp = client.api.pull('busybox', stream=True, decode=True)
            ... for line in resp:
            ...     print(json.dumps(line, indent=4))
            {
                "status": "Pulling image (latest) from busybox",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
            {
                "status": "Pulling image (latest) from busybox, endpoint: ...",
                "progressDetail": {},
                "id": "e72ac664f4f0"
            }
    
        """
        repository, image_tag = utils.parse_repository_tag(repository)
        tag = tag or image_tag or 'latest'
    
        if all_tags:
            tag = None
    
        registry, repo_name = auth.resolve_repository_name(repository)
    
        params = {
            'tag': tag,
            'fromImage': repository
        }
        headers = {}
    
        if auth_config is None:
            header = auth.get_config_header(self, registry)
            if header:
                headers['X-Registry-Auth'] = header
        else:
            log.debug('Sending supplied auth config')
            headers['X-Registry-Auth'] = auth.encode_header(auth_config)
    
        if platform is not None:
            if utils.version_lt(self._version, '1.32'):
                raise errors.InvalidVersion(
                    'platform was only introduced in API version 1.32'
                )
            params['platform'] = platform
    
        response = self._post(
            self._url('/images/create'), params=params, headers=headers,
            stream=stream, timeout=None
        )
    
>       self._raise_for_status(response)

.venv/lib/python3.13.../docker/api/image.py:429: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fddebf81950>
response = <Response [404]>

    def _raise_for_status(self, response):
        """Raises stored :class:`APIError`, if one occurred."""
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
>           raise create_api_error_from_http_exception(e) from e

.venv/lib/python3.13.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy')

    def create_api_error_from_http_exception(e):
        """
        Create a suitable APIError from requests.exceptions.HTTPError.
        """
        response = e.response
        try:
            explanation = response.json()['message']
        except ValueError:
            explanation = (response.text or '').strip()
        cls = APIError
        if response.status_code == 404:
            explanation_msg = (explanation or '').lower()
            if any(fragment in explanation_msg
                   for fragment in _image_not_found_explanation_fragments):
                cls = ImageNotFound
            else:
                cls = NotFound
>       raise cls(e, response=response, explanation=explanation) from e
E       docker.errors.NotFound: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy: Not Found ("manifest unknown")

.venv/lib/python3.13........./site-packages/docker/errors.py:39: NotFound

The above exception was the direct cause of the following exception:

self = <unittest.case._Outcome object at 0x7fddec8896a0>
test_case = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_controller>
subTest = False

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

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

self = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_controller>
result = <TestCaseFunction test_docker_controller>

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

.../hostedtoolcache/Python/3.13.3........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_controller>
method = <bound method TestProxyDocker.test_docker_controller of <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_controller>>

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

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

self = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_controller>

    @pytest.mark.timeout(120)
    def test_docker_controller(self):
        """test that deployment requires update"""
        controller = DockerController(self.outpost, self.service_connection)
>       controller.up()

tests/integration/test_proxy_docker.py:94: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.outposts.controllers.docker.DockerController object at 0x7fddebfcf250>
depth = 1

    def up(self, depth=1):
        if self.outpost.managed == MANAGED_OUTPOST:
            return None
        if depth >= DOCKER_MAX_ATTEMPTS:
            raise ControllerException("Giving up since we exceeded recursion limit.")
        self._migrate_container_name()
        try:
            container, has_been_created = self._get_container()
            if has_been_created:
                container.start()
                return None
            # Check if the container is out of date, delete it and retry
            if len(container.image.tags) > 0:
                should_image = self.try_pull_image()
                if should_image not in container.image.tags:  # pragma: no cover
                    self.logger.info(
                        "Container has mismatched image, re-creating...",
                        has=container.image.tags,
                        should=should_image,
                    )
                    self.down()
                    return self.up(depth + 1)
            # Check container's ports
            if self._comp_ports(container):
                self.logger.info("Container has mis-matched ports, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container values match our values
            if self._comp_env(container):
                self.logger.info("Container has outdated config, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container values match our values
            if self._comp_labels(container):
                self.logger.info("Container has outdated labels, re-creating...")
                self.down()
                return self.up(depth + 1)
            if (
                container.attrs.get("HostConfig", {})
                .get("RestartPolicy", {})
                .get("Name", "")
                .lower()
                != "unless-stopped"
            ):
                self.logger.info("Container has mis-matched restart policy, re-creating...")
                self.down()
                return self.up(depth + 1)
            # Check that container is healthy
            if container.status == "running" and container.attrs.get("State", {}).get(
                "Health", {}
            ).get("Status", "") not in ["healthy", "starting"]:
                # At this point we know the config is correct, but the container isn't healthy,
                # so we just restart it with the same config
                if has_been_created:
                    # Since we've just created the container, give it some time to start.
                    # If its still not up by then, restart it
                    self.logger.info("Container is unhealthy and new, giving it time to boot.")
                    sleep(60)
                self.logger.info("Container is unhealthy, restarting...")
                container.restart()
                return None
            # Check that container is running
            if container.status != "running":
                self.logger.info("Container is not running, restarting...")
                container.start()
                return None
            self.logger.info("Container is running")
            return None
        except DockerException as exc:
>           raise ControllerException(str(exc)) from exc
E           authentik.outposts.controllers.base.ControllerException: 404 Client Error for https://localhost:2376/v1.50/images/create?tag=2025.6.0&fromImage=ghcr.io%2Fgoauthentik%2Fproxy: Not Found ("manifest unknown")

.../outposts/controllers/docker.py:300: ControllerException

Add the following block to your .env file:

AUTHENTIK_IMAGE=ghcr.io/goauthentik/dev-server
AUTHENTIK_TAG=gh-51735c0dea0fc26893d375ecd0cfcc3391e6ee9d
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.

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-51735c0dea0fc26893d375ecd0cfcc3391e6ee9d

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