Date: 2025-12-14
Test File: resource.test/pytests/factory.core/test_ObjConnectionPool.py
Total Test Cases: 40+ tests across 10 test classes
Tests fundamental pool operations
| Test | Purpose | Validates |
|---|---|---|
test_pool_initialization |
Pool creates with correct size | Initial pool size = configured size |
test_acquire_connection |
Can acquire connection from pool | Connection is valid, stats updated |
test_release_connection |
Can release connection back to pool | Pool size increases, stats updated |
test_acquire_release_multiple |
Multiple acquire/release cycles work | Pool manages multiple connections |
test_context_manager |
Context manager works correctly | Auto-acquire on enter, auto-release on exit |
Run: pytest test_ObjConnectionPool.py::TestConnectionPoolBasics -v
Tests pool size limits and overflow behavior
| Test | Purpose | Validates |
|---|---|---|
test_pool_size_limit |
Pool respects max size | Blocks when pool exhausted (no overflow) |
test_overflow_connections |
Creates overflow connections | Exceeds base size up to overflow limit |
test_overflow_limit_reached |
Blocks when overflow exhausted | Total limit enforced |
test_overflow_connection_closed_on_release |
Overflow connections closed when pool full | Memory management |
Run: pytest test_ObjConnectionPool.py::TestConnectionPoolLimits -v
Key Scenarios:
Tests timeout behavior when pool exhausted
| Test | Purpose | Validates |
|---|---|---|
test_acquire_timeout_default |
Uses default pool timeout | Times out after configured period |
test_acquire_timeout_custom |
Respects custom timeout parameter | Override default timeout |
test_acquire_succeeds_after_release |
Waits for connection to become available | Blocking acquire works |
Run: pytest test_ObjConnectionPool.py::TestConnectionPoolTimeout -v
Timeout Behavior:
timeout=30)acquire(timeout=5)Tests connection recycling based on age
| Test | Purpose | Validates |
|---|---|---|
test_connection_recycling_disabled |
No recycling when recycle=0 |
Connections reused indefinitely |
test_connection_recycling_by_age |
Old connections recycled | Stale connections replaced |
test_young_connection_not_recycled |
Young connections kept | Only old connections recycled |
Run: pytest test_ObjConnectionPool.py::TestConnectionRecycling -v
Recycling Scenarios:
recycle=0: Disabled (connections never recycled)recycle=3600: Connections older than 1 hour recycledTests thread safety with concurrent access
| Test | Purpose | Validates |
|---|---|---|
test_concurrent_acquire_release |
Multiple threads acquire/release safely | No race conditions |
test_concurrent_context_manager |
Context manager thread-safe | Multiple threads use with pool |
test_no_connection_leaks |
No connections leak under concurrency | Pool state consistent |
test_threadpool_executor |
Works with ThreadPoolExecutor | Production pattern |
Run: pytest test_ObjConnectionPool.py::TestThreadSafety -v
Concurrency Patterns Tested:
Tests statistics tracking
| Test | Purpose | Validates |
|---|---|---|
test_stats_creation_tracking |
Tracks connections created | stats['created'] accurate |
test_stats_acquire_release_tracking |
Tracks acquire/release | Counters incremented |
test_stats_timeout_tracking |
Tracks timeouts | stats['timeouts'] incremented |
test_stats_recycling_tracking |
Tracks recycled connections | stats['recycled'] incremented |
test_stats_pool_size |
Reports pool size correctly | Current/max size accurate |
Run: pytest test_ObjConnectionPool.py::TestStatistics -v
Statistics Available:
stats = pool.get_stats()
{
'created': 10, # Total connections created
'recycled': 2, # Connections recycled
'acquired': 150, # Total acquires
'released': 150, # Total releases
'timeouts': 3, # Acquisition timeouts
'total_connections': 10, # Current total
'pool_size': 8, # Available in pool
'max_size': 15 # Maximum allowed
}
Tests error conditions and edge cases
| Test | Purpose | Validates |
|---|---|---|
test_release_none_connection |
Handles releasing None gracefully | No exception raised |
test_close_all_connections |
Closes all connections | Pool emptied |
test_acquire_after_close_all |
Can acquire after close_all | Pool recovers |
Run: pytest test_ObjConnectionPool.py::TestErrorHandling -v
Edge Cases:
Tests global pool singleton
| Test | Purpose | Validates |
|---|---|---|
test_get_global_pool_creates_once |
Singleton pattern works | Same instance returned |
test_get_global_pool_thread_safe |
Thread-safe initialization | No race on creation |
Run: pytest test_ObjConnectionPool.py::TestGlobalPool -v
Global Pool Pattern:
from ObjConnectionPool import get_global_pool
pool = get_global_pool(size=10) # Creates once
pool2 = get_global_pool() # Returns same instance
Stress tests for production readiness
| Test | Purpose | Load |
|---|---|---|
test_high_concurrency_stress |
High concurrent load | 50 threads × 50 ops = 2,500 ops |
test_rapid_acquire_release |
Rapid cycling | 10 threads × 100 cycles = 1,000 ops |
test_long_held_connections |
Varying hold times | 10 threads, 0.1-0.3s holds |
Run: pytest test_ObjConnectionPool.py::TestStressTesting -v
Performance Benchmarks:
Real-world usage patterns
| Test | Scenario | Validates |
|---|---|---|
test_fastapi_worker_simulation |
FastAPI workers | 100 requests, 20 workers |
test_background_task_queue |
Task queue processing | 50 tasks, 10 workers |
test_mixed_read_write_operations |
Mixed workload | 60 reads + 40 writes |
Run: pytest test_ObjConnectionPool.py::TestRealWorldScenarios -v
Real-World Patterns:
cd /home/vheyn/projects/axion
source dev-env/bin/activate
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py -v
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py::TestThreadSafety -v
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py::TestThreadSafety::test_concurrent_acquire_release -v
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py --cov=factory.core.ObjConnectionPool --cov-report=html
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py::TestStressTesting -v -s
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py -v -k "not stress"
All Tests Should:
Statistics Validation:
stats = pool.get_stats()
assert stats['acquired'] == stats['released'] # No leaks
assert stats['total_connections'] <= pool_size + max_overflow # Within limits
pool = ConnectionPool(
size=5, # 5 base connections
max_overflow=5, # Up to 5 extra
timeout=30, # 30s timeout
recycle=3600, # 1 hour recycle
config_name="primary"
)
Fast Tests (development):
pool = ConnectionPool(size=2, timeout=1)
Stress Tests (CI/CD):
pool = ConnectionPool(size=10, max_overflow=10, timeout=5)
Production Simulation:
pool = ConnectionPool(size=20, max_overflow=10, timeout=30, recycle=3600)
Symptom: Tests timeout or hang
Solutions:
size=10 instead of size=5timeout=10 instead of timeout=2max_connections settingSymptom: Cannot connect to database
Solutions:
config.yaml has correct credentialsprimary database config existsSymptom: assert stats['acquired'] == stats['released'] fails
Cause: Connection leak or timing issue
Solutions:
time.sleep(0.1)Pool Operations:
Concurrent Load:
Memory:
name: Connection Pool Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
mysql:
image: mariadb:latest
env:
MYSQL_ROOT_PASSWORD: test
MYSQL_DATABASE: testdb
ports:
- 3306:3306
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.12'
- name: Install dependencies
run: |
pip install -r resource.config/requirements.txt
- name: Run tests
run: |
pytest resource.test/pytests/factory.core/test_ObjConnectionPool.py -v
def test_my_scenario():
"""Test my specific use case"""
pool = ConnectionPool(size=10, config_name="primary")
# Your test logic
with pool as conn:
cursor = conn.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
assert result[0] == 1
Before all tests:
@pytest.fixture(scope="module")
def setup_pool():
pool = ConnectionPool(size=10)
yield pool
pool.close_all()
Use in tests:
def test_with_fixture(setup_pool):
conn = setup_pool.acquire()
# ... test ...
setup_pool.release(conn)
Total Tests: 40+
Coverage Areas: 10 major categories
Execution Time: ~15-30 seconds (all tests)
Success Rate: Should be 100% with properly configured database
Key Tests for Production:
TestThreadSafety - Ensures concurrent safetyTestStressTesting - Validates performanceTestRealWorldScenarios - Tests actual patternsTestConnectionPoolLimits - Prevents resource exhaustionNext Steps:
pytest test_ObjConnectionPool.py -vpool.get_stats()