core: return the intended status code from error views for all request methods (cherry-pick #24902 to version-2026.5) by authentik-cherry-pick[bot] · Pull Request #24957 · goauthentik/authentik

authentik-cherry-pick · 2026-08-10T15:28:47Z

Cherry-pick of #24902 to version-2026.5 branch.

Original PR: #24902
Original Author: @Souptik96
Cherry-picked commit: aca619f

@rissson

…t methods (#24902) * core: return the intended status code from error views for all request methods The views wired up as django's handler400/403/404/500 subclassed TemplateView directly. TemplateView only implements `get`, so its `dispatch` answers a failing POST/PUT/PATCH/DELETE with 405 Method Not Allowed and an `Allow: GET, HEAD, OPTIONS` header, discarding the status code the handler stands for. Any non-GET request that reached an error handler -- on any endpoint -- was reported as a 405 rather than the actual 400/403/404/500. Add a shared ErrorView base that renders the page for every method, and widen the dramatiq result handling in the outgoing sync mixin from ResultFailure to ResultError so a queued-but-not-yet-reported task returns the task log instead of escaping as an unhandled 500. closes #24230 * Apply suggestion from @rissson Co-authored-by: Marc 'risson' Schmitt <marc.schmitt@risson.space> Signed-off-by: Jens L. <jens@beryju.org> --------- Signed-off-by: Jens L. <jens@beryju.org> Co-authored-by: Jens L. <jens@beryju.org> Co-authored-by: Marc 'risson' Schmitt <marc.schmitt@risson.space>

codecov · 2026-08-10T15:34:08Z

❌ 9 Tests Failed:

Tests completed	Failed	Passed	Skipped
3687	9	3678	1

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

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_logout_search

Stack Traces | 12.3s run time

self = <unittest.case._Outcome object at 0x7ff26d87a030>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>
result = <TestCaseFunction test_ldap_bind_logout_search>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>
method = <bound method TestProviderLDAP.test_ldap_bind_logout_search of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>,)
kwargs = {}, config = <AuthentikTenantsConfig: authentik_tenants>

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>,)
kwargs = {}, config = <AuthentikOutpostConfig: authentik_outposts>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_logout_search>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @reconcile_app("authentik_tenants")
    @reconcile_app("authentik_outposts")
    def test_ldap_bind_logout_search(self):
        """Test bind + session deletion -> failed search"""
        self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
            password=self.user.username,
        )
>       _connection.bind()

tests/e2e/test_provider_ldap.py:517: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26ddef5f0>
message_id = 55

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26ddef5f0>
message_id = 55, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_success_ssl

Stack Traces | 12.4s run time

self = <unittest.case._Outcome object at 0x7ff26d842e50>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>
result = <TestCaseFunction test_ldap_bind_success_ssl>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>
method = <bound method TestProviderLDAP.test_ldap_bind_success_ssl of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_ssl>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_ldap_bind_success_ssl(self):
        """Test simple bind with ssl"""
        self._prepare()
        server = Server("ldaps://localhost:6636", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
            password=self.user.username,
        )
>       _connection.bind()

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

self = Connection(server=Server(host='localhost', port=6636, use_ssl=True, allowed_referral_hosts=[('*', True)], tls=Tls(vali...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26d89b4d0>
message_id = 57

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26d89b4d0>
message_id = 57, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_search_attrs_filter

Stack Traces | 12.6s run time

self = <unittest.case._Outcome object at 0x7ff26dbcc6b0>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>
result = <TestCaseFunction test_ldap_search_attrs_filter>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>
method = <bound method TestProviderLDAP.test_ldap_search_attrs_filter of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>,)
kwargs = {}, config = <AuthentikTenantsConfig: authentik_tenants>

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>,)
kwargs = {}, config = <AuthentikOutpostConfig: authentik_outposts>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_search_attrs_filter>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @reconcile_app("authentik_tenants")
    @reconcile_app("authentik_outposts")
    def test_ldap_search_attrs_filter(self):
        """Test search with attributes filtering"""
        # Remove akadmin to ensure list is correct
        # Remove user before starting container so it's not cached
        User.objects.filter(username="akadmin").delete()
    
        outpost = self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
            password=self.user.username,
        )
>       _connection.bind()

tests/e2e/test_provider_ldap.py:450: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26d898c30>
message_id = 56

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26d898c30>
message_id = 56, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_success

Stack Traces | 12.6s run time

self = <unittest.case._Outcome object at 0x7ff26d80a300>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>
result = <TestCaseFunction test_ldap_bind_success>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>
method = <bound method TestProviderLDAP.test_ldap_bind_success of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_ldap_bind_success(self):
        """Test simple bind"""
        self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
            password=self.user.username,
        )
