diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cac01d4d..e374d8f23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), does not change the default provider or ship any Rust driver binaries. ### Changed +- Pooled check-in skips transaction sanitation only after a prior successful + native rollback/autocommit restore and no subsequent statement allocation or + uncertain operation. Used connections still roll back explicit transactions + before parking; sanitation uses one native attribute probe and one GIL release. + Raw-handle exposure and arbitrary connection attributes disable the fast path; + successfully applied scalar login timeouts (including `connect(timeout=30)`) + do not permanently disable it. - Fetches reuse owned native metadata for stable columns within a result set; `fetchmany()` avoids the Python metadata-dictionary roundtrip. Re-execution, result transitions and statement/connection cleanup invalidate this metadata. diff --git a/mssql_python/connection.py b/mssql_python/connection.py index 1984f4979..db6c812fb 100644 --- a/mssql_python/connection.py +++ b/mssql_python/connection.py @@ -2194,35 +2194,14 @@ def close(self) -> None: # Close the connection even if cursor cleanup had issues try: if self._conn: - autocommit_error = None - rollback_error = None - manual_commit = False + # Native close owns transaction cleanup and fail-closed discard. + # Avoid duplicate attribute probes and untrusted rollback hints. try: - manual_commit = not self._conn.get_autocommit() - except RuntimeError as e: - autocommit_error = e - if manual_commit: - # End caller work before native close. Pooled connections are - # additionally restored to autocommit by native check-in, - # which atomically discards them if sanitation fails. - logger.debug("Rolling back uncommitted changes before closing connection.") - try: - self._conn.rollback() - except RuntimeError as e: - rollback_error = e - # TODO: Check potential race conditions in case of multithreaded scenarios - # Close the connection - try: - self._conn.close(manual_commit and rollback_error is None) + self._conn.close(rollback_before_disconnect=True) except RuntimeError as e: _raise_connection_error(e) finally: self._conn = None - if rollback_error is not None: - # Preserve prior DB-API error mapping after deterministic cleanup. - _raise_connection_error(rollback_error) - if autocommit_error is not None: - _raise_connection_error(autocommit_error) except Exception as e: logger.error(f"Error closing database connection: {e}") # Re-raise the connection close error as it's more critical diff --git a/mssql_python/pybind/CMakeLists.txt b/mssql_python/pybind/CMakeLists.txt index 77d599bd5..3be1cd0b0 100644 --- a/mssql_python/pybind/CMakeLists.txt +++ b/mssql_python/pybind/CMakeLists.txt @@ -393,3 +393,21 @@ if(APPLE) target_compile_definitions(ddbc_bindings PRIVATE MACOS_STRING_FIX) target_compile_options(ddbc_bindings PRIVATE -DAPPLE_SILICON) endif() + +# Exercise the real connection/pool/handle code against deterministic ODBC +# function pointers, without shipping test hooks in the extension. +option(BUILD_NATIVE_POOL_TESTS "Build database-free native pool sanitation tests" OFF) +if(BUILD_NATIVE_POOL_TESTS) + find_package(Python3 COMPONENTS Development REQUIRED) + get_target_property(pool_test_sources ddbc_bindings SOURCES) + add_executable(pool_sanitation_tests ${pool_test_sources} + ../../tests/native/pool_sanitation.cpp) + foreach(property INCLUDE_DIRECTORIES COMPILE_DEFINITIONS COMPILE_OPTIONS) + set_property(TARGET pool_sanitation_tests PROPERTY ${property} + $) + endforeach() + target_link_libraries(pool_sanitation_tests PRIVATE + Python3::Python simdutf::simdutf ${CMAKE_DL_LIBS}) + enable_testing() + add_test(NAME pool_sanitation COMMAND pool_sanitation_tests) +endif() diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp index 9c4304d2f..bde49044b 100644 --- a/mssql_python/pybind/connection/connection.cpp +++ b/mssql_python/pybind/connection/connection.cpp @@ -6,6 +6,7 @@ #include "utf_utils.h" #include #include +#include #include #include #include @@ -286,6 +287,8 @@ void Connection::clearResultMetadata() { void Connection::commit() { PERF_TIMER("Connection::commit"); + _poolClean = false; + _poolSessionReset = false; if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } @@ -303,6 +306,8 @@ void Connection::commit() { void Connection::rollback() { PERF_TIMER("Connection::rollback"); + _poolClean = false; + _poolSessionReset = false; if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } @@ -320,6 +325,10 @@ void Connection::rollback() { void Connection::setAutocommit(bool enable) { PERF_TIMER("Connection::setAutocommit"); + if (!enable) { + _poolClean = false; + _poolSessionReset = false; + } if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } @@ -336,7 +345,10 @@ void Connection::setAutocommit(bool enable) { ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, reinterpret_cast(static_cast(value)), 0); } - checkError(ret); + if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; + checkError(ret); + } if (value == SQL_AUTOCOMMIT_ON) { LOG("Autocommit enabled"); } else { @@ -354,12 +366,20 @@ bool Connection::getAutocommit() const { SQLINTEGER string_length; SQLRETURN ret = SQLGetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, &value, sizeof(value), &string_length); - checkError(ret); + if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; + checkError(ret); + } return value == SQL_AUTOCOMMIT_ON; } SqlHandlePtr Connection::allocStatementHandle() { PERF_TIMER("Connection::allocStatementHandle"); + // Invalidate before allocation (including failures), not by parsing SQL. + // Every execution/catalog/fetch path, including direct native calls, needs + // a statement handle. Retained handles also prevent re-establishing proof. + _poolClean = false; + _poolSessionReset = false; LOG("Allocating statement handle"); // Keep the wrapper outside the lock scope: unwinding a failed registration // frees the statement through the same cleanup gate. @@ -417,6 +437,16 @@ SqlHandlePtr Connection::allocStatementHandle() { } SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { + _poolClean = false; + _poolSessionReset = false; + // A scalar login timeout only bounds connection establishment; it cannot + // execute user work or retain a deferred buffer. Keep all other attributes + // conservative, and never clear an earlier permanent invalidation. + const bool scalarLoginTimeout = + attribute == SQL_ATTR_LOGIN_TIMEOUT && PyLong_CheckExact(value.ptr()); + if (!scalarLoginTimeout) { + _poolProofDisabled = true; + } clearResultMetadata(); LOG("Setting SQL attribute=%d", attribute); // SQLPOINTER ptr = nullptr; @@ -443,7 +473,21 @@ SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { if (py::isinstance(value)) { // Get the integer value - int64_t longValue = value.cast(); + int64_t longValue; + try { + longValue = value.cast(); + } catch (const py::cast_error&) { + _poolProofDisabled = true; + throw; + } catch (const py::error_already_set&) { + _poolProofDisabled = true; + throw; + } + if (scalarLoginTimeout && + (longValue < 0 || + static_cast(longValue) > std::numeric_limits::max())) { + _poolProofDisabled = true; + } SQLRETURN ret; { @@ -456,6 +500,7 @@ SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { } if (!SQL_SUCCEEDED(ret)) { + _poolProofDisabled = true; LOG("Failed to set integer attribute=%d, ret=%d", attribute, ret); } else { LOG("Set integer attribute=%d successfully", attribute); @@ -604,6 +649,10 @@ bool Connection::reset() { if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } + if (_poolClean && _poolSessionReset && !_poolProofDisabled) { + updateLastUsed(); + return true; + } clearResultMetadata(); LOG("Resetting connection via SQL_ATTR_RESET_CONNECTION"); // NOTE: SQL_ATTR_RESET_CONNECTION is a pool-checkin reset: it asks the @@ -627,6 +676,7 @@ bool Connection::reset() { (SQLPOINTER)SQL_RESET_CONNECTION_YES, SQL_IS_INTEGER); } if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; LOG("Failed to reset connection (ret=%d). Marking as dead.", ret); return false; } @@ -641,6 +691,7 @@ bool Connection::reset() { (SQLPOINTER)SQL_TXN_READ_COMMITTED, SQL_IS_INTEGER); } if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; LOG("Failed to reset transaction isolation level (ret=%d). Marking as dead.", ret); return false; } @@ -649,24 +700,97 @@ bool Connection::reset() { return true; } -void Connection::prepareForPool(bool transactionAlreadyRolledBack) { +void Connection::prepareForPool() { + PERF_TIMER("Connection::prepareForPool"); if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } + if (_poolClean && !_poolProofDisabled) { + PERF_TIMER("Connection::prepareForPool::clean"); + return; + } - // Explicit BEGIN TRANSACTION is valid while ODBC autocommit is on, but - // SQLEndTran does not end that transaction until the connection enters - // manual-commit mode. - if (getAutocommit()) { - setAutocommit(false); + _poolClean = false; + _poolSessionReset = false; + clearResultMetadata(); + SQLRETURN ret; + std::string statementError; + bool sessionReset = false; + { + // One GIL release and metadata invalidation for the whole sequence. + // Do not trust Python mode/rollback hints: native callers and set_attr + // can change the state, and autocommit permits explicit BEGIN TRAN. + py::gil_scoped_release release; + PERF_TIMER("Connection::prepareForPool::sanitize"); + SQLINTEGER mode = SQL_AUTOCOMMIT_OFF; + SQLINTEGER length = 0; + { + PERF_TIMER("Connection::prepareForPool::get_autocommit"); + ret = SQLGetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, + &mode, sizeof(mode), &length); + } + if (SQL_SUCCEEDED(ret) && mode == SQL_AUTOCOMMIT_ON) { + // SQLEndTran is a no-op in autocommit mode, but SQL Server still + // permits an explicit BEGIN TRANSACTION. Schedule the driver's + // session reset first so states such as SET NOEXEC ON cannot suppress + // the rollback batch that triggers it. + PERF_TIMER("Connection::prepareForPool::rollback_autocommit"); + ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_RESET_CONNECTION, + reinterpret_cast(SQL_RESET_CONNECTION_YES), + SQL_IS_INTEGER); + if (SQL_SUCCEEDED(ret)) { + SQLHANDLE statement = nullptr; + ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, _dbcHandle->get(), &statement); + if (SQL_SUCCEEDED(ret)) { + const std::u16string rollbackQuery = + u"IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION; " + u"SET TRANSACTION ISOLATION LEVEL READ COMMITTED"; + ret = SQLExecDirect_ptr(statement, + reinterpretU16stringAsSqlWChar(rollbackQuery), SQL_NTS); + if (!SQL_SUCCEEDED(ret)) { + ErrorInfo error = SQLReadError(SQL_HANDLE_STMT, statement, ret); + statementError = error.sqlState.length() == 5 + ? "SQLSTATE:" + error.sqlState + ":" + error.ddbcErrorMsg + : error.ddbcErrorMsg; + } + SQLRETURN freeRet = SQLFreeHandle_ptr(SQL_HANDLE_STMT, statement); + if (SQL_SUCCEEDED(ret) && !SQL_SUCCEEDED(freeRet)) { + ErrorInfo error = SQLReadError(SQL_HANDLE_STMT, statement, freeRet); + statementError = error.sqlState.length() == 5 + ? "SQLSTATE:" + error.sqlState + ":" + error.ddbcErrorMsg + : error.ddbcErrorMsg; + ret = freeRet; + } + sessionReset = SQL_SUCCEEDED(ret); + } + } + } else if (SQL_SUCCEEDED(ret)) { + PERF_TIMER("Connection::prepareForPool::rollback_manual"); + ret = SQLEndTran_ptr(SQL_HANDLE_DBC, _dbcHandle->get(), SQL_ROLLBACK); + // Never enable autocommit after a failed rollback: it could commit + // abandoned work. Manual mode can leave even an empty transaction open. + if (SQL_SUCCEEDED(ret)) { + PERF_TIMER("Connection::prepareForPool::autocommit_on"); + ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, + reinterpret_cast(SQL_AUTOCOMMIT_ON), 0); + } + } } - if (!transactionAlreadyRolledBack) { - rollback(); + if (!statementError.empty()) { + ThrowStdException(statementError); + } + checkError(ret); + _autocommit = true; + _poolSessionReset = sessionReset; + updateLastUsed(); + // A native statement alias can execute again without another allocation, + // even in a later lease. Only expired wrappers permit the fast path. + if (!_poolProofDisabled) { + std::lock_guard lock(_childHandlesMutex); + _poolClean = std::all_of( + _childStatementHandles.begin(), _childStatementHandles.end(), + [](const std::weak_ptr& handle) { return handle.expired(); }); } - // The SQL Server ODBC driver can leave an empty transaction visible after - // SQLEndTran while manual-commit mode remains enabled, so always park the - // physical connection in autocommit mode. - setAutocommit(true); } void Connection::updateLastUsed() { @@ -781,28 +905,38 @@ ConnectionHandle::~ConnectionHandle() { } } -void ConnectionHandle::close(bool transactionAlreadyRolledBack) { +void ConnectionHandle::close(bool rollbackBeforeDisconnect) { PERF_TIMER("ConnectionHandle::close"); if (!_conn) { ThrowStdException("Connection object is not initialized"); } - if (_usePool) { - try { - _conn->prepareForPool(transactionAlreadyRolledBack); - } catch (...) { - // Never retain a connection whose transaction state could not be - // sanitized. Discarding also releases this connection's reserved - // pool capacity. Preserve the original check-in error. - try { - ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn); - } catch (...) { + try { + if (_usePool) { + _conn->prepareForPool(); + } else { + // Preserve unpooled close semantics without doing pool sanitation. + if (rollbackBeforeDisconnect && !_conn->getAutocommit()) { + _conn->rollback(); } - _conn = nullptr; + _conn->disconnect(); + } + } catch (...) { + // A low-level unpooled close still allows the caller to recover from + // SQLDisconnect failure using the existing connection and children. + if (!_usePool && !rollbackBeforeDisconnect) { throw; } + // Never retain a connection whose transaction state could not be + // sanitized. Release capacity and preserve the original cleanup error. + try { + ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn); + } catch (...) { + } + _conn = nullptr; + throw; + } + if (_usePool) { ConnectionPoolManager::getInstance().returnConnection(_poolKey, _originPool, _conn); - } else { - _conn->disconnect(); } _conn = nullptr; } @@ -849,6 +983,12 @@ SqlHandlePtr ConnectionHandle::allocStatementHandle() { } py::object Connection::getInfo(SQLUSMALLINT infoType) const { + _poolClean = false; + _poolSessionReset = false; + if (infoType == SQL_DRIVER_HDBC || infoType == SQL_DRIVER_HENV || + infoType == SQL_DRIVER_HSTMT || infoType == SQL_DRIVER_HLIB) { + _poolProofDisabled = true; + } if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h index f43613842..c9dc6fab5 100644 --- a/mssql_python/pybind/connection/connection.h +++ b/mssql_python/pybind/connection/connection.h @@ -59,7 +59,7 @@ class Connection { bool getAutocommit() const; bool isAlive() const; bool reset(); - void prepareForPool(bool transactionAlreadyRolledBack = false); + void prepareForPool(); void updateLastUsed(); std::chrono::steady_clock::time_point lastUsed() const; @@ -108,6 +108,14 @@ class Connection { std::u16string _connStr; bool _fromPool = false; bool _autocommit = true; + // Only successful rollback + AUTOCOMMIT_ON establishes this proof. A new + // login or deferred SQL_ATTR_RESET_CONNECTION alone does not establish it. + mutable bool _poolClean = false; + // True only after check-in has triggered the driver reset and restored the + // default isolation level. This permits checkout to skip a duplicate reset. + mutable bool _poolSessionReset = false; + // Escaped raw handles and arbitrary/deferred attributes cannot be tracked. + mutable bool _poolProofDisabled = false; SqlHandlePtr _dbcHandle; std::chrono::steady_clock::time_point _lastUsed; // POSIX-epoch expiry (seconds) of the access token this connection last @@ -152,7 +160,7 @@ class ConnectionHandle { const py::object& tokenFactory = py::object()); ~ConnectionHandle(); - void close(bool transactionAlreadyRolledBack = false); + void close(bool rollbackBeforeDisconnect = false); void commit(); void rollback(); void setAutocommit(bool enabled); diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index f410631d4..4a8184bb9 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -6472,7 +6472,7 @@ PYBIND11_MODULE(ddbc_bindings, m) { py::arg("conn_str"), py::arg("use_pool"), py::arg("attrs_before") = py::dict(), py::arg("pool_key") = std::u16string(), py::arg("token_factory") = py::none()) .def("close", &ConnectionHandle::close, - py::arg("transaction_already_rolled_back") = false, "Close the connection") + py::arg("rollback_before_disconnect") = false, "Close the connection") .def("commit", &ConnectionHandle::commit, "Commit the current transaction") .def("rollback", &ConnectionHandle::rollback, "Rollback the current transaction") .def("set_autocommit", &ConnectionHandle::setAutocommit) diff --git a/tests/native/pool_sanitation.cpp b/tests/native/pool_sanitation.cpp new file mode 100644 index 000000000..6062ad021 --- /dev/null +++ b/tests/native/pool_sanitation.cpp @@ -0,0 +1,651 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +#include "connection/connection.h" +#include "connection/connection_pool.h" +#include "logger_bridge.hpp" +#include +#include +#include +#include +#include +#include + +namespace { +struct Handle { + SQLSMALLINT type; + SQLHANDLE parent; + bool autocommit = true; + bool transaction = false; + bool pendingWork = false; + bool resetPending = false; + SQLULEN loginTimeout = 0; +}; + +std::unordered_map> handles; +std::vector calls; +std::string failNext; +int commits = 0; +int logins = 0; +std::vector loginTimeouts; + +void require(bool condition, const std::string& message) { + if (!condition) { + throw std::runtime_error(message); + } +} + +SQLRETURN record(const std::string& operation) { + calls.push_back(operation); + if (operation == failNext) { + failNext.clear(); + return SQL_ERROR; + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API allocate(SQLSMALLINT type, SQLHANDLE parent, SQLHANDLE* out) { + auto ret = record(type == SQL_HANDLE_STMT ? "allocate_statement" : "allocate"); + if (!SQL_SUCCEEDED(ret)) { + return ret; + } + auto handle = std::make_unique(Handle{type, parent}); + *out = handle.get(); + handles.emplace(*out, std::move(handle)); + return SQL_SUCCESS; +} + +SQLRETURN SQL_API freeHandle(SQLSMALLINT, SQLHANDLE handle) { + auto ret = record("free"); + if (SQL_SUCCEEDED(ret)) { + handles.erase(handle); + } + return ret; +} + +SQLRETURN SQL_API setEnv(SQLHENV, SQLINTEGER, SQLPOINTER, SQLINTEGER) { + return SQL_SUCCESS; +} + +SQLRETURN SQL_API login(SQLHDBC dbc, SQLHWND, SQLWCHAR*, SQLSMALLINT, + SQLWCHAR*, SQLSMALLINT, SQLSMALLINT*, SQLUSMALLINT) { + ++logins; + loginTimeouts.push_back(handles.at(dbc)->loginTimeout); + return record("login"); +} + +SQLRETURN SQL_API setAttr(SQLHDBC dbc, SQLINTEGER attribute, SQLPOINTER value, SQLINTEGER length) { + std::string operation = "attribute"; + if (attribute == SQL_ATTR_AUTOCOMMIT) { + operation = value ? "on" : "off"; + } else if (attribute == SQL_ATTR_RESET_CONNECTION) { + operation = "reset"; + } else if (attribute == SQL_ATTR_TXN_ISOLATION) { + operation = "isolation"; + } + auto ret = record(operation); + if (!SQL_SUCCEEDED(ret)) { + return ret; + } + auto& handle = *handles.at(dbc); + if (attribute == SQL_ATTR_AUTOCOMMIT) { + if (value) { + if (handle.pendingWork) { + ++commits; + } + handle.transaction = handle.pendingWork = false; + } + handle.autocommit = value != nullptr; + } else if (attribute == SQL_ATTR_RESET_CONNECTION) { + handle.resetPending = true; + } else if (attribute == SQL_ATTR_LOGIN_TIMEOUT && length == SQL_IS_INTEGER) { + handle.loginTimeout = reinterpret_cast(value); + } + // A reset is deferred until execution. It must not establish clean proof. + return SQL_SUCCESS; +} + +SQLRETURN SQL_API getAttr(SQLHDBC dbc, SQLINTEGER attribute, SQLPOINTER value, + SQLINTEGER, SQLINTEGER*) { + auto ret = record(attribute == SQL_ATTR_AUTOCOMMIT ? "get" : "alive"); + if (SQL_SUCCEEDED(ret)) { + *static_cast(value) = attribute == SQL_ATTR_AUTOCOMMIT + ? (handles.at(dbc)->autocommit ? SQL_AUTOCOMMIT_ON : SQL_AUTOCOMMIT_OFF) + : SQL_CD_FALSE; + } + return ret; +} + +SQLRETURN SQL_API endTran(SQLSMALLINT, SQLHANDLE dbc, SQLSMALLINT completion) { + auto ret = record(completion == SQL_ROLLBACK ? "rollback" : "commit"); + if (SQL_SUCCEEDED(ret) && !handles.at(dbc)->autocommit) { + auto& handle = *handles.at(dbc); + if (completion == SQL_COMMIT && handle.pendingWork) { + ++commits; + } + handle.pendingWork = false; + // Model the empty manual-mode transaction that motivated PR #777. + handle.transaction = true; + } + return ret; +} + +SQLRETURN SQL_API executeDirect(SQLHSTMT statement, SQLWCHAR*, SQLINTEGER) { + auto& handle = *handles.at(handles.at(statement)->parent); + if (handle.resetPending) { + handle.transaction = handle.pendingWork = false; + handle.resetPending = false; + } + auto ret = record("rollback_batch"); + if (SQL_SUCCEEDED(ret)) { + handle.transaction = handle.pendingWork = false; + } + return ret; +} + +SQLRETURN SQL_API disconnect(SQLHDBC dbc) { + record("disconnect"); + if (handles.at(dbc)->pendingWork) { + return SQL_ERROR; + } + handles.at(dbc)->transaction = handles.at(dbc)->pendingWork = false; + for (auto it = handles.begin(); it != handles.end();) { + if (it->second->parent == dbc) { + it = handles.erase(it); + } else { + ++it; + } + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API getInfo(SQLHDBC dbc, SQLUSMALLINT, SQLPOINTER value, + SQLSMALLINT, SQLSMALLINT* length) { + *length = sizeof(dbc); + if (value) { + std::memcpy(value, &dbc, sizeof(dbc)); + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API diagnostic(SQLSMALLINT, SQLHANDLE, SQLSMALLINT recordNumber, + SQLWCHAR* state, SQLINTEGER* native, SQLWCHAR* message, + SQLSMALLINT, SQLSMALLINT* length) { + if (recordNumber != 1) { + return SQL_NO_DATA; + } + const SQLWCHAR sqlstate[] = {'H', 'Y', '0', '0', '0', 0}; + const SQLWCHAR text[] = {'f', 'a', 'i', 'l', 'e', 'd', 0}; + std::copy(std::begin(sqlstate), std::end(sqlstate), state); + std::copy(std::begin(text), std::end(text), message); + *native = 0; + *length = 6; + return SQL_SUCCESS; +} + +void expectCalls(std::initializer_list expected) { + std::vector wanted(expected.begin(), expected.end()); + if (calls != wanted) { + std::string actual; + for (const auto& call : calls) { + actual += call + " "; + } + throw std::runtime_error("Unexpected ODBC calls: " + actual); + } +} + +void expectFailure(const std::function& action) { + try { + action(); + } catch (const std::runtime_error&) { + return; + } + throw std::runtime_error("Expected native failure"); +} + +std::unique_ptr acquire(bool pooled = true, + const py::dict& attrs = py::dict()) { + return std::make_unique(u"native pool test", pooled, attrs); +} + +void checkParked() { + require(commits == 0, "Cleanup committed pending work"); + for (const auto& item : handles) { + if (item.second->type == SQL_HANDLE_DBC) { + require(item.second->autocommit && !item.second->transaction, + "Pooled DBC retained a transaction or manual mode"); + } + } +} + +void warm() { + auto connection = acquire(); + connection->close(); + checkParked(); +} + +void startWork(const SqlHandlePtr& statement) { + auto dbc = handles.at(statement->get())->parent; + handles.at(dbc)->transaction = handles.at(dbc)->pendingWork = true; +} + +void resetScenario() { + failNext.clear(); + auto& manager = ConnectionPoolManager::getInstance(); + manager.closePools(); + manager.configure(1, 30); + manager.setAccepting(true); + calls.clear(); + commits = logins = 0; + loginTimeouts.clear(); +} +} // namespace + +int main() { + py::scoped_interpreter interpreter; + mssql_python::logging::LoggerBridge::updateLevel(1000); + SQLAllocHandle_ptr = allocate; + SQLFreeHandle_ptr = freeHandle; + SQLSetEnvAttr_ptr = setEnv; + SQLDriverConnect_ptr = login; + SQLSetConnectAttr_ptr = setAttr; + SQLGetConnectAttr_ptr = getAttr; + SQLExecDirect_ptr = executeDirect; + SQLEndTran_ptr = endTran; + SQLDisconnect_ptr = disconnect; + SQLGetInfo_ptr = getInfo; + SQLGetDiagRec_ptr = diagnostic; + + int passed = 0; + auto run = [&](const char* name, const std::function& test) { + resetScenario(); + test(); + ++passed; + std::cout << "PASS " << name << '\n'; + }; + try { + run("new login is not clean proof; repeated empty leases skip all sanitation", [] { + auto connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + for (int i = 0; i < 100; ++i) { + calls.clear(); + connection = acquire(); + expectCalls({"alive"}); + connection->setAutocommit(true); + calls.clear(); + connection->close(true); + expectCalls({}); + checkParked(); + } + require(logins == 1, "Empty leases did not reuse the physical connection"); + }); + run("timeout=30 is applied before login and preserves repeated empty-lease fast path", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + auto connection = acquire(true, attrs); + connection->setAutocommit(true); + require(loginTimeouts == std::vector{30}, + "Login timeout was not applied as 30 before SQLDriverConnect"); + require(std::count(calls.begin(), calls.end(), "attribute") == 1, + "Login timeout was not applied exactly once"); + calls.clear(); + connection->close(true); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + for (int i = 0; i < 100; ++i) { + connection = acquire(true, attrs); + connection->setAutocommit(true); + calls.clear(); + connection->close(true); + expectCalls({}); + checkParked(); + } + require(logins == 1 && loginTimeouts == std::vector{30}, + "Timed leases did not reuse the physical connection with timeout 30"); + }); + run("valid scalar timeout cannot make dirty work clean", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + statement.reset(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + checkParked(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + for (bool timeoutFirst : {true, false}) { + run("scalar timeout never clears an unknown attribute's permanent invalidation", [timeoutFirst] { + py::dict attrs; + auto setTimeout = [&] { + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + }; + if (timeoutFirst) { + setTimeout(); + } + attrs[py::int_(SQL_ATTR_AUTOCOMMIT)] = py::int_(SQL_AUTOCOMMIT_ON); + if (!timeoutFirst) { + setTimeout(); + } + auto connection = acquire(true, attrs); + require(loginTimeouts == std::vector{30}, + "Timeout was lost when combined with another attribute"); + connection->close(); + connection = acquire(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const py::object& value : std::vector{ + py::str("30"), py::bytes("30"), py::bool_(true), + py::int_(-1), py::int_(uint64_t{1} << 32)}) { + run("nonscalar and out-of-range login timeouts permanently disable proof", [&value] { + warm(); + auto connection = acquire(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, value); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const py::object& value : std::vector{ + py::float_(30.0), py::eval("1 << 100")}) { + run("unsupported or overflowing timeout conversion fails closed", [&value] { + warm(); + auto connection = acquire(); + expectFailure([&] { connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, value); }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("failed login-timeout application releases capacity without login", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + failNext = "attribute"; + expectFailure([&] { acquire(true, attrs); }); + require(logins == 0, "Connected despite failed login-timeout application"); + auto connection = acquire(true, attrs); + require(loginTimeouts == std::vector{30}, + "Failed timeout application did not release pool capacity"); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + run("manual mode rolls back once and parks in autocommit", [] { + warm(); + auto connection = acquire(); + connection->setAutocommit(false); + calls.clear(); + connection->close(); + expectCalls({"get", "rollback", "on"}); + checkParked(); + }); + run("statement allocation invalidates even without execution", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + statement.reset(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + run("failed statement allocation invalidates clean proof", [] { + warm(); + auto connection = acquire(); + failNext = "allocate_statement"; + expectFailure([&] { connection->allocStatementHandle(); }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + run("explicit transaction and retained statement alias across leases", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + auto generation = statement->resultMetadata.snapshot().generation; + statement->resultMetadata.publish(generation, std::make_shared()); + connection->close(); + auto metadata = statement->resultMetadata.snapshot(); + require(!metadata.metadata && metadata.generation == generation + 1, + "Pool sanitation must invalidate result metadata exactly once"); + checkParked(); + connection = acquire(); + startWork(statement); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + checkParked(); + statement.reset(); + connection = acquire(); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + run("failed cursor free cannot enable the next lease's fast path", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + failNext = "free"; + require(statement->freeHandle() == SQL_ERROR, "Expected failed free"); + connection->close(); + connection = acquire(); + startWork(statement); + connection->close(); + checkParked(); + }); + for (const auto* operation : {"get", "reset", "allocate_statement", "rollback_batch", "free"}) { + run(operation, [operation] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + statement.reset(); + calls.clear(); + failNext = operation; + expectFailure([&] { connection->close(); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Failed sanitation did not disconnect"); + require(commits == 0, "Failure cleanup committed work"); + connection = acquire(); + require(logins == 2, "Discard did not release capacity / replace DBC"); + connection->close(); + checkParked(); + }); + } + for (const auto* operation : {"rollback", "on"}) { + run("manual-mode sanitation failure", [operation] { + auto connection = acquire(); + connection->setAutocommit(false); + calls.clear(); + failNext = operation; + expectFailure([&] { connection->close(); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Failed manual-mode sanitation did not disconnect"); + if (std::string(operation) == "rollback") { + require(std::find(calls.begin(), calls.end(), "on") == calls.end(), + "Enabled autocommit after failed rollback"); + } + require(commits == 0, "Failure cleanup committed work"); + connection = acquire(); + require(logins == 2, "Discard did not release capacity / replace DBC"); + connection->close(); + checkParked(); + }); + } + for (auto info : {SQL_DRIVER_HDBC, SQL_DRIVER_HENV, SQL_DRIVER_HSTMT, SQL_DRIVER_HLIB}) { + run("raw handle exposure permanently disables proof", [info] { + warm(); + auto connection = acquire(); + connection->getInfo(static_cast(info)); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("ordinary getinfo invalidates the current lease only", [] { + warm(); + auto connection = acquire(); + connection->getInfo(SQL_DBMS_NAME); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + run("arbitrary attrs_before disable proof", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_AUTOCOMMIT)] = py::int_(SQL_AUTOCOMMIT_OFF); + auto connection = acquire(true, attrs); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + for (bool fail : {false, true}) { + run("generic set_attr including failure disables proof", [fail] { + warm(); + auto connection = acquire(); + if (fail) { + failNext = "attribute"; + expectFailure([&] { connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); }); + } else { + connection->setAttr(SQL_ATTR_AUTOCOMMIT, py::int_(SQL_AUTOCOMMIT_OFF)); + } + calls.clear(); + connection->close(); + if (fail) { + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + } else { + expectCalls({"get", "rollback", "on"}); + } + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const auto* operation : {"commit", "rollback", "on", "off", "get"}) { + run("failed native operation invalidates clean proof", [operation] { + warm(); + auto connection = acquire(); + failNext = operation; + expectFailure([&] { + if (std::string(operation) == "commit") { + connection->commit(); + } else if (std::string(operation) == "rollback") { + connection->rollback(); + } else if (std::string(operation) == "get") { + connection->getAutocommit(); + } else { + connection->setAutocommit(std::string(operation) == "on"); + } + }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const auto* operation : {"reset", "isolation"}) { + run("failed deferred reset replaces the physical connection", [operation] { + auto connection = acquire(); + connection->setAutocommit(false); + connection->close(); + failNext = operation; + connection = acquire(); + require(logins == 2, "Failed reset was reused"); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("abandonment rolls back and releases capacity", [] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + connection.reset(); + require(commits == 0, "Abandoned connection committed work"); + connection = acquire(); + require(logins == 2, "Abandonment did not release capacity"); + connection->close(); + checkParked(); + }); + for (bool autocommit : {true, false}) { + run("unpooled close does not run pooled sanitation", [autocommit] { + auto connection = acquire(false); + connection->setAutocommit(autocommit); + calls.clear(); + connection->close(true); + if (autocommit) { + expectCalls({"get", "disconnect", "free"}); + } else { + expectCalls({"get", "rollback", "disconnect", "free"}); + } + }); + } + for (const auto* operation : {"get", "rollback"}) { + run("unpooled failure discards without pool sanitation", [operation] { + auto connection = acquire(false); + connection->setAutocommit(false); + failNext = operation; + expectFailure([&] { connection->close(true); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Unpooled failure did not disconnect"); + require(commits == 0, "Unpooled error cleanup committed work"); + }); + } + run("raw unpooled disconnect failure preserves connection and children", [] { + auto connection = acquire(false); + connection->setAutocommit(false); + auto statement = connection->allocStatementHandle(); + startWork(statement); + calls.clear(); + expectFailure([&] { connection->close(); }); + expectCalls({"disconnect"}); + require(handles.count(statement->get()) == 1, "Live child was freed"); + auto sibling = connection->allocStatementHandle(); + connection->rollback(); + connection->close(); + require(statement->isImplicitlyFreed() && sibling->isImplicitlyFreed(), + "Successful disconnect did not invalidate children"); + }); + ConnectionPoolManager::getInstance().closePools(); + std::cout << passed << " native pool tests passed\n"; + return 0; + } catch (const std::exception& error) { + std::cerr << "FAIL after " << passed << " tests: " << error.what() << '\n'; + failNext.clear(); + ConnectionPoolManager::getInstance().closePools(); + return 1; + } +} diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py index 371b9662f..318e3b0e6 100644 --- a/tests/test_006_exceptions.py +++ b/tests/test_006_exceptions.py @@ -282,7 +282,7 @@ def test_close_cleans_up_after_rollback_failure(): mock_conn = MagicMock() mock_conn.get_autocommit.return_value = False - mock_conn.rollback.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") + mock_conn.close.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn): conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;") @@ -290,8 +290,8 @@ def test_close_cleans_up_after_rollback_failure(): with pytest.raises(OperationalError, match="Communication link failure"): conn.close() - mock_conn.rollback.assert_called_once_with() - mock_conn.close.assert_called_once_with(False) + mock_conn.rollback.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) assert conn._conn is None assert conn.closed @@ -301,7 +301,7 @@ def test_close_cleans_up_after_autocommit_read_failure(): from unittest.mock import MagicMock, patch mock_conn = MagicMock() - mock_conn.get_autocommit.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") + mock_conn.close.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn): conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;") @@ -310,13 +310,14 @@ def test_close_cleans_up_after_autocommit_read_failure(): conn.close() mock_conn.rollback.assert_not_called() - mock_conn.close.assert_called_once_with(False) + mock_conn.get_autocommit.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) assert conn._conn is None assert conn.closed -def test_close_reports_successful_rollback_to_native_cleanup(): - """Native pool cleanup must not repeat a successful Python rollback.""" +def test_close_delegates_manual_transaction_cleanup_to_native(): + """Only native close probes the mode and rolls back, without Python hints.""" from unittest.mock import MagicMock, patch mock_conn = MagicMock() @@ -327,8 +328,9 @@ def test_close_reports_successful_rollback_to_native_cleanup(): conn.close() - mock_conn.rollback.assert_called_once_with() - mock_conn.close.assert_called_once_with(True) + mock_conn.get_autocommit.assert_not_called() + mock_conn.rollback.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) def test_autocommit_close_delegates_transaction_cleanup_to_native(): @@ -347,12 +349,13 @@ def test_autocommit_close_delegates_transaction_cleanup_to_native(): conn.close() mock_conn.rollback.assert_not_called() - mock_conn.close.assert_called_once_with(False) + mock_conn.get_autocommit.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) @pytest.mark.parametrize("preclose_failure", ["autocommit", "rollback"]) -def test_native_close_error_takes_precedence_over_preclose_failure(preclose_failure): - """The native close error wins, but the wrapper still releases its handle.""" +def test_native_close_error_does_not_need_python_preclose_calls(preclose_failure): + """Native cleanup is authoritative even if Python probes would have failed.""" from unittest.mock import MagicMock, patch mock_conn = MagicMock() @@ -369,11 +372,33 @@ def test_native_close_error_takes_precedence_over_preclose_failure(preclose_fail with pytest.raises(OperationalError, match="Native close failed"): conn.close() - mock_conn.close.assert_called_once_with(False) + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) + mock_conn.get_autocommit.assert_not_called() + mock_conn.rollback.assert_not_called() assert conn._conn is None assert conn.closed +def test_cursor_cleanup_failure_still_calls_native_close(): + from unittest.mock import MagicMock, patch + + native = MagicMock() + with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=native): + conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;") + broken = MagicMock(closed=False) + broken.close.side_effect = RuntimeError("cursor cleanup failed") + healthy = MagicMock(closed=False) + conn._cursors = {broken, healthy} + + conn.close() + + broken.close.assert_called_once_with() + healthy.close.assert_called_once_with() + native.close.assert_called_once_with(rollback_before_disconnect=True) + assert conn.closed + assert conn._conn is None + + def test_truncate_error_message_successful_cases(): """Test truncate_error_message with valid Microsoft messages for comparison.""" diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py index c8cafed8f..1bcdde646 100644 --- a/tests/test_009_pooling.py +++ b/tests/test_009_pooling.py @@ -83,6 +83,33 @@ def reset_pooling_state(): # ============================================================================= +def test_pooled_positive_login_timeout_reaches_native_attrs(): + """Exercise the Python boundary paired with native timeout call-count tests.""" + from unittest.mock import MagicMock, patch + from mssql_python.constants import ConstantsDDBC + + pooling(enabled=True, max_size=2, idle_timeout=30) + native = MagicMock() + with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=native) as create: + for _ in range(3): + connection = connect( + "Server=testserver;Database=mydb;Trusted_Connection=yes;", + timeout=30, + autocommit=True, + ) + assert create.call_args.args[1] is True + assert create.call_args.args[2] == {ConstantsDDBC.SQL_ATTR_LOGIN_TIMEOUT.value: 30} + assert connection.timeout == 0 # Query timeout is independent. + connection.close() + assert create.call_count == 3 + assert native.set_autocommit.call_count == 3 + native.set_autocommit.assert_called_with(True) + assert native.close.call_count == 3 + native.close.assert_called_with(rollback_before_disconnect=True) + native.get_autocommit.assert_not_called() + native.rollback.assert_not_called() + + def test_connection_pooling_basic(conn_str): """Test basic connection pooling functionality with multiple connections.""" # Enable pooling with small pool size @@ -172,6 +199,13 @@ def open_transaction_count(session_id): None, None, ), + ( + "explicit transaction with noexec", + True, + "BEGIN TRANSACTION; SELECT 1; SET NOEXEC ON", + None, + None, + ), ) expected_spid = None for name, autocommit, sql, params, action in scenarios: @@ -224,7 +258,8 @@ def open_transaction_count(session_id): ) -def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str): +@pytest.mark.parametrize("execution", ["execute", "executemany", "failed_execute"]) +def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str, execution): """Autocommit normalization must not commit an explicit SQL transaction.""" _run_in_subprocess( """ @@ -233,7 +268,8 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str import mssql_python conn_str = os.environ["DB_CONNECTION_STRING"] - table = "pytest_pool_explicit_autocommit_transaction" + execution = EXECUTION_MODE + table = f"pytest_pool_explicit_autocommit_transaction_{execution}" mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30) observer = mssql_python.connect(conn_str, autocommit=True) try: @@ -271,7 +307,38 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str ) sys.exit(77) - subject_cursor.execute(f"BEGIN TRANSACTION; INSERT INTO {table} VALUES (1)") + subject_cursor.close() + subject.close() + # Exercise proven-clean, no-statement leases before reintroducing + # explicit work. Inspect the parked session without borrowing it. + for _ in range(3): + empty = mssql_python.connect(conn_str, autocommit=True) + empty.close() + observer_cursor.execute( + "SELECT open_transaction_count " + "FROM sys.dm_exec_sessions WHERE session_id = ?", + [subject_spid], + ) + row = observer_cursor.fetchone() + assert row is not None and row[0] == 0 + + subject = mssql_python.connect(conn_str, autocommit=True) + subject_cursor = subject.cursor() + subject_cursor.execute("BEGIN TRANSACTION") + if execution == "executemany": + subject_cursor.executemany( + f"INSERT INTO {table} VALUES (?)", [(1,), (2,)] + ) + elif execution == "failed_execute": + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") + try: + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") + except mssql_python.IntegrityError: + pass + else: + raise AssertionError("Expected duplicate-key execution failure") + else: + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") subject_cursor.close() subject.close() @@ -303,7 +370,7 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str finally: observer.close() mssql_python.pooling(enabled=False) - """, + """.replace("EXECUTION_MODE", repr(execution)), conn_str, )