tests.integration.test_proxy_docker.TestProxyDocker::test_docker_static

Stack Traces | 122s run time

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

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

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

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

    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()

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

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

    def _callSetUp(self):
>       self.setUp()

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

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

    def setUp(self):
        super().setUp()
        self.ssl_folder = mkdtemp()
>       self.run_container(
            image="library/docker:dind",
            network_mode="host",
            privileged=True,
            healthcheck=Healthcheck(
                test=["CMD", "docker", "info"],
                interval=5 * 1_000 * 1_000_000,
                start_period=5 * 1_000 * 1_000_000,
            ),
            environment={"DOCKER_TLS_CERTDIR": "/ssl"},
            volumes={
                f"{self.ssl_folder}/": {
                    "bind": "/ssl",
                }
            },
        )

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

self = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_static>
specs = {'detach': True, 'environment': {'AUTHENTIK_HOST': 'http://localhost:41355', 'DOCKER_TLS_CERTDIR': '/ssl'}, 'healthche...terval': 5000000000, 'Timeout': None, 'Retries': None, 'StartPeriod': 5000000000}, 'image': 'library/docker:dind', ...}
container = <Container: 698a94133cde>
state = {'Dead': False, 'Error': '', 'ExitCode': 0, 'FinishedAt': '0001-01-01T00:00:00Z', ...}

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

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

self = <tests.integration.test_proxy_docker.TestProxyDocker testMethod=test_docker_static>
container = <Container: 698a94133cde>

    def wait_for_container(self, container: Container):
        """Check that container is health"""
        attempt = 0
        while True:
            container.reload()
            status = container.attrs.get("State", {}).get("Health", {}).get("Status")
            if status == "healthy":
                return container
>           sleep(1)

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

signum = 14
frame = <frame at 0x7fe9fcb09d80, file '.../tests/e2e/utils.py', line 95, code wait_for_container>

    def handler(signum, frame):
        __tracebackhide__ = True
>       timeout_sigalrm(item, settings)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12/site-packages/pytest_timeout.py:313: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

item = <TestCaseFunction test_docker_static>
settings = Settings(timeout=120.0, method='signal', func_only=False, disable_debugger_detection=False)

    def timeout_sigalrm(item, settings):
        """Dump stack of threads and raise an exception.
    
        This will output the stacks of any threads other then the
        current to stderr and then raise an AssertionError, thus
        terminating the test.
        """
        if not settings.disable_debugger_detection and is_debugging():
            return
        __tracebackhide__ = True
        nthreads = len(threading.enumerate())
        terminal = item.config.get_terminal_writer()
        if nthreads > 1:
            terminal.sep("+", title="Timeout")
        dump_stacks(terminal)
        if nthreads > 1:
            terminal.sep("+", title="Timeout")
>       pytest.fail("Timeout >%ss" % settings.timeout)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12/site-packages/pytest_timeout.py:498: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

reason = 'Timeout >120.0s', pytrace = True

    @_with_exception(Failed)
    def fail(reason: str = "", pytrace: bool = True) -> NoReturn:
        """Explicitly fail an executing test with the given message.
    
        :param reason:
            The message to show the user as reason for the failure.
    
        :param pytrace:
            If False, msg represents the full failure information and no
            python traceback will be reported.
    
        :raises pytest.fail.Exception:
            The exception that is raised.
        """
        __tracebackhide__ = True
>       raise Failed(msg=reason, pytrace=pytrace)
E       Failed: Timeout >120.0s

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/_pytest/outcomes.py:178: Failed

tests.integration.test_outpost_docker.OutpostDockerTests::test_docker_controller

Stack Traces | 122s run time

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

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

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../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.48........./containers/ak-outpost-test/json

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/requests/models.py:1024: HTTPError

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

self = <authentik.providers.proxy.controllers.docker.ProxyDockerController object at 0x7fea05b380b0>

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

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

self = <docker.models.containers.ContainerCollection object at 0x7fea05b3b320>
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)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/models/containers.py:954: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
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)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/utils/decorators.py:19: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
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
        )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/api/container.py:793: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
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)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/api/client.py:281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
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

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/api/client.py:277: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

