diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d09f635e..b6012bd6 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -65,10 +65,10 @@ jobs: pip install -r requirements.txt pip install -r requirements-dev.txt - - name: Install activerecord and testsuite from PyPI + - name: Install activerecord and testsuite from git run: | - pip install rhosocial-activerecord==1.0.0.dev21 - pip install rhosocial-activerecord-testsuite==1.0.0.dev8 + pip install git+https://github.com/rhosocial/python-activerecord@release/v1.0.0.dev22 + pip install git+https://github.com/rhosocial/python-activerecord-testsuite@release/v1.0.0.dev9 - name: Create postgres scenario config run: | diff --git a/changelog.d/19.added.md b/changelog.d/19.added.md new file mode 100644 index 00000000..0bf77e96 --- /dev/null +++ b/changelog.d/19.added.md @@ -0,0 +1 @@ +Added connection pool context awareness support for PostgreSQL backend, enabling proper connection management across sync/async contexts and transaction boundaries. diff --git a/tests/providers/basic_connection.py b/tests/providers/basic_connection.py new file mode 100644 index 00000000..c28dcb75 --- /dev/null +++ b/tests/providers/basic_connection.py @@ -0,0 +1,161 @@ +# tests/providers/basic_connection.py +""" +Concrete implementation of IBasicConnectionProvider for PostgreSQL backend. + +This provider sets up connection pools and models for testing +ActiveRecord context awareness. +""" +from typing import Type, Tuple, Optional, List + +from rhosocial.activerecord.model import ActiveRecord, AsyncActiveRecord +from rhosocial.activerecord.backend.impl.postgres import PostgresBackend +from rhosocial.activerecord.backend.impl.postgres.config import PostgresConnectionConfig +from rhosocial.activerecord.connection.pool import BackendPool, AsyncBackendPool, PoolConfig +from rhosocial.activerecord.backend.options import ExecutionOptions +from rhosocial.activerecord.backend.schema import StatementType + +from rhosocial.activerecord.testsuite.feature.basic.connection.interfaces import IBasicConnectionProvider +from .scenarios import get_scenario, get_enabled_scenarios + + +class SyncTestUser(ActiveRecord): + """Sync test user model for connection pool tests.""" + __table_name__ = "test_users" + id: Optional[int] = None + name: str + email: str + + +class AsyncTestUser(AsyncActiveRecord): + """Async test user model for connection pool tests.""" + __table_name__ = "test_users" + id: Optional[int] = None + name: str + email: str + + +class BasicConnectionProvider(IBasicConnectionProvider): + """ + PostgreSQL backend implementation for basic connection pool context tests. + """ + + def __init__(self): + self._active_backends: List = [] + self._active_async_backends: List = [] + + def get_test_scenarios(self) -> list: + """Returns available test scenarios.""" + return list(get_enabled_scenarios().keys()) + + def _create_test_table(self, backend): + """Create the test_users table.""" + try: + backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL)) + except Exception: + pass + + backend.execute(""" + CREATE TABLE test_users ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + email VARCHAR(255) NOT NULL + ) + """, options=ExecutionOptions(stmt_type=StatementType.DDL)) + + async def _create_test_table_async(self, backend): + """Create the test_users table asynchronously.""" + try: + await backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL)) + except Exception: + pass + + await backend.execute(""" + CREATE TABLE test_users ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + email VARCHAR(255) NOT NULL + ) + """, options=ExecutionOptions(stmt_type=StatementType.DDL)) + + def setup_sync_pool_and_model(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]: + """Setup sync connection pool and model for context tests.""" + _, config = get_scenario(scenario_name) + + # Create connection pool + pool_config = PoolConfig( + min_size=1, + max_size=5, + backend_factory=lambda: PostgresBackend(connection_config=config) + ) + pool = BackendPool.create(pool_config) + + # Create table + with pool.connection() as backend: + self._create_test_table(backend) + self._active_backends.append(backend) + + # Configure model to use a separate backend (not pool backend) + # This tests that model.backend() returns pool context backend + SyncTestUser.configure(config, PostgresBackend) + self._active_backends.append(SyncTestUser.__backend__) + + return pool, SyncTestUser + + async def setup_async_pool_and_model(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]: + """Setup async connection pool and model for context tests.""" + from rhosocial.activerecord.backend.impl.postgres import AsyncPostgresBackend + + _, config = get_scenario(scenario_name) + + # Create connection pool + pool_config = PoolConfig( + min_size=1, + max_size=5, + backend_factory=lambda: AsyncPostgresBackend(connection_config=config) + ) + + # Create pool + pool = await AsyncBackendPool.create(pool_config) + + # Create table + async with pool.connection() as backend: + await self._create_test_table_async(backend) + self._active_async_backends.append(backend) + + # Configure model + await AsyncTestUser.configure(config, AsyncPostgresBackend) + self._active_async_backends.append(AsyncTestUser.__backend__) + + return pool, AsyncTestUser + + def cleanup_sync(self, scenario_name: str, pool: BackendPool): + """Cleanup after sync tests.""" + pool.close(timeout=1.0) + + # Disconnect backends + for backend in self._active_backends: + try: + backend.disconnect() + except Exception: + pass + self._active_backends.clear() + + async def cleanup_async(self, scenario_name: str, pool: AsyncBackendPool): + """Cleanup after async tests.""" + await pool.close(timeout=1.0) + + # Disconnect backends + for backend in self._active_async_backends: + try: + await backend.disconnect() + except Exception: + pass + self._active_async_backends.clear() + + def setup_sync_pool_for_crud(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]: + """Setup sync connection pool for CRUD tests.""" + return self.setup_sync_pool_and_model(scenario_name) + + async def setup_async_pool_for_crud(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]: + """Setup async connection pool for CRUD tests.""" + return await self.setup_async_pool_and_model(scenario_name) diff --git a/tests/providers/query_connection.py b/tests/providers/query_connection.py new file mode 100644 index 00000000..985d390e --- /dev/null +++ b/tests/providers/query_connection.py @@ -0,0 +1,153 @@ +# tests/providers/query_connection.py +""" +Concrete implementation of IQueryConnectionProvider for PostgreSQL backend. + +This provider sets up connection pools and models for testing +query classes context awareness (ActiveQuery, CTEQuery, SetOperationQuery). +""" +from typing import Type, Tuple, Optional, List + +from rhosocial.activerecord.model import ActiveRecord, AsyncActiveRecord +from rhosocial.activerecord.backend.impl.postgres import PostgresBackend +from rhosocial.activerecord.backend.impl.postgres.config import PostgresConnectionConfig +from rhosocial.activerecord.connection.pool import BackendPool, AsyncBackendPool, PoolConfig +from rhosocial.activerecord.backend.options import ExecutionOptions +from rhosocial.activerecord.backend.schema import StatementType + +from rhosocial.activerecord.testsuite.feature.query.connection.interfaces import IQueryConnectionProvider +from .scenarios import get_scenario, get_enabled_scenarios + + +class SyncQueryTestUser(ActiveRecord): + """Sync test user model for query connection pool tests.""" + __table_name__ = "test_users" + id: Optional[int] = None + name: str + email: str + + +class AsyncQueryTestUser(AsyncActiveRecord): + """Async test user model for query connection pool tests.""" + __table_name__ = "test_users" + id: Optional[int] = None + name: str + email: str + + +class QueryConnectionProvider(IQueryConnectionProvider): + """ + PostgreSQL backend implementation for query connection pool context tests. + """ + + def __init__(self): + self._active_backends: List = [] + self._active_async_backends: List = [] + + def get_test_scenarios(self) -> list: + """Returns available test scenarios.""" + return list(get_enabled_scenarios().keys()) + + def _create_test_table(self, backend): + """Create the test_users table.""" + try: + backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL)) + except Exception: + pass + + backend.execute(""" + CREATE TABLE test_users ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + email VARCHAR(255) NOT NULL + ) + """, options=ExecutionOptions(stmt_type=StatementType.DDL)) + + async def _create_test_table_async(self, backend): + """Create the test_users table asynchronously.""" + try: + await backend.execute("DROP TABLE IF EXISTS test_users", options=ExecutionOptions(stmt_type=StatementType.DDL)) + except Exception: + pass + + await backend.execute(""" + CREATE TABLE test_users ( + id SERIAL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + email VARCHAR(255) NOT NULL + ) + """, options=ExecutionOptions(stmt_type=StatementType.DDL)) + + def setup_sync_pool_and_model(self, scenario_name: str) -> Tuple[BackendPool, Type[ActiveRecord]]: + """Setup sync connection pool and model for query context tests.""" + _, config = get_scenario(scenario_name) + + # Create connection pool + pool_config = PoolConfig( + min_size=1, + max_size=5, + backend_factory=lambda: PostgresBackend(connection_config=config) + ) + pool = BackendPool.create(pool_config) + + # Create table + with pool.connection() as backend: + self._create_test_table(backend) + self._active_backends.append(backend) + + # Configure model to use a separate backend (not pool backend) + # This tests that model.backend() returns pool context backend + SyncQueryTestUser.configure(config, PostgresBackend) + self._active_backends.append(SyncQueryTestUser.__backend__) + + return pool, SyncQueryTestUser + + async def setup_async_pool_and_model(self, scenario_name: str) -> Tuple[AsyncBackendPool, Type[AsyncActiveRecord]]: + """Setup async connection pool and model for query context tests.""" + from rhosocial.activerecord.backend.impl.postgres import AsyncPostgresBackend + + _, config = get_scenario(scenario_name) + + # Create connection pool + pool_config = PoolConfig( + min_size=1, + max_size=5, + backend_factory=lambda: AsyncPostgresBackend(connection_config=config) + ) + + # Create pool + pool = await AsyncBackendPool.create(pool_config) + + # Create table + async with pool.connection() as backend: + await self._create_test_table_async(backend) + self._active_async_backends.append(backend) + + # Configure model + await AsyncQueryTestUser.configure(config, AsyncPostgresBackend) + self._active_async_backends.append(AsyncQueryTestUser.__backend__) + + return pool, AsyncQueryTestUser + + def cleanup_sync(self, scenario_name: str, pool: BackendPool): + """Cleanup after sync tests.""" + pool.close(timeout=1.0) + + # Disconnect backends + for backend in self._active_backends: + try: + backend.disconnect() + except Exception: + pass + self._active_backends.clear() + + async def cleanup_async(self, scenario_name: str, pool: AsyncBackendPool): + """Cleanup after async tests.""" + await pool.close(timeout=1.0) + + # Disconnect backends + for backend in self._active_async_backends: + try: + await backend.disconnect() + except Exception: + pass + self._active_async_backends.clear() diff --git a/tests/providers/registry.py b/tests/providers/registry.py index 488ce965..69a1c98f 100644 --- a/tests/providers/registry.py +++ b/tests/providers/registry.py @@ -10,6 +10,8 @@ from .events import EventsProvider from .mixins import MixinsProvider from .query import QueryProvider +from .basic_connection import BasicConnectionProvider +from .query_connection import QueryConnectionProvider # Create a single, global instance of the ProviderRegistry for this backend. provider_registry = ProviderRegistry() @@ -20,3 +22,7 @@ provider_registry.register("feature.events.IEventsProvider", EventsProvider) provider_registry.register("feature.mixins.IMixinsProvider", MixinsProvider) provider_registry.register("feature.query.IQueryProvider", QueryProvider) + +# Register connection pool context awareness providers +provider_registry.register("feature.basic.connection.IBasicConnectionProvider", BasicConnectionProvider) +provider_registry.register("feature.query.connection.IQueryConnectionProvider", QueryConnectionProvider) diff --git a/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py new file mode 100644 index 00000000..f92b1c1d --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py @@ -0,0 +1,8 @@ +# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/conftest.py +""" +Pytest configuration for basic connection pool feature tests. + +This file imports fixtures from the corresponding testsuite, making them +available to the tests in this directory. +""" +from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py new file mode 100644 index 00000000..cdd54c73 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py @@ -0,0 +1,14 @@ +# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_context.py +""" +Basic ActiveRecord Context Test Module for PostgreSQL backend. + +This module imports and runs the shared tests from the testsuite package, +ensuring PostgreSQL backend compatibility for connection pool context awareness. +""" +from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import ( + sync_pool_and_model, + async_pool_and_model, +) + +# Import shared tests from testsuite package +from rhosocial.activerecord.testsuite.feature.basic.connection.test_active_record_context import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py new file mode 100644 index 00000000..61731018 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py @@ -0,0 +1,14 @@ +# tests/rhosocial/activerecord_postgres_test/feature/basic/connection/test_active_record_crud.py +""" +Basic ActiveRecord CRUD Test Module for PostgreSQL backend. + +This module imports and runs the shared tests from the testsuite package, +ensuring PostgreSQL backend compatibility for connection pool CRUD operations. +""" +from rhosocial.activerecord.testsuite.feature.basic.connection.conftest import ( + sync_pool_for_crud, + async_pool_for_crud, +) + +# Import shared tests from testsuite package +from rhosocial.activerecord.testsuite.feature.basic.connection.test_active_record_crud import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/connection/conftest.py b/tests/rhosocial/activerecord_postgres_test/feature/connection/conftest.py new file mode 100644 index 00000000..0f5c450e --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/connection/conftest.py @@ -0,0 +1,277 @@ +# tests/rhosocial/activerecord_postgres_test/feature/connection/conftest.py +""" +Pytest configuration for connection pool tests. + +This module provides fixtures for testing connection pools with PostgreSQL backends. +""" + +import asyncio +import sys +import os +from typing import Dict, Any, Generator + +import pytest +import pytest_asyncio +import yaml + +from rhosocial.activerecord.backend.impl.postgres import PostgresBackend, AsyncPostgresBackend +from rhosocial.activerecord.backend.impl.postgres.config import PostgresConnectionConfig +from rhosocial.activerecord.connection.pool import ( + PoolConfig, + BackendPool, + AsyncBackendPool, +) + + +def ensure_compatible_event_loop(): + if sys.platform == "win32": + policy = asyncio.get_event_loop_policy() + if not isinstance(policy, asyncio.WindowsSelectorEventLoopPolicy): + asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy) + + +ensure_compatible_event_loop() + +# --- Scenario Loading Logic --- + +SCENARIO_MAP: Dict[str, Dict[str, Any]] = {} + + +def register_scenario(name: str, config: Dict[str, Any]): + SCENARIO_MAP[name] = config + + +def _load_scenarios_from_config(): + """Load scenarios from a configuration file.""" + config_path = None + env_config_path = os.getenv("POSTGRES_SCENARIOS_CONFIG_PATH") + + if env_config_path and os.path.exists(env_config_path): + config_path = env_config_path + else: + default_path = os.path.join( + os.path.dirname(__file__), + "../../../../config", + "postgres_scenarios.yaml" + ) + if os.path.exists(default_path): + config_path = default_path + elif env_config_path: + print(f"Warning: Scenario file specified in POSTGRES_SCENARIOS_CONFIG_PATH not found: {env_config_path}") + return + + if not config_path: + raise FileNotFoundError( + "No PostgreSQL scenarios configuration file found. " + "Set POSTGRES_SCENARIOS_CONFIG_PATH or place postgres_scenarios.yaml in tests/config." + ) + + try: + with open(config_path, 'r', encoding='utf-8') as f: + config_data = yaml.safe_load(f) + + if 'scenarios' not in config_data: + raise ValueError(f"Configuration file {config_path} does not contain 'scenarios' key") + + for scenario_name, config in config_data['scenarios'].items(): + if config: + register_scenario(scenario_name, config) + + except ImportError: + raise ImportError("PyYAML is required to load scenario configuration files") + + +_load_scenarios_from_config() + + +def get_scenario_config(name: str) -> Dict[str, Any]: + if name not in SCENARIO_MAP: + if SCENARIO_MAP: + name = next(iter(SCENARIO_MAP)) + else: + raise ValueError("No scenarios registered or enabled") + return SCENARIO_MAP[name] + + +def get_scenario_names(): + return list(SCENARIO_MAP.keys()) + + +# --- Pool Fixtures --- + +def create_postgres_backend_factory(config_dict: Dict[str, Any]): + """Create a factory function that produces PostgresBackend instances.""" + def factory(): + config = PostgresConnectionConfig(**config_dict) + backend = PostgresBackend(connection_config=config) + backend.connect() + return backend + return factory + + +def create_async_postgres_backend_factory(config_dict: Dict[str, Any]): + """Create a factory function that produces AsyncPostgresBackend instances.""" + def factory(): + config = PostgresConnectionConfig(**config_dict) + backend = AsyncPostgresBackend(connection_config=config) + # Note: connect() will be called by the pool when needed + return backend + return factory + + +@pytest.fixture(scope="function") +def postgres_pool() -> Generator[BackendPool, None, None]: + """Create a BackendPool with PostgreSQL backends for testing.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + + config_dict = get_scenario_config(scenario_names[0]) + config_dict.pop('enabled', None) + + pool_config = PoolConfig( + min_size=1, + max_size=5, + timeout=30.0, + validate_on_borrow=True, + validation_query="SELECT 1", + backend_factory=create_postgres_backend_factory(config_dict), + ) + + pool = BackendPool.create(pool_config) + yield pool + pool.close(timeout=5.0, force=True) + + +@pytest_asyncio.fixture(scope="function") +async def async_postgres_pool() -> AsyncBackendPool: + """Create an AsyncBackendPool with PostgreSQL backends for testing.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + + config_dict = get_scenario_config(scenario_names[0]) + config_dict.pop('enabled', None) + + pool_config = PoolConfig( + min_size=1, + max_size=5, + timeout=30.0, + validate_on_borrow=True, + validation_query="SELECT 1", + backend_factory=create_async_postgres_backend_factory(config_dict), + ) + + pool = await AsyncBackendPool.create(pool_config) + yield pool + await pool.close(timeout=5.0, force=True) + + +@pytest.fixture(scope="function") +def postgres_pool_large() -> Generator[BackendPool, None, None]: + """Create a larger BackendPool for stress testing.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + + config_dict = get_scenario_config(scenario_names[0]) + config_dict.pop('enabled', None) + + pool_config = PoolConfig( + min_size=2, + max_size=10, + timeout=60.0, + validate_on_borrow=False, # Faster for stress tests + backend_factory=create_postgres_backend_factory(config_dict), + ) + + pool = BackendPool.create(pool_config) + yield pool + pool.close(timeout=5.0, force=True) + + +@pytest_asyncio.fixture(scope="function") +async def async_postgres_pool_large() -> AsyncBackendPool: + """Create a larger AsyncBackendPool for stress testing.""" + scenario_names = get_scenario_names() + if not scenario_names: + pytest.skip("No PostgreSQL scenarios configured") + + config_dict = get_scenario_config(scenario_names[0]) + config_dict.pop('enabled', None) + + pool_config = PoolConfig( + min_size=2, + max_size=10, + timeout=60.0, + validate_on_borrow=False, + backend_factory=create_async_postgres_backend_factory(config_dict), + ) + + pool = await AsyncBackendPool.create(pool_config) + yield pool + await pool.close(timeout=5.0, force=True) + + +# --- Table Setup Fixtures --- + +@pytest.fixture(scope="function") +def postgres_pool_with_tables(postgres_pool: BackendPool) -> Generator[BackendPool, None, None]: + """Create a pool with test tables initialized.""" + with postgres_pool.connection() as backend: + backend.execute("DROP TABLE IF EXISTS concurrent_test_users") + backend.execute("DROP TABLE IF EXISTS concurrent_test_posts") + backend.execute(""" + CREATE TABLE concurrent_test_users ( + id SERIAL PRIMARY KEY, + thread_id INTEGER, + name VARCHAR(255), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + backend.execute(""" + CREATE TABLE concurrent_test_posts ( + id SERIAL PRIMARY KEY, + thread_id INTEGER, + user_id INTEGER, + title VARCHAR(255), + content TEXT + ) + """) + + yield postgres_pool + + with postgres_pool.connection() as backend: + backend.execute("DROP TABLE IF EXISTS concurrent_test_posts") + backend.execute("DROP TABLE IF EXISTS concurrent_test_users") + + +@pytest_asyncio.fixture(scope="function") +async def async_postgres_pool_with_tables(async_postgres_pool: AsyncBackendPool) -> AsyncBackendPool: + """Create an async pool with test tables initialized.""" + async with async_postgres_pool.connection() as backend: + await backend.execute("DROP TABLE IF EXISTS concurrent_test_users") + await backend.execute("DROP TABLE IF EXISTS concurrent_test_posts") + await backend.execute(""" + CREATE TABLE concurrent_test_users ( + id SERIAL PRIMARY KEY, + task_id INTEGER, + name VARCHAR(255), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + await backend.execute(""" + CREATE TABLE concurrent_test_posts ( + id SERIAL PRIMARY KEY, + task_id INTEGER, + user_id INTEGER, + title VARCHAR(255), + content TEXT + ) + """) + + yield async_postgres_pool + + async with async_postgres_pool.connection() as backend: + await backend.execute("DROP TABLE IF EXISTS concurrent_test_posts") + await backend.execute("DROP TABLE IF EXISTS concurrent_test_users") diff --git a/tests/rhosocial/activerecord_postgres_test/feature/connection/test_connection_pool_concurrent.py b/tests/rhosocial/activerecord_postgres_test/feature/connection/test_connection_pool_concurrent.py new file mode 100644 index 00000000..ba6bd101 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/connection/test_connection_pool_concurrent.py @@ -0,0 +1,570 @@ +# tests/rhosocial/activerecord_postgres_test/feature/connection/test_connection_pool_concurrent.py +""" +Concurrent connection pool tests for PostgreSQL backend. + +Tests that verify: +1. Concurrently acquired connections are distinct +2. Concurrent operations don't cause deadlocks +3. Query results don't get mixed up between connections +""" + +import asyncio +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import List, Dict, Any, Tuple +from datetime import datetime + +import pytest +import pytest_asyncio + +from rhosocial.activerecord.connection.pool import ( + PoolConfig, + BackendPool, + AsyncBackendPool, +) + + +# ============================================================ +# Synchronous Concurrent Tests +# ============================================================ + +class TestBackendPoolConcurrent: + """Tests for concurrent operations with BackendPool (PostgreSQL).""" + + def test_concurrent_connections_are_distinct(self, postgres_pool: BackendPool): + """Verify that concurrently acquired connections are distinct. + + This test proves that when multiple threads acquire connections + simultaneously, each thread gets a different connection instance. + """ + num_threads = 5 + connection_ids: List[int] = [] + connection_ids_lock = threading.Lock() + barrier = threading.Barrier(num_threads) + errors: List[Exception] = [] + errors_lock = threading.Lock() + + def acquire_and_record(thread_id: int): + try: + # Synchronize all threads to start at the same time + barrier.wait() + + # Acquire connection + backend = postgres_pool.acquire() + try: + conn_id = id(backend) + with connection_ids_lock: + connection_ids.append((thread_id, conn_id)) + + # Hold connection briefly to ensure overlap + time.sleep(0.1) + finally: + postgres_pool.release(backend) + except Exception as e: + with errors_lock: + errors.append(e) + + # Run concurrent acquires + with ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [executor.submit(acquire_and_record, i) for i in range(num_threads)] + for future in as_completed(futures): + future.result() + + # Verify no errors + assert len(errors) == 0, f"Errors occurred: {errors}" + + # Verify all connections are distinct + assert len(connection_ids) == num_threads + unique_ids = set(conn_id for _, conn_id in connection_ids) + assert len(unique_ids) == num_threads, ( + f"Expected {num_threads} distinct connections, " + f"but got {len(unique_ids)}: {connection_ids}" + ) + + def test_concurrent_operations_no_deadlock(self, postgres_pool_with_tables: BackendPool): + """Verify that concurrent complex operations don't cause deadlock. + + This test simulates multiple threads performing complex operations + (insert, select, update) simultaneously on different connections, + proving no deadlock occurs. + """ + pool = postgres_pool_with_tables + num_threads = 4 + results: Dict[int, List[Any]] = {} + results_lock = threading.Lock() + start_barrier = threading.Barrier(num_threads) + errors: List[Tuple[int, Exception]] = [] + errors_lock = threading.Lock() + + def complex_operation(thread_id: int): + try: + # Synchronize all threads + start_barrier.wait() + + with pool.connection() as backend: + # Insert records for this thread + for i in range(10): + backend.execute( + "INSERT INTO concurrent_test_users (thread_id, name) VALUES (%s, %s)", + [thread_id, f"user_{thread_id}_{i}"] + ) + + # Query back the data + query_result = backend.execute( + "SELECT * FROM concurrent_test_users WHERE thread_id = %s", + [thread_id] + ) + + # Update some records + backend.execute( + "UPDATE concurrent_test_users SET name = %s WHERE thread_id = %s AND id <= %s", + [f"updated_{thread_id}", thread_id, thread_id + 5] + ) + + with results_lock: + results[thread_id] = query_result.data + + except Exception as e: + with errors_lock: + errors.append((thread_id, e)) + + # Run concurrent operations with timeout + with ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [executor.submit(complex_operation, i) for i in range(num_threads)] + + # Wait for all with timeout (deadlock detection) + for future in as_completed(futures, timeout=60): + future.result() + + # Verify no errors + assert len(errors) == 0, f"Errors occurred: {errors}" + + # Verify each thread got correct results + for thread_id in range(num_threads): + assert thread_id in results, f"Thread {thread_id} has no results" + assert len(results[thread_id]) == 10, ( + f"Thread {thread_id} expected 10 rows, got {len(results[thread_id])}" + ) + # Verify all rows belong to this thread + for row in results[thread_id]: + assert row['thread_id'] == thread_id, ( + f"Thread {thread_id} got wrong data: {row}" + ) + + def test_concurrent_query_results_not_mixed(self, postgres_pool_with_tables: BackendPool): + """Verify that concurrent queries on different connections don't mix results. + + This test proves that when multiple connections query different data + simultaneously, each connection gets only its own data without any mix-up. + """ + pool = postgres_pool_with_tables + num_threads = 5 + records_per_thread = 20 + + # First, set up shared data + with pool.connection() as backend: + # Insert data for each thread group + for thread_id in range(num_threads): + for i in range(records_per_thread): + backend.execute( + "INSERT INTO concurrent_test_posts (thread_id, title, content) VALUES (%s, %s, %s)", + [thread_id, f"title_{thread_id}_{i}", f"content_{thread_id}_{i}"] + ) + + # Now run concurrent reads + results: Dict[int, List[Dict]] = {} + results_lock = threading.Lock() + barrier = threading.Barrier(num_threads) + errors: List[Tuple[int, Exception]] = [] + errors_lock = threading.Lock() + + def read_thread_data(thread_id: int): + try: + # Synchronize start + barrier.wait() + + with pool.connection() as backend: + # Perform multiple reads with some computation + time.sleep(0.05) # Simulate processing + + result = backend.execute( + "SELECT * FROM concurrent_test_posts WHERE thread_id = %s ORDER BY id", + [thread_id] + ) + + time.sleep(0.05) # More processing + + with results_lock: + results[thread_id] = result.data + + except Exception as e: + with errors_lock: + errors.append((thread_id, e)) + + with ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [executor.submit(read_thread_data, i) for i in range(num_threads)] + for future in as_completed(futures, timeout=30): + future.result() + + # Verify no errors + assert len(errors) == 0, f"Errors occurred: {errors}" + + # Verify each thread got exactly its own data + for thread_id in range(num_threads): + assert thread_id in results + data = results[thread_id] + assert len(data) == records_per_thread, ( + f"Thread {thread_id}: expected {records_per_thread} records, " + f"got {len(data)}" + ) + + # Verify all records belong to this thread's group + for row in data: + assert row['thread_id'] == thread_id, ( + f"Thread {thread_id} got data from group {row['thread_id']}!" + ) + assert row['title'].startswith(f"title_{thread_id}_"), ( + f"Thread {thread_id} got wrong title: {row['title']}" + ) + + def test_concurrent_transactions_isolated(self, postgres_pool_with_tables: BackendPool): + """Verify that concurrent transactions on different connections are isolated. + + This test proves that uncommitted changes in one transaction + are not visible to other concurrent transactions. + """ + pool = postgres_pool_with_tables + num_threads = 3 + + # Insert initial data + with pool.transaction() as backend: + for i in range(num_threads): + backend.execute( + "INSERT INTO concurrent_test_users (thread_id, name) VALUES (%s, 'initial')", + [i] + ) + + results: Dict[int, Dict] = {} + results_lock = threading.Lock() + barrier = threading.Barrier(num_threads) + errors: List[Tuple[int, Exception]] = [] + errors_lock = threading.Lock() + + def transaction_test(thread_id: int): + try: + barrier.wait() + + with pool.transaction() as backend: + # Update own row + backend.execute( + "UPDATE concurrent_test_users SET name = %s WHERE thread_id = %s", + [f'modified_by_{thread_id}', thread_id] + ) + + # Read all rows - should only see committed data + # from other threads, not their uncommitted changes + time.sleep(0.2) # Allow other threads to make changes + + all_rows = backend.execute( + "SELECT * FROM concurrent_test_users ORDER BY thread_id" + ) + + own_row = backend.execute( + "SELECT * FROM concurrent_test_users WHERE thread_id = %s", + [thread_id] + ) + + with results_lock: + results[thread_id] = { + 'all_rows': all_rows.data, + 'own_row': own_row.data + } + + except Exception as e: + with errors_lock: + errors.append((thread_id, e)) + + with ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [executor.submit(transaction_test, i) for i in range(num_threads)] + for future in as_completed(futures, timeout=30): + future.result() + + # Verify no errors (no deadlock) + assert len(errors) == 0, f"Errors occurred: {errors}" + + # Verify each transaction saw its own changes + for thread_id in range(num_threads): + assert thread_id in results + own_row = results[thread_id]['own_row'] + assert len(own_row) == 1 + # Own row should show the modification + assert own_row[0]['name'] == f'modified_by_{thread_id}' + + def test_pool_stress_concurrent_acquire_release(self, postgres_pool_large: BackendPool): + """Stress test: rapidly acquire and release connections concurrently. + + This test verifies pool stability under high concurrent load. + """ + pool = postgres_pool_large + num_threads = 10 + operations_per_thread = 50 + + success_count = 0 + success_lock = threading.Lock() + errors: List[Tuple[int, int, Exception]] = [] + errors_lock = threading.Lock() + + def rapid_acquire_release(thread_id: int): + nonlocal success_count + for i in range(operations_per_thread): + try: + with pool.connection() as backend: + # Quick operation + backend.execute("SELECT 1") + time.sleep(0.001) # Tiny delay + + with success_lock: + success_count += 1 + except Exception as e: + with errors_lock: + errors.append((thread_id, i, e)) + + with ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [executor.submit(rapid_acquire_release, i) for i in range(num_threads)] + for future in as_completed(futures, timeout=120): + future.result() + + # Verify high success rate + expected = num_threads * operations_per_thread + success_rate = success_count / expected + assert success_rate >= 0.95, ( + f"Success rate too low: {success_rate:.2%} " + f"({success_count}/{expected}), errors: {errors[:5]}" + ) + + # Verify pool is still healthy + stats = pool.get_stats() + assert stats.current_in_use == 0, "Connections still in use after test" + + +# ============================================================ +# Asynchronous Concurrent Tests +# ============================================================ + +class TestAsyncBackendPoolConcurrent: + """Tests for concurrent operations with AsyncBackendPool (PostgreSQL).""" + + @pytest.mark.asyncio + async def test_concurrent_connections_are_distinct(self, async_postgres_pool: AsyncBackendPool): + """Verify that concurrently acquired async connections are distinct.""" + num_concurrent = 5 + connection_ids: List[int] = [] + connection_ids_lock = asyncio.Lock() + start_event = asyncio.Event() + + async def acquire_and_record(task_id: int): + # Wait for all tasks to be ready + await start_event.wait() + + async with async_postgres_pool.connection() as backend: + conn_id = id(backend) + async with connection_ids_lock: + connection_ids.append((task_id, conn_id)) + + # Hold connection briefly + await asyncio.sleep(0.1) + + # Start all tasks simultaneously + tasks = [acquire_and_record(i) for i in range(num_concurrent)] + start_event.set() # Release all tasks at once + await asyncio.gather(*tasks) + + # Verify all connections are distinct + assert len(connection_ids) == num_concurrent + unique_ids = set(conn_id for _, conn_id in connection_ids) + assert len(unique_ids) == num_concurrent, ( + f"Expected {num_concurrent} distinct connections, " + f"but got {len(unique_ids)}" + ) + + @pytest.mark.asyncio + async def test_concurrent_operations_no_deadlock(self, async_postgres_pool_with_tables: AsyncBackendPool): + """Verify that concurrent async operations don't cause deadlock.""" + pool = async_postgres_pool_with_tables + num_concurrent = 4 + results: Dict[int, List[Any]] = {} + results_lock = asyncio.Lock() + start_event = asyncio.Event() + + async def complex_operation(task_id: int): + await start_event.wait() + + async with pool.connection() as backend: + # Insert records + for i in range(10): + await backend.execute( + "INSERT INTO concurrent_test_users (task_id, name) VALUES (%s, %s)", + [task_id, f"user_{task_id}_{i}"] + ) + + # Query back + query_result = await backend.execute( + "SELECT * FROM concurrent_test_users WHERE task_id = %s", + [task_id] + ) + + async with results_lock: + results[task_id] = query_result.data + + # Run all tasks concurrently + tasks = [complex_operation(i) for i in range(num_concurrent)] + start_event.set() + await asyncio.gather(*tasks) + + # Verify results + for task_id in range(num_concurrent): + assert task_id in results + assert len(results[task_id]) == 10 + for row in results[task_id]: + assert row['task_id'] == task_id + + @pytest.mark.asyncio + async def test_concurrent_query_results_not_mixed(self, async_postgres_pool_with_tables: AsyncBackendPool): + """Verify that concurrent async queries don't mix results.""" + pool = async_postgres_pool_with_tables + num_concurrent = 5 + records_per_task = 20 + + # Setup shared data + async with pool.connection() as backend: + for task_id in range(num_concurrent): + for i in range(records_per_task): + await backend.execute( + "INSERT INTO concurrent_test_posts (task_id, title, content) VALUES (%s, %s, %s)", + [task_id, f"title_{task_id}_{i}", f"content_{task_id}_{i}"] + ) + + # Concurrent reads + results: Dict[int, List[Dict]] = {} + results_lock = asyncio.Lock() + start_event = asyncio.Event() + + async def read_task_data(task_id: int): + await start_event.wait() + + async with pool.connection() as backend: + await asyncio.sleep(0.05) + result = await backend.execute( + "SELECT * FROM concurrent_test_posts WHERE task_id = %s ORDER BY id", + [task_id] + ) + await asyncio.sleep(0.05) + + async with results_lock: + results[task_id] = result.data + + tasks = [read_task_data(i) for i in range(num_concurrent)] + start_event.set() + await asyncio.gather(*tasks) + + # Verify isolation + for task_id in range(num_concurrent): + assert task_id in results + data = results[task_id] + assert len(data) == records_per_task + for row in data: + assert row['task_id'] == task_id + assert row['title'].startswith(f"title_{task_id}_") + + @pytest.mark.asyncio + async def test_pool_stress_concurrent_acquire_release(self, async_postgres_pool_large: AsyncBackendPool): + """Stress test: rapid async acquire and release.""" + pool = async_postgres_pool_large + num_tasks = 10 + operations_per_task = 50 + + success_count = 0 + success_lock = asyncio.Lock() + errors: List[Tuple[int, int, Exception]] = [] + errors_lock = asyncio.Lock() + + async def rapid_acquire_release(task_id: int): + nonlocal success_count + for i in range(operations_per_task): + try: + async with pool.connection() as backend: + await backend.execute("SELECT 1") + await asyncio.sleep(0.001) + + async with success_lock: + success_count += 1 + except Exception as e: + async with errors_lock: + errors.append((task_id, i, e)) + + tasks = [rapid_acquire_release(i) for i in range(num_tasks)] + await asyncio.gather(*tasks) + + expected = num_tasks * operations_per_task + success_rate = success_count / expected + assert success_rate >= 0.95, ( + f"Success rate too low: {success_rate:.2%}, errors: {errors[:5]}" + ) + + stats = pool.get_stats() + assert stats.current_in_use == 0 + + @pytest.mark.asyncio + async def test_concurrent_with_semaphore_limit(self, async_postgres_pool: AsyncBackendPool): + """Test concurrent operations when pool size limits cause waiting. + + This test verifies that when more tasks than available connections + are running, they properly wait and don't deadlock. + """ + pool = async_postgres_pool + num_tasks = 10 + max_pool_size = pool.config.max_size + + concurrent_count = 0 + max_concurrent = 0 + count_lock = asyncio.Lock() + start_event = asyncio.Event() + completed: List[int] = [] + completed_lock = asyncio.Lock() + + async def tracked_operation(task_id: int): + nonlocal concurrent_count, max_concurrent + + await start_event.wait() + + async with pool.connection() as backend: + # Track concurrent execution + async with count_lock: + concurrent_count += 1 + max_concurrent = max(max_concurrent, concurrent_count) + + # Do some work + await backend.execute("SELECT 1") + await asyncio.sleep(0.1) + + async with count_lock: + concurrent_count -= 1 + + async with completed_lock: + completed.append(task_id) + + tasks = [tracked_operation(i) for i in range(num_tasks)] + start_event.set() + await asyncio.gather(*tasks) + + # Verify all tasks completed + assert len(completed) == num_tasks + + # Verify max concurrent didn't exceed pool size + assert max_concurrent <= max_pool_size, ( + f"Max concurrent {max_concurrent} exceeded pool size {max_pool_size}" + ) + + # Verify some waiting occurred (proves semaphore works) + assert max_concurrent > 0 diff --git a/tests/rhosocial/activerecord_postgres_test/feature/query/connection/conftest.py b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/conftest.py new file mode 100644 index 00000000..86492b05 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/conftest.py @@ -0,0 +1,8 @@ +# tests/rhosocial/activerecord_postgres_test/feature/query/connection/conftest.py +""" +Pytest configuration for query connection pool feature tests. + +This file imports fixtures from the corresponding testsuite, making them +available to the tests in this directory. +""" +from rhosocial.activerecord.testsuite.feature.query.connection.conftest import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_active_query_context.py b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_active_query_context.py new file mode 100644 index 00000000..4363b19d --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_active_query_context.py @@ -0,0 +1,14 @@ +# tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_active_query_context.py +""" +ActiveQuery Context Test Module for PostgreSQL backend. + +This module imports and runs the shared tests from the testsuite package, +ensuring PostgreSQL backend compatibility for ActiveQuery connection pool context awareness. +""" +from rhosocial.activerecord.testsuite.feature.query.connection.conftest import ( + sync_pool_and_model, + async_pool_and_model, +) + +# Import shared tests from testsuite package +from rhosocial.activerecord.testsuite.feature.query.connection.test_active_query_context import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_cte_query_context.py b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_cte_query_context.py new file mode 100644 index 00000000..94bfc1f9 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_cte_query_context.py @@ -0,0 +1,14 @@ +# tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_cte_query_context.py +""" +CTEQuery Context Test Module for PostgreSQL backend. + +This module imports and runs the shared tests from the testsuite package, +ensuring PostgreSQL backend compatibility for CTEQuery connection pool context awareness. +""" +from rhosocial.activerecord.testsuite.feature.query.connection.conftest import ( + sync_pool_and_model, + async_pool_and_model, +) + +# Import shared tests from testsuite package +from rhosocial.activerecord.testsuite.feature.query.connection.test_cte_query_context import * diff --git a/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_set_operation_context.py b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_set_operation_context.py new file mode 100644 index 00000000..3eef9b3e --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_set_operation_context.py @@ -0,0 +1,14 @@ +# tests/rhosocial/activerecord_postgres_test/feature/query/connection/test_set_operation_context.py +""" +SetOperationQuery Context Test Module for PostgreSQL backend. + +This module imports and runs the shared tests from the testsuite package, +ensuring PostgreSQL backend compatibility for SetOperationQuery connection pool context awareness. +""" +from rhosocial.activerecord.testsuite.feature.query.connection.conftest import ( + sync_pool_and_model, + async_pool_and_model, +) + +# Import shared tests from testsuite package +from rhosocial.activerecord.testsuite.feature.query.connection.test_set_operation_context import *