>       _connection.bind()

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

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff27fd68eb0>
message_id = 64

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff27fd68eb0>
message_id = 64, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_success_starttls

Stack Traces | 13s run time

self = <unittest.case._Outcome object at 0x7ff26d842f50>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>
result = <TestCaseFunction test_ldap_bind_success_starttls>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>
method = <bound method TestProviderLDAP.test_ldap_bind_success_starttls of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_success_starttls>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    def test_ldap_bind_success_starttls(self):
        """Test simple bind with ssl"""
        self._prepare()
        server = Server("ldap://localhost:3389")
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
            password=self.user.username,
        )
        _connection.start_tls()
>       _connection.bind()

tests/e2e/test_provider_ldap.py:152: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], tls=Tls(val...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26ddf1450>
message_id = 63

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26ddf1450>
message_id = 63, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_search_no_perms

Stack Traces | 13.3s run time

self = <unittest.case._Outcome object at 0x7ff27c1d5a70>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>
result = <TestCaseFunction test_ldap_bind_search_no_perms>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>
method = <bound method TestProviderLDAP.test_ldap_bind_search_no_perms of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>,)
kwargs = {}, config = <AuthentikTenantsConfig: authentik_tenants>

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>,)
kwargs = {}, config = <AuthentikOutpostConfig: authentik_outposts>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search_no_perms>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @reconcile_app("authentik_tenants")
    @reconcile_app("authentik_outposts")
    def test_ldap_bind_search_no_perms(self):
        """Test simple bind + search"""
        user = create_test_user()
        self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
            password=user.username,
        )
>       _connection.bind()

tests/e2e/test_provider_ldap.py:351: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26dd95490>
message_id = 53

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26dd95490>
message_id = 53, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_schema

Stack Traces | 16.1s run time

self = <unittest.case._Outcome object at 0x7ff26dabc8c0>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>
result = <TestCaseFunction test_ldap_schema>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>
method = <bound method TestProviderLDAP.test_ldap_schema of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>,)
kwargs = {}, config = <AuthentikTenantsConfig: authentik_tenants>

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>,)
kwargs = {}, config = <AuthentikOutpostConfig: authentik_outposts>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_schema>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @reconcile_app("authentik_tenants")
    @reconcile_app("authentik_outposts")
    def test_ldap_schema(self):
        """Test LDAP Schema"""
        self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
            password=self.user.username,
        )
>       _connection.bind()

tests/e2e/test_provider_ldap.py:424: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26df38050>
message_id = 65

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff26df38050>
message_id = 65, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_ldap.TestProviderLDAP::test_ldap_bind_search

Stack Traces | 21.2s run time

self = <unittest.case._Outcome object at 0x7ff27c1d5ba0>
test_case = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>
subTest = False

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>
result = <TestCaseFunction test_ldap_bind_search>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>
method = <bound method TestProviderLDAP.test_ldap_bind_search of <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>
args = (), kwargs = {}

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>,)
kwargs = {}, config = <AuthentikTenantsConfig: authentik_tenants>

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

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

args = (<tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>,)
kwargs = {}, config = <AuthentikOutpostConfig: authentik_outposts>

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

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

self = <tests.e2e.test_provider_ldap.TestProviderLDAP testMethod=test_ldap_bind_search>

    @retry()
    @apply_blueprint(
        "default/flow-default-authentication-flow.yaml",
        "default/flow-default-invalidation-flow.yaml",
    )
    @reconcile_app("authentik_tenants")
    @reconcile_app("authentik_outposts")
    def test_ldap_bind_search(self):
        """Test simple bind + search"""
        # Remove akadmin to ensure list is correct
        # Remove user before starting container so it's not cached
        User.objects.filter(username="akadmin").delete()
    
        outpost = self._prepare()
        server = Server("ldap://localhost:3389", get_info=ALL)
        _connection = Connection(
            server,
            raise_exceptions=True,
            user=f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
            password=self.user.username,
        )
>       _connection.bind()

