summaryrefslogtreecommitdiff
path: root/tests/backends/postgresql/tests.py
diff options
context:
space:
mode:
authorRaoni Timo <raoni@relume.io>2026-04-30 10:57:27 -0300
committerJacob Walls <jacobtylerwalls@gmail.com>2026-04-30 19:52:11 -0400
commit9f790ef1a0f356cf6342b5d57bbaeac35aed0d9f (patch)
tree1e14aebd1bce0831ca9353be104a219d430f71fd /tests/backends/postgresql/tests.py
parent8726605e2d48fd733ae224344a11cc2163012f2d (diff)
Fixed #37075 -- Allowed overriding the PostgreSQL pool's "check" callable.HEADmain
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/tests.py')
-rw-r--r--tests/backends/postgresql/tests.py27
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