diff options
| author | Raoni Timo <raoni@relume.io> | 2026-04-30 10:57:27 -0300 |
|---|---|---|
| committer | Jacob Walls <jacobtylerwalls@gmail.com> | 2026-04-30 19:52:11 -0400 |
| commit | 9f790ef1a0f356cf6342b5d57bbaeac35aed0d9f (patch) | |
| tree | 1e14aebd1bce0831ca9353be104a219d430f71fd /tests/backends/postgresql | |
| parent | 8726605e2d48fd733ae224344a11cc2163012f2d (diff) | |
Setting "check" in OPTIONS["pool"] previously raised TypeError because the
PostgreSQL backend always passed check= to ConnectionPool() and unpacked
**pool_options on top, regardless of CONN_HEALTH_CHECKS. The user's callable
now takes precedence via setdefault(); pool_options is copied first to avoid
mutating the user's settings dict.
Diffstat (limited to 'tests/backends/postgresql')
| -rw-r--r-- | tests/backends/postgresql/tests.py | 27 |
1 files changed, 27 insertions, 0 deletions
diff --git a/tests/backends/postgresql/tests.py b/tests/backends/postgresql/tests.py index e03c37cb57..9afa6020a3 100644 --- a/tests/backends/postgresql/tests.py +++ b/tests/backends/postgresql/tests.py @@ -326,6 +326,33 @@ class Tests(TestCase): new_connection.close_pool() @unittest.skipUnless(is_psycopg3, "psycopg3 specific test") + def test_pool_check_can_be_overridden(self): + def custom_check(conn): + pass + + new_connection = no_pool_connection(alias="default_pool") + new_connection.settings_dict["OPTIONS"]["pool"] = {"check": custom_check} + + for enable_checks in (True, False): + with self.subTest(CONN_HEALTH_CHECKS=enable_checks): + new_connection.settings_dict["CONN_HEALTH_CHECKS"] = enable_checks + try: + self.assertIs(new_connection.pool._check, custom_check) + finally: + new_connection.close_pool() + + @unittest.skipUnless(is_psycopg3, "psycopg3 specific test") + def test_pool_options_not_mutated(self): + pool_options = {"min_size": 0, "max_size": 2} + new_connection = no_pool_connection(alias="default_pool") + new_connection.settings_dict["OPTIONS"]["pool"] = pool_options + try: + self.assertIsNotNone(new_connection.pool) + finally: + new_connection.close_pool() + self.assertEqual(pool_options, {"min_size": 0, "max_size": 2}) + + @unittest.skipUnless(is_psycopg3, "psycopg3 specific test") def test_cannot_open_new_connection_in_atomic_block(self): new_connection = no_pool_connection(alias="default_pool") new_connection.settings_dict["OPTIONS"]["pool"] = True |