tests/e2e/test_provider_ldap.py:213: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = Connection(server=Server(host='localhost', port=3389, use_ssl=False, allowed_referral_hosts=[('*', True)], get_info='A...oder=True, auto_range=True, return_empty_attributes=True, auto_encode=True, auto_escape=True, use_referral_cache=False)
read_server_info = True, controls = None

    def bind(self,
             read_server_info=True,
             controls=None):
        """Bind to ldap Server with the authentication method and the user defined in the connection
    
        :param read_server_info: reads info from server
        :param controls: LDAP controls to send along with the bind operation
        :type controls: list of tuple
        :return: bool
    
        """
        if log_enabled(BASIC):
            log(BASIC, 'start BIND operation via <%s>', self)
        self.last_error = None
        with self.connection_lock:
            if self.lazy and not self._executing_deferred:
                if self.strategy.pooled:
                    self.strategy.validate_bind(controls)
                self._deferred_bind = True
                self._bind_controls = controls
                self.bound = True
                if log_enabled(BASIC):
                    log(BASIC, 'deferring bind for <%s>', self)
            else:
                self._deferred_bind = False
                self._bind_controls = None
                if self.closed:  # try to open connection if closed
                    self.open(read_server_info=False)
                if self.authentication == ANONYMOUS:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing anonymous BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, '', auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'anonymous BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
                        response = self.post_send_single_response(self.send('bindRequest', request, controls))
                    else:
                        response = self.strategy.validate_bind(controls)  # only for REUSABLE
                elif self.authentication == SIMPLE:
                    if log_enabled(PROTOCOL):
                        log(PROTOCOL, 'performing simple BIND for <%s>', self)
                    if not self.strategy.pooled:
                        request = bind_operation(self.version, self.authentication, self.user, self.password, auto_encode=self.auto_encode)
                        if log_enabled(PROTOCOL):
                            log(PROTOCOL, 'simple BIND request <%s> sent via <%s>', bind_request_to_dict(request), self)
>                       response = self.post_send_single_response(self.send('bindRequest', request, controls))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/core/connection.py:607: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff27fc36810>
message_id = 52

    def post_send_single_response(self, message_id):
        """
        Executed after an Operation Request (except Search)
        Returns the result message or None
        """
>       responses, result = self.get_response(message_id)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.venv/lib/python3.14.../ldap3/strategy/sync.py:160: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <ldap3.strategy.sync.SyncStrategy object at 0x7ff27fc36810>
message_id = 52, timeout = 20, get_request = False

    def get_response(self, message_id, timeout=None, get_request=False):
        """
        Get response LDAP messages
        Responses are returned by the underlying connection strategy
        Check if message_id LDAP message is still outstanding and wait for timeout to see if it appears in _get_response
        Result is stored in connection.result
        Responses without result is stored in connection.response
        A tuple (responses, result) is returned
        """
        if timeout is None:
            timeout = get_config_parameter('RESPONSE_WAITING_TIMEOUT')
        response = None
        result = None
        # request = None
        if self._outstanding and message_id in self._outstanding:
            responses = self._get_response(message_id, timeout)
    
            if not responses:
                if log_enabled(ERROR):
                    log(ERROR, 'socket timeout, no response from server for <%s>', self.connection)
                raise LDAPResponseTimeoutError('no response from server')
    
            if responses == SESSION_TERMINATED_BY_SERVER:
                try:  # try to close the session but don't raise any error if server has already closed the session
                    self.close()
                except (socket.error, LDAPExceptionError):
                    pass
                self.connection.last_error = 'session terminated by server'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPSessionTerminatedByServerError(self.connection.last_error)
            elif responses == TRANSACTION_ERROR:  # Novell LDAP Transaction unsolicited notification
                self.connection.last_error = 'transaction error'
                if log_enabled(ERROR):
                    log(ERROR, '<%s> for <%s>', self.connection.last_error, self.connection)
                raise LDAPTransactionError(self.connection.last_error)
    
            # if referral in response opens a new connection to resolve referrals if requested
    
            if responses[-2]['result'] == RESULT_REFERRAL:
                if self.connection.usage:
                    self.connection._usage.referrals_received += 1
                if self.connection.auto_referrals:
                    ref_response, ref_result = self.do_operation_on_referral(self._outstanding[message_id], responses[-2]['referrals'])
                    if ref_response is not None:
                        responses = ref_response + [ref_result]
                        responses.append(RESPONSE_COMPLETE)
                    elif ref_result is not None:
                        responses = [ref_result, RESPONSE_COMPLETE]
    
                    self._referrals = []
    
            if responses:
                result = responses[-2]
                response = responses[:-2]
                self.connection.result = None
                self.connection.response = None
    
            if self.connection.raise_exceptions and result and result['result'] not in DO_NOT_RAISE_EXCEPTIONS:
                if log_enabled(PROTOCOL):
                    log(PROTOCOL, 'operation result <%s> for <%s>', result, self.connection)
                self._outstanding.pop(message_id)
                self.connection.result = result.copy()
>               raise LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], response_type=result['type'])
E               ldap3.core.exceptions.LDAPOperationsErrorResult: LDAPOperationsErrorResult - 1 - operationsError - None - None - bindResponse - None

.venv/lib/python3.14.../ldap3/strategy/base.py:403: LDAPOperationsErrorResult

