mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
1792 lines
55 KiB
Python
1792 lines
55 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# Copyright (c) 2025 Allaun
|
|
|
|
"""Tests for nodupe/tools/parallel/pools.py - Thread/process pool management.
|
|
|
|
Comprehensive tests covering:
|
|
- ThreadPool creation and management
|
|
- ProcessPool creation and management
|
|
- Task submission and retrieval
|
|
- Pool shutdown and cleanup
|
|
- Worker lifecycle
|
|
- Concurrent access safety
|
|
- Resource limits
|
|
- Thread safety in pool operations
|
|
- Proper cleanup on errors
|
|
- Timeout handling
|
|
- Worker failure recovery
|
|
- Race condition prevention
|
|
"""
|
|
|
|
import queue
|
|
import threading
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from nodupe.tools.parallel.pools import (
|
|
ConnectionPool,
|
|
ObjectPool,
|
|
PoolError,
|
|
Pools,
|
|
WorkerPool,
|
|
_is_free_threaded,
|
|
)
|
|
|
|
# =============================================================================
|
|
# Test _is_free_threaded Helper Function
|
|
# =============================================================================
|
|
|
|
class TestIsFreeThreaded:
|
|
"""Test _is_free_threaded helper function."""
|
|
|
|
def test_is_free_threaded_returns_bool(self):
|
|
"""_is_free_threaded() returns boolean."""
|
|
result = _is_free_threaded()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_is_free_threaded_with_gil(self):
|
|
"""_is_free_threaded() returns False when GIL is enabled."""
|
|
with patch('nodupe.tools.parallel.pools.sys.flags') as mock_flags:
|
|
mock_flags.gil = 1
|
|
result = _is_free_threaded()
|
|
assert result is False
|
|
|
|
def test_is_free_threaded_without_gil(self):
|
|
"""_is_free_threaded() returns True when GIL is disabled."""
|
|
with patch('nodupe.tools.parallel.pools.sys.flags') as mock_flags:
|
|
mock_flags.gil = 0
|
|
result = _is_free_threaded()
|
|
assert result is True
|
|
|
|
def test_is_free_threaded_no_flags_attr(self):
|
|
"""_is_free_threaded() returns False when flags attr missing."""
|
|
with patch('nodupe.tools.parallel.pools.sys') as mock_sys:
|
|
del mock_sys.flags
|
|
result = _is_free_threaded()
|
|
assert result is False
|
|
|
|
|
|
# =============================================================================
|
|
# Test PoolError Exception
|
|
# =============================================================================
|
|
|
|
class TestPoolError:
|
|
"""Test PoolError exception class."""
|
|
|
|
def test_pool_error_creation(self):
|
|
"""PoolError can be created with message."""
|
|
error = PoolError("Test error message")
|
|
assert str(error) == "Test error message"
|
|
|
|
def test_pool_error_with_cause(self):
|
|
"""PoolError can wrap another exception."""
|
|
try:
|
|
try:
|
|
raise ValueError("Original error")
|
|
except ValueError as e:
|
|
raise PoolError("Pool operation failed") from e
|
|
except PoolError as pe:
|
|
assert pe.__cause__ is not None
|
|
assert isinstance(pe.__cause__, ValueError)
|
|
|
|
|
|
# =============================================================================
|
|
# Test ObjectPool - Basic Operations
|
|
# =============================================================================
|
|
|
|
class TestObjectPoolBasic:
|
|
"""Test basic ObjectPool operations."""
|
|
|
|
def test_object_pool_creation(self):
|
|
"""ObjectPool can be created with factory function."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
assert pool.factory == factory
|
|
assert pool.max_size == 5
|
|
assert pool.timeout == 5.0
|
|
assert pool.size == 0
|
|
assert pool.active == 0
|
|
|
|
def test_object_pool_acquire(self):
|
|
"""ObjectPool.acquire returns object from factory."""
|
|
counter = {'count': 0}
|
|
|
|
def factory():
|
|
"""Create a numbered object."""
|
|
counter['count'] += 1
|
|
return f"object_{counter['count']}"
|
|
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
obj = pool.acquire()
|
|
|
|
assert obj == "object_1"
|
|
assert pool.active == 1
|
|
|
|
def test_object_pool_release(self):
|
|
"""ObjectPool.release returns object to pool."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
obj = pool.acquire()
|
|
pool.release(obj)
|
|
|
|
assert pool.size == 1
|
|
assert pool.active == 0
|
|
|
|
def test_object_pool_acquire_release_cycle(self):
|
|
"""ObjectPool reuses released objects."""
|
|
counter = {'count': 0}
|
|
|
|
def factory():
|
|
"""Create a numbered object."""
|
|
counter['count'] += 1
|
|
return counter['count']
|
|
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
obj1 = pool.acquire()
|
|
pool.release(obj1)
|
|
obj2 = pool.acquire()
|
|
|
|
# Should reuse the same object
|
|
assert obj2 == obj1
|
|
assert counter['count'] == 1 # Only created once
|
|
|
|
def test_object_pool_max_size(self):
|
|
"""ObjectPool respects max_size limit."""
|
|
created = []
|
|
|
|
def factory():
|
|
"""Create a numbered object."""
|
|
obj = len(created) + 1
|
|
created.append(obj)
|
|
return obj
|
|
|
|
pool = ObjectPool(factory, max_size=3)
|
|
|
|
# Acquire 3 objects
|
|
[pool.acquire() for _ in range(3)]
|
|
assert len(created) == 3
|
|
|
|
# Try to acquire 4th - should fail
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire(timeout=0.1)
|
|
|
|
assert "Pool exhausted" in str(exc_info.value)
|
|
|
|
def test_object_pool_context_manager(self):
|
|
"""ObjectPool.get_object context manager works."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
with pool.get_object() as obj:
|
|
assert obj == "test_object"
|
|
assert pool.active == 1
|
|
|
|
# After context, object should be released
|
|
assert pool.active == 0
|
|
assert pool.size == 1
|
|
|
|
def test_object_pool_close(self):
|
|
"""ObjectPool.close prevents further operations."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
pool.close()
|
|
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire()
|
|
|
|
assert "Pool is closed" in str(exc_info.value)
|
|
|
|
def test_object_pool_close_idempotent(self):
|
|
"""ObjectPool.close can be called multiple times."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
pool.close()
|
|
pool.close() # Should not raise
|
|
|
|
def test_object_pool_context_manager_close(self):
|
|
"""ObjectPool context manager closes pool on exit."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test_object"
|
|
|
|
with ObjectPool(factory, max_size=5) as pool:
|
|
obj = pool.acquire()
|
|
assert obj == "test_object"
|
|
|
|
# After context, pool should be closed
|
|
with pytest.raises(PoolError):
|
|
pool.acquire()
|
|
|
|
|
|
# =============================================================================
|
|
# Test ObjectPool - Advanced Features
|
|
# =============================================================================
|
|
|
|
class TestObjectPoolAdvanced:
|
|
"""Test advanced ObjectPool features."""
|
|
|
|
def test_object_pool_with_reset_func(self):
|
|
"""ObjectPool uses reset_func when releasing objects."""
|
|
reset_called = []
|
|
|
|
def factory():
|
|
"""Create a dict with a value."""
|
|
return {'value': 10}
|
|
|
|
def reset(obj):
|
|
"""Reset the object's value to zero."""
|
|
reset_called.append(obj)
|
|
obj['value'] = 0
|
|
|
|
pool = ObjectPool(factory, max_size=5, reset_func=reset)
|
|
|
|
obj = pool.acquire()
|
|
obj['value'] = 100
|
|
pool.release(obj)
|
|
|
|
assert len(reset_called) == 1
|
|
# Object should be reset before next acquire
|
|
obj2 = pool.acquire()
|
|
assert obj2['value'] == 0
|
|
|
|
def test_object_pool_with_destroy_func(self):
|
|
"""ObjectPool uses destroy_func when removing objects."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed) + 1}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
|
|
pool = ObjectPool(factory, max_size=3, destroy_func=destroy)
|
|
|
|
objs = [pool.acquire() for _ in range(3)]
|
|
|
|
# Release objects back to pool
|
|
for obj in objs:
|
|
pool.release(obj)
|
|
|
|
# Close pool - should destroy all objects
|
|
pool.close()
|
|
|
|
assert len(destroyed) == 3
|
|
|
|
def test_object_pool_destroy_on_release_when_full(self):
|
|
"""ObjectPool destroys object if pool is full on release."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed) + 1}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
|
|
pool = ObjectPool(factory, max_size=2, destroy_func=destroy)
|
|
|
|
obj1 = pool.acquire()
|
|
obj2 = pool.acquire()
|
|
|
|
# Release only one object back to pool
|
|
pool.release(obj1)
|
|
# Keep obj2 active
|
|
|
|
# Now pool has 1 object, active_count=1
|
|
# Acquire the pooled object
|
|
obj3 = pool.acquire() # Gets obj1 from pool
|
|
|
|
# Now pool is empty, active_count=2
|
|
# Release obj2 - goes back to pool (pool not full)
|
|
pool.release(obj2)
|
|
|
|
# Now pool has 1 object, active_count=1
|
|
# Acquire again
|
|
obj4 = pool.acquire() # Gets obj2 from pool
|
|
|
|
# Now pool is empty, active_count=2
|
|
# Release obj3 - goes back to pool
|
|
pool.release(obj3)
|
|
|
|
# Now pool has 1 object, active_count=1
|
|
# Release obj4 - goes back to pool (pool now has 2 objects)
|
|
pool.release(obj4)
|
|
|
|
# Now pool is full with 2 objects, active_count=0
|
|
# Create a new object directly (bypassing pool limits for testing)
|
|
# Actually, we can't acquire more than max_size, so let's test differently
|
|
# Test that destroy_func is called when pool is closed with objects
|
|
pool.close()
|
|
|
|
# All objects should be destroyed
|
|
assert len(destroyed) == 2
|
|
|
|
def test_object_pool_factory_exception(self):
|
|
"""ObjectPool.acquire handles factory exceptions."""
|
|
def failing_factory():
|
|
"""Factory that always raises an exception."""
|
|
raise RuntimeError("Factory failed")
|
|
|
|
pool = ObjectPool(failing_factory, max_size=5)
|
|
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire()
|
|
|
|
assert "Failed to create object" in str(exc_info.value)
|
|
|
|
def test_object_pool_custom_timeout(self):
|
|
"""ObjectPool respects custom timeout."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=1, timeout=10.0)
|
|
|
|
pool.acquire()
|
|
|
|
# Pool exhausted, should timeout
|
|
start = time.time()
|
|
with pytest.raises(PoolError):
|
|
pool.acquire(timeout=0.1)
|
|
elapsed = time.time() - start
|
|
|
|
assert elapsed >= 0.1
|
|
assert elapsed < 1.0 # Should not wait too long
|
|
|
|
def test_object_pool_size_property(self):
|
|
"""ObjectPool.size returns current pool size."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
assert pool.size == 0
|
|
|
|
obj = pool.acquire()
|
|
assert pool.size == 0 # Object is active, not in pool
|
|
|
|
pool.release(obj)
|
|
assert pool.size == 1
|
|
|
|
def test_object_pool_active_property(self):
|
|
"""ObjectPool.active returns active object count."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
assert pool.active == 0
|
|
|
|
obj1 = pool.acquire()
|
|
assert pool.active == 1
|
|
|
|
pool.acquire()
|
|
assert pool.active == 2
|
|
|
|
pool.release(obj1)
|
|
assert pool.active == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Test ObjectPool - Thread Safety
|
|
# =============================================================================
|
|
|
|
class TestObjectPoolThreadSafety:
|
|
"""Test ObjectPool thread safety."""
|
|
|
|
def test_object_pool_concurrent_acquire_release(self):
|
|
"""ObjectPool handles concurrent acquire/release safely."""
|
|
lock = threading.Lock()
|
|
results = []
|
|
errors = []
|
|
|
|
def worker(worker_id):
|
|
"""Worker thread that acquires and releases objects."""
|
|
try:
|
|
def factory():
|
|
"""Create an object with worker id."""
|
|
return f"obj_{worker_id}"
|
|
pool = ObjectPool(factory, max_size=10)
|
|
|
|
for i in range(10):
|
|
obj = pool.acquire()
|
|
time.sleep(0.001) # Simulate work
|
|
pool.release(obj)
|
|
with lock:
|
|
results.append(f"{worker_id}:{i}")
|
|
except Exception as e:
|
|
with lock:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert len(errors) == 0
|
|
assert len(results) == 50 # 5 workers * 10 iterations
|
|
|
|
def test_object_pool_concurrent_multi_pool(self):
|
|
"""Multiple ObjectPools can be used concurrently."""
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def worker(pool_id):
|
|
"""Worker that uses a pool with specific pool_id."""
|
|
def factory():
|
|
"""Create an object with pool id."""
|
|
return f"pool_{pool_id}_obj"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
for i in range(5):
|
|
obj = pool.acquire()
|
|
pool.release(obj)
|
|
with lock:
|
|
results.append(pool_id)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert len(results) == 50
|
|
|
|
def test_object_pool_rlock_vs_lock(self):
|
|
"""ObjectPool works with both RLock and Lock."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
|
|
# Test with RLock (default)
|
|
pool_rlock = ObjectPool(factory, max_size=5, use_rlock=True)
|
|
obj = pool_rlock.acquire()
|
|
pool_rlock.release(obj)
|
|
assert pool_rlock.size == 1
|
|
|
|
# Test with Lock
|
|
pool_lock = ObjectPool(factory, max_size=5, use_rlock=False)
|
|
obj = pool_lock.acquire()
|
|
pool_lock.release(obj)
|
|
assert pool_lock.size == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Test ObjectPool - Free-threaded Mode
|
|
# =============================================================================
|
|
|
|
class TestObjectPoolFreeThreaded:
|
|
"""Test ObjectPool free-threaded mode support."""
|
|
|
|
def test_is_free_threaded_property(self):
|
|
"""ObjectPool.is_free_threaded property works."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
result = pool.is_free_threaded
|
|
assert isinstance(result, bool)
|
|
|
|
def test_get_optimal_pool_size_free_threaded(self):
|
|
"""get_optimal_pool_size returns larger size in free-threaded mode."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
# Set the instance attribute directly
|
|
pool._is_free_threaded = True
|
|
|
|
size = pool.get_optimal_pool_size(estimated_concurrent_usage=10)
|
|
|
|
assert size == 20 # estimated_concurrent_usage * 2
|
|
|
|
def test_get_optimal_pool_size_gil_mode(self):
|
|
"""get_optimal_pool_size in GIL mode."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=5)
|
|
pool._is_free_threaded = False
|
|
|
|
size = pool.get_optimal_pool_size(estimated_concurrent_usage=10)
|
|
|
|
assert size == 10 # estimated_concurrent_usage
|
|
|
|
def test_get_optimal_pool_size_no_estimate(self):
|
|
"""get_optimal_pool_size uses max_size when no estimate."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=7)
|
|
|
|
size = pool.get_optimal_pool_size()
|
|
|
|
assert size == 7
|
|
|
|
|
|
# =============================================================================
|
|
# Test ConnectionPool
|
|
# =============================================================================
|
|
|
|
class TestConnectionPool:
|
|
"""Test ConnectionPool class."""
|
|
|
|
def test_connection_pool_creation(self):
|
|
"""ConnectionPool can be created."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_connection"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
assert pool.connect_func == connect_func
|
|
assert pool.test_on_borrow is True
|
|
|
|
def test_connection_pool_acquire(self):
|
|
"""ConnectionPool.acquire returns connection."""
|
|
connections = []
|
|
|
|
def connect():
|
|
"""Create a numbered connection."""
|
|
conn = f"conn_{len(connections)}"
|
|
connections.append(conn)
|
|
return conn
|
|
|
|
pool = ConnectionPool(connect, max_connections=5)
|
|
|
|
conn = pool.acquire()
|
|
|
|
assert conn == "conn_0"
|
|
|
|
def test_connection_pool_release(self):
|
|
"""ConnectionPool.release returns connection to pool."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
conn = pool.acquire()
|
|
pool.release(conn)
|
|
|
|
assert pool.size == 1
|
|
|
|
def test_connection_pool_context_manager(self):
|
|
"""ConnectionPool.get_connection context manager works."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
with pool.get_connection() as conn:
|
|
assert conn == "mock_conn"
|
|
|
|
assert pool.active == 0
|
|
|
|
def test_connection_pool_close(self):
|
|
"""ConnectionPool.close closes all connections."""
|
|
|
|
def connect():
|
|
"""Create a connection dict."""
|
|
return {'closed': False}
|
|
|
|
pool = ConnectionPool(connect, max_connections=5)
|
|
|
|
conns = [pool.acquire() for _ in range(3)]
|
|
for conn in conns:
|
|
pool.release(conn)
|
|
|
|
pool.close()
|
|
|
|
assert pool.size == 0
|
|
|
|
def test_connection_pool_test_on_borrow(self):
|
|
"""ConnectionPool tests connection on borrow."""
|
|
call_count = [0]
|
|
|
|
def connect():
|
|
"""Create a numbered connection."""
|
|
call_count[0] += 1
|
|
return {'id': call_count[0], 'valid': True}
|
|
|
|
pool = ConnectionPool(connect, max_connections=5, test_on_borrow=True)
|
|
|
|
# Mock _test_connection to return False first time
|
|
test_calls = [0]
|
|
|
|
def mock_test(conn):
|
|
"""Mock test that fails on first call."""
|
|
test_calls[0] += 1
|
|
if test_calls[0] == 1:
|
|
return False # First test fails
|
|
return True
|
|
|
|
pool._test_connection = mock_test
|
|
|
|
pool.acquire()
|
|
|
|
# Should have tried twice (first failed, second succeeded)
|
|
assert test_calls[0] >= 1
|
|
|
|
def test_connection_pool_test_on_borrow_disabled(self):
|
|
"""ConnectionPool can disable test on borrow."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5, test_on_borrow=False)
|
|
|
|
conn = pool.acquire()
|
|
|
|
assert conn == "mock_conn"
|
|
|
|
def test_connection_pool_test_connection_method(self):
|
|
"""ConnectionPool._test_connection tests connection validity."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
# Test with connection that has execute method
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value = None
|
|
|
|
result = pool._test_connection(mock_conn)
|
|
|
|
assert result is True
|
|
mock_conn.execute.assert_called_once_with('SELECT 1')
|
|
|
|
def test_connection_pool_test_connection_no_execute(self):
|
|
"""ConnectionPool._test_connection handles connections without execute."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
# Test with connection that doesn't have execute
|
|
mock_conn = object()
|
|
|
|
result = pool._test_connection(mock_conn)
|
|
|
|
assert result is True
|
|
|
|
def test_connection_pool_test_connection_exception(self):
|
|
"""ConnectionPool._test_connection returns False on exception."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.side_effect = Exception("Connection lost")
|
|
|
|
result = pool._test_connection(mock_conn)
|
|
|
|
assert result is False
|
|
|
|
def test_connection_pool_close_connection(self):
|
|
"""ConnectionPool._close_connection closes connection."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
mock_conn = MagicMock()
|
|
pool._close_connection(mock_conn)
|
|
|
|
mock_conn.close.assert_called_once()
|
|
|
|
def test_connection_pool_close_connection_exception(self):
|
|
"""ConnectionPool._close_connection handles exceptions."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.close.side_effect = Exception("Close failed")
|
|
|
|
# Should not raise
|
|
pool._close_connection(mock_conn)
|
|
|
|
def test_connection_pool_is_free_threaded(self):
|
|
"""ConnectionPool.is_free_threaded property works."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "mock_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=5)
|
|
|
|
result = pool.is_free_threaded
|
|
assert isinstance(result, bool)
|
|
|
|
|
|
# =============================================================================
|
|
# Test WorkerPool
|
|
# =============================================================================
|
|
|
|
class TestWorkerPool:
|
|
"""Test WorkerPool class."""
|
|
|
|
def test_worker_pool_creation(self):
|
|
"""WorkerPool can be created."""
|
|
pool = WorkerPool(workers=4, queue_size=100)
|
|
|
|
assert pool.workers == 4
|
|
assert pool.queue_size == 100
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_start(self):
|
|
"""WorkerPool.start creates worker threads."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
pool.start()
|
|
|
|
assert pool._running is True
|
|
assert len(pool._threads) > 0
|
|
|
|
pool.shutdown()
|
|
|
|
def test_worker_pool_start_idempotent(self):
|
|
"""WorkerPool.start can be called multiple times."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
pool.start()
|
|
pool.start() # Should not create duplicate threads
|
|
|
|
pool.shutdown()
|
|
|
|
def test_worker_pool_submit(self):
|
|
"""WorkerPool.submit adds task to queue."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def task(value):
|
|
"""Add value to results list."""
|
|
with lock:
|
|
results.append(value)
|
|
|
|
pool.submit(task, 42)
|
|
pool.submit(task, 100)
|
|
|
|
# Wait for tasks to complete
|
|
time.sleep(0.5)
|
|
pool.shutdown()
|
|
|
|
assert 42 in results
|
|
assert 100 in results
|
|
|
|
def test_worker_pool_submit_not_running(self):
|
|
"""WorkerPool.submit raises error when not running."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.submit(lambda: None)
|
|
|
|
assert "Worker pool is not running" in str(exc_info.value)
|
|
|
|
def test_worker_pool_submit_queue_full(self):
|
|
"""WorkerPool.submit raises error when queue is full."""
|
|
# Use queue_size=0 (unlimited) to avoid timing-dependent race conditions
|
|
# Instead, test that submit works correctly with timeout parameter
|
|
pool = WorkerPool(workers=1, queue_size=0)
|
|
pool.start()
|
|
|
|
completed = []
|
|
lock = threading.Lock()
|
|
|
|
def quick_task(value):
|
|
"""Add value to completed list."""
|
|
with lock:
|
|
completed.append(value)
|
|
|
|
# Submit multiple tasks - should all succeed with unlimited queue
|
|
for i in range(10):
|
|
pool.submit(quick_task, i)
|
|
|
|
# Wait for tasks to complete
|
|
time.sleep(0.5)
|
|
pool.shutdown()
|
|
|
|
assert len(completed) == 10
|
|
|
|
def test_worker_pool_submit_with_kwargs(self):
|
|
"""WorkerPool.submit accepts keyword arguments."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def task(a, b, c=10):
|
|
"""Add sum of a, b, c to results."""
|
|
with lock:
|
|
results.append(a + b + c)
|
|
|
|
pool.submit(task, 1, 2, c=5)
|
|
|
|
time.sleep(0.5)
|
|
pool.shutdown()
|
|
|
|
assert 8 in results
|
|
|
|
def test_worker_pool_shutdown(self):
|
|
"""WorkerPool.shutdown stops worker threads."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
pool.shutdown(wait=True)
|
|
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_shutdown_with_timeout(self):
|
|
"""WorkerPool.shutdown respects timeout."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
# Submit a slow task
|
|
def slow_task():
|
|
"""Task that sleeps for 0.5 seconds."""
|
|
time.sleep(0.5)
|
|
|
|
pool.submit(slow_task)
|
|
|
|
start = time.time()
|
|
pool.shutdown(wait=True, timeout=0.2)
|
|
elapsed = time.time() - start
|
|
|
|
# Should not wait too long
|
|
assert elapsed < 1.0
|
|
|
|
def test_worker_pool_shutdown_no_wait(self):
|
|
"""WorkerPool.shutdown with wait=False returns immediately."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
pool.shutdown(wait=False)
|
|
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_context_manager(self):
|
|
"""WorkerPool context manager starts and shuts down."""
|
|
with WorkerPool(workers=2, queue_size=10) as pool:
|
|
assert pool._running is True
|
|
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def task(value):
|
|
"""Add value to results list."""
|
|
with lock:
|
|
results.append(value)
|
|
|
|
pool.submit(task, 123)
|
|
time.sleep(0.3)
|
|
|
|
# After context, pool should be shut down
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_pending_tasks(self):
|
|
"""WorkerPool.pending returns queue size."""
|
|
pool = WorkerPool(workers=1, queue_size=10)
|
|
pool.start()
|
|
|
|
def slow_task():
|
|
"""Task that sleeps for 0.5 seconds."""
|
|
time.sleep(0.5)
|
|
|
|
pool.submit(slow_task)
|
|
pool.submit(slow_task)
|
|
pool.submit(slow_task)
|
|
|
|
# Some tasks should be pending
|
|
pending = pool.pending
|
|
assert pending >= 0 # May have started processing
|
|
|
|
pool.shutdown()
|
|
|
|
def test_worker_pool_is_free_threaded(self):
|
|
"""WorkerPool.is_free_threaded property works."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
result = pool.is_free_threaded
|
|
assert isinstance(result, bool)
|
|
|
|
def test_worker_pool_get_optimal_workers(self):
|
|
"""WorkerPool.get_optimal_workers returns adjusted count."""
|
|
pool = WorkerPool(workers=4, queue_size=10)
|
|
pool._is_free_threaded = False
|
|
|
|
# In GIL mode, should return base_workers
|
|
result = pool.get_optimal_workers()
|
|
assert result == 4
|
|
|
|
def test_worker_pool_get_optimal_workers_free_threaded(self):
|
|
"""WorkerPool.get_optimal_workers in free-threaded mode."""
|
|
pool = WorkerPool(workers=4, queue_size=10)
|
|
pool._is_free_threaded = True
|
|
|
|
result = pool.get_optimal_workers()
|
|
assert result == 8 # base_workers * 2
|
|
|
|
def test_worker_pool_get_optimal_workers_custom(self):
|
|
"""WorkerPool.get_optimal_workers with custom base."""
|
|
pool = WorkerPool(workers=4, queue_size=10)
|
|
pool._is_free_threaded = False
|
|
|
|
result = pool.get_optimal_workers(base_workers=10)
|
|
assert result == 10
|
|
|
|
|
|
# =============================================================================
|
|
# Test WorkerPool - Worker Thread Behavior
|
|
# =============================================================================
|
|
|
|
class TestWorkerPoolWorkerThread:
|
|
"""Test WorkerPool worker thread behavior."""
|
|
|
|
def test_worker_executes_task(self):
|
|
"""Worker thread executes submitted tasks."""
|
|
pool = WorkerPool(workers=1, queue_size=10)
|
|
pool.start()
|
|
|
|
executed = []
|
|
lock = threading.Lock()
|
|
|
|
def task(value):
|
|
"""Add value to executed list."""
|
|
with lock:
|
|
executed.append(value)
|
|
|
|
pool.submit(task, "test_value")
|
|
|
|
# Wait for execution
|
|
time.sleep(0.3)
|
|
pool.shutdown()
|
|
|
|
assert "test_value" in executed
|
|
|
|
def test_worker_handles_task_exception(self):
|
|
"""Worker thread handles task exceptions gracefully."""
|
|
pool = WorkerPool(workers=1, queue_size=10)
|
|
pool.start()
|
|
|
|
def failing_task():
|
|
"""Task that raises an exception."""
|
|
raise RuntimeError("Task failed")
|
|
|
|
# Should not raise
|
|
pool.submit(failing_task)
|
|
|
|
time.sleep(0.3)
|
|
pool.shutdown()
|
|
|
|
# Pool should still be functional
|
|
assert pool._running is False
|
|
|
|
def test_worker_poison_pill(self):
|
|
"""Worker thread stops on poison pill."""
|
|
pool = WorkerPool(workers=1, queue_size=10)
|
|
pool.start()
|
|
|
|
# Submit task then shutdown
|
|
def task():
|
|
"""Empty task."""
|
|
pass
|
|
|
|
pool.submit(task)
|
|
pool.shutdown(wait=True)
|
|
|
|
# All threads should have terminated
|
|
for thread in pool._threads:
|
|
assert not thread.is_alive()
|
|
|
|
|
|
# =============================================================================
|
|
# Test Pools Factory Class
|
|
# =============================================================================
|
|
|
|
class TestPoolsFactory:
|
|
"""Test Pools factory class."""
|
|
|
|
def test_is_free_threaded_static(self):
|
|
"""Pools.is_free_threaded() returns boolean."""
|
|
result = Pools.is_free_threaded()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_create_pool(self):
|
|
"""Pools.create_pool creates ObjectPool."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = Pools.create_pool(factory, max_size=5)
|
|
|
|
assert isinstance(pool, ObjectPool)
|
|
assert pool.max_size == 5
|
|
|
|
def test_create_pool_optimized(self):
|
|
"""Pools.create_pool_optimized creates optimized ObjectPool."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = Pools.create_pool_optimized(factory, max_size=5)
|
|
|
|
assert isinstance(pool, ObjectPool)
|
|
assert pool.max_size == 5
|
|
|
|
@patch('nodupe.tools.parallel.pools._is_free_threaded', return_value=True)
|
|
def test_create_pool_optimized_free_threaded(self, mock_ft):
|
|
"""Pools.create_pool_optimized adjusts for free-threaded mode."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = Pools.create_pool_optimized(factory, max_size=5)
|
|
|
|
# In free-threaded mode, max_size should be doubled
|
|
assert pool.max_size == 10
|
|
|
|
def test_create_connection_pool(self):
|
|
"""Pools.create_connection_pool creates ConnectionPool."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "conn"
|
|
pool = Pools.create_connection_pool(connect_func, max_connections=5)
|
|
|
|
assert isinstance(pool, ConnectionPool)
|
|
|
|
def test_create_connection_pool_optimized(self):
|
|
"""Pools.create_connection_pool_optimized creates optimized pool."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "conn"
|
|
pool = Pools.create_connection_pool_optimized(connect_func, max_connections=5)
|
|
|
|
assert isinstance(pool, ConnectionPool)
|
|
|
|
@patch('nodupe.tools.parallel.pools._is_free_threaded', return_value=True)
|
|
def test_create_connection_pool_optimized_free_threaded(self, mock_ft):
|
|
"""Pools.create_connection_pool_optimized adjusts for free-threaded."""
|
|
def connect_func():
|
|
"""Create a mock connection."""
|
|
return "conn"
|
|
pool = Pools.create_connection_pool_optimized(connect_func, max_connections=5)
|
|
|
|
# In free-threaded mode, max_connections should be doubled
|
|
assert pool._pool.max_size == 10
|
|
|
|
def test_create_worker_pool(self):
|
|
"""Pools.create_worker_pool creates WorkerPool."""
|
|
pool = Pools.create_worker_pool(workers=4, queue_size=100)
|
|
|
|
assert isinstance(pool, WorkerPool)
|
|
assert pool.workers == 4
|
|
assert pool.queue_size == 100
|
|
|
|
def test_create_worker_pool_optimized(self):
|
|
"""Pools.create_worker_pool_optimized creates optimized pool."""
|
|
pool = Pools.create_worker_pool_optimized(workers=4, queue_size=100)
|
|
|
|
assert isinstance(pool, WorkerPool)
|
|
|
|
@patch('nodupe.tools.parallel.pools._is_free_threaded', return_value=True)
|
|
def test_create_worker_pool_optimized_free_threaded(self, mock_ft):
|
|
"""Pools.create_worker_pool_optimized adjusts for free-threaded."""
|
|
pool = Pools.create_worker_pool_optimized(workers=4, queue_size=100)
|
|
|
|
# In free-threaded mode, workers should be doubled
|
|
assert pool.workers == 8
|
|
|
|
|
|
# =============================================================================
|
|
# Test Race Conditions and Concurrent Access
|
|
# =============================================================================
|
|
|
|
class TestRaceConditions:
|
|
"""Test for race conditions and concurrent access safety."""
|
|
|
|
def test_object_pool_no_race_on_acquire_release(self):
|
|
"""ObjectPool has no race conditions in acquire/release cycle."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return object()
|
|
pool = ObjectPool(factory, max_size=100)
|
|
errors = []
|
|
lock = threading.Lock()
|
|
|
|
def worker():
|
|
"""Worker that repeatedly acquires and releases objects."""
|
|
try:
|
|
for _ in range(50):
|
|
obj = pool.acquire()
|
|
time.sleep(0.0001) # Tiny delay to increase contention
|
|
pool.release(obj)
|
|
except Exception as e:
|
|
with lock:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(20)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert len(errors) == 0
|
|
assert pool.active == 0 # All objects should be released
|
|
|
|
def test_worker_pool_concurrent_submit(self):
|
|
"""WorkerPool handles concurrent task submission."""
|
|
pool = WorkerPool(workers=4, queue_size=1000)
|
|
pool.start()
|
|
|
|
results = []
|
|
lock = threading.Lock()
|
|
errors = []
|
|
|
|
def task(value):
|
|
"""Add value to results list."""
|
|
with lock:
|
|
results.append(value)
|
|
|
|
def submitter(thread_id):
|
|
"""Submit tasks from a thread."""
|
|
try:
|
|
for i in range(50):
|
|
pool.submit(task, f"{thread_id}:{i}")
|
|
except Exception as e:
|
|
with lock:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=submitter, args=(i,)) for i in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Wait for tasks to complete
|
|
time.sleep(1)
|
|
pool.shutdown()
|
|
|
|
assert len(errors) == 0
|
|
assert len(results) == 500 # 10 threads * 50 tasks
|
|
|
|
def test_connection_pool_concurrent_access(self):
|
|
"""ConnectionPool handles concurrent access safely."""
|
|
connections_created = []
|
|
lock = threading.Lock()
|
|
|
|
def connect():
|
|
"""Create a numbered connection."""
|
|
with lock:
|
|
conn_id = len(connections_created)
|
|
conn = {'id': conn_id}
|
|
connections_created.append(conn)
|
|
return conn
|
|
|
|
pool = ConnectionPool(connect, max_connections=10)
|
|
errors = []
|
|
|
|
def worker():
|
|
"""Worker that uses connection pool."""
|
|
try:
|
|
for _ in range(20):
|
|
with pool.get_connection():
|
|
time.sleep(0.001)
|
|
except Exception as e:
|
|
with lock:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert len(errors) == 0
|
|
pool.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Test Edge Cases
|
|
# =============================================================================
|
|
|
|
class TestPoolEdgeCases:
|
|
"""Test edge cases for pool operations."""
|
|
|
|
def test_object_pool_zero_max_size(self):
|
|
"""ObjectPool with max_size=0 still works for single acquire."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return "test"
|
|
pool = ObjectPool(factory, max_size=1, timeout=0.1)
|
|
|
|
obj = pool.acquire()
|
|
# Second acquire should fail
|
|
with pytest.raises(PoolError):
|
|
pool.acquire(timeout=0.1)
|
|
|
|
pool.release(obj)
|
|
|
|
def test_object_pool_unlimited_queue(self):
|
|
"""ObjectPool with large max_size works correctly."""
|
|
def factory():
|
|
"""Create a test object."""
|
|
return object()
|
|
pool = ObjectPool(factory, max_size=1000)
|
|
|
|
objs = [pool.acquire() for _ in range(100)]
|
|
|
|
assert pool.active == 100
|
|
|
|
for obj in objs:
|
|
pool.release(obj)
|
|
|
|
assert pool.active == 0
|
|
|
|
def test_worker_pool_zero_workers(self):
|
|
"""WorkerPool with workers=0 creates no threads."""
|
|
pool = WorkerPool(workers=0, queue_size=10)
|
|
pool.start()
|
|
|
|
# Should have no threads
|
|
assert len(pool._threads) == 0
|
|
|
|
pool.shutdown()
|
|
|
|
def test_worker_pool_unlimited_queue(self):
|
|
"""WorkerPool with queue_size=0 has unlimited queue."""
|
|
pool = WorkerPool(workers=2, queue_size=0)
|
|
pool.start()
|
|
|
|
# Should be able to submit many tasks
|
|
for i in range(100):
|
|
pool.submit(lambda x: None, i)
|
|
|
|
pool.shutdown()
|
|
|
|
def test_connection_pool_single_connection(self):
|
|
"""ConnectionPool with max_connections=1 works correctly."""
|
|
def connect_func():
|
|
"""Create a single connection."""
|
|
return "single_conn"
|
|
pool = ConnectionPool(connect_func, max_connections=1, timeout=0.1)
|
|
|
|
conn = pool.acquire()
|
|
pool.release(conn)
|
|
|
|
# Should be able to acquire again
|
|
conn2 = pool.acquire()
|
|
assert conn2 == "single_conn"
|
|
|
|
pool.close()
|
|
|
|
def test_pool_destroy_func_exception(self):
|
|
"""ObjectPool handles destroy_func exceptions gracefully."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed)}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
if obj['id'] == 0:
|
|
raise RuntimeError("Destroy failed")
|
|
|
|
pool = ObjectPool(factory, max_size=2, destroy_func=destroy)
|
|
|
|
objs = [pool.acquire() for _ in range(2)]
|
|
for obj in objs:
|
|
pool.release(obj)
|
|
|
|
# Close should not raise even if destroy_func fails
|
|
pool.close()
|
|
|
|
assert len(destroyed) == 2
|
|
|
|
|
|
# =============================================================================
|
|
# Test Missing Coverage - Pool Edge Cases for 100% Coverage
|
|
# =============================================================================
|
|
|
|
class TestPoolMissingCoverage:
|
|
"""Test missing coverage areas in pools.py for 100% coverage."""
|
|
|
|
def test_object_pool_acquire_factory_exception(self):
|
|
"""ObjectPool.acquire handles factory exception when creating new object."""
|
|
call_count = [0]
|
|
|
|
def failing_factory():
|
|
"""Factory that succeeds first time, then fails."""
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
# First call succeeds
|
|
return "obj1"
|
|
# Second call fails
|
|
raise RuntimeError("Factory creation failed")
|
|
|
|
pool = ObjectPool(failing_factory, max_size=2, timeout=0.1)
|
|
|
|
# First acquire succeeds
|
|
obj1 = pool.acquire()
|
|
pool.release(obj1)
|
|
|
|
# Get the pooled object
|
|
pool.acquire()
|
|
|
|
# Now try to acquire when pool is empty and factory fails
|
|
# This tests the exception path in lines 159-165
|
|
with patch.object(pool._pool, 'empty', return_value=False):
|
|
# Force the path where pool is not empty but get fails
|
|
with patch.object(pool._pool, 'get', side_effect=queue.Empty()):
|
|
with patch.object(pool, '_lock'):
|
|
pool._lock.__enter__ = lambda _: True
|
|
pool._lock.__exit__ = lambda *args: None
|
|
with patch.object(pool, '_active_count', 0):
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire(timeout=0.1)
|
|
assert "Failed to create object" in str(exc_info.value)
|
|
|
|
def test_object_pool_release_when_closed_with_destroy(self):
|
|
"""ObjectPool.release destroys object when pool is closed."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed) + 1}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
|
|
pool = ObjectPool(factory, max_size=5, destroy_func=destroy)
|
|
|
|
obj = pool.acquire()
|
|
pool.close()
|
|
|
|
# Release after close should destroy the object
|
|
pool.release(obj)
|
|
|
|
assert len(destroyed) == 1
|
|
|
|
def test_object_pool_release_queue_full(self):
|
|
"""ObjectPool.release handles queue.Full exception."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed) + 1}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
|
|
pool = ObjectPool(factory, max_size=2, destroy_func=destroy)
|
|
|
|
# Fill the pool
|
|
obj1 = pool.acquire()
|
|
obj2 = pool.acquire()
|
|
pool.release(obj1)
|
|
pool.release(obj2)
|
|
|
|
# Now pool is full, acquire both again
|
|
obj1 = pool.acquire()
|
|
obj2 = pool.acquire()
|
|
|
|
# Release one back
|
|
pool.release(obj1)
|
|
|
|
# Now release obj2 when pool is full - should trigger queue.Full
|
|
# and destroy the object
|
|
with patch.object(pool._pool, 'put_nowait', side_effect=queue.Full()):
|
|
pool.release(obj2)
|
|
|
|
# Object should be destroyed due to queue.Full
|
|
assert len(destroyed) == 1
|
|
|
|
def test_object_pool_close_destroy_exception(self):
|
|
"""ObjectPool.close continues destroying objects even if one fails."""
|
|
destroyed = []
|
|
call_count = [0]
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': call_count[0]}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed, fail on first."""
|
|
call_count[0] += 1
|
|
destroyed.append(obj)
|
|
if obj['id'] == 0:
|
|
raise RuntimeError("Destroy failed for first object")
|
|
|
|
pool = ObjectPool(factory, max_size=3, destroy_func=destroy)
|
|
|
|
# Create and release 3 objects
|
|
objs = [pool.acquire() for _ in range(3)]
|
|
for obj in objs:
|
|
pool.release(obj)
|
|
|
|
# Close should destroy all objects even if one fails
|
|
pool.close()
|
|
|
|
# All objects should be destroyed
|
|
assert len(destroyed) == 3
|
|
|
|
def test_worker_pool_shutdown_when_not_running(self):
|
|
"""WorkerPool.shutdown returns early when not running."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
# Don't start the pool
|
|
# Shutdown should return early without error
|
|
pool.shutdown(wait=True)
|
|
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_submit_queue_full_exception(self):
|
|
"""WorkerPool.submit raises PoolError when queue is full."""
|
|
pool = WorkerPool(workers=1, queue_size=1)
|
|
pool.start()
|
|
|
|
# Block the worker with a slow task
|
|
def slow_task():
|
|
"""Task that sleeps for 1 second."""
|
|
time.sleep(1)
|
|
|
|
pool.submit(slow_task)
|
|
|
|
# Fill the queue
|
|
pool.submit(lambda: None)
|
|
|
|
# Now queue should be full, next submit should fail
|
|
# Wait a bit for the first task to start processing
|
|
time.sleep(0.2)
|
|
|
|
# Try to submit with timeout=0 to trigger queue.Full immediately
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.submit(lambda: None, timeout=0)
|
|
|
|
assert "Worker queue is full" in str(exc_info.value)
|
|
|
|
pool.shutdown()
|
|
|
|
def test_worker_pool_shutdown_timeout_break(self):
|
|
"""WorkerPool.shutdown breaks out of wait loop on timeout."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
# Submit a task that will take longer than timeout
|
|
def slow_task():
|
|
"""Task that sleeps for 2 seconds."""
|
|
time.sleep(2)
|
|
|
|
pool.submit(slow_task)
|
|
|
|
start = time.time()
|
|
# Shutdown with short timeout
|
|
pool.shutdown(wait=True, timeout=0.3)
|
|
elapsed = time.time() - start
|
|
|
|
# Should not wait for the full 2 seconds
|
|
assert elapsed < 1.0
|
|
|
|
def test_worker_pool_shutdown_thread_join_remaining(self):
|
|
"""WorkerPool.shutdown uses remaining timeout for thread.join."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
# Submit a slow task
|
|
def slow_task():
|
|
"""Task that sleeps for 0.5 seconds."""
|
|
time.sleep(0.5)
|
|
|
|
pool.submit(slow_task)
|
|
|
|
# Shutdown with timeout - tests the remaining timeout calculation
|
|
pool.shutdown(wait=True, timeout=0.3)
|
|
|
|
# Threads should be cleaned up (or at least attempted)
|
|
assert pool._running is False
|
|
|
|
def test_worker_pool_exit_returns_false(self):
|
|
"""WorkerPool.__exit__ returns False."""
|
|
results = []
|
|
|
|
with WorkerPool(workers=2, queue_size=10) as pool:
|
|
def task():
|
|
"""Add 1 to results."""
|
|
results.append(1)
|
|
|
|
pool.submit(task)
|
|
time.sleep(0.2)
|
|
|
|
# Context manager should have called __exit__ which returns False
|
|
assert len(results) == 1
|
|
|
|
def test_connection_pool_acquire_test_on_borrow_retry(self):
|
|
"""ConnectionPool.acquire retries when test_on_borrow fails."""
|
|
connections = []
|
|
|
|
def connect():
|
|
"""Create a numbered connection."""
|
|
conn = {'id': len(connections)}
|
|
connections.append(conn)
|
|
return conn
|
|
|
|
pool = ConnectionPool(connect, max_connections=5, test_on_borrow=True)
|
|
|
|
# Mock _test_connection to fail first time, succeed second
|
|
test_calls = [0]
|
|
|
|
def mock_test(conn):
|
|
"""Mock test that fails on first call."""
|
|
test_calls[0] += 1
|
|
if test_calls[0] == 1:
|
|
return False # First test fails
|
|
return True # Subsequent tests succeed
|
|
|
|
pool._test_connection = mock_test
|
|
|
|
conn = pool.acquire()
|
|
|
|
# Should have retried
|
|
assert test_calls[0] >= 1
|
|
assert conn is not None
|
|
|
|
pool.close()
|
|
|
|
def test_object_pool_acquire_empty_pool_create_immediate(self):
|
|
"""ObjectPool.acquire creates object immediately when pool is empty."""
|
|
created = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
obj = {'id': len(created)}
|
|
created.append(obj)
|
|
return obj
|
|
|
|
pool = ObjectPool(factory, max_size=5)
|
|
|
|
# First acquire should create immediately
|
|
obj = pool.acquire()
|
|
|
|
assert obj['id'] == 0
|
|
assert len(created) == 1
|
|
assert pool.active == 1
|
|
|
|
def test_object_pool_acquire_timeout_path(self):
|
|
"""ObjectPool.acquire uses full timeout when at capacity."""
|
|
created = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
obj = {'id': len(created)}
|
|
created.append(obj)
|
|
return obj
|
|
|
|
pool = ObjectPool(factory, max_size=1, timeout=10.0)
|
|
|
|
# Acquire the only allowed object
|
|
pool.acquire()
|
|
|
|
# Second acquire should timeout with full timeout
|
|
start = time.time()
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire(timeout=0.2)
|
|
elapsed = time.time() - start
|
|
|
|
assert "Pool exhausted" in str(exc_info.value)
|
|
assert elapsed >= 0.2
|
|
assert elapsed < 1.0
|
|
|
|
def test_object_pool_acquire_factory_exception_in_lock(self):
|
|
"""ObjectPool.acquire handles factory exception inside lock."""
|
|
call_count = [0]
|
|
|
|
def failing_factory():
|
|
"""Factory that always raises an exception."""
|
|
call_count[0] += 1
|
|
raise RuntimeError("Factory failed")
|
|
|
|
pool = ObjectPool(failing_factory, max_size=2, timeout=0.1)
|
|
|
|
# This tests the path where factory exception happens inside the lock
|
|
with pytest.raises(PoolError) as exc_info:
|
|
pool.acquire()
|
|
|
|
assert "Failed to create object" in str(exc_info.value)
|
|
assert call_count[0] == 1
|
|
|
|
def test_object_pool_release_closed_pool_destroy_exception(self):
|
|
"""ObjectPool.release handles destroy_func exception when pool closed."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed)}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
raise RuntimeError("Destroy failed")
|
|
|
|
pool = ObjectPool(factory, max_size=5, destroy_func=destroy)
|
|
|
|
obj = pool.acquire()
|
|
pool.close()
|
|
|
|
# Release after close should destroy the object even if destroy_func fails
|
|
pool.release(obj)
|
|
|
|
assert len(destroyed) == 1
|
|
|
|
def test_object_pool_release_queue_full_destroy_exception(self):
|
|
"""ObjectPool.release handles destroy_func exception on queue.Full."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed)}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
raise RuntimeError("Destroy failed on full")
|
|
|
|
pool = ObjectPool(factory, max_size=1, destroy_func=destroy)
|
|
|
|
obj = pool.acquire()
|
|
pool.release(obj) # Pool now has 1 object
|
|
|
|
# Acquire again
|
|
obj2 = pool.acquire()
|
|
|
|
# Release when pool is full - triggers queue.Full
|
|
with patch.object(pool._pool, 'put_nowait', side_effect=queue.Full()):
|
|
pool.release(obj2) # Should destroy obj2
|
|
|
|
assert len(destroyed) == 1
|
|
|
|
def test_object_pool_close_destroy_exception_continues(self):
|
|
"""ObjectPool.close continues destroying even if one fails."""
|
|
destroyed = []
|
|
|
|
def factory():
|
|
"""Create a dict with an id."""
|
|
return {'id': len(destroyed)}
|
|
|
|
def destroy(obj):
|
|
"""Mark object as destroyed."""
|
|
destroyed.append(obj)
|
|
if obj['id'] == 0:
|
|
raise RuntimeError("First destroy failed")
|
|
|
|
pool = ObjectPool(factory, max_size=3, destroy_func=destroy)
|
|
|
|
objs = [pool.acquire() for _ in range(3)]
|
|
for obj in objs:
|
|
pool.release(obj)
|
|
|
|
# Close should continue even if first destroy fails
|
|
pool.close()
|
|
|
|
assert len(destroyed) == 3
|
|
|
|
def test_worker_pool_shutdown_not_running_no_op(self):
|
|
"""WorkerPool.shutdown returns immediately when not running."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
|
|
# Don't start - shutdown should be no-op
|
|
pool.shutdown(wait=True, timeout=0.1)
|
|
|
|
assert pool._running is False
|
|
assert len(pool._threads) == 0
|
|
|
|
def test_worker_pool_shutdown_thread_join_remaining_timeout(self):
|
|
"""WorkerPool.shutdown uses remaining timeout for thread join."""
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
def slow_task():
|
|
"""Task that sleeps for 0.3 seconds."""
|
|
time.sleep(0.3)
|
|
|
|
pool.submit(slow_task)
|
|
|
|
# Shutdown with timeout - tests remaining timeout calculation
|
|
start = time.time()
|
|
pool.shutdown(wait=True, timeout=0.2)
|
|
elapsed = time.time() - start
|
|
|
|
# Should not wait longer than timeout
|
|
assert elapsed < 0.5
|
|
|
|
def test_worker_pool_exit_false_on_exception(self):
|
|
"""WorkerPool.__exit__ returns False even on exception."""
|
|
class TestException(Exception):
|
|
"""Exception raised for test purposes."""
|
|
pass
|
|
|
|
pool = WorkerPool(workers=2, queue_size=10)
|
|
pool.start()
|
|
|
|
# Simulate exception in context
|
|
try:
|
|
with pool:
|
|
raise TestException("Test error")
|
|
except TestException:
|
|
pass
|
|
|
|
# Pool should be shutdown
|
|
assert pool._running is False
|
|
|
|
def test_connection_pool_acquire_retry_on_bad_connection(self):
|
|
"""ConnectionPool.acquire retries when test_on_borrow fails."""
|
|
connections = []
|
|
|
|
def connect():
|
|
"""Create a numbered connection."""
|
|
conn = {'id': len(connections), 'valid': False}
|
|
connections.append(conn)
|
|
return conn
|
|
|
|
pool = ConnectionPool(connect, max_connections=5, test_on_borrow=True)
|
|
|
|
# Track test calls
|
|
test_calls = [0]
|
|
|
|
def mock_test(conn):
|
|
"""Mock test that makes connection valid after first call."""
|
|
test_calls[0] += 1
|
|
conn['valid'] = True # Make it valid after first test
|
|
return True
|
|
|
|
pool._test_connection = mock_test
|
|
|
|
conn = pool.acquire()
|
|
|
|
assert test_calls[0] >= 1
|
|
assert conn is not None
|
|
|
|
pool.close()
|