ci: test with postgres read-replica

[BeryJu]

Details

Read-replicas were previously only manually tested

---

Checklist

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

If an API change has been made

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

If changes to the frontend have been made

• The code has been formatted (make web)

If applicable

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

[netlify]

✅ Deploy Preview for authentik-storybook ready!

Name	Link
🔨 Latest commit	9b04f13
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/68d413535bfc68000769f16c
😎 Deploy Preview	https://deploy-preview-16977--authentik-storybook.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

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

[netlify]

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	9b04f13
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-docs/deploys/68d413531071c10008a3462c

[netlify]

✅ Deploy Preview for authentik-integrations ready!

Name	Link
🔨 Latest commit	9b04f13
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/68d41353e3f8900008fff437
😎 Deploy Preview	https://deploy-preview-16977--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.

[codecov]

❌ 3 Tests Failed:

Tests completed	Failed	Passed	Skipped
411	3	408	0

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

authentik.sources.kerberos.tests.test_sync.TestKerberosSync::test_tasks

Stack Traces | 0.002s run time

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
>           return getattr(self._connections, alias)

.venv/lib/python3.13.../django/utils/connection.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgiref.local.Local object at 0x7f0c83b5a710>, key = '_'

    def __getattr__(self, key):
        with self._lock_storage() as storage:
>           return getattr(storage, key)
E           AttributeError: '_thread._local' object has no attribute '_'

.venv/lib/python3.13.../site-packages/asgiref/local.py:121: AttributeError

During handling of the above exception, another exception occurred:

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_tasks>
result = <TestCaseFunction test_tasks>, debug = False

    def _setup_and_call(self, result, debug=False):
        """
        Perform the following in order: pre-setup, run test, post-teardown,
        skipping pre/post hooks if test is set to be skipped.
    
        If debug=True, reraise any errors in setup and use super().debug()
        instead of __call__() to run the test.
        """
        testMethod = getattr(self, self._testMethodName)
        skipped = getattr(self.__class__, "__unittest_skip__", False) or getattr(
            testMethod, "__unittest_skip__", False
        )
    
        # Convert async test methods.
        if iscoroutinefunction(testMethod):
            setattr(self, self._testMethodName, async_to_sync(testMethod))
    
        if not skipped:
            try:
>               self._pre_setup()

.venv/lib/python3.13.../django/test/testcases.py:360: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_tasks>

    def _pre_setup(self):
        """
        Perform pre-test setup:
        * If the class has an 'available_apps' attribute, restrict the app
          registry to these applications, then fire the post_migrate signal --
          it must run with the correct set of applications for the test case.
        * If the class has a 'fixtures' attribute, install those fixtures.
        """
        super()._pre_setup()
        if self.available_apps is not None:
            apps.set_available_apps(self.available_apps)
            setting_changed.send(
                sender=settings._wrapped.__class__,
                setting="INSTALLED_APPS",
                value=self.available_apps,
                enter=True,
            )
            for db_name in self._databases_names(include_mirrors=False):
                emit_post_migrate_signal(verbosity=0, interactive=False, db=db_name)
        try:
>           self._fixture_setup()

.venv/lib/python3.13.../django/test/testcases.py:1128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_tasks>

    def _fixture_setup(self):
>       if not self._databases_support_transactions():

.venv/lib/python3.13.../django/test/testcases.py:1430: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'authentik.sources.kerberos.tests.test_sync.TestKerberosSync'>

    @classmethod
    def _databases_support_transactions(cls):
>       return connections_support_transactions(cls.databases)

.venv/lib/python3.13.../django/test/testcases.py:1370: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

aliases = '__all__'

    def connections_support_transactions(aliases=None):
        """
        Return whether or not all (or specified) connections support
        transactions.
        """
        conns = (
            connections.all()
            if aliases is None
            else (connections[alias] for alias in aliases)
        )
>       return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <generator object connections_support_transactions.<locals>.<genexpr> at 0x7f0c7327e810>

>   return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <str_ascii_iterator object at 0x7f0c7430f610>

>       else (connections[alias] for alias in aliases)
    )