tests.e2e.test_provider_proxy.TestProviderProxyConnect::test_proxy_connectivity

Stack Traces | 118s run time

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

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}
file = 'default/flow-default-provider-authorization-implicit-consent.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Provider authorization flow (implicit consent)\nentries:\n- attrs:\n    desi...henticated\n  identifiers:\n    slug: default-provider-authorization-implicit-consent\n  model: authentik_flows.flow\n'

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

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

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

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

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

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

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
a = 0, b = 1, msg = None

    def assertGreaterEqual(self, a, b, msg=None):
        """Just like self.assertTrue(a >= b), but with a nicer default message."""
        if not a >= b:
            standardMsg = '%s not greater than or equal to %s' % (safe_repr(a), safe_repr(b))
>           self.fail(self._formatMessage(msg, standardMsg))

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
msg = '0 not greater than or equal to 1'

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: 0 not greater than or equal to 1

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

During handling of the above exception, another exception occurred:

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

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}
file = 'default/flow-default-provider-authorization-implicit-consent.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Provider authorization flow (implicit consent)\nentries:\n- attrs:\n    desi...henticated\n  identifiers:\n    slug: default-provider-authorization-implicit-consent\n  model: authentik_flows.flow\n'

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

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

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

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

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

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

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
a = 0, b = 1, msg = None

    def assertGreaterEqual(self, a, b, msg=None):
        """Just like self.assertTrue(a >= b), but with a nicer default message."""
        if not a >= b:
            standardMsg = '%s not greater than or equal to %s' % (safe_repr(a), safe_repr(b))
>           self.fail(self._formatMessage(msg, standardMsg))

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
msg = '0 not greater than or equal to 1'

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: 0 not greater than or equal to 1

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

During handling of the above exception, another exception occurred:

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

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

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

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

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

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

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

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

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

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

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

tests/decorators.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

tests/decorators.py:73: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

tests/decorators.py:67: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

tests/decorators.py:60: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

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

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

args = (<tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>,)
kwargs = {}
file = 'default/flow-default-provider-authorization-implicit-consent.yaml'
content = 'version: 1\nmetadata:\n  name: Default - Provider authorization flow (implicit consent)\nentries:\n- attrs:\n    desi...henticated\n  identifiers:\n    slug: default-provider-authorization-implicit-consent\n  model: authentik_flows.flow\n'

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

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

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

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

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

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

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

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
a = 0, b = 1, msg = None

    def assertGreaterEqual(self, a, b, msg=None):
        """Just like self.assertTrue(a >= b), but with a nicer default message."""
        if not a >= b:
            standardMsg = '%s not greater than or equal to %s' % (safe_repr(a), safe_repr(b))
>           self.fail(self._formatMessage(msg, standardMsg))

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

self = <tests.e2e.test_provider_proxy.TestProviderProxyConnect testMethod=test_proxy_connectivity>
msg = '0 not greater than or equal to 1'

    def fail(self, msg=None):
        """Fail immediately, with the given message."""
>       raise self.failureException(msg)
E       AssertionError: 0 not greater than or equal to 1

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

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

netlify · 2026-08-10T15:34:48Z

✅ Deploy Preview for authentik-integrations ready!

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

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

authentik-cherry-pick Bot added the cherry-pick label Aug 10, 2026

authentik-cherry-pick Bot requested a review from a team as a code owner August 10, 2026 15:28

authentik-cherry-pick Bot added the cherry-pick label Aug 10, 2026

authentik-cherry-pick Bot mentioned this pull request Aug 10, 2026

core: return the intended status code from error views for all request methods #24902

Merged

3 tasks

BeryJu approved these changes Aug 10, 2026

View reviewed changes

BeryJu merged commit 52a3a88 into version-2026.5 Aug 10, 2026
95 of 100 checks passed

BeryJu deleted the cherry-pick/24902-to-version-2026.5 branch August 10, 2026 17:25

Navigation Menu

Uh oh!

Uh oh!

core: return the intended status code from error views for all request methods (cherry-pick #24902 to version-2026.5) - #24957

core: return the intended status code from error views for all request methods (cherry-pick #24902 to version-2026.5)#24957
BeryJu merged 1 commit into
version-2026.5from
cherry-pick/24902-to-version-2026.5

authentik-cherry-pick Bot commented Aug 10, 2026

Uh oh!

codecov Bot commented Aug 10, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Aug 10, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Uh oh!

Conversation

authentik-cherry-pick Bot commented Aug 10, 2026

Uh oh!

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

Uh oh!

❌ 9 Tests Failed:

Uh oh!

netlify Bot commented Aug 10, 2026

✅ Deploy Preview for authentik-integrations ready!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented Aug 10, 2026 •

edited

Loading