Skip to content
Merged
Prev Previous commit
Next Next commit
Add test for pool release overflow path (coverage lines 107-110)
Add test_pool_release_overflow_disconnects_outside_mutex to exercise the
ConnectionPool::release() overflow path where a connection is returned to
a pool that is already at max_size. This triggers the should_disconnect
branch that disconnects outside the mutex.

The disconnect-without-GIL path (connection.cpp lines 160-161) is a C++
destructor safety guard only reachable during interpreter shutdown when
the last shared_ptr drops without the GIL held. This cannot be reliably
exercised from a Python test since __del__/GC always holds the GIL.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
  • Loading branch information
saurabh500 and Copilot committed Apr 6, 2026
commit 83995e43e596d4adc441dca9a7d9e51bf84926bb
29 changes: 29 additions & 0 deletions tests/test_009_pooling.py
Original file line number Diff line number Diff line change
Expand Up @@ -278,6 +278,35 @@ def try_overflow():
c.close()


def test_pool_release_overflow_disconnects_outside_mutex(conn_str):
"""Test that releasing a connection when pool is full disconnects it correctly.

When a connection is returned to a pool that is already at max_size,
the connection must be disconnected. This exercises the overflow path in
ConnectionPool::release() (connection_pool.cpp lines 107-110) where
should_disconnect is set and disconnect happens outside the mutex.
"""
pooling(max_size=1, idle_timeout=30)

# Open two connections — both succeed because the pool issues slots
conn1 = connect(conn_str)
conn2 = connect(conn_str)

# Close conn1 first — returned to the pool (pool now has 1 idle entry)
conn1.close()

# Close conn2 — pool is full (1 idle already), so this connection
# must be disconnected rather than pooled (overflow path).
conn2.close()

# Verify the pool is still functional
Comment thread
saurabh500 marked this conversation as resolved.
conn3 = connect(conn_str)
cursor = conn3.cursor()
cursor.execute("SELECT 1")
assert cursor.fetchone()[0] == 1
conn3.close()


@pytest.mark.skip("Flaky test - idle timeout behavior needs investigation")
def test_pool_idle_timeout_removes_connections(conn_str):
"""Test that idle_timeout removes connections from the pool after the timeout."""
Expand Down
Loading