.venv/lib/python3.13.../django/test/testcases.py:1281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
            return getattr(self._connections, alias)
        except AttributeError:
            if alias not in self.settings:
>               raise self.exception_class(f"The connection '{alias}' doesn't exist.")
E               django.utils.connection.ConnectionDoesNotExist: The connection '_' doesn't exist.

.venv/lib/python3.13.../django/utils/connection.py:61: ConnectionDoesNotExist

authentik.sources.kerberos.tests.test_sync.TestKerberosSync::test_sync_mapping

Stack Traces | 0.006s run time

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
>           return getattr(self._connections, alias)

.venv/lib/python3.13.../django/utils/connection.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgiref.local.Local object at 0x7f0c83b5a710>, key = '_'

    def __getattr__(self, key):
        with self._lock_storage() as storage:
>           return getattr(storage, key)
E           AttributeError: '_thread._local' object has no attribute '_'

.venv/lib/python3.13.../site-packages/asgiref/local.py:121: AttributeError

During handling of the above exception, another exception occurred:

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_sync_mapping>
result = <TestCaseFunction test_sync_mapping>, debug = False

    def _setup_and_call(self, result, debug=False):
        """
        Perform the following in order: pre-setup, run test, post-teardown,
        skipping pre/post hooks if test is set to be skipped.
    
        If debug=True, reraise any errors in setup and use super().debug()
        instead of __call__() to run the test.
        """
        testMethod = getattr(self, self._testMethodName)
        skipped = getattr(self.__class__, "__unittest_skip__", False) or getattr(
            testMethod, "__unittest_skip__", False
        )
    
        # Convert async test methods.
        if iscoroutinefunction(testMethod):
            setattr(self, self._testMethodName, async_to_sync(testMethod))
    
        if not skipped:
            try:
>               self._pre_setup()

.venv/lib/python3.13.../django/test/testcases.py:360: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_sync_mapping>

    def _pre_setup(self):
        """
        Perform pre-test setup:
        * If the class has an 'available_apps' attribute, restrict the app
          registry to these applications, then fire the post_migrate signal --
          it must run with the correct set of applications for the test case.
        * If the class has a 'fixtures' attribute, install those fixtures.
        """
        super()._pre_setup()
        if self.available_apps is not None:
            apps.set_available_apps(self.available_apps)
            setting_changed.send(
                sender=settings._wrapped.__class__,
                setting="INSTALLED_APPS",
                value=self.available_apps,
                enter=True,
            )
            for db_name in self._databases_names(include_mirrors=False):
                emit_post_migrate_signal(verbosity=0, interactive=False, db=db_name)
        try:
>           self._fixture_setup()

.venv/lib/python3.13.../django/test/testcases.py:1128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_sync_mapping>

    def _fixture_setup(self):
>       if not self._databases_support_transactions():

.venv/lib/python3.13.../django/test/testcases.py:1430: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'authentik.sources.kerberos.tests.test_sync.TestKerberosSync'>

    @classmethod
    def _databases_support_transactions(cls):
>       return connections_support_transactions(cls.databases)

.venv/lib/python3.13.../django/test/testcases.py:1370: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

aliases = '__all__'

    def connections_support_transactions(aliases=None):
        """
        Return whether or not all (or specified) connections support
        transactions.
        """
        conns = (
            connections.all()
            if aliases is None
            else (connections[alias] for alias in aliases)
        )
>       return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <generator object connections_support_transactions.<locals>.<genexpr> at 0x7f0c72cc9700>

>   return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <str_ascii_iterator object at 0x7f0c7371e6e0>

>       else (connections[alias] for alias in aliases)
    )

.venv/lib/python3.13.../django/test/testcases.py:1281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
            return getattr(self._connections, alias)
        except AttributeError:
            if alias not in self.settings:
>               raise self.exception_class(f"The connection '{alias}' doesn't exist.")
E               django.utils.connection.ConnectionDoesNotExist: The connection '_' doesn't exist.

.venv/lib/python3.13.../django/utils/connection.py:61: ConnectionDoesNotExist

authentik.sources.kerberos.tests.test_sync.TestKerberosSync::test_default_mappings

