Test D is mislabeled as 'pool exhaustion' but never triggers it: 10 concurrent get_user_with_posts calls on a pool of 5 each hold the connection for only ~10ms (asyncio.sleep(0.01)), so all 10 complete within the 2s acquire timeout and PoolExhaustedError is never raised or asserted.
No positive acquire-timeout test: nothing holds max_size connections and then asserts a (max_size+1)th acquire raises PoolExhaustedError, so the timeout path is unverified.
MockConnection.execute's injection guard ('if "'" in query and not any string params') is inverted/naive — it flags any query containing a quote when no string params are present, which would false-positive on legitimate DDL.
_in_use_count in MockPool is maintained as a separate counter from the semaphore — correct today but a manual-invariant drift hazard; in_use should be derived from the semaphore.
▮ RECOMMENDED USE
Production-shaped async data-access layer with correct pooling, parameterization, pagination, and transactional rollback — a reliable template for a real Postgres-backed service.
▮ REFACTORED PATCH
# 1. Add a real acquire-timeout / pool-exhaustion test:
async def test_pool_exhaustion():
state = MockDatabaseState()
pool = MockPool(max_size=5, state=state)
held = [await pool.acquire() for _ in range(5)]
try:
with pytest.raises(PoolExhaustedError):
await asyncio.wait_for(pool.acquire(), timeout=3.0)
finally:
for c in held: await c.release()
assert pool.in_use == 0
# 2. Derive in_use from the semaphore so the counter cannot drift:
@property
def in_use(self) -> int:
return self.max_size - self._semaphore._value