e = HTTPError('404 Client Error: Not Found for url: https://localhost:2376/v1.48........./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.48........./containers/ak-outpost-test/json: Not Found ("No such container: ak-outpost-test")

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/docker/errors.py:39: NotFound

During handling of the above exception, another exception occurred:

self = <unittest.case._Outcome object at 0x7fea06fbd5b0>
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.12.9............................../x64/lib/python3.12/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()

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

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

    def _callSetUp(self):
>       self.setUp()

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

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

    def setUp(self):
        super().setUp()
        self.ssl_folder = mkdtemp()
        self.run_container(
            image="library/docker:dind",
            network_mode="host",
            privileged=True,
            healthcheck=Healthcheck(
                test=["CMD", "docker", "info"],
                interval=5 * 1_000 * 1_000_000,
                start_period=5 * 1_000 * 1_000_000,
            ),
            environment={"DOCKER_TLS_CERTDIR": "/ssl"},
            volumes={
                f"{self.ssl_folder}/": {
                    "bind": "/ssl",
                }
            },
        )
        # Ensure that local connection have been created
        outpost_connection_discovery()
        self.provider: ProxyProvider = ProxyProvider.objects.create(
            name="test",
            internal_host="http://localhost",
            external_host="http://localhost",
            authorization_flow=create_test_flow(),
        )
        with (
            open(f"{self.ssl_folder}/client/cert.pem", encoding="utf8") as cert,
            open(f"{self.ssl_folder}/client/key.pem", encoding="utf8") as key,
        ):
            authentication_kp = CertificateKeyPair.objects.create(
                name="docker-authentication",
                certificate_data=cert.read(),
                key_data=key.read(),
            )
        with open(f"{self.ssl_folder}/client/ca.pem", encoding="utf8") as authority:
            verification_kp = CertificateKeyPair.objects.create(
                name="docker-verification",
                certificate_data=authority.read(),
            )
        self.service_connection = DockerServiceConnection.objects.create(
            url="https://localhost:2376",
            tls_verification=verification_kp,
            tls_authentication=authentication_kp,
        )
>       self.outpost: Outpost = Outpost.objects.create(
            name="test",
            type=OutpostType.PROXY,
            service_connection=self.service_connection,
            _config=default_outpost_config(self.live_server_url),
        )

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

self = <django.db.models.manager.Manager object at 0x7fea0623bbf0>, args = ()
kwargs = {'_config': {'authentik_host': 'http://localhost:45507', 'authentik_host_browser': '', 'authentik_host_insecure': Fals...'name': 'test', 'service_connection': <DockerServiceConnection: Docker Service-Connection >, 'type': OutpostType.PROXY}

    @wraps(method)
    def manager_method(self, *args, **kwargs):
>       return getattr(self.get_queryset(), name)(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../db/models/manager.py:87: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <QuerySet []>
kwargs = {'_config': {'authentik_host': 'http://localhost:45507', 'authentik_host_browser': '', 'authentik_host_insecure': Fals...'name': 'test', 'service_connection': <DockerServiceConnection: Docker Service-Connection >, 'type': OutpostType.PROXY}
reverse_one_to_one_fields = frozenset(), obj = <Outpost: Outpost test>

    def create(self, **kwargs):
        """
        Create a new object with the given kwargs, saving it to the database
        and returning the created object.
        """
        reverse_one_to_one_fields = frozenset(kwargs).intersection(
            self.model._meta._reverse_one_to_one_field_names
        )
        if reverse_one_to_one_fields:
            raise ValueError(
                "The following fields do not exist in this model: %s"
                % ", ".join(reverse_one_to_one_fields)
            )
    
        obj = self.model(**kwargs)
        self._for_write = True
>       obj.save(force_insert=True, using=self.db)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../db/models/query.py:679: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Outpost: Outpost test>, force_insert = True, force_update = False
using = 'default', update_fields = None

    def save(
        self, force_insert=False, force_update=False, using=None, update_fields=None
    ):
        """
        Save the current instance. Override this in a subclass if you want to
        control the saving process.
    
        The 'force_insert' and 'force_update' parameters can be used to insist
        that the "save" must be an SQL insert or update (or equivalent for
        non-SQL backends), respectively. Normally, they should not be set.
        """
        self._prepare_related_fields_for_save(operation_name="save")
    
        using = using or router.db_for_write(self.__class__, instance=self)
        if force_insert and (force_update or update_fields):
            raise ValueError("Cannot force both insert and updating in model saving.")
    
        deferred_non_generated_fields = {
            f.attname
            for f in self._meta.concrete_fields
            if f.attname not in self.__dict__ and f.generated is False
        }
        if update_fields is not None:
            # If update_fields is empty, skip the save. We do also check for
            # no-op saves later on for inheritance cases. This bailout is
            # still needed for skipping signal sending.
            if not update_fields:
                return
    
            update_fields = frozenset(update_fields)
            field_names = self._meta._non_pk_concrete_field_names
            non_model_fields = update_fields.difference(field_names)
    
            if non_model_fields:
                raise ValueError(
                    "The following fields do not exist in this model, are m2m "
                    "fields, or are non-concrete fields: %s"
                    % ", ".join(non_model_fields)
                )
    
        # If saving to the same database, and this model is deferred, then
        # automatically do an "update_fields" save on the loaded fields.
        elif (
            not force_insert
            and deferred_non_generated_fields
            and using == self._state.db
        ):
            field_names = set()
            for field in self._meta.concrete_fields:
                if not field.primary_key and not hasattr(field, "through"):
                    field_names.add(field.attname)
            loaded_fields = field_names.difference(deferred_non_generated_fields)
            if loaded_fields:
                update_fields = frozenset(loaded_fields)
    
>       self.save_base(
            using=using,
            force_insert=force_insert,
            force_update=force_update,
            update_fields=update_fields,
        )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../db/models/base.py:822: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Outpost: Outpost test>, raw = False
force_insert = (<class 'authentik.outposts.models.Outpost'>,)
force_update = False, using = 'default', update_fields = None

    def save_base(
        self,
        raw=False,
        force_insert=False,
        force_update=False,
        using=None,
        update_fields=None,
    ):
        """
        Handle the parts of saving which should be done only once per save,
        yet need to be done in raw saves, too. This includes some sanity
        checks and signal sending.
    
        The 'raw' argument is telling save_base not to save any parent
        models and not to do any changes to the values before save. This
        is used by fixture loading.
        """
        using = using or router.db_for_write(self.__class__, instance=self)
        assert not (force_insert and (force_update or update_fields))
        assert update_fields is None or update_fields
        cls = origin = self.__class__
        # Skip proxies, but keep the origin as the proxy model.
        if cls._meta.proxy:
            cls = cls._meta.concrete_model
        meta = cls._meta
        if not meta.auto_created:
            pre_save.send(
                sender=origin,
                instance=self,
                raw=raw,
                using=using,
                update_fields=update_fields,
            )
        # A transaction isn't needed if one query is issued.
        if meta.parents:
            context_manager = transaction.atomic(using=using, savepoint=False)
        else:
            context_manager = transaction.mark_for_rollback_on_error(using=using)
        with context_manager:
            parent_inserted = False
            if not raw:
                # Validate force insert only when parents are inserted.
                force_insert = self._validate_force_insert(force_insert)
                parent_inserted = self._save_parents(
                    cls, using, update_fields, force_insert
                )
            updated = self._save_table(
                raw,
                cls,
                force_insert or parent_inserted,
                force_update,
                using,
                update_fields,
            )
        # Store the database on which the object was saved
        self._state.db = using
        # Once saved, this is no longer a to-be-added instance.
        self._state.adding = False
    
        # Signal that the save is complete
        if not meta.auto_created:
>           post_save.send(
                sender=origin,
                instance=self,
                created=(not updated),
                update_fields=update_fields,
                raw=raw,
                using=using,
            )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../db/models/base.py:924: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.models.signals.ModelSignal object at 0x7fea0f0e2b10>
sender = <class 'authentik.outposts.models.Outpost'>
named = {'created': True, 'instance': <Outpost: Outpost test>, 'raw': False, 'update_fields': None, ...}
responses = [(<function invalidate_flow_cache at 0x7fea05ba8cc0>, None)]
sync_receivers = [<function invalidate_flow_cache at 0x7fea05ba8cc0>, <function post_save_update at 0x7fea056c8b80>, <function ssf_device_post_save at 0x7fea056c8400>]
receiver = <function post_save_update at 0x7fea056c8b80>, response = None

    def send(self, sender, **named):
        """
        Send signal from sender to all connected receivers.
    
        If any receiver raises an error, the error propagates back through send,
        terminating the dispatch loop. So it's possible that all receivers
        won't be called if an error is raised.
    
        If any receivers are asynchronous, they are called after all the
        synchronous receivers via a single call to async_to_sync(). They are
        also executed concurrently with asyncio.gather().
    
        Arguments:
    
            sender
                The sender of the signal. Either a specific object or None.
    
            named
                Named arguments which will be passed to receivers.
    
        Return a list of tuple pairs [(receiver, response), ... ].
        """
        if (
            not self.receivers
            or self.sender_receivers_cache.get(sender) is NO_RECEIVERS
        ):
            return []
        responses = []
        sync_receivers, async_receivers = self._live_receivers(sender)
        for receiver in sync_receivers:
>           response = receiver(signal=self, sender=sender, **named)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../django/dispatch/dispatcher.py:189: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ()
kwargs = {'created': True, 'instance': <Outpost: Outpost test>, 'raw': False, 'sender': <class 'authentik.outposts.models.Outpost'>, ...}
signal_name = 'authentik.outposts.signals.post_save_update'
span = <Span(op='event.django', description:'authentik.outposts.signals.post_save_update', trace_id='dd49d13c18484d68aab45163e198d1c9', span_id='a918c7ee2d4bd750', parent_span_id=None, sampled=None, origin='auto.http.django')>

    @wraps(receiver)
    def wrapper(*args, **kwargs):
        # type: (Any, Any) -> Any
        signal_name = _get_receiver_name(receiver)
        with sentry_sdk.start_span(
            op=OP.EVENT_DJANGO,
            name=signal_name,
            origin=DjangoIntegration.origin,
        ) as span:
            span.set_data("signal", signal_name)
>           return receiver(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/django/signals_handlers.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

sender = <class 'authentik.outposts.models.Outpost'>
instance = <Outpost: Outpost test>, created = True
_ = <Token: ak-outpost-07315da9-5d11-4f3f-b39f-5d530d21589a-api>

    @receiver(post_save)
    def post_save_update(sender, instance: Model, created: bool, **_):
        """If an Outpost is saved, Ensure that token is created/updated
    
        If an OutpostModel, or a model that is somehow connected to an OutpostModel is saved,
        we send a message down the relevant OutpostModels WS connection to trigger an update"""
        if instance.__module__ == "django.db.migrations.recorder":
            return
        if instance.__module__ == "__fake__":
            return
        if not isinstance(instance, UPDATE_TRIGGERING_MODELS):
            return
        if isinstance(instance, Outpost) and created:
            LOGGER.info("New outpost saved, ensuring initial token and user are created")
            _ = instance.token
>       outpost_post_save.delay(class_to_path(instance.__class__), instance.pk)

authentik/outposts/signals.py:67: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>
args = ('authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a'))
kwargs = {}

    def delay(self, *args, **kwargs):
        """Star argument version of :meth:`apply_async`.
    
        Does not support the extra options enabled by :meth:`apply_async`.
    
        Arguments:
            *args (Any): Positional arguments passed on to the task.
            **kwargs (Any): Keyword arguments passed on to the task.
        Returns:
            celery.result.AsyncResult: Future promise.
        """
>       return self.apply_async(args, kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:444: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>, ('authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')), {})
kwargs = {'headers': {'_schema_name': 'public'}}
integration = <sentry_sdk.integrations.celery.CeleryIntegration object at 0x7fea08f3d910>
kwarg_headers = {}, propagate_traces = True
task_name = 'authentik.outposts.tasks.outpost_post_save'
task_started_from_beat = False
span_mgr = <Span(op='queue.submit.celery', description:'authentik.outposts.tasks.outpost_post_save', trace_id='dd49d13c18484d68aab45163e198d1c9', span_id='b70c36b0b97424d3', parent_span_id='a918c7ee2d4bd750', sampled=None, origin='auto.queue.celery')>
span = <Span(op='queue.submit.celery', description:'authentik.outposts.tasks.outpost_post_save', trace_id='dd49d13c18484d68aab45163e198d1c9', span_id='b70c36b0b97424d3', parent_span_id='a918c7ee2d4bd750', sampled=None, origin='auto.queue.celery')>

    @wraps(f)
    def apply_async(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        # Note: kwargs can contain headers=None, so no setdefault!
        # Unsure which backend though.
        integration = sentry_sdk.get_client().get_integration(CeleryIntegration)
        if integration is None:
            return f(*args, **kwargs)
    
        kwarg_headers = kwargs.get("headers") or {}
        propagate_traces = kwarg_headers.pop(
            "sentry-propagate-traces", integration.propagate_traces
        )
    
        if not propagate_traces:
            return f(*args, **kwargs)
    
        if isinstance(args[0], Task):
            task_name = args[0].name  # type: str
        elif len(args) > 1 and isinstance(args[1], str):
            task_name = args[1]
        else:
            task_name = "<unknown Celery task>"
    
        task_started_from_beat = sentry_sdk.get_isolation_scope()._name == "celery-beat"
    
        span_mgr = (
            sentry_sdk.start_span(
                op=OP.QUEUE_SUBMIT_CELERY,
                name=task_name,
                origin=CeleryIntegration.origin,
            )
            if not task_started_from_beat
            else NoOpMgr()
        )  # type: Union[Span, NoOpMgr]
    
        with span_mgr as span:
            kwargs["headers"] = _update_celery_task_headers(
                kwarg_headers, span, integration.monitor_beat_tasks
            )
>           return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:289: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>
args = ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')]
kwargs = {}, task_id = None, producer = None, link = None, link_error = None
shadow = None
options = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
check_arguments = <function outpost_post_save at 0x7fea05717880>
preopts = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
app = <CeleryApp authentik at 0x7fea0e8b23f0>, eager_producer = <Producer: None>
serializer = 'json'
body = (('authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')), {})
content_type = 'application/json'

    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                    link=None, link_error=None, shadow=None, **options):
        """Apply tasks asynchronously by sending a message.
    
        Arguments:
            args (Tuple): The positional arguments to pass on to the task.
    
            kwargs (Dict): The keyword arguments to pass on to the task.
    
            countdown (float): Number of seconds into the future that the
                task should execute.  Defaults to immediate execution.
    
            eta (~datetime.datetime): Absolute time and date of when the task
                should be executed.  May not be specified if `countdown`
                is also supplied.
    
            expires (float, ~datetime.datetime): Datetime or
                seconds in the future for the task should expire.
                The task won't be executed after the expiration time.
    
            shadow (str): Override task name used in logs/monitoring.
                Default is retrieved from :meth:`shadow_name`.
    
            connection (kombu.Connection): Re-use existing broker connection
                instead of acquiring one from the connection pool.
    
            retry (bool): If enabled sending of the task message will be
                retried in the event of connection loss or failure.
                Default is taken from the :setting:`task_publish_retry`
                setting.  Note that you need to handle the
                producer/connection manually for this to work.
    
            retry_policy (Mapping): Override the retry policy used.
                See the :setting:`task_publish_retry_policy` setting.
    
            time_limit (int): If set, overrides the default time limit.
    
            soft_time_limit (int): If set, overrides the default soft
                time limit.
    
            queue (str, kombu.Queue): The queue to route the task to.
                This must be a key present in :setting:`task_queues`, or
                :setting:`task_create_missing_queues` must be
                enabled.  See :ref:`guide-routing` for more
                information.
    
            exchange (str, kombu.Exchange): Named custom exchange to send the
                task to.  Usually not used in combination with the ``queue``
                argument.
    
            routing_key (str): Custom routing key used to route the task to a
                worker server.  If in combination with a ``queue`` argument
                only used to specify custom routing keys to topic exchanges.
    
            priority (int): The task priority, a number between 0 and 9.
                Defaults to the :attr:`priority` attribute.
    
            serializer (str): Serialization method to use.
                Can be `pickle`, `json`, `yaml`, `msgpack` or any custom
                serialization method that's been registered
                with :mod:`kombu.serialization.registry`.
                Defaults to the :attr:`serializer` attribute.
    
            compression (str): Optional compression method
                to use.  Can be one of ``zlib``, ``bzip2``,
                or any custom compression methods registered with
                :func:`kombu.compression.register`.
                Defaults to the :setting:`task_compression` setting.
    
            link (Signature): A single, or a list of tasks signatures
                to apply if the task returns successfully.
    
            link_error (Signature): A single, or a list of task signatures
                to apply if an error occurs while executing the task.
    
            producer (kombu.Producer): custom producer to use when publishing
                the task.
    
            add_to_parent (bool): If set to True (default) and the task
                is applied while executing another task, then the result
                will be appended to the parent tasks ``request.children``
                attribute.  Trailing can also be disabled by default using the
                :attr:`trail` attribute
    
            ignore_result (bool): If set to `False` (default) the result
                of a task will be stored in the backend. If set to `True`
                the result will not be stored. This can also be set
                using the :attr:`ignore_result` in the `app.task` decorator.
    
            publisher (kombu.Producer): Deprecated alias to ``producer``.
    
            headers (Dict): Message headers to be included in the message.
    
        Returns:
            celery.result.AsyncResult: Promise of future evaluation.
    
        Raises:
            TypeError: If not enough arguments are passed, or too many
                arguments are passed.  Note that signature checks may
                be disabled by specifying ``@task(typing=False)``.
            kombu.exceptions.OperationalError: If a connection to the
               transport cannot be made, or if the connection is lost.
    
        Note:
            Also supports all keyword arguments supported by
            :meth:`kombu.Producer.publish`.
        """
        if self.typing:
            try:
                check_arguments = self.__header__
            except AttributeError:  # pragma: no cover
                pass
            else:
                check_arguments(*(args or ()), **(kwargs or {}))
    
        if self.__v2_compat__:
            shadow = shadow or self.shadow_name(self(), args, kwargs, options)
        else:
            shadow = shadow or self.shadow_name(args, kwargs, options)
    
        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts
    
        options.setdefault('ignore_result', self.ignore_result)
        if self.priority:
            options.setdefault('priority', self.priority)
    
        app = self._get_app()
        if app.conf.task_always_eager:
            with app.producer_or_acquire(producer) as eager_producer:
                serializer = options.get('serializer')
                if serializer is None:
                    if eager_producer.serializer:
                        serializer = eager_producer.serializer
                    else:
                        serializer = app.conf.task_serializer
                body = args, kwargs
                content_type, content_encoding, data = serialization.dumps(
                    body, serializer,
                )
                args, kwargs = serialization.loads(
                    data, content_type, content_encoding,
                    accept=[content_type]
                )
            with denied_join_result():
>               return self.apply(args, kwargs, task_id=task_id or uuid(),
                                  link=link, link_error=link_error, **options)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:591: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>
args = ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')]
kwargs = {}, arg = ()
kw = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}

    def apply(self, args=None, kwargs=None, *arg, **kw):
        if celery.VERSION[0] < 4:
            kwargs = kwargs or {}
            self._add_current_schema(kwargs)
    
        else:
            # Celery 4.0 introduced strong typing and the `headers` meta dict.
            self._update_headers(kw)
>       return super(TenantTask, self).apply(args, kwargs, *arg, **kw)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12....../site-packages/tenant_schemas_celery/task.py:75: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>
args = ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')]
kwargs = {}, link = None, link_error = None
task_id = '0c49cb78-77dc-4b51-97f5-fa2ef4133548', retries = 0, throw = False
logfile = None, loglevel = None, headers = {'_schema_name': 'public'}
options = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
build_tracer = <function _patch_build_tracer.<locals>.sentry_build_tracer at 0x7fea07d1b060>
app = <CeleryApp authentik at 0x7fea0e8b23f0>
task = <@task: authentik.outposts.tasks.outpost_post_save of authentik at 0x7fea0e8b23f0>
request = {'callbacks': None, 'delivery_info': {'exchange': None, 'is_eager': True, 'priority': None, 'routing_key': None}, 'errbacks': None, 'headers': {'_schema_name': 'public'}, ...}
tb = None
tracer = <function build_tracer.<locals>.trace_task at 0x7fea0558cf40>

    def apply(self, args=None, kwargs=None,
              link=None, link_error=None,
              task_id=None, retries=None, throw=None,
              logfile=None, loglevel=None, headers=None, **options):
        """Execute this task locally, by blocking until the task returns.
    
        Arguments:
            args (Tuple): positional arguments passed on to the task.
            kwargs (Dict): keyword arguments passed on to the task.
            throw (bool): Re-raise task exceptions.
                Defaults to the :setting:`task_eager_propagates` setting.
    
        Returns:
            celery.result.EagerResult: pre-evaluated result.
        """
        # trace imports Task, so need to import inline.
        from celery.app.trace import build_tracer
    
        app = self._get_app()
        args = args or ()
        kwargs = kwargs or {}
        task_id = task_id or uuid()
        retries = retries or 0
        if throw is None:
            throw = app.conf.task_eager_propagates
    
        # Make sure we get the task instance, not class.
        task = app._tasks[self.name]
    
        request = {
            'id': task_id,
            'task': self.name,
            'retries': retries,
            'is_eager': True,
            'logfile': logfile,
            'loglevel': loglevel or 0,
            'hostname': gethostname(),
            'callbacks': maybe_list(link),
            'errbacks': maybe_list(link_error),
            'headers': headers,
            'ignore_result': options.get('ignore_result', False),
            'delivery_info': {
                'is_eager': True,
                'exchange': options.get('exchange'),
                'routing_key': options.get('routing_key'),
                'priority': options.get('priority'),
            }
        }
        if 'stamped_headers' in options:
            request['stamped_headers'] = maybe_list(options['stamped_headers'])
            request['stamps'] = {
                header: maybe_list(options.get(header, [])) for header in request['stamped_headers']
            }
    
        tb = None
        tracer = build_tracer(
            task.name, task, eager=True,
            propagate=throw, app=self._get_app(),
        )
>       ret = tracer(task_id, args, kwargs, request)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:819: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('0c49cb78-77dc-4b51-97f5-fa2ef4133548', ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d215...is_eager': True, 'priority': None, 'routing_key': None}, 'errbacks': None, 'headers': {'_schema_name': 'public'}, ...})
kwargs = {}

    def runner(*args: "P.args", **kwargs: "P.kwargs"):
        # type: (...) -> R
        if sentry_sdk.get_client().get_integration(integration) is None:
            return original_function(*args, **kwargs)
    
>       return sentry_patched_function(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12............/site-packages/sentry_sdk/utils.py:1783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('0c49cb78-77dc-4b51-97f5-fa2ef4133548', ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d215...is_eager': True, 'priority': None, 'routing_key': None}, 'errbacks': None, 'headers': {'_schema_name': 'public'}, ...})
kwargs = {}
scope = <Scope id=0x7fea057deea0 name=celery type=ScopeType.ISOLATION>
transaction = <Transaction(name='authentik.outposts.tasks.outpost_post_save', op='queue.task.celery', trace_id='6c891cc3d258408caf4618db3922782c', span_id='b1f92fbe7c7b7326', parent_span_id=None, sampled=False, source='task', origin='auto.queue.celery')>
headers = {'_schema_name': 'public'}

    @wraps(f)
    @ensure_integration_enabled(CeleryIntegration, f)
    def _inner(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        with isolation_scope() as scope:
            scope._name = "celery"
            scope.clear_breadcrumbs()
            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))
    
            transaction = None
    
            # Celery task objects are not a thing to be trusted. Even
            # something such as attribute access can fail.
            with capture_internal_exceptions():
                headers = args[3].get("headers") or {}
                transaction = continue_trace(
                    headers,
                    op=OP.QUEUE_TASK_CELERY,
                    name="unknown celery task",
                    source=TRANSACTION_SOURCE_TASK,
                    origin=CeleryIntegration.origin,
                )
                transaction.name = task.name
                transaction.set_status(SPANSTATUS.OK)
    
            if transaction is None:
                return f(*args, **kwargs)
    
            with sentry_sdk.start_transaction(
                transaction,
                custom_sampling_context={
                    "celery_job": {
                        "task": task.name,
                        # for some reason, args[1] is a list if non-empty but a
                        # tuple if empty
                        "args": list(args[1]),
                        "kwargs": args[2],
                    }
                },
            ):
>               return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

uuid = '0c49cb78-77dc-4b51-97f5-fa2ef4133548'
args = ['authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')]
kwargs = {}
request = {'callbacks': None, 'delivery_info': {'exchange': None, 'is_eager': True, 'priority': None, 'routing_key': None}, 'errbacks': None, 'headers': {'_schema_name': 'public'}, ...}

    def trace_task(uuid, args, kwargs, request=None):
        # R      - is the possibly prepared return value.
        # I      - is the Info object.
        # T      - runtime
        # Rstr   - textual representation of return value
        # retval - is the always unmodified return value.
        # state  - is the resulting task state.
    
        # This function is very long because we've unrolled all the calls
        # for performance reasons, and because the function is so long
        # we want the main variables (I, and R) to stand out visually from the
        # the rest of the variables, so breaking PEP8 is worth it ;)
        R = I = T = Rstr = retval = state = None
        task_request = None
        time_start = monotonic()
        try:
            try:
                kwargs.items
            except AttributeError:
                raise InvalidTaskError(
                    'Task keyword arguments is not a mapping')
    
            task_request = Context(request or {}, args=args,
                                   called_directly=False, kwargs=kwargs)
    
            redelivered = (task_request.delivery_info
                           and task_request.delivery_info.get('redelivered', False))
            if deduplicate_successful_tasks and redelivered:
                if task_request.id in successful_requests:
                    return trace_ok_t(R, I, T, Rstr)
                r = AsyncResult(task_request.id, app=app)
    
                try:
                    state = r.state
                except BackendGetMetaError:
                    pass
                else:
                    if state == SUCCESS:
                        info(LOG_IGNORED, {
                            'id': task_request.id,
                            'name': get_task_name(task_request, name),
                            'description': 'Task already completed successfully.'
                        })
                        return trace_ok_t(R, I, T, Rstr)
    
            push_task(task)
            root_id = task_request.root_id or uuid
            task_priority = task_request.delivery_info.get('priority') if \
                inherit_parent_priority else None
            push_request(task_request)
            try:
                # -*- PRE -*-
                if prerun_receivers:
                    send_prerun(sender=task, task_id=uuid, task=task,
                                args=args, kwargs=kwargs)
                loader_task_init(uuid, task)
                if track_started:
                    task.backend.store_result(
                        uuid, {'pid': pid, 'hostname': hostname}, STARTED,
                        request=task_request,
                    )
    
                # -*- TRACE -*-
                try:
                    if task_before_start:
                        task_before_start(uuid, args, kwargs)
    
>                   R = retval = fun(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/trace.py:453: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a'))
kwargs = {}

    def runner(*args: "P.args", **kwargs: "P.kwargs"):
        # type: (...) -> R
        if sentry_sdk.get_client().get_integration(integration) is None:
            return original_function(*args, **kwargs)
    
>       return sentry_patched_function(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12............/site-packages/sentry_sdk/utils.py:1783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('authentik.outposts.models.Outpost', UUID('07315da9-5d11-4f3f-b39f-5d530d21589a'))
kwargs = {}
span = <Span(op='queue.process', description:'authentik.outposts.tasks.outpost_post_save', trace_id='6c891cc3d258408caf4618db3922782c', span_id='8162e8afcc66a4ed', parent_span_id='b1f92fbe7c7b7326', sampled=False, origin='auto.queue.celery')>
latency = 0.0013689994812011719

    @ensure_integration_enabled(CeleryIntegration, f)
    def _inner(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        try:
            with sentry_sdk.start_span(
                op=OP.QUEUE_PROCESS,
                name=task.name,
                origin=CeleryIntegration.origin,
            ) as span:
                _set_messaging_destination_name(task, span)
    
                latency = None
                with capture_internal_exceptions():
                    if (
                        task.request.headers is not None
                        and "sentry-task-enqueued-time" in task.request.headers
                    ):
                        latency = _now_seconds_since_epoch() - task.request.headers.pop(
                            "sentry-task-enqueued-time"
                        )
    
                if latency is not None:
                    span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
    
                with capture_internal_exceptions():
                    span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, task.request.id)
    
                with capture_internal_exceptions():
                    span.set_data(
                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, task.request.retries
                    )
    
                with capture_internal_exceptions():
                    span.set_data(
                        SPANDATA.MESSAGING_SYSTEM,
                        task.app.connection().transport.driver_type,
                    )
    
>               return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:410: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

model_class = 'authentik.outposts.models.Outpost'
model_pk = UUID('07315da9-5d11-4f3f-b39f-5d530d21589a')

    @CELERY_APP.task()
    def outpost_post_save(model_class: str, model_pk: Any):
        """If an Outpost is saved, Ensure that token is created/updated
    
        If an OutpostModel, or a model that is somehow connected to an OutpostModel is saved,
        we send a message down the relevant OutpostModels WS connection to trigger an update"""
        model: Model = path_to_class(model_class)
        try:
            instance = model.objects.get(pk=model_pk)
        except model.DoesNotExist:
            LOGGER.warning("Model does not exist", model=model, pk=model_pk)
            return
    
        if isinstance(instance, Outpost):
            LOGGER.debug("Trigger reconcile for outpost", instance=instance)
>           outpost_controller.delay(str(instance.pk))

authentik/outposts/tasks.py:193: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
args = ('07315da9-5d11-4f3f-b39f-5d530d21589a',), kwargs = {}

    def delay(self, *args, **kwargs):
        """Star argument version of :meth:`apply_async`.
    
        Does not support the extra options enabled by :meth:`apply_async`.
    
        Arguments:
            *args (Any): Positional arguments passed on to the task.
            **kwargs (Any): Keyword arguments passed on to the task.
        Returns:
            celery.result.AsyncResult: Future promise.
        """
>       return self.apply_async(args, kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:444: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>, ('07315da9-5d11-4f3f-b39f-5d530d21589a',), {})
kwargs = {'headers': {'_schema_name': 'public', 'baggage': 'sentry-trace_id=6c891cc3d258408caf4618db3922782c,sentry-environment...3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}}
integration = <sentry_sdk.integrations.celery.CeleryIntegration object at 0x7fea08f3d910>
kwarg_headers = {}, propagate_traces = True
task_name = 'authentik.outposts.tasks.outpost_controller'
task_started_from_beat = False
span_mgr = <Span(op='queue.submit.celery', description:'authentik.outposts.tasks.outpost_controller', trace_id='6c891cc3d258408ca...db3922782c', span_id='b747743d30c2a3c8', parent_span_id='8162e8afcc66a4ed', sampled=False, origin='auto.queue.celery')>
span = <Span(op='queue.submit.celery', description:'authentik.outposts.tasks.outpost_controller', trace_id='6c891cc3d258408ca...db3922782c', span_id='b747743d30c2a3c8', parent_span_id='8162e8afcc66a4ed', sampled=False, origin='auto.queue.celery')>

    @wraps(f)
    def apply_async(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        # Note: kwargs can contain headers=None, so no setdefault!
        # Unsure which backend though.
        integration = sentry_sdk.get_client().get_integration(CeleryIntegration)
        if integration is None:
            return f(*args, **kwargs)
    
        kwarg_headers = kwargs.get("headers") or {}
        propagate_traces = kwarg_headers.pop(
            "sentry-propagate-traces", integration.propagate_traces
        )
    
        if not propagate_traces:
            return f(*args, **kwargs)
    
        if isinstance(args[0], Task):
            task_name = args[0].name  # type: str
        elif len(args) > 1 and isinstance(args[1], str):
            task_name = args[1]
        else:
            task_name = "<unknown Celery task>"
    
        task_started_from_beat = sentry_sdk.get_isolation_scope()._name == "celery-beat"
    
        span_mgr = (
            sentry_sdk.start_span(
                op=OP.QUEUE_SUBMIT_CELERY,
                name=task_name,
                origin=CeleryIntegration.origin,
            )
            if not task_started_from_beat
            else NoOpMgr()
        )  # type: Union[Span, NoOpMgr]
    
        with span_mgr as span:
            kwargs["headers"] = _update_celery_task_headers(
                kwarg_headers, span, integration.monitor_beat_tasks
            )
>           return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:289: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
args = ['07315da9-5d11-4f3f-b39f-5d530d21589a'], kwargs = {}, task_id = None
producer = None, link = None, link_error = None, shadow = None
options = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
check_arguments = functools.partial(<function outpost_controller at 0x7fea056cb6a0>, <object object at 0x7fea0aaf7840>)
preopts = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
app = <CeleryApp authentik at 0x7fea0e8b23f0>, eager_producer = <Producer: None>
serializer = 'json', body = (('07315da9-5d11-4f3f-b39f-5d530d21589a',), {})
content_type = 'application/json'

    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                    link=None, link_error=None, shadow=None, **options):
        """Apply tasks asynchronously by sending a message.
    
        Arguments:
            args (Tuple): The positional arguments to pass on to the task.
    
            kwargs (Dict): The keyword arguments to pass on to the task.
    
            countdown (float): Number of seconds into the future that the
                task should execute.  Defaults to immediate execution.
    
            eta (~datetime.datetime): Absolute time and date of when the task
                should be executed.  May not be specified if `countdown`
                is also supplied.
    
            expires (float, ~datetime.datetime): Datetime or
                seconds in the future for the task should expire.
                The task won't be executed after the expiration time.
    
            shadow (str): Override task name used in logs/monitoring.
                Default is retrieved from :meth:`shadow_name`.
    
            connection (kombu.Connection): Re-use existing broker connection
                instead of acquiring one from the connection pool.
    
            retry (bool): If enabled sending of the task message will be
                retried in the event of connection loss or failure.
                Default is taken from the :setting:`task_publish_retry`
                setting.  Note that you need to handle the
                producer/connection manually for this to work.
    
            retry_policy (Mapping): Override the retry policy used.
                See the :setting:`task_publish_retry_policy` setting.
    
            time_limit (int): If set, overrides the default time limit.
    
            soft_time_limit (int): If set, overrides the default soft
                time limit.
    
            queue (str, kombu.Queue): The queue to route the task to.
                This must be a key present in :setting:`task_queues`, or
                :setting:`task_create_missing_queues` must be
                enabled.  See :ref:`guide-routing` for more
                information.
    
            exchange (str, kombu.Exchange): Named custom exchange to send the
                task to.  Usually not used in combination with the ``queue``
                argument.
    
            routing_key (str): Custom routing key used to route the task to a
                worker server.  If in combination with a ``queue`` argument
                only used to specify custom routing keys to topic exchanges.
    
            priority (int): The task priority, a number between 0 and 9.
                Defaults to the :attr:`priority` attribute.
    
            serializer (str): Serialization method to use.
                Can be `pickle`, `json`, `yaml`, `msgpack` or any custom
                serialization method that's been registered
                with :mod:`kombu.serialization.registry`.
                Defaults to the :attr:`serializer` attribute.
    
            compression (str): Optional compression method
                to use.  Can be one of ``zlib``, ``bzip2``,
                or any custom compression methods registered with
                :func:`kombu.compression.register`.
                Defaults to the :setting:`task_compression` setting.
    
            link (Signature): A single, or a list of tasks signatures
                to apply if the task returns successfully.
    
            link_error (Signature): A single, or a list of task signatures
                to apply if an error occurs while executing the task.
    
            producer (kombu.Producer): custom producer to use when publishing
                the task.
    
            add_to_parent (bool): If set to True (default) and the task
                is applied while executing another task, then the result
                will be appended to the parent tasks ``request.children``
                attribute.  Trailing can also be disabled by default using the
                :attr:`trail` attribute
    
            ignore_result (bool): If set to `False` (default) the result
                of a task will be stored in the backend. If set to `True`
                the result will not be stored. This can also be set
                using the :attr:`ignore_result` in the `app.task` decorator.
    
            publisher (kombu.Producer): Deprecated alias to ``producer``.
    
            headers (Dict): Message headers to be included in the message.
    
        Returns:
            celery.result.AsyncResult: Promise of future evaluation.
    
        Raises:
            TypeError: If not enough arguments are passed, or too many
                arguments are passed.  Note that signature checks may
                be disabled by specifying ``@task(typing=False)``.
            kombu.exceptions.OperationalError: If a connection to the
               transport cannot be made, or if the connection is lost.
    
        Note:
            Also supports all keyword arguments supported by
            :meth:`kombu.Producer.publish`.
        """
        if self.typing:
            try:
                check_arguments = self.__header__
            except AttributeError:  # pragma: no cover
                pass
            else:
                check_arguments(*(args or ()), **(kwargs or {}))
    
        if self.__v2_compat__:
            shadow = shadow or self.shadow_name(self(), args, kwargs, options)
        else:
            shadow = shadow or self.shadow_name(args, kwargs, options)
    
        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts
    
        options.setdefault('ignore_result', self.ignore_result)
        if self.priority:
            options.setdefault('priority', self.priority)
    
        app = self._get_app()
        if app.conf.task_always_eager:
            with app.producer_or_acquire(producer) as eager_producer:
                serializer = options.get('serializer')
                if serializer is None:
                    if eager_producer.serializer:
                        serializer = eager_producer.serializer
                    else:
                        serializer = app.conf.task_serializer
                body = args, kwargs
                content_type, content_encoding, data = serialization.dumps(
                    body, serializer,
                )
                args, kwargs = serialization.loads(
                    data, content_type, content_encoding,
                    accept=[content_type]
                )
            with denied_join_result():
>               return self.apply(args, kwargs, task_id=task_id or uuid(),
                                  link=link, link_error=link_error, **options)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:591: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
args = ['07315da9-5d11-4f3f-b39f-5d530d21589a'], kwargs = {}, arg = ()
kw = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}

    def apply(self, args=None, kwargs=None, *arg, **kw):
        if celery.VERSION[0] < 4:
            kwargs = kwargs or {}
            self._add_current_schema(kwargs)
    
        else:
            # Celery 4.0 introduced strong typing and the `headers` meta dict.
            self._update_headers(kw)
>       return super(TenantTask, self).apply(args, kwargs, *arg, **kw)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12....../site-packages/tenant_schemas_celery/task.py:75: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
args = ['07315da9-5d11-4f3f-b39f-5d530d21589a'], kwargs = {}, link = None
link_error = None, task_id = 'ce52eeb6-d60b-4613-b0ab-8f667e4996a1', retries = 0
throw = False, logfile = None, loglevel = None
headers = {'_schema_name': 'public', 'baggage': 'sentry-trace_id=6c891cc3d258408caf4618db3922782c,sentry-environment=testing,sen...c3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}
options = {'compression': None, 'delivery_mode': None, 'exchange': None, 'expires': None, ...}
build_tracer = <function _patch_build_tracer.<locals>.sentry_build_tracer at 0x7fea07d1b060>
app = <CeleryApp authentik at 0x7fea0e8b23f0>
task = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
request = {'callbacks': None, 'delivery_info': {'exchange': None, 'is_eager': True, 'priority': None, 'routing_key': None}, 'err...408caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, ...}
tb = None
tracer = <function build_tracer.<locals>.trace_task at 0x7fea0558f920>

    def apply(self, args=None, kwargs=None,
              link=None, link_error=None,
              task_id=None, retries=None, throw=None,
              logfile=None, loglevel=None, headers=None, **options):
        """Execute this task locally, by blocking until the task returns.
    
        Arguments:
            args (Tuple): positional arguments passed on to the task.
            kwargs (Dict): keyword arguments passed on to the task.
            throw (bool): Re-raise task exceptions.
                Defaults to the :setting:`task_eager_propagates` setting.
    
        Returns:
            celery.result.EagerResult: pre-evaluated result.
        """
        # trace imports Task, so need to import inline.
        from celery.app.trace import build_tracer
    
        app = self._get_app()
        args = args or ()
        kwargs = kwargs or {}
        task_id = task_id or uuid()
        retries = retries or 0
        if throw is None:
            throw = app.conf.task_eager_propagates
    
        # Make sure we get the task instance, not class.
        task = app._tasks[self.name]
    
        request = {
            'id': task_id,
            'task': self.name,
            'retries': retries,
            'is_eager': True,
            'logfile': logfile,
            'loglevel': loglevel or 0,
            'hostname': gethostname(),
            'callbacks': maybe_list(link),
            'errbacks': maybe_list(link_error),
            'headers': headers,
            'ignore_result': options.get('ignore_result', False),
            'delivery_info': {
                'is_eager': True,
                'exchange': options.get('exchange'),
                'routing_key': options.get('routing_key'),
                'priority': options.get('priority'),
            }
        }
        if 'stamped_headers' in options:
            request['stamped_headers'] = maybe_list(options['stamped_headers'])
            request['stamps'] = {
                header: maybe_list(options.get(header, [])) for header in request['stamped_headers']
            }
    
        tb = None
        tracer = build_tracer(
            task.name, task, eager=True,
            propagate=throw, app=self._get_app(),
        )
>       ret = tracer(task_id, args, kwargs, request)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/task.py:819: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('ce52eeb6-d60b-4613-b0ab-8f667e4996a1', ['07315da9-5d11-4f3f-b39f-5d530d21589a'], {}, {'callbacks': None, 'delivery_i...08caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, ...})
kwargs = {}

    def runner(*args: "P.args", **kwargs: "P.kwargs"):
        # type: (...) -> R
        if sentry_sdk.get_client().get_integration(integration) is None:
            return original_function(*args, **kwargs)
    
>       return sentry_patched_function(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12............/site-packages/sentry_sdk/utils.py:1783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('ce52eeb6-d60b-4613-b0ab-8f667e4996a1', ['07315da9-5d11-4f3f-b39f-5d530d21589a'], {}, {'callbacks': None, 'delivery_i...08caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, ...})
kwargs = {}
scope = <Scope id=0x7fea057de5e0 name=celery type=ScopeType.ISOLATION>
transaction = <Transaction(name='authentik.outposts.tasks.outpost_controller', op='queue.task.celery', trace_id='6c891cc3d258408caf4...an_id='8f0292928f5b1211', parent_span_id='b747743d30c2a3c8', sampled=False, source='task', origin='auto.queue.celery')>
headers = {'_schema_name': 'public', 'baggage': 'sentry-trace_id=6c891cc3d258408caf4618db3922782c,sentry-environment=testing,sen...c3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}

    @wraps(f)
    @ensure_integration_enabled(CeleryIntegration, f)
    def _inner(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        with isolation_scope() as scope:
            scope._name = "celery"
            scope.clear_breadcrumbs()
            scope.add_event_processor(_make_event_processor(task, *args, **kwargs))
    
            transaction = None
    
            # Celery task objects are not a thing to be trusted. Even
            # something such as attribute access can fail.
            with capture_internal_exceptions():
                headers = args[3].get("headers") or {}
                transaction = continue_trace(
                    headers,
                    op=OP.QUEUE_TASK_CELERY,
                    name="unknown celery task",
                    source=TRANSACTION_SOURCE_TASK,
                    origin=CeleryIntegration.origin,
                )
                transaction.name = task.name
                transaction.set_status(SPANSTATUS.OK)
    
            if transaction is None:
                return f(*args, **kwargs)
    
            with sentry_sdk.start_transaction(
                transaction,
                custom_sampling_context={
                    "celery_job": {
                        "task": task.name,
                        # for some reason, args[1] is a list if non-empty but a
                        # tuple if empty
                        "args": list(args[1]),
                        "kwargs": args[2],
                    }
                },
            ):
>               return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

uuid = 'ce52eeb6-d60b-4613-b0ab-8f667e4996a1'
args = ['07315da9-5d11-4f3f-b39f-5d530d21589a'], kwargs = {}
request = {'callbacks': None, 'delivery_info': {'exchange': None, 'is_eager': True, 'priority': None, 'routing_key': None}, 'err...408caf4618db3922782c-b747743d30c2a3c8-0'}, 'sentry-trace': '6c891cc3d258408caf4618db3922782c-b747743d30c2a3c8-0'}, ...}

    def trace_task(uuid, args, kwargs, request=None):
        # R      - is the possibly prepared return value.
        # I      - is the Info object.
        # T      - runtime
        # Rstr   - textual representation of return value
        # retval - is the always unmodified return value.
        # state  - is the resulting task state.
    
        # This function is very long because we've unrolled all the calls
        # for performance reasons, and because the function is so long
        # we want the main variables (I, and R) to stand out visually from the
        # the rest of the variables, so breaking PEP8 is worth it ;)
        R = I = T = Rstr = retval = state = None
        task_request = None
        time_start = monotonic()
        try:
            try:
                kwargs.items
            except AttributeError:
                raise InvalidTaskError(
                    'Task keyword arguments is not a mapping')
    
            task_request = Context(request or {}, args=args,
                                   called_directly=False, kwargs=kwargs)
    
            redelivered = (task_request.delivery_info
                           and task_request.delivery_info.get('redelivered', False))
            if deduplicate_successful_tasks and redelivered:
                if task_request.id in successful_requests:
                    return trace_ok_t(R, I, T, Rstr)
                r = AsyncResult(task_request.id, app=app)
    
                try:
                    state = r.state
                except BackendGetMetaError:
                    pass
                else:
                    if state == SUCCESS:
                        info(LOG_IGNORED, {
                            'id': task_request.id,
                            'name': get_task_name(task_request, name),
                            'description': 'Task already completed successfully.'
                        })
                        return trace_ok_t(R, I, T, Rstr)
    
            push_task(task)
            root_id = task_request.root_id or uuid
            task_priority = task_request.delivery_info.get('priority') if \
                inherit_parent_priority else None
            push_request(task_request)
            try:
                # -*- PRE -*-
                if prerun_receivers:
                    send_prerun(sender=task, task_id=uuid, task=task,
                                args=args, kwargs=kwargs)
                loader_task_init(uuid, task)
                if track_started:
                    task.backend.store_result(
                        uuid, {'pid': pid, 'hostname': hostname}, STARTED,
                        request=task_request,
                    )
    
                # -*- TRACE -*-
                try:
                    if task_before_start:
                        task_before_start(uuid, args, kwargs)
    
>                   R = retval = fun(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../celery/app/trace.py:453: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('07315da9-5d11-4f3f-b39f-5d530d21589a',), kwargs = {}

    def runner(*args: "P.args", **kwargs: "P.kwargs"):
        # type: (...) -> R
        if sentry_sdk.get_client().get_integration(integration) is None:
            return original_function(*args, **kwargs)
    
>       return sentry_patched_function(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12............/site-packages/sentry_sdk/utils.py:1783: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = ('07315da9-5d11-4f3f-b39f-5d530d21589a',), kwargs = {}
span = <Span(op='queue.process', description:'authentik.outposts.tasks.outpost_controller', trace_id='6c891cc3d258408caf4618db3922782c', span_id='a58828e7189c4fb4', parent_span_id='8f0292928f5b1211', sampled=False, origin='auto.queue.celery')>
latency = 0.0013566017150878906

    @ensure_integration_enabled(CeleryIntegration, f)
    def _inner(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        try:
            with sentry_sdk.start_span(
                op=OP.QUEUE_PROCESS,
                name=task.name,
                origin=CeleryIntegration.origin,
            ) as span:
                _set_messaging_destination_name(task, span)
    
                latency = None
                with capture_internal_exceptions():
                    if (
                        task.request.headers is not None
                        and "sentry-task-enqueued-time" in task.request.headers
                    ):
                        latency = _now_seconds_since_epoch() - task.request.headers.pop(
                            "sentry-task-enqueued-time"
                        )
    
                if latency is not None:
                    span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
    
                with capture_internal_exceptions():
                    span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, task.request.id)
    
                with capture_internal_exceptions():
                    span.set_data(
                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, task.request.retries
                    )
    
                with capture_internal_exceptions():
                    span.set_data(
                        SPANDATA.MESSAGING_SYSTEM,
                        task.app.connection().transport.driver_type,
                    )
    
>               return f(*args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../integrations/celery/__init__.py:410: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <@task: authentik.outposts.tasks.outpost_controller of authentik at 0x7fea0e8b23f0>
outpost_pk = '07315da9-5d11-4f3f-b39f-5d530d21589a', action = 'up'
from_cache = False

    @CELERY_APP.task(bind=True, base=SystemTask)
    def outpost_controller(
        self: SystemTask, outpost_pk: str, action: str = "up", from_cache: bool = False
    ):
        """.../update/monitor/delete the deployment of an Outpost"""
        logs = []
        if from_cache:
            outpost: Outpost = cache.get(CACHE_KEY_OUTPOST_DOWN % outpost_pk)
            LOGGER.debug("Getting outpost from cache to delete")
        else:
            outpost: Outpost = Outpost.objects.filter(pk=outpost_pk).first()
            LOGGER.debug("Getting outpost from DB")
        if not outpost:
            LOGGER.warning("No outpost")
            return
        self.set_uid(slugify(outpost.name))
        try:
            controller_type = controller_for_outpost(outpost)
            if not controller_type:
                return
            with controller_type(outpost, outpost.service_connection) as controller:
                LOGGER.debug("---------------Outpost Controller logs starting----------------")
>               logs = getattr(controller, f"{action}_with_logs")()

authentik/outposts/tasks.py:153: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.docker.ProxyDockerController object at 0x7fea05b380b0>

    def up_with_logs(self) -> list[LogEvent]:
        """Call .up() but capture all log output and return it."""
        with capture_logs() as logs:
>           self.up()

.../outposts/controllers/base.py:69: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.docker.ProxyDockerController object at 0x7fea05b380b0>
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:237: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.docker.ProxyDockerController object at 0x7fea05b380b0>

    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:196: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.providers.proxy.controllers.docker.ProxyDockerController object at 0x7fea05b380b0>

    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:185: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.models.images.ImageCollection object at 0x7fea05b3a810>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-website-docs-move-rac-to-open-source', all_tags = False, kwargs = {}
image_tag = 'gh-website-docs-move-rac-to-open-source'

    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
        )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/models/images.py:464: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
repository = 'ghcr.io/goauthentik/dev-proxy'
tag = 'gh-website-docs-move-rac-to-open-source', 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
        )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/api/image.py:424: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
args = ('https://localhost:2376/v1.48/images/create',)
kwargs = {'headers': {}, 'params': {'fromImage': 'ghcr.io/goauthentik/dev-proxy', 'tag': 'gh-website-docs-move-rac-to-open-source'}, 'stream': True, 'timeout': None}

    def inner(self, *args, **kwargs):
        if 'HttpHeaders' in self._general_configs:
            if not kwargs.get('headers'):
                kwargs['headers'] = self._general_configs['HttpHeaders']
            else:
                kwargs['headers'].update(self._general_configs['HttpHeaders'])
>       return f(self, *args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/utils/decorators.py:44: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
url = 'https://localhost:2376/v1.48/images/create'
kwargs = {'headers': {}, 'params': {'fromImage': 'ghcr.io/goauthentik/dev-proxy', 'tag': 'gh-website-docs-move-rac-to-open-source'}, 'stream': True, 'timeout': None}

    @update_headers
    def _post(self, url, **kwargs):
>       return self.post(url, **self._set_request_timeout(kwargs))

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../docker/api/client.py:242: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
url = 'https://localhost:2376/v1.48/images/create', data = None, json = None
kwargs = {'headers': {}, 'params': {'fromImage': 'ghcr.io/goauthentik/dev-proxy', 'tag': 'gh-website-docs-move-rac-to-open-source'}, 'stream': True, 'timeout': None}

    def post(self, url, data=None, json=None, **kwargs):
        r"""Sends a POST request. Returns :class:`Response` object.
    
        :param url: URL for the new :class:`Request` object.
        :param data: (optional) Dictionary, list of tuples, bytes, or file-like
            object to send in the body of the :class:`Request`.
        :param json: (optional) json to send in the body of the :class:`Request`.
        :param \*\*kwargs: Optional arguments that ``request`` takes.
        :rtype: requests.Response
        """
    
>       return self.request("POST", url, data=data, json=json, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12........./site-packages/requests/sessions.py:637: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>, method = 'POST'
url = 'https://localhost:2376/v1.48/images/create'
params = {'fromImage': 'ghcr.io/goauthentik/dev-proxy', 'tag': 'gh-website-docs-move-rac-to-open-source'}
data = None, headers = {}, cookies = None, files = None, auth = None
timeout = None, allow_redirects = True, proxies = {}, hooks = None
stream = True, verify = None, cert = None, json = None

    def request(
        self,
        method,
        url,
        params=None,
        data=None,
        headers=None,
        cookies=None,
        files=None,
        auth=None,
        timeout=None,
        allow_redirects=True,
        proxies=None,
        hooks=None,
        stream=None,
        verify=None,
        cert=None,
        json=None,
    ):
        """Constructs a :class:`Request <Request>`, prepares it and sends it.
        Returns :class:`Response <Response>` object.
    
        :param method: method for the new :class:`Request` object.
        :param url: URL for the new :class:`Request` object.
        :param params: (optional) Dictionary or bytes to be sent in the query
            string for the :class:`Request`.
        :param data: (optional) Dictionary, list of tuples, bytes, or file-like
            object to send in the body of the :class:`Request`.
        :param json: (optional) json to send in the body of the
            :class:`Request`.
        :param headers: (optional) Dictionary of HTTP Headers to send with the
            :class:`Request`.
        :param cookies: (optional) Dict or CookieJar object to send with the
            :class:`Request`.
        :param files: (optional) Dictionary of ``'filename': file-like-objects``
            for multipart encoding upload.
        :param auth: (optional) Auth tuple or callable to enable
            Basic/Digest/Custom HTTP Auth.
        :param timeout: (optional) How long to wait for the server to send
            data before giving up, as a float, or a :ref:`(connect timeout,
            read timeout) <timeouts>` tuple.
        :type timeout: float or tuple
        :param allow_redirects: (optional) Set to True by default.
        :type allow_redirects: bool
        :param proxies: (optional) Dictionary mapping protocol or protocol and
            hostname to the URL of the proxy.
        :param hooks: (optional) Dictionary mapping hook name to one event or
            list of events, event must be callable.
        :param stream: (optional) whether to immediately download the response
            content. Defaults to ``False``.
        :param verify: (optional) Either a boolean, in which case it controls whether we verify
            the server's TLS certificate, or a string, in which case it must be a path
            to a CA bundle to use. Defaults to ``True``. When set to
            ``False``, requests will accept any TLS certificate presented by
            the server, and will ignore hostname mismatches and/or expired
            certificates, which will make your application vulnerable to
            man-in-the-middle (MitM) attacks. Setting verify to ``False``
            may be useful during local development or testing.
        :param cert: (optional) if String, path to ssl client cert file (.pem).
            If Tuple, ('cert', 'key') pair.
        :rtype: requests.Response
        """
        # Create the Request.
        req = Request(
            method=method.upper(),
            url=url,
            headers=headers,
            files=files,
            data=data or {},
            json=json,
            params=params or {},
            auth=auth,
            cookies=cookies,
            hooks=hooks,
        )
        prep = self.prepare_request(req)
    
        proxies = proxies or {}
    
        settings = self.merge_environment_settings(
            prep.url, proxies, stream, verify, cert
        )
    
        # Send the request.
        send_kwargs = {
            "timeout": timeout,
            "allow_redirects": allow_redirects,
        }
        send_kwargs.update(settings)
>       resp = self.send(prep, **send_kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12........./site-packages/requests/sessions.py:589: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <docker.api.client.APIClient object at 0x7fea05b3b380>
request = <PreparedRequest [POST]>
kwargs = {'cert': ('/tmp/a57ced53d59e411a94236ad21ff7dd4e-cert.pem', '/tmp/a57ced53d59e411a94236ad21ff7dd4e-key.pem'), 'proxies': OrderedDict(), 'stream': True, 'timeout': None, ...}
allow_redirects = True, stream = True, hooks = {'response': []}
adapter = <requests.adapters.HTTPAdapter object at 0x7fea05b3b650>
start = 1740400079.799228

    def send(self, request, **kwargs):
        """Send a given PreparedRequest.
    
        :rtype: requests.Response
        """
        # Set defaults that the hooks can utilize to ensure they always have
        # the correct parameters to reproduce the previous request.
        kwargs.setdefault("stream", self.stream)
        kwargs.setdefault("verify", self.verify)
        kwargs.setdefault("cert", self.cert)
        if "proxies" not in kwargs:
            kwargs["proxies"] = resolve_proxies(request, self.proxies, self.trust_env)
    
        # It's possible that users might accidentally send a Request object.
        # Guard against that specific failure case.
        if isinstance(request, Request):
            raise ValueError("You can only send PreparedRequests.")
    
        # Set up variables needed for resolve_redirects and dispatching of hooks
        allow_redirects = kwargs.pop("allow_redirects", True)
        stream = kwargs.get("stream")
        hooks = request.hooks
    
        # Get the appropriate adapter to use
        adapter = self.get_adapter(url=request.url)
    
        # Start time (approximately) of the request
        start = preferred_clock()
    
        # Send the request
>       r = adapter.send(request, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12........./site-packages/requests/sessions.py:703: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <requests.adapters.HTTPAdapter object at 0x7fea05b3b650>
request = <PreparedRequest [POST]>, stream = True
timeout = Timeout(connect=None, read=None, total=None), verify = None
cert = ('/tmp/a57ced53d59e411a94236ad21ff7dd4e-cert.pem', '/tmp/a57ced53d59e411a94236ad21ff7dd4e-key.pem')
proxies = OrderedDict()

    def send(
        self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None
    ):
        """Sends PreparedRequest object. Returns Response object.
    
        :param request: The :class:`PreparedRequest <PreparedRequest>` being sent.
        :param stream: (optional) Whether to stream the request content.
        :param timeout: (optional) How long to wait for the server to send
            data before giving up, as a float, or a :ref:`(connect timeout,
            read timeout) <timeouts>` tuple.
        :type timeout: float or tuple or urllib3 Timeout object
        :param verify: (optional) Either a boolean, in which case it controls whether
            we verify the server's TLS certificate, or a string, in which case it
            must be a path to a CA bundle to use
        :param cert: (optional) Any user-provided SSL certificate to be trusted.
        :param proxies: (optional) The proxies dictionary to apply to the request.
        :rtype: requests.Response
        """
    
        try:
            conn = self.get_connection_with_tls_context(
                request, verify, proxies=proxies, cert=cert
            )
        except LocationValueError as e:
            raise InvalidURL(e, request=request)
    
        self.cert_verify(conn, request.url, verify, cert)
        url = self.request_url(request, proxies)
        self.add_headers(
            request,
            stream=stream,
            timeout=timeout,
            verify=verify,
            cert=cert,
            proxies=proxies,
        )
    
        chunked = not (request.body is None or "Content-Length" in request.headers)
    
        if isinstance(timeout, tuple):
            try:
                connect, read = timeout
                timeout = TimeoutSauce(connect=connect, read=read)
            except ValueError:
                raise ValueError(
                    f"Invalid timeout {timeout}. Pass a (connect, read) timeout tuple, "
                    f"or a single float to set both timeouts to the same value."
                )
        elif isinstance(timeout, TimeoutSauce):
            pass
        else:
            timeout = TimeoutSauce(connect=timeout, read=timeout)
    
        try:
>           resp = conn.urlopen(
                method=request.method,
                url=url,
                body=request.body,
                headers=request.headers,
                redirect=False,
                assert_same_host=False,
                preload_content=False,
                decode_content=False,
                retries=self.max_retries,
                timeout=timeout,
                chunked=chunked,
            )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/requests/adapters.py:667: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <urllib3.connectionpool.HTTPSConnectionPool object at 0x7fea05b3b680>
method = 'POST'
url = '/v1.48/images/create?tag=gh-website-docs-move-rac-to-open-source&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy'
body = None
headers = {'User-Agent': 'docker-sdk-python/7.1.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '0'}
retries = Retry(total=0, connect=None, read=False, redirect=None, status=None)
redirect = False, assert_same_host = False
timeout = Timeout(connect=None, read=None, total=None), pool_timeout = None
release_conn = False, chunked = False, body_pos = None, preload_content = False
decode_content = False, response_kw = {}
parsed_url = Url(scheme=None, auth=None, host=None, port=None, path='/v1.48/images/create', query='tag=gh-website-docs-move-rac-to-open-source&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy', fragment=None)
destination_scheme = None, conn = None, release_this_conn = True
http_tunnel_required = False, err = None, clean_exit = False

    def urlopen(  # type: ignore[override]
        self,
        method: str,
        url: str,
        body: _TYPE_BODY | None = None,
        headers: typing.Mapping[str, str] | None = None,
        retries: Retry | bool | int | None = None,
        redirect: bool = True,
        assert_same_host: bool = True,
        timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
        pool_timeout: int | None = None,
        release_conn: bool | None = None,
        chunked: bool = False,
        body_pos: _TYPE_BODY_POSITION | None = None,
        preload_content: bool = True,
        decode_content: bool = True,
        **response_kw: typing.Any,
    ) -> BaseHTTPResponse:
        """
        Get a connection from the pool and perform an HTTP request. This is the
        lowest level call for making a request, so you'll need to specify all
        the raw details.
    
        .. note::
    
           More commonly, it's appropriate to use a convenience method
           such as :meth:`request`.
    
        .. note::
    
           `release_conn` will only behave as expected if
           `preload_content=False` because we want to make
           `preload_content=False` the default behaviour someday soon without
           breaking backwards compatibility.
    
        :param method:
            HTTP request method (such as GET, POST, PUT, etc.)
    
        :param url:
            The URL to perform the request on.
    
        :param body:
            Data to send in the request body, either :class:`str`, :class:`bytes`,
            an iterable of :class:`str`/:class:`bytes`, or a file-like object.
    
        :param headers:
            Dictionary of custom headers to send, such as User-Agent,
            If-None-Match, etc. If None, pool headers are used. If provided,
            these headers completely replace any pool-specific headers.
    
        :param retries:
            Configure the number of retries to allow before raising a
            :class:`~urllib3.exceptions.MaxRetryError` exception.
    
            If ``None`` (default) will retry 3 times, see ``Retry.DEFAULT``. Pass a
            :class:`~urllib3.util.retry.Retry` object for fine-grained control
            over different types of retries.
            Pass an integer number to retry connection errors that many times,
            but no other types of errors. Pass zero to never retry.
    
            If ``False``, then retries are disabled and any exception is raised
            immediately. Also, instead of raising a MaxRetryError on redirects,
            the redirect response will be returned.
    
        :type retries: :class:`~urllib3.util.retry.Retry`, False, or an int.
    
        :param redirect:
            If True, automatically handle redirects (status codes 301, 302,
            303, 307, 308). Each redirect counts as a retry. Disabling retries
            will disable redirect, too.
    
        :param assert_same_host:
            If ``True``, will make sure that the host of the pool requests is
            consistent else will raise HostChangedError. When ``False``, you can
            use the pool on an HTTP proxy and request foreign hosts.
    
        :param timeout:
            If specified, overrides the default timeout for this one
            request. It may be a float (in seconds) or an instance of
            :class:`urllib3.util.Timeout`.
    
        :param pool_timeout:
            If set and the pool is set to block=True, then this method will
            block for ``pool_timeout`` seconds and raise EmptyPoolError if no
            connection is available within the time period.
    
        :param bool preload_content:
            If True, the response's body will be preloaded into memory.
    
        :param bool decode_content:
            If True, will attempt to decode the body based on the
            'content-encoding' header.
    
        :param release_conn:
            If False, then the urlopen call will not release the connection
            back into the pool once a response is received (but will release if
            you read the entire contents of the response such as when
            `preload_content=True`). This is useful if you're not preloading
            the response's content immediately. You will need to call
            ``r.release_conn()`` on the response ``r`` to return the connection
            back into the pool. If None, it takes the value of ``preload_content``
            which defaults to ``True``.
    
        :param bool chunked:
            If True, urllib3 will send the body using chunked transfer
            encoding. Otherwise, urllib3 will send the body using the standard
            content-length form. Defaults to False.
    
        :param int body_pos:
            Position to seek to in file-like body in the event of a retry or
            redirect. Typically this won't need to be set because urllib3 will
            auto-populate the value when needed.
        """
        parsed_url = parse_url(url)
        destination_scheme = parsed_url.scheme
    
        if headers is None:
            headers = self.headers
    
        if not isinstance(retries, Retry):
            retries = Retry.from_int(retries, redirect=redirect, default=self.retries)
    
        if release_conn is None:
            release_conn = preload_content
    
        # Check host
        if assert_same_host and not self.is_same_host(url):
            raise HostChangedError(self, url, retries)
    
        # Ensure that the URL we're connecting to is properly encoded
        if url.startswith("/"):
            url = to_str(_encode_target(url))
        else:
            url = to_str(parsed_url.url)
    
        conn = None
    
        # Track whether `conn` needs to be released before
        # returning/raising/recursing. Update this variable if necessary, and
        # leave `release_conn` constant throughout the function. That way, if
        # the function recurses, the original value of `release_conn` will be
        # passed down into the recursive call, and its value will be respected.
        #
        # See issue #651 [1] for details.
        #
        # [1] <https://github..../urllib3/issues/651>
        release_this_conn = release_conn
    
        http_tunnel_required = connection_requires_http_tunnel(
            self.proxy, self.proxy_config, destination_scheme
        )
    
        # Merge the proxy headers. Only done when not using HTTP CONNECT. We
        # have to copy the headers dict so we can safely change it without those
        # changes being reflected in anyone else's copy.
        if not http_tunnel_required:
            headers = headers.copy()  # type: ignore[attr-defined]
            headers.update(self.proxy_headers)  # type: ignore[union-attr]
    
        # Must keep the exception bound to a separate variable or else Python 3
        # complains about UnboundLocalError.
        err = None
    
        # Keep track of whether we cleanly exited the except block. This
        # ensures we do proper cleanup in finally.
        clean_exit = False
    
        # Rewind body position, if needed. Record current position
        # for future rewinds in the event of a redirect/retry.
        body_pos = set_file_position(body, body_pos)
    
        try:
            # Request a connection from the queue.
            timeout_obj = self._get_timeout(timeout)
            conn = self._get_conn(timeout=pool_timeout)
    
            conn.timeout = timeout_obj.connect_timeout  # type: ignore[assignment]
    
            # Is this a closed/new connection that requires CONNECT tunnelling?
            if self.proxy is not None and http_tunnel_required and conn.is_closed:
                try:
                    self._prepare_proxy(conn)
                except (BaseSSLError, OSError, SocketTimeout) as e:
                    self._raise_timeout(
                        err=e, url=self.proxy.url, timeout_value=conn.timeout
                    )
                    raise
    
            # If we're going to release the connection in ``finally:``, then
            # the response doesn't need to know about the connection. Otherwise
            # it will also try to release it and we'll have a double-release
            # mess.
            response_conn = conn if not release_conn else None
    
            # Make the request on the HTTPConnection object
>           response = self._make_request(
                conn,
                method,
                url,
                timeout=timeout_obj,
                body=body,
                headers=headers,
                chunked=chunked,
                retries=retries,
                response_conn=response_conn,
                preload_content=preload_content,
                decode_content=decode_content,
                **response_kw,
            )

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12....../site-packages/urllib3/connectionpool.py:787: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <urllib3.connectionpool.HTTPSConnectionPool object at 0x7fea05b3b680>
conn = <urllib3.connection.HTTPSConnection object at 0x7fea05b3a540>
method = 'POST'
url = '/v1.48/images/create?tag=gh-website-docs-move-rac-to-open-source&fromImage=ghcr.io%2Fgoauthentik%2Fdev-proxy'
body = None
headers = {'User-Agent': 'docker-sdk-python/7.1.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '0'}
retries = Retry(total=0, connect=None, read=False, redirect=None, status=None)
timeout = Timeout(connect=None, read=None, total=None), chunked = False
response_conn = <urllib3.connection.HTTPSConnection object at 0x7fea05b3a540>
preload_content = False, decode_content = False, enforce_content_length = True

    def _make_request(
        self,
        conn: BaseHTTPConnection,
        method: str,
        url: str,
        body: _TYPE_BODY | None = None,
        headers: typing.Mapping[str, str] | None = None,
        retries: Retry | None = None,
        timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
        chunked: bool = False,
        response_conn: BaseHTTPConnection | None = None,
        preload_content: bool = True,
        decode_content: bool = True,
        enforce_content_length: bool = True,
    ) -> BaseHTTPResponse:
        """
        Perform a request on a given urllib connection object taken from our
        pool.
    
        :param conn:
            a connection from one of our connection pools
    
        :param method:
            HTTP request method (such as GET, POST, PUT, etc.)
    
        :param url:
            The URL to perform the request on.
    
        :param body:
            Data to send in the request body, either :class:`str`, :class:`bytes`,
            an iterable of :class:`str`/:class:`bytes`, or a file-like object.
    
        :param headers:
            Dictionary of custom headers to send, such as User-Agent,
            If-None-Match, etc. If None, pool headers are used. If provided,
            these headers completely replace any pool-specific headers.
    
        :param retries:
            Configure the number of retries to allow before raising a
            :class:`~urllib3.exceptions.MaxRetryError` exception.
    
            Pass ``None`` to retry until you receive a response. Pass a
            :class:`~urllib3.util.retry.Retry` object for fine-grained control
            over different types of retries.
            Pass an integer number to retry connection errors that many times,
            but no other types of errors. Pass zero to never retry.
    
            If ``False``, then retries are disabled and any exception is raised
            immediately. Also, instead of raising a MaxRetryError on redirects,
            the redirect response will be returned.
    
        :type retries: :class:`~urllib3.util.retry.Retry`, False, or an int.
    
        :param timeout:
            If specified, overrides the default timeout for this one
            request. It may be a float (in seconds) or an instance of
            :class:`urllib3.util.Timeout`.
    
        :param chunked:
            If True, urllib3 will send the body using chunked transfer
            encoding. Otherwise, urllib3 will send the body using the standard
            content-length form. Defaults to False.
    
        :param response_conn:
            Set this to ``None`` if you will handle releasing the connection or
            set the connection to have the response release it.
    
        :param preload_content:
          If True, the response's body will be preloaded during construction.
    
        :param decode_content:
            If True, will attempt to decode the body based on the
            'content-encoding' header.
    
        :param enforce_content_length:
            Enforce content length checking. Body returned by server must match
            value of Content-Length header, if present. Otherwise, raise error.
        """
        self.num_requests += 1
    
        timeout_obj = self._get_timeout(timeout)
        timeout_obj.start_connect()
        conn.timeout = Timeout.resolve_default_timeout(timeout_obj.connect_timeout)
    
        try:
            # Trigger any extra validation we need to do.
            try:
                self._validate_conn(conn)
            except (SocketTimeout, BaseSSLError) as e:
                self._raise_timeout(err=e, url=url, timeout_value=conn.timeout)
                raise
    
        # _validate_conn() starts the connection to an HTTPS proxy
        # so we need to wrap errors with 'ProxyError' here too.
        except (
            OSError,
            NewConnectionError,
            TimeoutError,
            BaseSSLError,
            CertificateError,
            SSLError,
        ) as e:
            new_e: Exception = e
            if isinstance(e, (BaseSSLError, CertificateError)):
                new_e = SSLError(e)
            # If the connection didn't successfully connect to it's proxy
            # then there
            if isinstance(
                new_e, (OSError, NewConnectionError, TimeoutError, SSLError)
            ) and (conn and conn.proxy and not conn.has_connected_to_proxy):
                new_e = _wrap_proxy_error(new_e, conn.proxy.scheme)
            raise new_e
    
        # conn.request() calls http.client.*.request, not the method in
        # urllib3.request. It also calls makefile (recv) on the socket.
        try:
            conn.request(
                method,
                url,
                body=body,
                headers=headers,
                chunked=chunked,
                preload_content=preload_content,
                decode_content=decode_content,
                enforce_content_length=enforce_content_length,
            )
    
        # We are swallowing BrokenPipeError (errno.EPIPE) since the server is
        # legitimately able to close the connection after sending a valid response.
        # With this behaviour, the received response is still readable.
        except BrokenPipeError:
            pass
        except OSError as e:
            # MacOS/Linux
            # EPROTOTYPE and ECONNRESET are needed on macOS
            # https://erickt.github..../11/19/adventures-in-debugging-a-potential-osx-kernel-bug/
            # Condition changed later to emit ECONNRESET instead of only EPROTOTYPE.
            if e.errno != errno.EPROTOTYPE and e.errno != errno.ECONNRESET:
                raise
    
        # Reset the timeout for the recv() on the socket
        read_timeout = timeout_obj.read_timeout
    
        if not conn.is_closed:
            # In Python 3 socket.py will catch EAGAIN and return None when you
            # try and read into the file pointer created by http.client, which
            # instead raises a BadStatusLine exception. Instead of catching
            # the exception and assuming all BadStatusLine exceptions are read
            # timeouts, check for a zero timeout before making the request.
            if read_timeout == 0:
                raise ReadTimeoutError(
                    self, url, f"Read timed out. (read timeout={read_timeout})"
                )
            conn.timeout = read_timeout
    
        # Receive the response from the server
        try:
>           response = conn.getresponse()

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12....../site-packages/urllib3/connectionpool.py:534: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <urllib3.connection.HTTPSConnection object at 0x7fea05b3a540>

    def getresponse(  # type: ignore[override]
        self,
    ) -> HTTPResponse:
        """
        Get the response from the server.
    
        If the HTTPConnection is in the correct state, returns an instance of HTTPResponse or of whatever object is returned by the response_class variable.
    
        If a request has not been sent or if a previous response has not be handled, ResponseNotReady is raised. If the HTTP response indicates that the connection should be closed, then it will be closed before the response is returned. When the connection is closed, the underlying socket is closed.
        """
        # Raise the same error as http.client.HTTPConnection
        if self._response_options is None:
            raise ResponseNotReady()
    
        # Reset this attribute for being used again.
        resp_options = self._response_options
        self._response_options = None
    
        # Since the connection's timeout value may have been updated
        # we need to set the timeout on the socket.
        self.sock.settimeout(self.timeout)
    
        # This is needed here to avoid circular import errors
        from .response import HTTPResponse
    
        # Save a reference to the shutdown function before ownership is passed
        # to httplib_response
        # TODO should we implement it everywhere?
        _shutdown = getattr(self.sock, "shutdown", None)
    
        # Get the response from http.client.HTTPConnection
>       httplib_response = super().getresponse()

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/urllib3/connection.py:516: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <urllib3.connection.HTTPSConnection object at 0x7fea05b3a540>, args = ()
kwargs = {}
span = <Span(op='http.client', description:'POST https://localhost:2376/v1.48/images/create', trace_id='6c891cc3d258408caf461...82c', span_id='a1c40aa3099ea91e', parent_span_id='a58828e7189c4fb4', sampled=False, origin='auto.http.stdlib.httplib')>

    def getresponse(self, *args, **kwargs):
        # type: (HTTPConnection, *Any, **Any) -> Any
        span = getattr(self, "_sentrysdk_span", None)
    
        if span is None:
            return real_getresponse(self, *args, **kwargs)
    
        try:
>           rv = real_getresponse(self, *args, **kwargs)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../sentry_sdk/integrations/stdlib.py:131: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <urllib3.connection.HTTPSConnection object at 0x7fea05b3a540>

    def getresponse(self):
        """Get the response from the server.
    
        If the HTTPConnection is in the correct state, returns an
        instance of HTTPResponse or of whatever object is returned by
        the response_class variable.
    
        If a request has not been sent or if a previous response has
        not be handled, ResponseNotReady is raised.  If the HTTP
        response indicates that the connection should be closed, then
        it will be closed before the response is returned.  When the
        connection is closed, the underlying socket is closed.
        """
    
        # if a prior response has been completed, then forget about it.
        if self.__response and self.__response.isclosed():
            self.__response = None
    
        # if a prior response exists, then it must be completed (otherwise, we
        # cannot read this response's header to determine the connection-close
        # behavior)
        #
        # note: if a prior response existed, but was connection-close, then the
        # socket and response were made independent of this HTTPConnection
        # object since a new request requires that we open a whole new
        # connection
        #
        # this means the prior response had one of two states:
        #   1) will_close: this connection was reset and the prior socket and
        #                  response operate independently
        #   2) persistent: the response was retained and we await its
        #                  isclosed() status to become true.
        #
        if self.__state != _CS_REQ_SENT or self.__response:
            raise ResponseNotReady(self.__state)
    
        if self.debuglevel > 0:
            response = self.response_class(self.sock, self.debuglevel,
                                           method=self._method)
        else:
            response = self.response_class(self.sock, method=self._method)
    
        try:
            try:
>               response.begin()

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/http/client.py:1430: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <http.client.HTTPResponse object at 0x7fea06a5e3b0>

    def begin(self):
        if self.headers is not None:
            # we've already started reading the response
            return
    
        # read until we get a non-100 response
        while True:
>           version, status, reason = self._read_status()

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/http/client.py:331: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <http.client.HTTPResponse object at 0x7fea06a5e3b0>

    def _read_status(self):
>       line = str(self.fp.readline(_MAXLINE + 1), "iso-8859-1")

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/http/client.py:292: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <socket.SocketIO object at 0x7fea06a5e560>
b = <memory at 0x7fea06fc8f40>

    def readinto(self, b):
        """Read up to len(b) bytes into the writable buffer *b* and return
        the number of bytes read.  If the socket is non-blocking and no bytes
        are available, None is returned.
    
        If *b* is non-empty, a 0 return value indicates that the connection
        was shutdown at the other end.
        """
        self._checkClosed()
        self._checkReadable()
        if self._timeout_occurred:
            raise OSError("cannot read from timed out object")
        while True:
            try:
>               return self._sock.recv_into(b)

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/socket.py:720: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ssl.SSLSocket [closed] fd=-1, family=10, type=1, proto=6>
buffer = <memory at 0x7fea06fc8f40>, nbytes = 8192, flags = 0

    def recv_into(self, buffer, nbytes=None, flags=0):
        self._checkClosed()
        if nbytes is None:
            if buffer is not None:
                with memoryview(buffer) as view:
                    nbytes = view.nbytes
                if not nbytes:
                    nbytes = 1024
            else:
                nbytes = 1024
        if self._sslobj is not None:
            if flags != 0:
                raise ValueError(
                  "non-zero flags not allowed in calls to recv_into() on %s" %
                  self.__class__)
>           return self.read(nbytes, buffer)

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/ssl.py:1251: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ssl.SSLSocket [closed] fd=-1, family=10, type=1, proto=6>, len = 8192
buffer = <memory at 0x7fea06fc8f40>

    def read(self, len=1024, buffer=None):
        """Read up to LEN bytes and return them.
        Return zero-length string on EOF."""
    
        self._checkClosed()
        if self._sslobj is None:
            raise ValueError("Read on closed or unwrapped SSL socket.")
        try:
            if buffer is not None:
>               return self._sslobj.read(len, buffer)

.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/ssl.py:1103: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

signum = 14
frame = <frame at 0x7fea06162260, file '.../hostedtoolcache/Python/3.12.9............................../x64/lib/python3.12/ssl.py', line 1106, code read>

    def handler(signum, frame):
        __tracebackhide__ = True
>       timeout_sigalrm(item, settings)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12/site-packages/pytest_timeout.py:313: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

item = <TestCaseFunction test_docker_controller>
settings = Settings(timeout=120.0, method='signal', func_only=False, disable_debugger_detection=False)

    def timeout_sigalrm(item, settings):
        """Dump stack of threads and raise an exception.
    
        This will output the stacks of any threads other then the
        current to stderr and then raise an AssertionError, thus
        terminating the test.
        """
        if not settings.disable_debugger_detection and is_debugging():
            return
        __tracebackhide__ = True
        nthreads = len(threading.enumerate())
        terminal = item.config.get_terminal_writer()
        if nthreads > 1:
            terminal.sep("+", title="Timeout")
        dump_stacks(terminal)
        if nthreads > 1:
            terminal.sep("+", title="Timeout")
>       pytest.fail("Timeout >%ss" % settings.timeout)

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12/site-packages/pytest_timeout.py:498: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

reason = 'Timeout >120.0s', pytrace = True

    @_with_exception(Failed)
    def fail(reason: str = "", pytrace: bool = True) -> NoReturn:
        """Explicitly fail an executing test with the given message.
    
        :param reason:
            The message to show the user as reason for the failure.
    
        :param pytrace:
            If False, msg represents the full failure information and no
            python traceback will be reported.
    
        :raises pytest.fail.Exception:
            The exception that is raised.
        """
        __tracebackhide__ = True
>       raise Failed(msg=reason, pytrace=pytrace)
E       Failed: Timeout >120.0s

../../../..../pypoetry/virtualenvs/authentik-xvtLQ9eE-py3.12/lib/python3.12.../site-packages/_pytest/outcomes.py:178: Failed