Stack Traces | 18.6s run time

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
>           return getattr(self._connections, alias)

.venv/lib/python3.13.../django/utils/connection.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <asgiref.local.Local object at 0x7f0c83b5a710>, key = '_'

    def __getattr__(self, key):
        with self._lock_storage() as storage:
>           return getattr(storage, key)
E           AttributeError: '_thread._local' object has no attribute '_'

.venv/lib/python3.13.../site-packages/asgiref/local.py:121: AttributeError

During handling of the above exception, another exception occurred:

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_default_mappings>
result = <TestCaseFunction test_default_mappings>, debug = False

    def _setup_and_call(self, result, debug=False):
        """
        Perform the following in order: pre-setup, run test, post-teardown,
        skipping pre/post hooks if test is set to be skipped.
    
        If debug=True, reraise any errors in setup and use super().debug()
        instead of __call__() to run the test.
        """
        testMethod = getattr(self, self._testMethodName)
        skipped = getattr(self.__class__, "__unittest_skip__", False) or getattr(
            testMethod, "__unittest_skip__", False
        )
    
        # Convert async test methods.
        if iscoroutinefunction(testMethod):
            setattr(self, self._testMethodName, async_to_sync(testMethod))
    
        if not skipped:
            try:
>               self._pre_setup()

.venv/lib/python3.13.../django/test/testcases.py:360: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_default_mappings>

    def _pre_setup(self):
        """
        Perform pre-test setup:
        * If the class has an 'available_apps' attribute, restrict the app
          registry to these applications, then fire the post_migrate signal --
          it must run with the correct set of applications for the test case.
        * If the class has a 'fixtures' attribute, install those fixtures.
        """
        super()._pre_setup()
        if self.available_apps is not None:
            apps.set_available_apps(self.available_apps)
            setting_changed.send(
                sender=settings._wrapped.__class__,
                setting="INSTALLED_APPS",
                value=self.available_apps,
                enter=True,
            )
            for db_name in self._databases_names(include_mirrors=False):
                emit_post_migrate_signal(verbosity=0, interactive=False, db=db_name)
        try:
>           self._fixture_setup()

.venv/lib/python3.13.../django/test/testcases.py:1128: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.sources.kerberos.tests.test_sync.TestKerberosSync testMethod=test_default_mappings>

    def _fixture_setup(self):
>       if not self._databases_support_transactions():

.venv/lib/python3.13.../django/test/testcases.py:1430: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

cls = <class 'authentik.sources.kerberos.tests.test_sync.TestKerberosSync'>

    @classmethod
    def _databases_support_transactions(cls):
>       return connections_support_transactions(cls.databases)

.venv/lib/python3.13.../django/test/testcases.py:1370: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

aliases = '__all__'

    def connections_support_transactions(aliases=None):
        """
        Return whether or not all (or specified) connections support
        transactions.
        """
        conns = (
            connections.all()
            if aliases is None
            else (connections[alias] for alias in aliases)
        )
>       return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <generator object connections_support_transactions.<locals>.<genexpr> at 0x7f0c72d3dff0>

>   return all(conn.features.supports_transactions for conn in conns)

.venv/lib/python3.13.../django/test/testcases.py:1283: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <str_ascii_iterator object at 0x7f0c74846740>

>       else (connections[alias] for alias in aliases)
    )

.venv/lib/python3.13.../django/test/testcases.py:1281: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.utils.ConnectionHandler object at 0x7f0c83bc97f0>, alias = '_'

    def __getitem__(self, alias):
        try:
            return getattr(self._connections, alias)
        except AttributeError:
            if alias not in self.settings:
>               raise self.exception_class(f"The connection '{alias}' doesn't exist.")
E               django.utils.connection.ConnectionDoesNotExist: The connection '_' doesn't exist.

.venv/lib/python3.13.../django/utils/connection.py:61: ConnectionDoesNotExist

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

[rissson]

We need to define a policy for those tests. I'm thinking one of:

• keep the earliest postgres version support and the latest one, so 14 and 17
• the two latest ones, so 16 and 17