FIX: GH-Issue (#532) - Login failures are raised as RuntimeError instead of a mssql_python exception

mssql_python/connection.py

@@ -38,6 +38,7 @@
  InternalError,
  ProgrammingError,
  NotSupportedError,
+ sqlstate_to_exception,
 )
 from mssql_python.auth import extract_auth_type, process_connection_string
 from mssql_python.constants import ConstantsDDBC, GetInfoConstants
@@ -57,6 +58,42 @@
 # Note: "utf-16" with BOM is NOT included as it's problematic for SQL_WCHAR
 UTF16_ENCODINGS: frozenset[str] = frozenset(["utf-16le", "utf-16be"])
 
+_SQLSTATE_RE = re.compile(r"^SQLSTATE:([A-Z0-9]{0,5}):(.*)", re.DOTALL)
+
+
+def _raise_connection_error(e: RuntimeError) -> None:
+ """Map a RuntimeError from the C++ pybind layer to the correct DB-API 2.0 exception.
+
+ Connection::checkError() throws "SQLSTATE:XXXXX:<odbc_message>" so the SQLSTATE
+ can be mapped via sqlstate_to_exception(), consistent with cursor-level error handling.
+ """
+ error_msg = str(e)
+ match = _SQLSTATE_RE.match(error_msg)
+ if match:
+ sqlstate, ddbc_error = match.group(1), match.group(2)
+ # Handle malformed SQLSTATE prefix (empty or invalid code)
+ if not sqlstate or len(sqlstate) != 5:
+ logger.error("Connection error (malformed SQLSTATE): %s", ddbc_error)
+ raise OperationalError(
+ driver_error="Connection operation failed",
+ ddbc_error=ddbc_error,
+ ) from None
+ exc = sqlstate_to_exception(sqlstate, ddbc_error)
+ if exc is None:
+ logger.error("Unknown SQLSTATE %s, raising DatabaseError", sqlstate)
+ raise DatabaseError(
+ driver_error=f"An error occurred with SQLSTATE code: {sqlstate}",
+ ddbc_error=ddbc_error,
+ ) from None
+ logger.error("Connection error (SQLSTATE %s): %s", sqlstate, ddbc_error)
+ raise exc from None
+ # Fallback: no SQLSTATE prefix — e.g. "Connection handle not allocated"
+ logger.error("Connection error: %s", error_msg)
+ raise OperationalError(
+ driver_error="Connection operation failed",
+ ddbc_error=error_msg,
+ ) from None
+
 
 def _validate_utf16_wchar_compatibility(
  encoding: str, wchar_type: int, context: str = "SQL_WCHAR"
@@ -333,9 +370,12 @@ def __init__(
  if not PoolingManager.is_initialized():
  PoolingManager.enable()
  self._pooling = PoolingManager.is_enabled()
- self._conn = ddbc_bindings.Connection(
- self.connection_str, self._pooling, self._attrs_before
- )
+ try:
+ self._conn = ddbc_bindings.Connection(
+ self.connection_str, self._pooling, self._attrs_before
+ )
+ except RuntimeError as e:
+ _raise_connection_error(e)
  self.setautocommit(autocommit)
 
  # Register this connection for cleanup before Python shutdown
@@ -456,7 +496,10 @@ def autocommit(self) -> bool:
  Returns:
  bool: True if autocommit is enabled, False otherwise.
  """
- return self._conn.get_autocommit()
+ try:
+ return self._conn.get_autocommit()
+ except RuntimeError as e:
+ _raise_connection_error(e)
 
  @autocommit.setter
  def autocommit(self, value: bool) -> None:
@@ -496,7 +539,10 @@ def setautocommit(self, value: bool = False) -> None:
  Raises:
  DatabaseError: If there is an error while setting the autocommit mode.
  """
- self._conn.set_autocommit(value)
+ try:
+ self._conn.set_autocommit(value)
+ except RuntimeError as e:
+ _raise_connection_error(e)
 
  def setencoding(self, encoding: Optional[str] = None, ctype: Optional[int] = None) -> None:
  """
@@ -1491,7 +1537,10 @@ def commit(self) -> None:
  )
 
  # Commit the current transaction
- self._conn.commit()
+ try:
+ self._conn.commit()
+ except RuntimeError as e:
+ _raise_connection_error(e)
  logger.info("Transaction committed successfully.")
 
  def rollback(self) -> None:
@@ -1514,7 +1563,10 @@ def rollback(self) -> None:
  )
 
  # Roll back the current transaction
- self._conn.rollback()
+ try:
+ self._conn.rollback()
+ except RuntimeError as e:
+ _raise_connection_error(e)
  logger.info("Transaction rolled back successfully.")
 
  def close(self) -> None:
@@ -1570,7 +1622,11 @@ def close(self) -> None:
  # For autocommit True, this is not necessary as each statement is
  # committed immediately
  logger.debug("Rolling back uncommitted changes before closing connection.")
- self._conn.rollback()
+ try:
+ self._conn.rollback()
+ except RuntimeError as e:
+ # Handle C++ layer RuntimeError with proper DB-API exception mapping
+ _raise_connection_error(e)
  # TODO: Check potential race conditions in case of multithreaded scenarios
  # Close the connection
  self._conn.close()

mssql_python/pybind/connection/connection.cpp

@@ -170,8 +170,17 @@ void Connection::disconnect() {
 // DB spec compliant
 void Connection::checkError(SQLRETURN ret) const {
  if (!SQL_SUCCEEDED(ret)) {
+ // Format: "SQLSTATE:XXXXX:<odbc_error_message>" — parsed by _raise_connection_error()
  ErrorInfo err = SQLCheckError_Wrap(SQL_HANDLE_DBC, _dbcHandle, ret);
- ThrowStdException(err.ddbcErrorMsg);
+ std::string sqlState = err.sqlState;
+ std::string errorMsg = err.ddbcErrorMsg;
+ // Only add SQLSTATE prefix if we have a valid 5-character code
+ if (sqlState.length() == 5) {
+ ThrowStdException("SQLSTATE:" + sqlState + ":" + errorMsg);
+ } else {
+ // No valid SQLSTATE (e.g., SQL_INVALID_HANDLE) — throw clean error message
+ ThrowStdException(errorMsg);
+ }
  }
 }
 

tests/test_006_exceptions.py

@@ -190,6 +190,92 @@ def test_connection_error():
  assert "Incomplete specification" in str(excinfo.value) or "has no value" in str(excinfo.value)
 
 
+def test_connect_runtime_error_mapped_to_correct_dbapi_exception():
+ """Regression test for https://github.com/microsoft/mssql-python/issues/532.
+
+ Connection failures from the C++ pybind layer (RuntimeError) must be mapped
+ to the correct DB-API 2.0 exception via the embedded SQLSTATE code.
+ This covers connect(), commit(), and rollback() — all of which go through
+ Connection::checkError() in the C++ layer.
+ """
+ from unittest.mock import MagicMock, patch
+
+ # SQLSTATE 28000 = login failed -> OperationalError
+ with patch(
+ "mssql_python.connection.ddbc_bindings.Connection",
+ side_effect=RuntimeError("SQLSTATE:28000:Login failed for user 'baduser'."),
+ ):
+ with pytest.raises(OperationalError) as exc_info:
+ connect("Server=testserver;Database=mydb;UID=baduser;PWD=wrongpassword;")
+ assert "Login failed for user" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # SQLSTATE IM002 = driver not found -> OperationalError (per SQLSTATE mapping)
+ with patch(
+ "mssql_python.connection.ddbc_bindings.Connection",
+ side_effect=RuntimeError(
+ "SQLSTATE:IM002:Data source name not found and no default driver specified"
+ ),
+ ):
+ with pytest.raises(OperationalError) as exc_info:
+ connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ assert "no default driver" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # No SQLSTATE prefix -> fallback OperationalError
+ with patch(
+ "mssql_python.connection.ddbc_bindings.Connection",
+ side_effect=RuntimeError("Connection handle not allocated"),
+ ):
+ with pytest.raises(OperationalError) as exc_info:
+ connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ assert "Connection handle not allocated" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # Unmapped SQLSTATE (sqlstate_to_exception returns None) -> DatabaseError
+ with patch(
+ "mssql_python.connection.ddbc_bindings.Connection",
+ side_effect=RuntimeError("SQLSTATE:99999:Unknown error with unmapped code"),
+ ):
+ with pytest.raises(DatabaseError) as exc_info:
+ connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ assert "Unknown error with unmapped code" in exc_info.value.ddbc_error
+ assert "SQLSTATE code: 99999" in exc_info.value.driver_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # Malformed SQLSTATE (empty code) -> OperationalError
+ with patch(
+ "mssql_python.connection.ddbc_bindings.Connection",
+ side_effect=RuntimeError("SQLSTATE::Invalid handle!"),
+ ):
+ with pytest.raises(OperationalError) as exc_info:
+ connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ assert "Invalid handle!" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # commit() failure -> OperationalError via same _raise_connection_error path
+ mock_conn = MagicMock()
+ mock_conn.commit.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure")
+ mock_conn.get_autocommit.return_value = False
+ with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+ conn = connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ with pytest.raises(OperationalError) as exc_info:
+ conn.commit()
+ assert "Communication link failure" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+ # rollback() failure -> OperationalError via same _raise_connection_error path
+ mock_conn2 = MagicMock()
+ mock_conn2.rollback.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure")
+ mock_conn2.get_autocommit.return_value = False
+ with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn2):
+ conn2 = connect("Server=testserver;Database=mydb;UID=u;PWD=p;")
+ with pytest.raises(OperationalError) as exc_info:
+ conn2.rollback()
+ assert "Communication link failure" in exc_info.value.ddbc_error
+ assert not isinstance(exc_info.value, RuntimeError)
+
+
 def test_truncate_error_message_successful_cases():
  """Test truncate_error_message with valid Microsoft messages for comparison."""