diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a9ed7c51..0ad39791 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -63,8 +63,8 @@ jobs: - name: Install activerecord and testsuite run: | - pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" # NOSONAR - name: Install dependencies run: | @@ -278,8 +278,8 @@ jobs: - name: Install activerecord and testsuite run: | - pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" # NOSONAR - name: Install dependencies run: | @@ -385,8 +385,8 @@ jobs: - name: Install activerecord and testsuite run: | - pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" # NOSONAR - name: Install dependencies run: | @@ -488,8 +488,8 @@ jobs: - name: Install activerecord and testsuite run: | - pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" # NOSONAR - pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + pip install "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" # NOSONAR + pip install "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" # NOSONAR - name: Install dependencies run: | diff --git a/changelog.d/69.added.md b/changelog.d/69.added.md new file mode 100644 index 00000000..d0cd7b3c --- /dev/null +++ b/changelog.d/69.added.md @@ -0,0 +1 @@ +Classified PostgreSQL driver errors into the semantic error taxonomy (data, record, policy, concurrency, infrastructure) and aligned `execute` / `execute_many` with the execute-phase error contract: framework errors pass through, auto-commit failures raise `TransactionError`, and only transient connection failures are retried. Added contract tests covering the mapping and phases. diff --git a/changelog.d/69.fixed.md b/changelog.d/69.fixed.md new file mode 100644 index 00000000..34bb0e87 --- /dev/null +++ b/changelog.d/69.fixed.md @@ -0,0 +1 @@ +Fixed the unused-indexes listing: `indisvalid` is now read from `pg_index` (joined on `indexrelid`) instead of `pg_stat_user_indexes`, which carries only scan counters. Naming that column on the statistics view made the server reject the whole statement (`column i.indisvalid does not exist`), so `list_unused_indexes()` and `introspect unused-indexes` could not answer on any real PostgreSQL server. diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py index 829b811c..693fafcc 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py @@ -31,12 +31,6 @@ import psycopg from psycopg import AsyncConnection from psycopg.errors import Error as PsycopgError -from psycopg.errors import ( - IntegrityError as PsycopgIntegrityError, - OperationalError as PsycopgOperationalError, - ProgrammingError as PsycopgProgrammingError, - DeadlockDetected as PsycopgDeadlockError, -) from rhosocial.activerecord.backend.base import AsyncStorageBackend from rhosocial.activerecord.backend.base.execution_logging import ( @@ -49,10 +43,8 @@ from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, - DeadlockError, - IntegrityError, OperationalError, - QueryError, + TransactionError, ) from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType from rhosocial.activerecord.backend.result import QueryResult @@ -436,21 +428,27 @@ async def execute_many( result = QueryResult(affected_rows=affected_rows, data=None, duration=duration) + # Auto-commit before reporting success: a commit failure must surface + # as TransactionError instead of a silently uncommitted batch. + await self._handle_auto_commit_if_needed() log_batch_summary(self, sql, params_list, affected_rows, duration, options) return result - except PsycopgIntegrityError as e: - log_execution_error(self, e, options) - raise IntegrityError(str(e)) from e - except PsycopgError as e: - log_execution_error(self, e, options) - raise DatabaseError(str(e)) from e except Exception as e: log_execution_error(self, e, options) - raise QueryError(str(e)) from e + if isinstance(e, DatabaseError): + raise + await self._handle_error(e) + # _handle_error() is required to raise. Keep a defensive fallback so + # a backend that returns cannot turn a failure into a None result. + raise DatabaseError(str(e)) from e finally: if cursor: - await cursor.close() + try: + await cursor.close() + except Exception as close_error: + # Closing must never replace the error already propagating. + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") async def get_server_version(self) -> Optional[Tuple[int, int, int]]: """Get PostgreSQL server version asynchronously. @@ -490,7 +488,10 @@ async def get_server_version(self) -> Optional[Tuple[int, int, int]]: return None finally: if cursor: - await cursor.close() + try: + await cursor.close() + except Exception as close_error: + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") async def introspect_and_adapt(self) -> None: """Introspect backend and adapt backend instance to actual server capabilities. @@ -678,77 +679,112 @@ async def execute( for attempt in range(max_retries + 1): try: return await super().execute(sql, params, options=options) - except OperationalError as e: - error_msg = str(e) - if self._is_connection_error_message(error_msg) and attempt < max_retries: + except OperationalError as exc: + error_msg = str(exc) + # Prefer the framework's transient hint. Only fall back to + # message matching for errors that carry no hint at all + # (transient is None); a non-transient error is never retried. + retryable = exc.transient is True or ( + exc.transient is None and self._is_connection_error_message(error_msg) + ) + if retryable and attempt < max_retries: self.log(logging.WARNING, f"Connection error on attempt {attempt + 1}, retrying... Error: {error_msg}") if await self._reconnect(): continue - # If reconnection fails, break and raise the error + # Reconnection failed: re-raise instead of leaving the loop + # and turning the failure into a None result. self.log(logging.WARNING, "Failed to reconnect after connection error") - break raise async def _handle_error(self, error: Exception) -> None: """Handle PostgreSQL-specific errors asynchronously.""" + # Framework errors are final: they are never remapped into a driver + # error. Re-raising here also keeps errors raised before any SQL + # (TypeConversionError, ValueError, ...) from triggering a rollback. + if isinstance(error, DatabaseError): + self.log(logging.DEBUG, f"Framework error passed through: {str(error)}") + raise error + + await self._try_rollback_transaction(error) + error_msg = str(error) + mapped = self._map_postgres_error(error) + if mapped is not None: + framework_error, log_label = mapped + self.log(logging.DEBUG, f"{log_label}: {error_msg}") + raise framework_error from error + self.log(logging.DEBUG, f"Unexpected error: {error_msg}") + raise error - # Try to rollback transaction to recover from aborted state - # PostgreSQL requires explicit ROLLBACK after errors in a transaction - await self._try_rollback_transaction() - - if isinstance(error, PsycopgIntegrityError): - if "duplicate key value violates unique constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Unique constraint violation: {error_msg}") - raise IntegrityError(f"Unique constraint violation: {error_msg}") from error - elif "violates foreign key constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Foreign key constraint violation: {error_msg}") - raise IntegrityError(f"Foreign key constraint violation: {error_msg}") from error - self.log(logging.DEBUG, f"Integrity error: {error_msg}") - raise IntegrityError(error_msg) from error - elif isinstance(error, PsycopgOperationalError): - if "deadlock detected" in error_msg.lower(): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - self.log(logging.DEBUG, f"Operational error: {error_msg}") - raise OperationalError(error_msg) from error - elif isinstance(error, PsycopgProgrammingError): - self.log(logging.DEBUG, f"Programming error: {error_msg}") - raise QueryError(error_msg) from error - elif isinstance(error, PsycopgDeadlockError): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - elif isinstance(error, PsycopgError): - self.log(logging.DEBUG, f"PostgreSQL error: {error_msg}") - raise DatabaseError(error_msg) from error - else: - self.log(logging.DEBUG, f"Unexpected error: {error_msg}") - raise error + @staticmethod + def _is_rollback_error(error: BaseException) -> bool: + """Return True when the error may have aborted a server transaction. + + Only driver errors (the statement reached PostgreSQL) and transaction + errors need recovery. A framework error raised before any SQL, such as + a ``TypeConversionError`` or ``ValueError`` during parameter + preparation, leaves the transaction untouched. + """ + if isinstance(error, TransactionError): + return True + if isinstance(error, DatabaseError): + return False + return isinstance(error, PsycopgError) - async def _try_rollback_transaction(self) -> None: + async def _try_rollback_transaction(self, error: Optional[BaseException] = None) -> None: """Attempt to rollback transaction to recover from aborted state asynchronously. PostgreSQL requires explicit ROLLBACK after query errors in a transaction. - This method attempts to recover by rolling back the transaction, regardless - of the transaction manager's state. + Only driver and transaction errors are recovered; framework errors raised + before SQL must not disturb the transaction. Outside a framework + transaction the connection runs in autocommit mode, so there is nothing + to roll back and a direct connection rollback is never attempted. + + A failed rollback is logged at ERROR and the original error keeps + propagating. + + Args: + error: The error that triggered recovery. ``None`` keeps the + historical behaviour for callers that recover without one. """ + if error is not None and not self._is_rollback_error(error): + self.log(logging.DEBUG, f"Skipping rollback for non-driver error: {error!r}") + return + + if not self.in_transaction: + self.log(logging.DEBUG, "No active transaction to roll back after error") + return + try: - # First, try the transaction manager if active - if self.in_transaction: - self.log(logging.DEBUG, "Attempting to rollback transaction after error") - await self._transaction_manager.rollback() - - # Also try direct connection rollback to handle edge cases - # This ensures we recover from aborted transaction state - if self._connection and not self._connection.closed: - try: - await self._connection.rollback() - self.log(logging.DEBUG, "Direct connection rollback completed") - except Exception as direct_error: - self.log(logging.DEBUG, f"Direct rollback not needed or failed: {direct_error}") - except Exception as rollback_error: - self.log(logging.WARNING, f"Failed to rollback transaction: {rollback_error}") + self.log(logging.DEBUG, "Attempting to rollback transaction after error") + await self._transaction_manager.rollback() + except Exception: + # Never let recovery hide the original failure, but do not swallow + # the recovery failure silently: an aborted transaction left as-is + # is a real problem for the next statement. + self.log( + logging.ERROR, + "Failed to rollback transaction after error; " + "the transaction state may be inconsistent", + exc_info=True, + ) + return + + # Inside a transaction the manager already issued ROLLBACK. Reset the + # driver-level transaction as well, because the connection's own state + # is independent of the manager's bookkeeping. + if self._connection and not getattr(self._connection, "closed", False): + try: + await self._connection.rollback() + self.log(logging.DEBUG, "Direct connection rollback completed") + except Exception: + self.log( + logging.ERROR, + "Failed to rollback connection after error; " + "the connection state may be inconsistent", + exc_info=True, + ) async def _handle_auto_commit(self) -> None: """Handle auto commit based on PostgreSQL connection and transaction state asynchronously. @@ -756,35 +792,40 @@ async def _handle_auto_commit(self) -> None: This method will commit the current connection if: 1. The connection exists and is open 2. There is no active transaction managed by transaction_manager + 3. autocommit is disabled in the connection configuration It's used by insert/update/delete operations to ensure changes are persisted immediately when auto_commit=True is specified. + + Raises: + TransactionError: If the commit fails. A DML result must not be + reported as successful when its commit never reached the server. """ + if not self._connection: + return + + if self.in_transaction: + return + + # For PostgreSQL, if autocommit is disabled, we need to commit explicitly + if self.config.autocommit: + return + try: - # Check if connection exists - if not self._connection: - return - - # Check if we're not in an active transaction - if not self.in_transaction: - # For PostgreSQL, if autocommit is disabled, we need to commit explicitly - if not self.config.autocommit: - await self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") - except Exception as e: - # Just log the error but don't raise - this is a convenience feature - self.log(logging.WARNING, f"Failed to auto-commit: {str(e)}") + await self._connection.commit() + except Exception as exc: + raise TransactionError("Auto-commit failed") from exc + self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") async def _handle_auto_commit_if_needed(self) -> None: """ Handle auto-commit for PostgreSQL asynchronously. PostgreSQL respects the autocommit setting, but we also need to handle explicit commits. + Delegates to :meth:`_handle_auto_commit` so a commit failure surfaces as + ``TransactionError`` on every path that reaches the execute result. """ - if not self.in_transaction and self._connection: - if not self.config.autocommit: - await self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") + await self._handle_auto_commit() # region Advisory Lock Methods diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/backend.py index 879237ee..b25a1cd4 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/backend.py @@ -20,12 +20,6 @@ import psycopg from psycopg.errors import Error as PsycopgError -from psycopg.errors import ( - IntegrityError as PsycopgIntegrityError, - OperationalError as PsycopgOperationalError, - ProgrammingError as PsycopgProgrammingError, - DeadlockDetected as PsycopgDeadlockError, -) from rhosocial.activerecord.backend.base import StorageBackend from rhosocial.activerecord.backend.base.execution_logging import ( @@ -38,10 +32,8 @@ from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, - DeadlockError, - IntegrityError, OperationalError, - QueryError, + TransactionError, ) from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType from rhosocial.activerecord.backend.result import QueryResult @@ -415,21 +407,27 @@ def execute_many( result = QueryResult(affected_rows=affected_rows, data=None, duration=duration) + # Auto-commit before reporting success: a commit failure must surface + # as TransactionError instead of a silently uncommitted batch. + self._handle_auto_commit_if_needed() log_batch_summary(self, sql, params_list, affected_rows, duration, options) return result - except PsycopgIntegrityError as e: - log_execution_error(self, e, options) - raise IntegrityError(str(e)) from e - except PsycopgError as e: - log_execution_error(self, e, options) - raise DatabaseError(str(e)) from e except Exception as e: log_execution_error(self, e, options) - raise QueryError(str(e)) from e + if isinstance(e, DatabaseError): + raise + self._handle_error(e) + # _handle_error() is required to raise. Keep a defensive fallback so + # a backend that returns cannot turn a failure into a None result. + raise DatabaseError(str(e)) from e finally: if cursor: - cursor.close() + try: + cursor.close() + except Exception as close_error: + # Closing must never replace the error already propagating. + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") def get_server_version(self) -> Optional[Tuple[int, int, int]]: """Get PostgreSQL server version. @@ -469,7 +467,10 @@ def get_server_version(self) -> Optional[Tuple[int, int, int]]: return None finally: if cursor: - cursor.close() + try: + cursor.close() + except Exception as close_error: + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") def introspect_and_adapt(self) -> None: """Introspect backend and adapt backend instance to actual server capabilities. @@ -657,77 +658,112 @@ def execute( for attempt in range(max_retries + 1): try: return super().execute(sql, params, options=options) - except OperationalError as e: - error_msg = str(e) - if self._is_connection_error_message(error_msg) and attempt < max_retries: + except OperationalError as exc: + error_msg = str(exc) + # Prefer the framework's transient hint. Only fall back to + # message matching for errors that carry no hint at all + # (transient is None); a non-transient error is never retried. + retryable = exc.transient is True or ( + exc.transient is None and self._is_connection_error_message(error_msg) + ) + if retryable and attempt < max_retries: self.log(logging.WARNING, f"Connection error on attempt {attempt + 1}, retrying... Error: {error_msg}") if self._reconnect(): continue - # If reconnection fails, break and raise the error + # Reconnection failed: re-raise instead of leaving the loop + # and turning the failure into a None result. self.log(logging.WARNING, "Failed to reconnect after connection error") - break raise def _handle_error(self, error: Exception) -> None: """Handle PostgreSQL-specific errors.""" + # Framework errors are final: they are never remapped into a driver + # error. Re-raising here also keeps errors raised before any SQL + # (TypeConversionError, ValueError, ...) from triggering a rollback. + if isinstance(error, DatabaseError): + self.log(logging.DEBUG, f"Framework error passed through: {str(error)}") + raise error + + self._try_rollback_transaction(error) + error_msg = str(error) + mapped = self._map_postgres_error(error) + if mapped is not None: + framework_error, log_label = mapped + self.log(logging.DEBUG, f"{log_label}: {error_msg}") + raise framework_error from error + self.log(logging.DEBUG, f"Unexpected error: {error_msg}") + raise error - # Try to rollback transaction to recover from aborted state - # PostgreSQL requires explicit ROLLBACK after errors in a transaction - self._try_rollback_transaction() - - if isinstance(error, PsycopgIntegrityError): - if "duplicate key value violates unique constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Unique constraint violation: {error_msg}") - raise IntegrityError(f"Unique constraint violation: {error_msg}") from error - elif "violates foreign key constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Foreign key constraint violation: {error_msg}") - raise IntegrityError(f"Foreign key constraint violation: {error_msg}") from error - self.log(logging.DEBUG, f"Integrity error: {error_msg}") - raise IntegrityError(error_msg) from error - elif isinstance(error, PsycopgOperationalError): - if "deadlock detected" in error_msg.lower(): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - self.log(logging.DEBUG, f"Operational error: {error_msg}") - raise OperationalError(error_msg) from error - elif isinstance(error, PsycopgProgrammingError): - self.log(logging.DEBUG, f"Programming error: {error_msg}") - raise QueryError(error_msg) from error - elif isinstance(error, PsycopgDeadlockError): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - elif isinstance(error, PsycopgError): - self.log(logging.DEBUG, f"PostgreSQL error: {error_msg}") - raise DatabaseError(error_msg) from error - else: - self.log(logging.DEBUG, f"Unexpected error: {error_msg}") - raise error + @staticmethod + def _is_rollback_error(error: BaseException) -> bool: + """Return True when the error may have aborted a server transaction. + + Only driver errors (the statement reached PostgreSQL) and transaction + errors need recovery. A framework error raised before any SQL, such as + a ``TypeConversionError`` or ``ValueError`` during parameter + preparation, leaves the transaction untouched. + """ + if isinstance(error, TransactionError): + return True + if isinstance(error, DatabaseError): + return False + return isinstance(error, PsycopgError) - def _try_rollback_transaction(self) -> None: + def _try_rollback_transaction(self, error: Optional[BaseException] = None) -> None: """Attempt to rollback transaction to recover from aborted state. PostgreSQL requires explicit ROLLBACK after query errors in a transaction. - This method attempts to recover by rolling back the transaction, regardless - of the transaction manager's state. + Only driver and transaction errors are recovered; framework errors raised + before SQL must not disturb the transaction. Outside a framework + transaction the connection runs in autocommit mode, so there is nothing + to roll back and a direct connection rollback is never attempted. + + A failed rollback is logged at ERROR and the original error keeps + propagating. + + Args: + error: The error that triggered recovery. ``None`` keeps the + historical behaviour for callers that recover without one. """ + if error is not None and not self._is_rollback_error(error): + self.log(logging.DEBUG, f"Skipping rollback for non-driver error: {error!r}") + return + + if not self.in_transaction: + self.log(logging.DEBUG, "No active transaction to roll back after error") + return + try: - # First, try the transaction manager if active - if self.in_transaction: - self.log(logging.DEBUG, "Attempting to rollback transaction after error") - self._transaction_manager.rollback() - - # Also try direct connection rollback to handle edge cases - # This ensures we recover from aborted transaction state - if self._connection and not self._connection.closed: - try: - self._connection.rollback() - self.log(logging.DEBUG, "Direct connection rollback completed") - except Exception as direct_error: - self.log(logging.DEBUG, f"Direct rollback not needed or failed: {direct_error}") - except Exception as rollback_error: - self.log(logging.WARNING, f"Failed to rollback transaction: {rollback_error}") + self.log(logging.DEBUG, "Attempting to rollback transaction after error") + self._transaction_manager.rollback() + except Exception: + # Never let recovery hide the original failure, but do not swallow + # the recovery failure silently: an aborted transaction left as-is + # is a real problem for the next statement. + self.log( + logging.ERROR, + "Failed to rollback transaction after error; " + "the transaction state may be inconsistent", + exc_info=True, + ) + return + + # Inside a transaction the manager already issued ROLLBACK. Reset the + # driver-level transaction as well, because the connection's own state + # is independent of the manager's bookkeeping. + if self._connection and not getattr(self._connection, "closed", False): + try: + self._connection.rollback() + self.log(logging.DEBUG, "Direct connection rollback completed") + except Exception: + self.log( + logging.ERROR, + "Failed to rollback connection after error; " + "the connection state may be inconsistent", + exc_info=True, + ) def _handle_auto_commit(self) -> None: """Handle auto commit based on PostgreSQL connection and transaction state. @@ -735,35 +771,40 @@ def _handle_auto_commit(self) -> None: This method will commit the current connection if: 1. The connection exists and is open 2. There is no active transaction managed by transaction_manager + 3. autocommit is disabled in the connection configuration It's used by insert/update/delete operations to ensure changes are persisted immediately when auto_commit=True is specified. + + Raises: + TransactionError: If the commit fails. A DML result must not be + reported as successful when its commit never reached the server. """ + if not self._connection: + return + + if self.in_transaction: + return + + # For PostgreSQL, if autocommit is disabled, we need to commit explicitly + if self.config.autocommit: + return + try: - # Check if connection exists - if not self._connection: - return - - # Check if we're not in an active transaction - if not self.in_transaction: - # For PostgreSQL, if autocommit is disabled, we need to commit explicitly - if not self.config.autocommit: - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") - except Exception as e: - # Just log the error but don't raise - this is a convenience feature - self.log(logging.WARNING, f"Failed to auto-commit: {str(e)}") + self._connection.commit() + except Exception as exc: + raise TransactionError("Auto-commit failed") from exc + self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") def _handle_auto_commit_if_needed(self) -> None: """ Handle auto-commit for PostgreSQL. PostgreSQL respects the autocommit setting, but we also need to handle explicit commits. + Delegates to :meth:`_handle_auto_commit` so a commit failure surfaces as + ``TransactionError`` on every path that reaches the execute result. """ - if not self.in_transaction and self._connection: - if not self.config.autocommit: - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") + self._handle_auto_commit() # region Advisory Lock Methods diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/base.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/base.py index a654efd9..56ae2ca3 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/base.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/base.py @@ -5,9 +5,28 @@ containing methods that are shared between sync and async backends. """ -from typing import Dict, Tuple, Type, TYPE_CHECKING +from typing import Dict, Optional, Tuple, Type, TYPE_CHECKING import logging +from psycopg.errors import ( + Error as PsycopgError, + IntegrityError as PsycopgIntegrityError, + OperationalError as PsycopgOperationalError, + ProgrammingError as PsycopgProgrammingError, + DeadlockDetected as PsycopgDeadlockError, +) +from rhosocial.activerecord.backend.errors import ( + ConnectionError, + ConcurrencyError, + DatabaseError, + DeadlockError, + IntegrityError, + LockError, + OperationalError, + QueryError, + attach_diagnostics, +) + if TYPE_CHECKING: from rhosocial.activerecord.backend.type_adapter import SQLTypeAdapter from rhosocial.activerecord.backend.protocols import ConcurrencyHint @@ -52,6 +71,152 @@ class PostgresBackendMixin: "57P04", # DATABASE_DROPPED } + # Fallback message markers for driver errors that carry no SQLSTATE. Text + # matching is the last resort after SQLSTATE and driver exception class. + _CONNECTION_ERROR_MESSAGE_PATTERNS = ( + "connection", + "connect", + "closed", + "terminated", + "unexpectedly", + "broken pipe", + "reset by peer", + "server closed", + "server has gone away", + "consuming input failed", + ) + _TIMEOUT_ERROR_MESSAGE_PATTERNS = ( + "timeout", + "timed out", + "canceling statement", + "cancelling statement", + "query canceled", + "query cancelled", + "statement canceled", + "statement cancelled", + ) + + def _map_postgres_error(self, error: Exception) -> Optional[Tuple[DatabaseError, str]]: + """Translate a psycopg exception into the framework error taxonomy. + + Recognition order is SQLSTATE, driver exception class, then message + text. The returned tuple is ``(framework_error, log_label)``; the + error already carries diagnostics (operation, SQLSTATE, driver code, + backend and transient hint). Non-psycopg exceptions return ``None`` + so the caller can re-raise them unchanged. + + psycopg 3.x: specific SQLSTATE classes (``UniqueViolation``, + ``DeadlockDetected``, ...) expose ``sqlstate`` on the exception. + ``pgcode`` is absent from psycopg 3, so the SQLSTATE doubles as the + driver code when no dedicated numeric code is available. + """ + error_msg = str(error) + lower_msg = error_msg.lower() + sqlstate = getattr(error, "sqlstate", None) + sqlstate = str(sqlstate).upper() if sqlstate else None + driver_code = getattr(error, "pgcode", None) or sqlstate + + def mapped( + framework_error: DatabaseError, + log_label: str, + transient: Optional[bool] = None, + ) -> Tuple[DatabaseError, str]: + return ( + attach_diagnostics( + framework_error, + operation="execute", + sqlstate=sqlstate, + driver_code=driver_code, + backend="postgres", + transient=transient, + ), + log_label, + ) + + if sqlstate == "23505": + return mapped( + IntegrityError(f"Unique constraint violation: {error_msg}"), + "Unique constraint violation", + ) + if sqlstate == "23503": + return mapped( + IntegrityError(f"Foreign key constraint violation: {error_msg}"), + "Foreign key constraint violation", + ) + if sqlstate == "23502": + return mapped( + IntegrityError(f"Not-null constraint violation: {error_msg}"), + "Not-null constraint violation", + ) + if sqlstate == "23514": + return mapped( + IntegrityError(f"Check constraint violation: {error_msg}"), + "Check constraint violation", + ) + if sqlstate == "40P01": + return mapped(DeadlockError(error_msg), "Deadlock error", transient=True) + if sqlstate == "55P03": + return mapped(LockError(error_msg), "Lock not available", transient=True) + if sqlstate == "40001": + return mapped(ConcurrencyError(error_msg), "Serialization failure", transient=True) + if sqlstate == "25006": + # A read-only transaction refuses writes by policy, not by + # transient infrastructure failure: no retry hint. + return mapped(OperationalError(error_msg), "Read-only transaction", transient=False) + if sqlstate == "57014": + return mapped(OperationalError(error_msg), "Query canceled or timeout", transient=True) + if sqlstate in {"42601", "42P01", "42703", "42883"}: + return mapped(QueryError(error_msg), "Programming error") + if sqlstate and sqlstate.startswith("23"): + return mapped(IntegrityError(error_msg), "Integrity error") + if sqlstate in self.CONNECTION_ERROR_SQLSTATES: + return mapped(OperationalError(error_msg), "Operational error", transient=True) + if sqlstate and sqlstate.startswith("28"): + # Invalid authorization / password: retrying cannot succeed. + return mapped(ConnectionError(error_msg), "Connection error", transient=False) + if sqlstate and sqlstate.startswith("57"): + # Operator intervention (shutdown, cannot connect now) is + # transient; the connection may come back. + return mapped(OperationalError(error_msg), "Operational error", transient=True) + + if isinstance(error, PsycopgDeadlockError): + return mapped(DeadlockError(error_msg), "Deadlock error", transient=True) + if isinstance(error, PsycopgIntegrityError): + if "duplicate key value violates unique constraint" in lower_msg: + return mapped( + IntegrityError(f"Unique constraint violation: {error_msg}"), + "Unique constraint violation", + ) + if "violates foreign key constraint" in lower_msg: + return mapped( + IntegrityError(f"Foreign key constraint violation: {error_msg}"), + "Foreign key constraint violation", + ) + if "null value in column" in lower_msg or "not-null constraint" in lower_msg: + return mapped( + IntegrityError(f"Not-null constraint violation: {error_msg}"), + "Not-null constraint violation", + ) + if "check constraint" in lower_msg: + return mapped( + IntegrityError(f"Check constraint violation: {error_msg}"), + "Check constraint violation", + ) + return mapped(IntegrityError(error_msg), "Integrity error") + if isinstance(error, PsycopgOperationalError): + if "deadlock detected" in lower_msg: + return mapped(DeadlockError(error_msg), "Deadlock error", transient=True) + if any(marker in lower_msg for marker in self._CONNECTION_ERROR_MESSAGE_PATTERNS): + return mapped(OperationalError(error_msg), "Operational error", transient=True) + if any(marker in lower_msg for marker in self._TIMEOUT_ERROR_MESSAGE_PATTERNS): + return mapped(OperationalError(error_msg), "Operational error", transient=True) + return mapped(OperationalError(error_msg), "Operational error") + if isinstance(error, PsycopgProgrammingError): + return mapped(QueryError(error_msg), "Programming error") + if isinstance(error, PsycopgError): + return mapped(DatabaseError(error_msg), "PostgreSQL error") + return None + def get_default_schema(self) -> str: """Get the default schema name for PostgreSQL. diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py index 6dd035bf..24d60476 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py @@ -20,12 +20,6 @@ import psycopg from psycopg.errors import Error as PsycopgError -from psycopg.errors import ( - IntegrityError as PsycopgIntegrityError, - OperationalError as PsycopgOperationalError, - ProgrammingError as PsycopgProgrammingError, - DeadlockDetected as PsycopgDeadlockError, -) from rhosocial.activerecord.backend.base import StorageBackend from rhosocial.activerecord.backend.base.execution_logging import ( @@ -38,10 +32,8 @@ from rhosocial.activerecord.backend.errors import ( ConnectionError, DatabaseError, - DeadlockError, - IntegrityError, OperationalError, - QueryError, + TransactionError, ) from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType from rhosocial.activerecord.backend.result import QueryResult @@ -415,21 +407,27 @@ def execute_many( result = QueryResult(affected_rows=affected_rows, data=None, duration=duration) + # Auto-commit before reporting success: a commit failure must surface + # as TransactionError instead of a silently uncommitted batch. + self._handle_auto_commit_if_needed() log_batch_summary(self, sql, params_list, affected_rows, duration, options) return result - except PsycopgIntegrityError as e: - log_execution_error(self, e, options) - raise IntegrityError(str(e)) from e - except PsycopgError as e: - log_execution_error(self, e, options) - raise DatabaseError(str(e)) from e except Exception as e: log_execution_error(self, e, options) - raise QueryError(str(e)) from e + if isinstance(e, DatabaseError): + raise + self._handle_error(e) + # _handle_error() is required to raise. Keep a defensive fallback so + # a backend that returns cannot turn a failure into a None result. + raise DatabaseError(str(e)) from e finally: if cursor: - cursor.close() + try: + cursor.close() + except Exception as close_error: + # Closing must never replace the error already propagating. + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") def get_server_version(self) -> Optional[Tuple[int, int, int]]: """Get PostgreSQL server version. @@ -469,7 +467,10 @@ def get_server_version(self) -> Optional[Tuple[int, int, int]]: return None finally: if cursor: - cursor.close() + try: + cursor.close() + except Exception as close_error: + self.log(logging.WARNING, f"Failed to close cursor: {close_error}") def introspect_and_adapt(self) -> None: """Introspect backend and adapt backend instance to actual server capabilities. @@ -657,77 +658,112 @@ def execute( for attempt in range(max_retries + 1): try: return super().execute(sql, params, options=options) - except OperationalError as e: - error_msg = str(e) - if self._is_connection_error_message(error_msg) and attempt < max_retries: + except OperationalError as exc: + error_msg = str(exc) + # Prefer the framework's transient hint. Only fall back to + # message matching for errors that carry no hint at all + # (transient is None); a non-transient error is never retried. + retryable = exc.transient is True or ( + exc.transient is None and self._is_connection_error_message(error_msg) + ) + if retryable and attempt < max_retries: self.log(logging.WARNING, f"Connection error on attempt {attempt + 1}, retrying... Error: {error_msg}") if self._reconnect(): continue - # If reconnection fails, break and raise the error + # Reconnection failed: re-raise instead of leaving the loop + # and turning the failure into a None result. self.log(logging.WARNING, "Failed to reconnect after connection error") - break raise def _handle_error(self, error: Exception) -> None: """Handle PostgreSQL-specific errors.""" + # Framework errors are final: they are never remapped into a driver + # error. Re-raising here also keeps errors raised before any SQL + # (TypeConversionError, ValueError, ...) from triggering a rollback. + if isinstance(error, DatabaseError): + self.log(logging.DEBUG, f"Framework error passed through: {str(error)}") + raise error + + self._try_rollback_transaction(error) + error_msg = str(error) + mapped = self._map_postgres_error(error) + if mapped is not None: + framework_error, log_label = mapped + self.log(logging.DEBUG, f"{log_label}: {error_msg}") + raise framework_error from error + self.log(logging.DEBUG, f"Unexpected error: {error_msg}") + raise error - # Try to rollback transaction to recover from aborted state - # PostgreSQL requires explicit ROLLBACK after errors in a transaction - self._try_rollback_transaction() - - if isinstance(error, PsycopgIntegrityError): - if "duplicate key value violates unique constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Unique constraint violation: {error_msg}") - raise IntegrityError(f"Unique constraint violation: {error_msg}") from error - elif "violates foreign key constraint" in error_msg.lower(): - self.log(logging.DEBUG, f"Foreign key constraint violation: {error_msg}") - raise IntegrityError(f"Foreign key constraint violation: {error_msg}") from error - self.log(logging.DEBUG, f"Integrity error: {error_msg}") - raise IntegrityError(error_msg) from error - elif isinstance(error, PsycopgOperationalError): - if "deadlock detected" in error_msg.lower(): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - self.log(logging.DEBUG, f"Operational error: {error_msg}") - raise OperationalError(error_msg) from error - elif isinstance(error, PsycopgProgrammingError): - self.log(logging.DEBUG, f"Programming error: {error_msg}") - raise QueryError(error_msg) from error - elif isinstance(error, PsycopgDeadlockError): - self.log(logging.DEBUG, f"Deadlock error: {error_msg}") - raise DeadlockError(error_msg) from error - elif isinstance(error, PsycopgError): - self.log(logging.DEBUG, f"PostgreSQL error: {error_msg}") - raise DatabaseError(error_msg) from error - else: - self.log(logging.DEBUG, f"Unexpected error: {error_msg}") - raise error + @staticmethod + def _is_rollback_error(error: BaseException) -> bool: + """Return True when the error may have aborted a server transaction. + + Only driver errors (the statement reached PostgreSQL) and transaction + errors need recovery. A framework error raised before any SQL, such as + a ``TypeConversionError`` or ``ValueError`` during parameter + preparation, leaves the transaction untouched. + """ + if isinstance(error, TransactionError): + return True + if isinstance(error, DatabaseError): + return False + return isinstance(error, PsycopgError) - def _try_rollback_transaction(self) -> None: + def _try_rollback_transaction(self, error: Optional[BaseException] = None) -> None: """Attempt to rollback transaction to recover from aborted state. PostgreSQL requires explicit ROLLBACK after query errors in a transaction. - This method attempts to recover by rolling back the transaction, regardless - of the transaction manager's state. + Only driver and transaction errors are recovered; framework errors raised + before SQL must not disturb the transaction. Outside a framework + transaction the connection runs in autocommit mode, so there is nothing + to roll back and a direct connection rollback is never attempted. + + A failed rollback is logged at ERROR and the original error keeps + propagating. + + Args: + error: The error that triggered recovery. ``None`` keeps the + historical behaviour for callers that recover without one. """ + if error is not None and not self._is_rollback_error(error): + self.log(logging.DEBUG, f"Skipping rollback for non-driver error: {error!r}") + return + + if not self.in_transaction: + self.log(logging.DEBUG, "No active transaction to roll back after error") + return + try: - # First, try the transaction manager if active - if self.in_transaction: - self.log(logging.DEBUG, "Attempting to rollback transaction after error") - self._transaction_manager.rollback() - - # Also try direct connection rollback to handle edge cases - # This ensures we recover from aborted transaction state - if self._connection and not self._connection.closed: - try: - self._connection.rollback() - self.log(logging.DEBUG, "Direct connection rollback completed") - except Exception as direct_error: - self.log(logging.DEBUG, f"Direct rollback not needed or failed: {direct_error}") - except Exception as rollback_error: - self.log(logging.WARNING, f"Failed to rollback transaction: {rollback_error}") + self.log(logging.DEBUG, "Attempting to rollback transaction after error") + self._transaction_manager.rollback() + except Exception: + # Never let recovery hide the original failure, but do not swallow + # the recovery failure silently: an aborted transaction left as-is + # is a real problem for the next statement. + self.log( + logging.ERROR, + "Failed to rollback transaction after error; " + "the transaction state may be inconsistent", + exc_info=True, + ) + return + + # Inside a transaction the manager already issued ROLLBACK. Reset the + # driver-level transaction as well, because the connection's own state + # is independent of the manager's bookkeeping. + if self._connection and not getattr(self._connection, "closed", False): + try: + self._connection.rollback() + self.log(logging.DEBUG, "Direct connection rollback completed") + except Exception: + self.log( + logging.ERROR, + "Failed to rollback connection after error; " + "the connection state may be inconsistent", + exc_info=True, + ) def _handle_auto_commit(self) -> None: """Handle auto commit based on PostgreSQL connection and transaction state. @@ -735,35 +771,40 @@ def _handle_auto_commit(self) -> None: This method will commit the current connection if: 1. The connection exists and is open 2. There is no active transaction managed by transaction_manager + 3. autocommit is disabled in the connection configuration It's used by insert/update/delete operations to ensure changes are persisted immediately when auto_commit=True is specified. + + Raises: + TransactionError: If the commit fails. A DML result must not be + reported as successful when its commit never reached the server. """ + if not self._connection: + return + + if self.in_transaction: + return + + # For PostgreSQL, if autocommit is disabled, we need to commit explicitly + if self.config.autocommit: + return + try: - # Check if connection exists - if not self._connection: - return - - # Check if we're not in an active transaction - if not self.in_transaction: - # For PostgreSQL, if autocommit is disabled, we need to commit explicitly - if not self.config.autocommit: - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") - except Exception as e: - # Just log the error but don't raise - this is a convenience feature - self.log(logging.WARNING, f"Failed to auto-commit: {str(e)}") + self._connection.commit() + except Exception as exc: + raise TransactionError("Auto-commit failed") from exc + self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") def _handle_auto_commit_if_needed(self) -> None: """ Handle auto-commit for PostgreSQL. PostgreSQL respects the autocommit setting, but we also need to handle explicit commits. + Delegates to :meth:`_handle_auto_commit` so a commit failure surfaces as + ``TransactionError`` on every path that reaches the execute result. """ - if not self.in_transaction and self._connection: - if not self.config.autocommit: - self._connection.commit() - self.log(logging.DEBUG, "Auto-committed operation (not in active transaction)") + self._handle_auto_commit() # region Advisory Lock Methods diff --git a/src/rhosocial/activerecord/backend/impl/postgres/mixins/introspection.py b/src/rhosocial/activerecord/backend/impl/postgres/mixins/introspection.py index e32cd985..fea0dd3f 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/mixins/introspection.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/mixins/introspection.py @@ -2886,8 +2886,14 @@ def format_unused_indexes_query_statement( PostgreSQL ships no vendor view for this, so the family is computed from the counters: ``idx_scan = 0`` over valid indexes in the requested schema. Only ``idx_scan`` speaks -- an index is what makes - a scan indexed, so the other counters do not apply -- and - ``indisvalid`` filters the indexes the planner would never choose. + a scan indexed, so the other counters do not apply. + + The two conditions answer different questions, so they cannot share a + relation: ``i.idx_scan = 0`` is "no scan in the statistics window", + while ``ix.indisvalid`` is "the planner would select it". The + statistics view carries only counters, so the validity flag has to + come from ``pg_index``, joined on ``indexrelid`` (the index oid on + both sides -- ``relid`` is the table oid and does not match). What the listing is not: proof. It reports "no index scan in the window the statistics remember", and that window ends at the last @@ -2912,16 +2918,25 @@ def format_unused_indexes_query_statement( ComparisonPredicate( self, "eq", - self._catalog_column("indisvalid", "i"), + self._catalog_column("indisvalid", "ix"), Literal(self, True, inline=True), ), ] - indexes = self._catalog_relation("pg_stat_user_indexes", "i") - source = JoinClause( + indexes = JoinClause( self, - left_table=indexes, - right_table=self._catalog_relation("pg_namespace", "n"), + left_table=self._catalog_relation("pg_stat_user_indexes", "i"), + right_table=self._catalog_relation("pg_index", "ix"), + join_type="JOIN", + condition=ComparisonPredicate( + self, + "eq", + self._catalog_column("indexrelid", "i"), + self._catalog_column("indexrelid", "ix"), + ), + ) + source = indexes.join( + self._catalog_relation("pg_namespace", "n"), join_type="JOIN", condition=ComparisonPredicate( self, diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/errors/test_error_taxonomy_contract.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/errors/test_error_taxonomy_contract.py new file mode 100644 index 00000000..66d3dab7 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/errors/test_error_taxonomy_contract.py @@ -0,0 +1,9 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/errors/test_error_taxonomy_contract.py +"""Run the shared error taxonomy contract against this backend. + +The testsuite module imports only the core taxonomy, so no PostgreSQL +connection is needed. Importing the module is the whole wrapper: pytest +collects the classes and methods it defines. +""" + +from rhosocial.activerecord.testsuite.feature.backend.errors.test_error_taxonomy import * # noqa: F401,F403 diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_introspection_unused_indexes.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_introspection_unused_indexes.py new file mode 100644 index 00000000..c16cf586 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_introspection_unused_indexes.py @@ -0,0 +1,85 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_introspection_unused_indexes.py +"""Unused-indexes listing: statement shape and live execution. + +The family reads two catalog relations that answer two different questions: +``pg_stat_user_indexes`` carries the counters (``idx_scan``), while the +validity flag ``indisvalid`` lives on ``pg_index``. Pointing the flag at the +statistics view makes the server reject the whole statement at parse time +(``column i.indisvalid does not exist``), so the shape assertions below pin +the relation each column belongs to, and the live cases prove the statement +parses on a real server. +""" + +import pytest + +from rhosocial.activerecord.backend.expression.introspection import ( + UnusedIndexesQueryStatement, +) +from rhosocial.activerecord.backend.impl.postgres.dialect import PostgresDialect +from rhosocial.activerecord.backend.introspection.types import UnusedIndexInfo + + +def _sql(version=(17, 0, 0), schema="public"): + return UnusedIndexesQueryStatement( + PostgresDialect(version=version), schema=schema + ).to_sql() + + +class TestUnusedIndexesStatementShape: + """The rendered statement must name columns on the relations that own them.""" + + def test_validity_comes_from_pg_index(self): + sql, _ = _sql() + assert "pg_index AS ix" in sql + assert "ix.indisvalid = TRUE" in sql + assert "i.indisvalid" not in sql + + def test_index_join_uses_the_index_oid_on_both_sides(self): + sql, _ = _sql() + assert "i.indexrelid = ix.indexrelid" in sql + # ``relid`` is the table oid; joining on it would pair the wrong rows. + assert "i.relid = ix" not in sql + + def test_counter_filter_stays_on_the_statistics_view(self): + sql, _ = _sql() + assert "i.idx_scan = %s" in sql + assert "i.idx_scan" in sql + + def test_schema_filter_and_projections_are_unchanged(self): + sql, params = _sql(schema="public") + assert "n.nspname = %s" in sql + assert "n.nspname AS \"schema_name\"" in sql + assert "i.relname AS \"table_name\"" in sql + assert "i.indexrelname AS \"index_name\"" in sql + assert params == ("public", 0) + + +class TestUnusedIndexesLive: + """The statement has to parse on a real PostgreSQL server.""" + + def test_list_unused_indexes_answers_for_a_real_schema(self, backend_with_tables): + indexes = backend_with_tables.introspector.list_unused_indexes(schema=None) + assert isinstance(indexes, list) + for index in indexes: + assert isinstance(index, UnusedIndexInfo) + assert index.schema + assert index.table_name + assert index.index_name + + +class TestAsyncUnusedIndexesLive: + """Asynchronous counterpart of the live case (same method name).""" + + @pytest.mark.asyncio + async def test_list_unused_indexes_answers_for_a_real_schema( + self, async_backend_with_tables + ): + indexes = await async_backend_with_tables.introspector.list_unused_indexes( + schema=None + ) + assert isinstance(indexes, list) + for index in indexes: + assert isinstance(index, UnusedIndexInfo) + assert index.schema + assert index.table_name + assert index.index_name diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_driver_error_mapping.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_driver_error_mapping.py new file mode 100644 index 00000000..650369c4 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_driver_error_mapping.py @@ -0,0 +1,403 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/test_driver_error_mapping.py +"""PostgreSQL driver error mapping unit tests. + +These tests instantiate psycopg exception classes and drive the backend's +``_handle_error`` on an unconnected backend. No PostgreSQL server is required. + +psycopg 3.x: SQLSTATE-bearing classes (``UniqueViolation``, +``DeadlockDetected``, ...) populate ``sqlstate`` at construction, which is +exactly what the mapper reads first. +""" + +import pytest + +psycopg_errors = pytest.importorskip("psycopg.errors") + +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + ConnectionError, + ConcurrencyError, + DatabaseError, + DeadlockError, + IntegrityError, + LockError, + OperationalError, + QueryError, +) +from rhosocial.activerecord.backend.impl.postgres.backend import ( # noqa: E402 + AsyncPostgresBackend, + PostgresBackend, +) + + +def _assert_mapped( + driver_error, + framework_error, + expected_type, + expected_sqlstate, + expected_retryable, + expected_transient, + message_fragment, +): + assert isinstance(framework_error, expected_type) + assert framework_error.sqlstate == expected_sqlstate + assert framework_error.driver_code == expected_sqlstate + assert framework_error.backend == "postgres" + assert framework_error.operation == "execute" + assert framework_error.retryable is expected_retryable + assert framework_error.transient is expected_transient + assert framework_error.__cause__ is driver_error + if message_fragment is not None: + assert message_fragment in str(framework_error) + + +def _expect_mapped( + backend, + driver_error, + expected_type, + expected_sqlstate, + expected_retryable, + expected_transient, + message_fragment, +): + with pytest.raises(expected_type) as info: + backend._handle_error(driver_error) + _assert_mapped( + driver_error, + info.value, + expected_type, + expected_sqlstate, + expected_retryable, + expected_transient, + message_fragment, + ) + + +async def _expect_mapped_async( + backend, + driver_error, + expected_type, + expected_sqlstate, + expected_retryable, + expected_transient, + message_fragment, +): + with pytest.raises(expected_type) as info: + await backend._handle_error(driver_error) + _assert_mapped( + driver_error, + info.value, + expected_type, + expected_sqlstate, + expected_retryable, + expected_transient, + message_fragment, + ) + + +SQLSTATE_CASES = [ + pytest.param( + lambda e: e.UniqueViolation( + "duplicate key value violates unique constraint users_email_key" + ), + IntegrityError, + "23505", + False, + None, + "Unique constraint violation", + id="unique-violation", + ), + pytest.param( + lambda e: e.ForeignKeyViolation( + "insert or update violates foreign key constraint fk_orders_user" + ), + IntegrityError, + "23503", + False, + None, + "Foreign key constraint violation", + id="foreign-key-violation", + ), + pytest.param( + lambda e: e.NotNullViolation( + 'null value in column "email" violates not-null constraint' + ), + IntegrityError, + "23502", + False, + None, + "Not-null constraint violation", + id="not-null-violation", + ), + pytest.param( + lambda e: e.CheckViolation('new row violates check constraint "age_positive"'), + IntegrityError, + "23514", + False, + None, + "Check constraint violation", + id="check-violation", + ), + pytest.param( + lambda e: e.DeadlockDetected("deadlock detected"), + DeadlockError, + "40P01", + True, + True, + "deadlock detected", + id="deadlock", + ), + pytest.param( + lambda e: e.SerializationFailure( + "could not serialize access due to concurrent update" + ), + ConcurrencyError, + "40001", + True, + True, + "could not serialize access", + id="serialization-failure", + ), + pytest.param( + lambda e: e.LockNotAvailable("could not obtain lock on row in relation orders"), + LockError, + "55P03", + True, + True, + "could not obtain lock", + id="lock-not-available", + ), + pytest.param( + lambda e: e.ReadOnlySqlTransaction( + "cannot execute UPDATE in a read-only transaction" + ), + OperationalError, + "25006", + False, + False, + "read-only transaction", + id="read-only-transaction", + ), + pytest.param( + lambda e: e.SyntaxError('syntax error at or near "SELEC"'), + QueryError, + "42601", + False, + None, + "syntax error", + id="syntax-error", + ), + pytest.param( + lambda e: e.UndefinedTable('relation "missing_table" does not exist'), + QueryError, + "42P01", + False, + None, + "does not exist", + id="undefined-table", + ), + pytest.param( + lambda e: e.UndefinedColumn('column "missing_column" does not exist'), + QueryError, + "42703", + False, + None, + "does not exist", + id="undefined-column", + ), + pytest.param( + lambda e: e.UndefinedFunction("function missing_fn(integer) does not exist"), + QueryError, + "42883", + False, + None, + "does not exist", + id="undefined-function", + ), + pytest.param( + lambda e: e.QueryCanceled("canceling statement due to statement timeout"), + OperationalError, + "57014", + True, + True, + "canceling statement", + id="query-canceled", + ), + pytest.param( + lambda e: e.ConnectionFailure("server closed the connection unexpectedly"), + OperationalError, + "08006", + True, + True, + "server closed the connection", + id="connection-failure", + ), + pytest.param( + lambda e: e.InvalidPassword('password authentication failed for user "app"'), + ConnectionError, + "28P01", + False, + False, + "password authentication failed", + id="invalid-password", + ), + pytest.param( + lambda e: e.Error("some PostgreSQL failure"), + DatabaseError, + None, + False, + None, + "some PostgreSQL failure", + id="generic-error", + ), +] + + +@pytest.fixture +def backend(): + """An unconnected synchronous backend.""" + return PostgresBackend(database="error_mapping_unit") + + +@pytest.fixture +def async_backend(): + """An unconnected asynchronous backend.""" + return AsyncPostgresBackend(database="error_mapping_unit") + + +class TestSyncDriverErrorMapping: + @pytest.mark.parametrize( + "factory,expected_type,sqlstate,retryable,transient,message_fragment", + SQLSTATE_CASES, + ) + def test_sqlstate_and_class_mapping( + self, + backend, + factory, + expected_type, + sqlstate, + retryable, + transient, + message_fragment, + ): + driver_error = factory(psycopg_errors) + _expect_mapped( + backend, + driver_error, + expected_type, + sqlstate, + retryable, + transient, + message_fragment, + ) + + def test_message_fallback_for_deadlock(self, backend): + driver_error = psycopg_errors.OperationalError("deadlock detected") + _expect_mapped( + backend, + driver_error, + DeadlockError, + None, + True, + True, + "deadlock detected", + ) + + def test_message_fallback_for_connection_loss(self, backend): + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + _expect_mapped( + backend, + driver_error, + OperationalError, + None, + True, + True, + "server closed the connection", + ) + + def test_message_fallback_for_integrity_constraints(self, backend): + errors = [ + ( + psycopg_errors.IntegrityError( + "duplicate key value violates unique constraint users_pkey" + ), + "Unique constraint violation", + ), + ( + psycopg_errors.IntegrityError( + "insert violates foreign key constraint fk_orders_user" + ), + "Foreign key constraint violation", + ), + ] + for driver_error, fragment in errors: + _expect_mapped( + backend, + driver_error, + IntegrityError, + None, + False, + None, + fragment, + ) + + def test_non_psycopg_error_is_reraised(self, backend): + class UnexpectedError(Exception): + pass + + driver_error = UnexpectedError("unexpected failure") + with pytest.raises(UnexpectedError) as info: + backend._handle_error(driver_error) + assert info.value is driver_error + + +class TestAsyncDriverErrorMapping: + @pytest.mark.asyncio + @pytest.mark.parametrize( + "factory,expected_type,sqlstate,retryable,transient,message_fragment", + SQLSTATE_CASES, + ) + async def test_sqlstate_and_class_mapping( + self, + async_backend, + factory, + expected_type, + sqlstate, + retryable, + transient, + message_fragment, + ): + driver_error = factory(psycopg_errors) + await _expect_mapped_async( + async_backend, + driver_error, + expected_type, + sqlstate, + retryable, + transient, + message_fragment, + ) + + @pytest.mark.asyncio + async def test_message_fallback_for_deadlock(self, async_backend): + driver_error = psycopg_errors.OperationalError("deadlock detected") + await _expect_mapped_async( + async_backend, + driver_error, + DeadlockError, + None, + True, + True, + "deadlock detected", + ) + + @pytest.mark.asyncio + async def test_non_psycopg_error_is_reraised(self, async_backend): + class UnexpectedError(Exception): + pass + + driver_error = UnexpectedError("unexpected failure") + with pytest.raises(UnexpectedError) as info: + await async_backend._handle_error(driver_error) + assert info.value is driver_error diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_execute_error_contract.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_execute_error_contract.py new file mode 100644 index 00000000..ec48f4be --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_execute_error_contract.py @@ -0,0 +1,624 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/test_execute_error_contract.py +"""PostgreSQL execute-phase error contract tests. + +No PostgreSQL server is required: the tests instantiate unconnected backends +and drive them with fake connections, cursors and transaction managers. The +psycopg test cases are skipped when the driver is not installed. +""" + +import logging + +import pytest + +psycopg_errors = pytest.importorskip("psycopg.errors") + +from rhosocial.activerecord.backend.base import ( # noqa: E402 + AsyncStorageBackend, + StorageBackend, +) +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + DeadlockError, + OperationalError, + TransactionError, + TypeConversionError, +) +from rhosocial.activerecord.backend.impl.postgres.backend import ( # noqa: E402 + AsyncPostgresBackend, + PostgresBackend, +) +from rhosocial.activerecord.backend.result import QueryResult # noqa: E402 + + +class _FakeCursor: + """Minimal synchronous cursor that records nothing.""" + + def __init__(self, execute_error=None): + self.closed = False + self.rowcount = 0 + self.execute_error = execute_error + + def execute(self, *args, **kwargs): + if self.execute_error is not None: + raise self.execute_error + return None + + def close(self): + self.closed = True + + +class _FakeConnection: + """Synchronous connection double with a controllable commit/rollback.""" + + closed = False + broken = False + + def __init__(self, commit_error=None, cursor=None): + self.commit_error = commit_error + self.commit_calls = 0 + self.rollback_calls = 0 + self._cursor = cursor or _FakeCursor() + + def cursor(self): + return self._cursor + + def commit(self): + self.commit_calls += 1 + if self.commit_error is not None: + raise self.commit_error + + def rollback(self): + self.rollback_calls += 1 + + +class _FakeTransactionManager: + """Transaction manager double exposing only what the backend reads.""" + + def __init__(self, active=True, rollback_error=None): + self.is_active = active + self.rollback_error = rollback_error + self.rollback_calls = 0 + + def rollback(self): + self.rollback_calls += 1 + if self.rollback_error is not None: + raise self.rollback_error + + +class _AsyncFakeCursor: + def __init__(self, execute_error=None): + self.closed = False + self.rowcount = 0 + self.execute_error = execute_error + + async def execute(self, *args, **kwargs): + if self.execute_error is not None: + raise self.execute_error + return None + + async def close(self): + self.closed = True + + +class _AsyncFakeConnection: + closed = False + broken = False + + def __init__(self, commit_error=None, cursor=None): + self.commit_error = commit_error + self.commit_calls = 0 + self.rollback_calls = 0 + self._cursor = cursor or _AsyncFakeCursor() + + def cursor(self): + return self._cursor + + async def commit(self): + self.commit_calls += 1 + if self.commit_error is not None: + raise self.commit_error + + async def rollback(self): + self.rollback_calls += 1 + + +class _AsyncFakeTransactionManager: + def __init__(self, active=True, rollback_error=None): + self.is_active = active + self.rollback_error = rollback_error + self.rollback_calls = 0 + + async def rollback(self): + self.rollback_calls += 1 + if self.rollback_error is not None: + raise self.rollback_error + + +def _operational_error(message, transient): + error = OperationalError(message) + error.transient = transient + return error + + +@pytest.fixture +def backend(): + """An unconnected synchronous backend.""" + return PostgresBackend(database="execute_contract_unit") + + +@pytest.fixture +def async_backend(): + """An unconnected asynchronous backend.""" + return AsyncPostgresBackend(database="execute_contract_unit") + + +class TestSyncAutoCommitContract: + @pytest.mark.parametrize( + "method_name", + ["_handle_auto_commit", "_handle_auto_commit_if_needed"], + ) + def test_commit_failure_raises_transaction_error(self, backend, method_name): + commit_error = RuntimeError("connection lost during commit") + backend._connection = _FakeConnection(commit_error=commit_error) + backend._transaction_manager = _FakeTransactionManager(active=False) + + with pytest.raises(TransactionError) as info: + getattr(backend, method_name)() + + assert info.value.__cause__ is commit_error + assert str(info.value) == "Auto-commit failed" + + def test_autocommit_config_still_skips_commit(self, backend): + backend._connection = _FakeConnection(commit_error=RuntimeError("must not commit")) + backend._transaction_manager = _FakeTransactionManager(active=False) + backend.config.autocommit = True + + backend._handle_auto_commit() + + def test_active_transaction_is_not_auto_committed(self, backend): + backend._connection = _FakeConnection(commit_error=RuntimeError("must not commit")) + backend._transaction_manager = _FakeTransactionManager(active=True) + + backend._handle_auto_commit() + + +class TestSyncErrorPhaseContract: + def test_framework_error_is_not_remapped_and_does_not_roll_back(self, backend): + manager = _FakeTransactionManager(active=True) + backend._transaction_manager = manager + error = TypeConversionError("cannot convert value") + + with pytest.raises(TypeConversionError) as info: + backend._handle_error(error) + + assert info.value is error + assert manager.rollback_calls == 0 + + def test_prepare_parameters_failure_does_not_roll_back(self, backend, monkeypatch): + manager = _FakeTransactionManager(active=True) + backend._connection = _FakeConnection() + backend._transaction_manager = manager + conversion_error = TypeConversionError("cannot convert value") + + def failing_prepare(params, param_adapters): + raise conversion_error + + monkeypatch.setattr(backend, "prepare_parameters", failing_prepare) + + with pytest.raises(TypeConversionError) as info: + backend.execute("SELECT 1", (1,)) + + assert info.value is conversion_error + assert manager.rollback_calls == 0 + + def test_driver_error_in_transaction_rolls_back_once(self, backend): + manager = _FakeTransactionManager(active=True) + connection = _FakeConnection() + backend._transaction_manager = manager + backend._connection = connection + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with pytest.raises(OperationalError) as info: + backend._handle_error(driver_error) + + assert isinstance(info.value, OperationalError) + assert info.value.__cause__ is driver_error + # The transaction manager rollback is the authoritative recovery path. + assert manager.rollback_calls == 1 + + def test_driver_error_outside_transaction_does_not_roll_back(self, backend): + manager = _FakeTransactionManager(active=False) + connection = _FakeConnection() + backend._transaction_manager = manager + backend._connection = connection + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with pytest.raises(OperationalError): + backend._handle_error(driver_error) + + assert manager.rollback_calls == 0 + assert connection.rollback_calls == 0 + + def test_rollback_failure_is_logged_at_error_and_original_propagates( + self, backend, caplog + ): + rollback_error = RuntimeError("rollback exploded") + manager = _FakeTransactionManager(active=True, rollback_error=rollback_error) + backend._transaction_manager = manager + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with caplog.at_level( + logging.ERROR, logger="rhosocial.activerecord.backend.postgres" + ): + with pytest.raises(OperationalError) as info: + backend._handle_error(driver_error) + + assert manager.rollback_calls == 1 + assert info.value.__cause__ is driver_error + error_records = [record for record in caplog.records if record.levelno == logging.ERROR] + assert any("rollback" in record.getMessage().lower() for record in error_records) + + +class TestSyncRetryContract: + def test_transient_operational_error_is_retried(self, backend, monkeypatch): + result = QueryResult(affected_rows=1) + calls = {"execute": 0, "reconnect": 0} + + def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + if calls["execute"] == 1: + raise _operational_error( + "server closed the connection unexpectedly", transient=True + ) + return result + + def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(StorageBackend, "execute", fake_execute) + monkeypatch.setattr(backend, "_reconnect", fake_reconnect) + + assert backend.execute("SELECT 1") is result + assert calls["execute"] == 2 + assert calls["reconnect"] == 1 + + def test_non_transient_operational_error_is_not_retried(self, backend, monkeypatch): + calls = {"execute": 0, "reconnect": 0} + failure = _operational_error("password authentication failed", transient=False) + + def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + raise failure + + def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(StorageBackend, "execute", fake_execute) + monkeypatch.setattr(backend, "_reconnect", fake_reconnect) + + with pytest.raises(OperationalError) as info: + backend.execute("SELECT 1") + + assert info.value is failure + assert calls["execute"] == 1 + assert calls["reconnect"] == 0 + + def test_untagged_connection_message_still_retries(self, backend, monkeypatch): + result = QueryResult(affected_rows=1) + calls = {"execute": 0, "reconnect": 0} + + def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + if calls["execute"] == 1: + raise _operational_error( + "server closed the connection unexpectedly", transient=None + ) + return result + + def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(StorageBackend, "execute", fake_execute) + monkeypatch.setattr(backend, "_reconnect", fake_reconnect) + + assert backend.execute("SELECT 1") is result + assert calls["execute"] == 2 + assert calls["reconnect"] == 1 + + +class TestSyncExecuteManyContract: + def test_commit_failure_raises_transaction_error(self, backend): + commit_error = RuntimeError("commit failed") + cursor = _FakeCursor() + backend._connection = _FakeConnection(commit_error=commit_error, cursor=cursor) + backend._transaction_manager = _FakeTransactionManager(active=False) + + with pytest.raises(TransactionError) as info: + backend.execute_many("UPDATE users SET name = %s", [("alice",)]) + + assert info.value.__cause__ is commit_error + assert str(info.value) == "Auto-commit failed" + assert cursor.closed is True + + def test_commit_is_attempted_on_success(self, backend): + connection = _FakeConnection() + backend._connection = connection + backend._transaction_manager = _FakeTransactionManager(active=False) + + result = backend.execute_many( + "UPDATE users SET name = %s", [("alice",), ("bob",)] + ) + + assert result.affected_rows == 0 + assert connection.commit_calls == 1 + + def test_deadlock_maps_to_deadlock_error(self, backend): + cursor = _FakeCursor( + execute_error=psycopg_errors.DeadlockDetected("deadlock detected") + ) + connection = _FakeConnection(cursor=cursor) + manager = _FakeTransactionManager(active=False) + backend._connection = connection + backend._transaction_manager = manager + + with pytest.raises(DeadlockError) as info: + backend.execute_many("UPDATE users SET name = %s", [("alice",)]) + + assert isinstance(info.value, DeadlockError) + assert manager.rollback_calls == 0 + assert connection.commit_calls == 0 + assert cursor.closed is True + + +class TestAsyncAutoCommitContract: + @pytest.mark.asyncio + @pytest.mark.parametrize( + "method_name", + ["_handle_auto_commit", "_handle_auto_commit_if_needed"], + ) + async def test_commit_failure_raises_transaction_error(self, async_backend, method_name): + commit_error = RuntimeError("connection lost during commit") + async_backend._connection = _AsyncFakeConnection(commit_error=commit_error) + async_backend._transaction_manager = _AsyncFakeTransactionManager(active=False) + + with pytest.raises(TransactionError) as info: + await getattr(async_backend, method_name)() + + assert info.value.__cause__ is commit_error + assert str(info.value) == "Auto-commit failed" + + @pytest.mark.asyncio + async def test_autocommit_config_still_skips_commit(self, async_backend): + async_backend._connection = _AsyncFakeConnection( + commit_error=RuntimeError("must not commit") + ) + async_backend._transaction_manager = _AsyncFakeTransactionManager(active=False) + async_backend.config.autocommit = True + + await async_backend._handle_auto_commit() + + +class TestAsyncErrorPhaseContract: + @pytest.mark.asyncio + async def test_framework_error_is_not_remapped_and_does_not_roll_back( + self, async_backend + ): + manager = _AsyncFakeTransactionManager(active=True) + async_backend._transaction_manager = manager + error = TypeConversionError("cannot convert value") + + with pytest.raises(TypeConversionError) as info: + await async_backend._handle_error(error) + + assert info.value is error + assert manager.rollback_calls == 0 + + @pytest.mark.asyncio + async def test_prepare_parameters_failure_does_not_roll_back( + self, async_backend, monkeypatch + ): + manager = _AsyncFakeTransactionManager(active=True) + async_backend._connection = _AsyncFakeConnection() + async_backend._transaction_manager = manager + conversion_error = TypeConversionError("cannot convert value") + + def failing_prepare(params, param_adapters): + raise conversion_error + + monkeypatch.setattr(async_backend, "prepare_parameters", failing_prepare) + + with pytest.raises(TypeConversionError) as info: + await async_backend.execute("SELECT 1", (1,)) + + assert info.value is conversion_error + assert manager.rollback_calls == 0 + + @pytest.mark.asyncio + async def test_driver_error_in_transaction_rolls_back_once(self, async_backend): + manager = _AsyncFakeTransactionManager(active=True) + connection = _AsyncFakeConnection() + async_backend._transaction_manager = manager + async_backend._connection = connection + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with pytest.raises(OperationalError) as info: + await async_backend._handle_error(driver_error) + + assert isinstance(info.value, OperationalError) + assert info.value.__cause__ is driver_error + # The transaction manager rollback is the authoritative recovery path. + assert manager.rollback_calls == 1 + + @pytest.mark.asyncio + async def test_driver_error_outside_transaction_does_not_roll_back(self, async_backend): + manager = _AsyncFakeTransactionManager(active=False) + connection = _AsyncFakeConnection() + async_backend._transaction_manager = manager + async_backend._connection = connection + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with pytest.raises(OperationalError): + await async_backend._handle_error(driver_error) + + assert manager.rollback_calls == 0 + assert connection.rollback_calls == 0 + + @pytest.mark.asyncio + async def test_rollback_failure_is_logged_at_error_and_original_propagates( + self, async_backend, caplog + ): + rollback_error = RuntimeError("rollback exploded") + manager = _AsyncFakeTransactionManager(active=True, rollback_error=rollback_error) + async_backend._transaction_manager = manager + driver_error = psycopg_errors.OperationalError( + "server closed the connection unexpectedly" + ) + + with caplog.at_level( + logging.ERROR, logger="rhosocial.activerecord.backend.postgres" + ): + with pytest.raises(OperationalError) as info: + await async_backend._handle_error(driver_error) + + assert manager.rollback_calls == 1 + assert info.value.__cause__ is driver_error + error_records = [record for record in caplog.records if record.levelno == logging.ERROR] + assert any("rollback" in record.getMessage().lower() for record in error_records) + + +class TestAsyncRetryContract: + @pytest.mark.asyncio + async def test_transient_operational_error_is_retried(self, async_backend, monkeypatch): + result = QueryResult(affected_rows=1) + calls = {"execute": 0, "reconnect": 0} + + async def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + if calls["execute"] == 1: + raise _operational_error( + "server closed the connection unexpectedly", transient=True + ) + return result + + async def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(AsyncStorageBackend, "execute", fake_execute) + monkeypatch.setattr(async_backend, "_reconnect", fake_reconnect) + + assert await async_backend.execute("SELECT 1") is result + assert calls["execute"] == 2 + assert calls["reconnect"] == 1 + + @pytest.mark.asyncio + async def test_non_transient_operational_error_is_not_retried( + self, async_backend, monkeypatch + ): + calls = {"execute": 0, "reconnect": 0} + failure = _operational_error("password authentication failed", transient=False) + + async def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + raise failure + + async def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(AsyncStorageBackend, "execute", fake_execute) + monkeypatch.setattr(async_backend, "_reconnect", fake_reconnect) + + with pytest.raises(OperationalError) as info: + await async_backend.execute("SELECT 1") + + assert info.value is failure + assert calls["execute"] == 1 + assert calls["reconnect"] == 0 + + @pytest.mark.asyncio + async def test_untagged_connection_message_still_retries(self, async_backend, monkeypatch): + result = QueryResult(affected_rows=1) + calls = {"execute": 0, "reconnect": 0} + + async def fake_execute(self, sql, params=None, *, options=None): + calls["execute"] += 1 + if calls["execute"] == 1: + raise _operational_error( + "server closed the connection unexpectedly", transient=None + ) + return result + + async def fake_reconnect(): + calls["reconnect"] += 1 + return True + + monkeypatch.setattr(AsyncStorageBackend, "execute", fake_execute) + monkeypatch.setattr(async_backend, "_reconnect", fake_reconnect) + + assert await async_backend.execute("SELECT 1") is result + assert calls["execute"] == 2 + assert calls["reconnect"] == 1 + + +class TestAsyncExecuteManyContract: + @pytest.mark.asyncio + async def test_commit_failure_raises_transaction_error(self, async_backend): + commit_error = RuntimeError("commit failed") + cursor = _AsyncFakeCursor() + async_backend._connection = _AsyncFakeConnection( + commit_error=commit_error, cursor=cursor + ) + async_backend._transaction_manager = _AsyncFakeTransactionManager(active=False) + + with pytest.raises(TransactionError) as info: + await async_backend.execute_many("UPDATE users SET name = %s", [("alice",)]) + + assert info.value.__cause__ is commit_error + assert str(info.value) == "Auto-commit failed" + assert cursor.closed is True + + @pytest.mark.asyncio + async def test_commit_is_attempted_on_success(self, async_backend): + connection = _AsyncFakeConnection() + async_backend._connection = connection + async_backend._transaction_manager = _AsyncFakeTransactionManager(active=False) + + result = await async_backend.execute_many( + "UPDATE users SET name = %s", [("alice",), ("bob",)] + ) + + assert result.affected_rows == 0 + assert connection.commit_calls == 1 + + @pytest.mark.asyncio + async def test_deadlock_maps_to_deadlock_error(self, async_backend): + cursor = _AsyncFakeCursor( + execute_error=psycopg_errors.DeadlockDetected("deadlock detected") + ) + connection = _AsyncFakeConnection(cursor=cursor) + manager = _AsyncFakeTransactionManager(active=False) + async_backend._connection = connection + async_backend._transaction_manager = manager + + with pytest.raises(DeadlockError) as info: + await async_backend.execute_many( + "UPDATE users SET name = %s", [("alice",)] + ) + + assert isinstance(info.value, DeadlockError) + assert manager.rollback_calls == 0 + assert connection.commit_calls == 0 + assert cursor.closed is True diff --git a/tests/rhosocial/activerecord_postgres_test/feature/logging/test_execution_logging_unit.py b/tests/rhosocial/activerecord_postgres_test/feature/logging/test_execution_logging_unit.py index 5df170ec..a7345057 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/logging/test_execution_logging_unit.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/logging/test_execution_logging_unit.py @@ -143,11 +143,15 @@ class FakeConnection: def __init__(self, cursor) -> None: self._cursor = cursor + self.commit_calls = 0 self.rollback_calls = 0 def cursor(self): return self._cursor + def commit(self) -> None: + self.commit_calls += 1 + def rollback(self) -> None: self.rollback_calls += 1 @@ -155,6 +159,9 @@ def rollback(self) -> None: class FakeAsyncConnection(FakeConnection): """Minimal asynchronous connection, open and healthy enough for ``_get_cursor``.""" + async def commit(self) -> None: + self.commit_calls += 1 + async def rollback(self) -> None: self.rollback_calls += 1 @@ -319,8 +326,10 @@ def test_handle_error_maps_driver_errors(self, sync_backend, driver_error, monke sync_backend._handle_error(error) assert fragment in _messages(caplog) - # Active transaction rolls back through the manager, then the connection. - assert connection.rollback_calls == 1 + # Driver errors recover through the manager and reset the connection; + # a non-driver error (UnexpectedError) must leave the transaction alone. + expected_rollback_calls = 0 if isinstance(error, UnexpectedError) else 1 + assert connection.rollback_calls == expected_rollback_calls def test_defer_constraint_logs_deferred_name(self, sync_backend, monkeypatch, caplog): """defer_constraint logs the accepted constraint once it is registered.""" @@ -413,7 +422,10 @@ async def _noop() -> None: await async_backend._handle_error(error) assert fragment in _messages(caplog) - assert connection.rollback_calls == 1 + # Driver errors recover through the manager and reset the connection; + # a non-driver error (UnexpectedError) must leave the transaction alone. + expected_rollback_calls = 0 if isinstance(error, UnexpectedError) else 1 + assert connection.rollback_calls == expected_rollback_calls @pytest.mark.asyncio async def test_defer_constraint_logs_deferred_name(self, async_backend, monkeypatch, caplog):