diff --git a/CHANGELOG.md b/CHANGELOG.md index f4ebfb81..d4f9787e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,29 @@ 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 +- **GH-769 deprecation policy:** The misplaced `GetInfoConstants` members + `SQL_TXN_ISOLATION_LEVEL`, `SQL_CONCURRENCY`, `SQL_ROWSET_SIZE`, `SQL_ROW_NUMBER`, + `SQL_IC_UPPER`, `SQL_IC_LOWER`, `SQL_IC_SENSITIVE`, `SQL_IC_MIXED`, and + `SQL_SQL92_ENTRY_SQL`, `SQL_SQL92_INTERMEDIATE_SQL`, `SQL_SQL92_FULL_SQL` + remain available with their original values throughout **1.x**. Existing enum + attribute access, module-level imports, and `get_info_constants()` dictionary + lookups remain available. Removal is deferred to **2.0 or later**, only after + maintainer approval and an explicit migration notice; this is not a scheduled removal. + Deprecation is documented rather than emitting runtime warnings. + These names are not information-type names, and passing their integers to + `getinfo()` still requests unrelated information. They cannot be rejected by + value without also rejecting legitimate information types with the same IDs. + For migration, use `SQL_ATTR_TXN_ISOLATION` for the connection attribute; + use `ConstantsDDBC` or module-level names for legacy statement options and + `SQL_IC_*` response values, not as `getinfo()` requests. To interpret + `SQL_SQL_CONFORMANCE`, use `SQL_SC_SQL92_ENTRY` (1), + `SQL_SC_FIPS127_2_TRANSITIONAL` (2), `SQL_SC_SQL92_INTERMEDIATE` (4), and + `SQL_SC_SQL92_FULL` (8). The deprecated `SQL_SQL92_*_SQL` names retain + **127/128/129** solely for compatibility; they are not conformance flags. + For information types whose IDs are corrected, previously persisted value-based + enum pickles and raw IDs cannot identify their original meaning; rebuild them + from the intended information-type names. Name-based enum pickles resolve + retained names to their corrected values. - Connection strings and string connection parameters that contain a NUL (`\x00`) character are now rejected up front with `InterfaceError` instead of being silently truncated at the NUL by the underlying driver. @@ -69,6 +92,28 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), before; users should call `cursor.setinputsizes()` to work around this. ### Fixed +- **GH-769:** Corrected 11 `GetInfoConstants` IDs for scalar functions, outer + joins, driver handles, cursor attributes, catalog support, and parameter + descriptions. Added the ODBC name `SQL_TIMEDATE_FUNCTIONS` as an alias of + `SQL_DATETIME_FUNCTIONS` at 52, preserving the existing canonical `.name`. + For registered information types, + `getinfo()` now uses ODBC return types instead of guessing from the bytes: numeric + information is decoded as unsigned integers (including + `SQL_SQL_CONFORMANCE`, `SQL_CURSOR_SENSITIVITY`, and + `SQL_MAX_IDENTIFIER_LEN`), and character results use the Unicode path. + An immutable return-type registry enforces the exact ODBC numeric width, + including pointer-sized handles. Wrong-width or truncated numeric payloads + raise `DatabaseError` rather than returning a guessed value. Legacy non-byte + numeric payloads are not truncated or coerced from bool to int; convertible + decimal strings still return integers. + Also corrects decoding for the unlisted standard IDs `SQL_DBMS_NAME` (17), + `SQL_DBMS_VER` (18), `SQL_XOPEN_CLI_YEAR` (10000), `SQL_ASYNC_MODE` (10021), and + `SQL_CREATE_ASSERTION` (127). Other unlisted raw IDs retain their existing behavior. + Driver Manager-only handle queries can still be unsupported by the native + provider; correcting their IDs does not add Driver Manager support. Native + retrieval errors, including timeout and connection-loss errors, continue to be + logged and return `None`. Providers may return cached metadata after connection + loss: `getinfo()` is not a connection-health check. - **GH-740:** A Python `Decimal` whose value falls in the SQL Server MONEY / SMALLMONEY range is now bound as `SQL_NUMERIC` with its own precision and scale on both `execute()` paths (native detection, and the legacy path reached when diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index e0a9ea06..e859d77c 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -190,6 +190,11 @@ def _cleanup_connections(): SQL_ATTR_LOGIN_TIMEOUT, SQL_ATTR_PACKET_SIZE, SQL_ATTR_TXN_ISOLATION, + SQL_TXN_ISOLATION_LEVEL, + # Legacy statement options + SQL_ROWSET_SIZE, + SQL_CONCURRENCY, + SQL_ROW_NUMBER, # Transaction isolation levels SQL_TXN_READ_UNCOMMITTED, SQL_TXN_READ_COMMITTED, @@ -216,9 +221,6 @@ def _cleanup_connections(): SQL_IDENTIFIER_CASE, SQL_IDENTIFIER_QUOTE_CHAR, SQL_SPECIAL_CHARACTERS, - SQL_SQL92_ENTRY_SQL, - SQL_SQL92_INTERMEDIATE_SQL, - SQL_SQL92_FULL_SQL, SQL_SUBQUERIES, SQL_EXPRESSIONS_IN_ORDERBY, SQL_CORRELATION_NAME, @@ -239,10 +241,10 @@ def _cleanup_connections(): SQL_TXN_ISOLATION_OPTION, SQL_DEFAULT_TXN_ISOLATION, SQL_MULTIPLE_ACTIVE_TXN, - SQL_TXN_ISOLATION_LEVEL, SQL_NUMERIC_FUNCTIONS, SQL_STRING_FUNCTIONS, SQL_DATETIME_FUNCTIONS, + SQL_TIMEDATE_FUNCTIONS, SQL_SYSTEM_FUNCTIONS, SQL_CONVERT_FUNCTIONS, SQL_LIKE_ESCAPE_CLAUSE, @@ -284,9 +286,6 @@ def _cleanup_connections(): SQL_SCROLL_OPTIONS, SQL_SCROLL_CONCURRENCY, SQL_FETCH_DIRECTION, - SQL_ROWSET_SIZE, - SQL_CONCURRENCY, - SQL_ROW_NUMBER, SQL_STATIC_SENSITIVITY, SQL_BATCH_SUPPORT, SQL_BATCH_ROW_COUNT, @@ -309,10 +308,19 @@ def _cleanup_connections(): SQL_QUALIFIER_USAGE, SQL_TIMEDATE_ADD_INTERVALS, SQL_TIMEDATE_DIFF_INTERVALS, + # SQLGetInfo return values (not information types) SQL_IC_UPPER, SQL_IC_LOWER, SQL_IC_SENSITIVE, SQL_IC_MIXED, + SQL_SC_SQL92_ENTRY, + SQL_SC_FIPS127_2_TRANSITIONAL, + SQL_SC_SQL92_INTERMEDIATE, + SQL_SC_SQL92_FULL, + # Deprecated legacy values, not SQL conformance flags + SQL_SQL92_ENTRY_SQL, + SQL_SQL92_INTERMEDIATE_SQL, + SQL_SQL92_FULL_SQL, ) __all__ = [ @@ -403,6 +411,11 @@ def _cleanup_connections(): "SQL_ATTR_LOGIN_TIMEOUT", "SQL_ATTR_PACKET_SIZE", "SQL_ATTR_TXN_ISOLATION", + "SQL_TXN_ISOLATION_LEVEL", + # Legacy statement options + "SQL_ROWSET_SIZE", + "SQL_CONCURRENCY", + "SQL_ROW_NUMBER", # Transaction isolation levels "SQL_TXN_READ_UNCOMMITTED", "SQL_TXN_READ_COMMITTED", @@ -429,9 +442,6 @@ def _cleanup_connections(): "SQL_IDENTIFIER_CASE", "SQL_IDENTIFIER_QUOTE_CHAR", "SQL_SPECIAL_CHARACTERS", - "SQL_SQL92_ENTRY_SQL", - "SQL_SQL92_INTERMEDIATE_SQL", - "SQL_SQL92_FULL_SQL", "SQL_SUBQUERIES", "SQL_EXPRESSIONS_IN_ORDERBY", "SQL_CORRELATION_NAME", @@ -452,10 +462,10 @@ def _cleanup_connections(): "SQL_TXN_ISOLATION_OPTION", "SQL_DEFAULT_TXN_ISOLATION", "SQL_MULTIPLE_ACTIVE_TXN", - "SQL_TXN_ISOLATION_LEVEL", "SQL_NUMERIC_FUNCTIONS", "SQL_STRING_FUNCTIONS", "SQL_DATETIME_FUNCTIONS", + "SQL_TIMEDATE_FUNCTIONS", "SQL_SYSTEM_FUNCTIONS", "SQL_CONVERT_FUNCTIONS", "SQL_LIKE_ESCAPE_CLAUSE", @@ -497,9 +507,6 @@ def _cleanup_connections(): "SQL_SCROLL_OPTIONS", "SQL_SCROLL_CONCURRENCY", "SQL_FETCH_DIRECTION", - "SQL_ROWSET_SIZE", - "SQL_CONCURRENCY", - "SQL_ROW_NUMBER", "SQL_STATIC_SENSITIVITY", "SQL_BATCH_SUPPORT", "SQL_BATCH_ROW_COUNT", @@ -522,10 +529,19 @@ def _cleanup_connections(): "SQL_QUALIFIER_USAGE", "SQL_TIMEDATE_ADD_INTERVALS", "SQL_TIMEDATE_DIFF_INTERVALS", + # SQLGetInfo return values (not information types) "SQL_IC_UPPER", "SQL_IC_LOWER", "SQL_IC_SENSITIVE", "SQL_IC_MIXED", + "SQL_SC_SQL92_ENTRY", + "SQL_SC_FIPS127_2_TRANSITIONAL", + "SQL_SC_SQL92_INTERMEDIATE", + "SQL_SC_SQL92_FULL", + # Deprecated legacy values, not SQL conformance flags + "SQL_SQL92_ENTRY_SQL", + "SQL_SQL92_INTERMEDIATE_SQL", + "SQL_SQL92_FULL_SQL", # API level globals "apilevel", "paramstyle", diff --git a/mssql_python/connection.py b/mssql_python/connection.py index 10aec710..21f564ad 100644 --- a/mssql_python/connection.py +++ b/mssql_python/connection.py @@ -15,6 +15,8 @@ import re import codecs import warnings +import struct +from types import MappingProxyType from typing import Any, Dict, Optional, Union, List, Tuple, Callable, Protocol, TYPE_CHECKING import threading @@ -87,8 +89,121 @@ def get_token(self, scope: str) -> Any: # Add SQL_WMETADATA constant for metadata decoding configuration SQL_WMETADATA: int = -99 # Special flag for column name decoding -# Threshold to determine if an info type is string-based -INFO_TYPE_STRING_THRESHOLD: int = 10000 +INFO_TYPE_STRING_THRESHOLD: int = 10000 # Legacy fallback for unlisted information types + +# ODBC integers have fixed widths and native byte order; handles are pointer-sized. +_SQLUSMALLINT = struct.Struct("=H") +_SQLUINTEGER = struct.Struct("=I") +_SQLPOINTER = struct.Struct("P") + +# SQLGetInfoW return types. Only type information is shared, never connection metadata. +_GETINFO_RETURN_TYPES = MappingProxyType( + { + GetInfoConstants.SQL_DATA_SOURCE_NAME.value: str, + GetInfoConstants.SQL_DATABASE_NAME.value: str, + GetInfoConstants.SQL_DRIVER_NAME.value: str, + GetInfoConstants.SQL_DRIVER_VER.value: str, + GetInfoConstants.SQL_SERVER_NAME.value: str, + GetInfoConstants.SQL_USER_NAME.value: str, + GetInfoConstants.SQL_DRIVER_ODBC_VER.value: str, + GetInfoConstants.SQL_IDENTIFIER_QUOTE_CHAR.value: str, + GetInfoConstants.SQL_CATALOG_NAME_SEPARATOR.value: str, + GetInfoConstants.SQL_CATALOG_TERM.value: str, + GetInfoConstants.SQL_SCHEMA_TERM.value: str, + GetInfoConstants.SQL_TABLE_TERM.value: str, + GetInfoConstants.SQL_KEYWORDS.value: str, + GetInfoConstants.SQL_PROCEDURE_TERM.value: str, + GetInfoConstants.SQL_SPECIAL_CHARACTERS.value: str, + GetInfoConstants.SQL_SEARCH_PATTERN_ESCAPE.value: str, + GetInfoConstants.SQL_ACCESSIBLE_PROCEDURES.value: str, + GetInfoConstants.SQL_ACCESSIBLE_TABLES.value: str, + GetInfoConstants.SQL_DATA_SOURCE_READ_ONLY.value: str, + GetInfoConstants.SQL_EXPRESSIONS_IN_ORDERBY.value: str, + GetInfoConstants.SQL_LIKE_ESCAPE_CLAUSE.value: str, + GetInfoConstants.SQL_MULTIPLE_ACTIVE_TXN.value: str, + GetInfoConstants.SQL_NEED_LONG_DATA_LEN.value: str, + GetInfoConstants.SQL_PROCEDURES.value: str, + GetInfoConstants.SQL_CATALOG_NAME.value: str, + GetInfoConstants.SQL_COLUMN_ALIAS.value: str, + GetInfoConstants.SQL_DESCRIBE_PARAMETER.value: str, + GetInfoConstants.SQL_ORDER_BY_COLUMNS_IN_SELECT.value: str, + GetInfoConstants.SQL_OUTER_JOINS.value: str, + GetInfoConstants.SQL_MULT_RESULT_SETS.value: str, + GetInfoConstants.SQL_DRIVER_HLIB.value: _SQLPOINTER, + GetInfoConstants.SQL_DRIVER_HENV.value: _SQLPOINTER, + GetInfoConstants.SQL_DRIVER_HDBC.value: _SQLPOINTER, + GetInfoConstants.SQL_SQL_CONFORMANCE.value: _SQLUINTEGER, + GetInfoConstants.SQL_IDENTIFIER_CASE.value: _SQLUSMALLINT, + GetInfoConstants.SQL_SUBQUERIES.value: _SQLUINTEGER, + GetInfoConstants.SQL_CORRELATION_NAME.value: _SQLUSMALLINT, + GetInfoConstants.SQL_CATALOG_USAGE.value: _SQLUINTEGER, + GetInfoConstants.SQL_SCHEMA_USAGE.value: _SQLUINTEGER, + GetInfoConstants.SQL_TXN_CAPABLE.value: _SQLUSMALLINT, + GetInfoConstants.SQL_TXN_ISOLATION_OPTION.value: _SQLUINTEGER, + GetInfoConstants.SQL_DEFAULT_TXN_ISOLATION.value: _SQLUINTEGER, + GetInfoConstants.SQL_NUMERIC_FUNCTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_STRING_FUNCTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_TIMEDATE_FUNCTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_SYSTEM_FUNCTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_CONVERT_FUNCTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_MAX_COLUMN_NAME_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_TABLE_NAME_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_SCHEMA_NAME_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_CATALOG_NAME_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_IDENTIFIER_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_STATEMENT_LEN.value: _SQLUINTEGER, + GetInfoConstants.SQL_MAX_CHAR_LITERAL_LEN.value: _SQLUINTEGER, + GetInfoConstants.SQL_MAX_BINARY_LITERAL_LEN.value: _SQLUINTEGER, + GetInfoConstants.SQL_MAX_COLUMNS_IN_TABLE.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_COLUMNS_IN_SELECT.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_COLUMNS_IN_GROUP_BY.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_COLUMNS_IN_ORDER_BY.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_COLUMNS_IN_INDEX.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_TABLES_IN_SELECT.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_CONCURRENT_ACTIVITIES.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_DRIVER_CONNECTIONS.value: _SQLUSMALLINT, + GetInfoConstants.SQL_MAX_ROW_SIZE.value: _SQLUINTEGER, + GetInfoConstants.SQL_MAX_USER_NAME_LEN.value: _SQLUSMALLINT, + GetInfoConstants.SQL_GETDATA_EXTENSIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_CURSOR_COMMIT_BEHAVIOR.value: _SQLUSMALLINT, + GetInfoConstants.SQL_CURSOR_ROLLBACK_BEHAVIOR.value: _SQLUSMALLINT, + GetInfoConstants.SQL_CURSOR_SENSITIVITY.value: _SQLUINTEGER, + GetInfoConstants.SQL_BOOKMARK_PERSISTENCE.value: _SQLUINTEGER, + GetInfoConstants.SQL_DYNAMIC_CURSOR_ATTRIBUTES1.value: _SQLUINTEGER, + GetInfoConstants.SQL_DYNAMIC_CURSOR_ATTRIBUTES2.value: _SQLUINTEGER, + GetInfoConstants.SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES1.value: _SQLUINTEGER, + GetInfoConstants.SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES2.value: _SQLUINTEGER, + GetInfoConstants.SQL_STATIC_CURSOR_ATTRIBUTES1.value: _SQLUINTEGER, + GetInfoConstants.SQL_STATIC_CURSOR_ATTRIBUTES2.value: _SQLUINTEGER, + GetInfoConstants.SQL_KEYSET_CURSOR_ATTRIBUTES1.value: _SQLUINTEGER, + GetInfoConstants.SQL_KEYSET_CURSOR_ATTRIBUTES2.value: _SQLUINTEGER, + GetInfoConstants.SQL_SCROLL_OPTIONS.value: _SQLUINTEGER, + GetInfoConstants.SQL_SCROLL_CONCURRENCY.value: _SQLUINTEGER, + GetInfoConstants.SQL_FETCH_DIRECTION.value: _SQLUINTEGER, + GetInfoConstants.SQL_STATIC_SENSITIVITY.value: _SQLUINTEGER, + GetInfoConstants.SQL_BATCH_SUPPORT.value: _SQLUINTEGER, + GetInfoConstants.SQL_BATCH_ROW_COUNT.value: _SQLUINTEGER, + GetInfoConstants.SQL_PARAM_ARRAY_ROW_COUNTS.value: _SQLUINTEGER, + GetInfoConstants.SQL_PARAM_ARRAY_SELECTS.value: _SQLUINTEGER, + GetInfoConstants.SQL_POSITIONED_STATEMENTS.value: _SQLUINTEGER, + GetInfoConstants.SQL_GROUP_BY.value: _SQLUSMALLINT, + GetInfoConstants.SQL_OJ_CAPABILITIES.value: _SQLUINTEGER, + GetInfoConstants.SQL_QUOTED_IDENTIFIER_CASE.value: _SQLUSMALLINT, + GetInfoConstants.SQL_CONCAT_NULL_BEHAVIOR.value: _SQLUSMALLINT, + GetInfoConstants.SQL_NULL_COLLATION.value: _SQLUSMALLINT, + GetInfoConstants.SQL_ALTER_TABLE.value: _SQLUINTEGER, + GetInfoConstants.SQL_UNION.value: _SQLUINTEGER, + GetInfoConstants.SQL_DDL_INDEX.value: _SQLUINTEGER, + GetInfoConstants.SQL_TIMEDATE_ADD_INTERVALS.value: _SQLUINTEGER, + GetInfoConstants.SQL_TIMEDATE_DIFF_INTERVALS.value: _SQLUINTEGER, + # Standard information types not currently exposed by GetInfoConstants. + 17: str, # SQL_DBMS_NAME + 18: str, # SQL_DBMS_VER + 10000: str, # SQL_XOPEN_CLI_YEAR + 10021: _SQLUINTEGER, # SQL_ASYNC_MODE + 127: _SQLUINTEGER, # SQL_CREATE_ASSERTION + } +) # UTF-16 encoding variants that should use SQL_WCHAR by default # Note: "utf-16" with BOM is NOT included as it's problematic for SQL_WCHAR @@ -1825,10 +1940,22 @@ def getinfo(self, info_type: int) -> Union[str, int, bool, None]: Returns: The requested information. The type of the returned value depends - on the information requested. It will be a string, integer, or boolean. + on the information requested. For registered ODBC types, character values (including + "Y"/"N") return strings; numeric values and bitmasks return unsigned + integers. Native retrieval failures, including unsupported types, + timeouts, and connection loss, are logged and return None. + + Note: + SQL_DRIVER_HDBC, SQL_DRIVER_HENV and SQL_DRIVER_HLIB are implemented + by the ODBC Driver Manager, which this driver bypasses. Correct IDs + do not imply that the selected native provider supports these queries. + A native provider may return cached metadata even after connection loss; + getinfo() is not a connection-health check. + Deprecated GetInfoConstants names are retained for API compatibility, + not as valid requests. Their integer values may request unrelated information. Raises: - DatabaseError: If there is an error retrieving the information. + DatabaseError: If a numeric byte result does not match its ODBC type's width. InterfaceError: If the connection is closed. """ if self._closed: @@ -1843,17 +1970,14 @@ def getinfo(self, info_type: int) -> Union[str, int, bool, None]: # Check for invalid info_type values if info_type < 0: - logger.debug( - "warning", - f"Invalid info_type: {info_type}. Must be a positive integer.", - ) + logger.debug("Invalid info_type: %d. Must be non-negative.", info_type) return None # Get the raw result from the C++ layer try: raw_result = self._conn.get_info(info_type) except Exception as e: # pylint: disable=broad-exception-caught - # Log the error and return None for invalid info types + # Preserve the legacy logged-None contract for native retrieval failures. logger.warning(f"getinfo({info_type}) failed: {e}") return None @@ -1870,70 +1994,19 @@ def getinfo(self, info_type: int) -> Union[str, int, bool, None]: data = raw_result["data"] length = raw_result["length"] - # Debug logging to understand the issue better logger.debug( - "debug", - f"getinfo: info_type={info_type}, length={length}, data_type={type(data)}", + "getinfo: info_type=%d, length=%r, data_type=%s", + info_type, + length, + type(data), ) - # Define constants for different return types - # String types - these return strings in pyodbc - string_type_constants = { - GetInfoConstants.SQL_DATA_SOURCE_NAME.value, - GetInfoConstants.SQL_DATABASE_NAME.value, - GetInfoConstants.SQL_DRIVER_NAME.value, - GetInfoConstants.SQL_DRIVER_VER.value, - GetInfoConstants.SQL_SERVER_NAME.value, - GetInfoConstants.SQL_USER_NAME.value, - GetInfoConstants.SQL_DRIVER_ODBC_VER.value, - GetInfoConstants.SQL_IDENTIFIER_QUOTE_CHAR.value, - GetInfoConstants.SQL_CATALOG_NAME_SEPARATOR.value, - GetInfoConstants.SQL_CATALOG_TERM.value, - GetInfoConstants.SQL_SCHEMA_TERM.value, - GetInfoConstants.SQL_TABLE_TERM.value, - GetInfoConstants.SQL_KEYWORDS.value, - GetInfoConstants.SQL_PROCEDURE_TERM.value, - GetInfoConstants.SQL_SPECIAL_CHARACTERS.value, - GetInfoConstants.SQL_SEARCH_PATTERN_ESCAPE.value, - } - - # Boolean 'Y'/'N' types - yn_type_constants = { - GetInfoConstants.SQL_ACCESSIBLE_PROCEDURES.value, - GetInfoConstants.SQL_ACCESSIBLE_TABLES.value, - GetInfoConstants.SQL_DATA_SOURCE_READ_ONLY.value, - GetInfoConstants.SQL_EXPRESSIONS_IN_ORDERBY.value, - GetInfoConstants.SQL_LIKE_ESCAPE_CLAUSE.value, - GetInfoConstants.SQL_MULTIPLE_ACTIVE_TXN.value, - GetInfoConstants.SQL_NEED_LONG_DATA_LEN.value, - GetInfoConstants.SQL_PROCEDURES.value, - } - - # Numeric type constants that return integers - numeric_type_constants = { - GetInfoConstants.SQL_MAX_COLUMN_NAME_LEN.value, - GetInfoConstants.SQL_MAX_TABLE_NAME_LEN.value, - GetInfoConstants.SQL_MAX_SCHEMA_NAME_LEN.value, - GetInfoConstants.SQL_MAX_CATALOG_NAME_LEN.value, - GetInfoConstants.SQL_MAX_IDENTIFIER_LEN.value, - GetInfoConstants.SQL_MAX_STATEMENT_LEN.value, - GetInfoConstants.SQL_MAX_DRIVER_CONNECTIONS.value, - GetInfoConstants.SQL_NUMERIC_FUNCTIONS.value, - GetInfoConstants.SQL_STRING_FUNCTIONS.value, - GetInfoConstants.SQL_DATETIME_FUNCTIONS.value, - GetInfoConstants.SQL_TXN_CAPABLE.value, - GetInfoConstants.SQL_DEFAULT_TXN_ISOLATION.value, - GetInfoConstants.SQL_CURSOR_COMMIT_BEHAVIOR.value, - } - - # Determine the type of information we're dealing with - is_string_type = ( - info_type > INFO_TYPE_STRING_THRESHOLD or info_type in string_type_constants + # Explicit numeric types take precedence over the legacy high-ID + # string fallback (e.g. SQL_MAX_IDENTIFIER_LEN is numeric at 10005). + return_type = _GETINFO_RETURN_TYPES.get(info_type) + is_string_type = return_type is str or ( + return_type is None and info_type > INFO_TYPE_STRING_THRESHOLD ) - is_yn_type = info_type in yn_type_constants - is_numeric_type = info_type in numeric_type_constants - - # Process the data based on type if is_string_type: # For string data, ensure we properly handle the byte array if isinstance(data, bytes): @@ -1958,85 +2031,33 @@ def getinfo(self, info_type: int) -> Union[str, int, bool, None]: else: # If it's not bytes, return as is return data - elif is_yn_type: - # For Y/N types, pyodbc returns a string 'Y' or 'N' - if isinstance(data, bytes) and length >= 1: - byte_val = data[0] - if byte_val in (b"Y"[0], b"y"[0], 1): - return "Y" - return "N" - # If it's not a byte or we can't determine, default to 'N' - return "N" - elif is_numeric_type: - # Handle numeric types based on length + elif isinstance(return_type, struct.Struct): if isinstance(data, bytes): - # Map byte length → signed int size - int_sizes = { - 1: lambda d: int(d[0]), - 2: lambda d: int.from_bytes(d[:2], "little", signed=True), - 4: lambda d: int.from_bytes(d[:4], "little", signed=True), - 8: lambda d: int.from_bytes(d[:8], "little", signed=True), - } - - # Direct numeric conversion if supported length - if length in int_sizes: - result = int_sizes[length](data) - return int(result) - - # Helper: check if all chars are digits - def is_digit_bytes(b: bytes) -> bool: - return all(c in b"0123456789" for c in b) - - # Helper: check if bytes are ASCII-printable or NUL padded - def is_printable_bytes(b: bytes) -> bool: - return all(32 <= c <= 126 or c == 0 for c in b) - - chunk = data[:length] - - # Try interpret as integer string - if is_digit_bytes(chunk): - return int(chunk) - - # Try decode as ASCII/UTF-8 string - if is_printable_bytes(chunk): - str_val = chunk.decode("utf-8", errors="replace").rstrip("\0") - return int(str_val) if str_val.isdigit() else str_val - - # For 16-bit values that might be returned for max lengths - if length == 2: - return int.from_bytes(data[:2], "little", signed=True) - - # For 32-bit values (common for bitwise flags) - if length == 4: - return int.from_bytes(data[:4], "little", signed=True) - - # Fallback: try to convert to int if possible - try: - if length <= 8: - return int.from_bytes(data[:length], "little", signed=True) - except Exception: - pass - - # Last resort: return as integer if all else fails - try: - return int.from_bytes(data[: min(length, 8)], "little", signed=True) - except Exception: - return 0 - elif isinstance(data, (int, float)): - # Already numeric - return int(data) - else: - # Try to convert to int if it's a string + if ( + type(length) is not int + or length != return_type.size + or len(data) < return_type.size + ): + raise DatabaseError( + driver_error=f"Invalid numeric result length for getinfo({info_type})", + ddbc_error=( + f"Expected {return_type.size} bytes; " + f"got length={length} with {len(data)} bytes of data" + ), + ) + return return_type.unpack_from(data)[0] + # Legacy non-byte payloads must not lose precision or change bool to int. + if isinstance(data, str) and data.isdecimal(): try: - if isinstance(data, str) and data.isdigit(): - return int(data) - except Exception: - pass - - # Return as is if we can't convert - return data + return int(data) + except ValueError: + logger.debug( + "Numeric getinfo compatibility value exceeds the integer conversion " + "limit; returning it unchanged" + ) + return data - # For other types, try to determine the most appropriate type + # Preserve legacy handling for unregistered, driver-specific info types. if isinstance(data, bytes): # Try to convert to string first try: diff --git a/mssql_python/constants.py b/mssql_python/constants.py index fecb9257..54a51b9c 100644 --- a/mssql_python/constants.py +++ b/mssql_python/constants.py @@ -149,6 +149,7 @@ class ConstantsDDBC(Enum): SQL_ATTR_PACKET_SIZE = 112 SQL_ATTR_QUIET_MODE = 111 SQL_ATTR_TXN_ISOLATION = 108 + SQL_TXN_ISOLATION_LEVEL = SQL_ATTR_TXN_ISOLATION # Legacy Python spelling SQL_ATTR_TRACE = 104 SQL_ATTR_TRACEFILE = 105 SQL_ATTR_TRANSLATE_LIB = 106 @@ -190,10 +191,32 @@ class ConstantsDDBC(Enum): # Query Timeout Constants SQL_ATTR_QUERY_TIMEOUT = 0 + # Legacy statement options, not SQLGetInfo information types + SQL_CONCURRENCY = 7 + SQL_ROWSET_SIZE = 9 + SQL_ROW_NUMBER = 14 + + # SQLGetInfo return values, not information types + SQL_IC_UPPER = 1 + SQL_IC_LOWER = 2 + SQL_IC_SENSITIVE = 3 + SQL_IC_MIXED = 4 + SQL_SC_SQL92_ENTRY = 1 + SQL_SC_FIPS127_2_TRANSITIONAL = 2 + SQL_SC_SQL92_INTERMEDIATE = 4 + SQL_SC_SQL92_FULL = 8 + + # Deprecated compatibility values; use SQL_SC_* for SQL_SQL_CONFORMANCE results. + SQL_SQL92_ENTRY_SQL = 127 + SQL_SQL92_INTERMEDIATE_SQL = 128 + SQL_SQL92_FULL_SQL = 129 + class GetInfoConstants(Enum): - """ - These constants are used with various methods like getinfo(). + """ODBC information-type IDs plus deprecated compatibility members. + + The deprecated members at the end retain their original values throughout + 1.x, but are not information-type names. See CHANGELOG.md for migration. """ # Driver and database information @@ -201,8 +224,8 @@ class GetInfoConstants(Enum): SQL_DRIVER_VER = 7 SQL_DRIVER_ODBC_VER = 77 SQL_DRIVER_HLIB = 76 - SQL_DRIVER_HENV = 75 - SQL_DRIVER_HDBC = 74 + SQL_DRIVER_HENV = 4 + SQL_DRIVER_HDBC = 3 SQL_DATA_SOURCE_NAME = 2 SQL_DATABASE_NAME = 16 SQL_SERVER_NAME = 13 @@ -214,9 +237,6 @@ class GetInfoConstants(Enum): SQL_IDENTIFIER_CASE = 28 SQL_IDENTIFIER_QUOTE_CHAR = 29 SQL_SPECIAL_CHARACTERS = 94 - SQL_SQL92_ENTRY_SQL = 127 - SQL_SQL92_INTERMEDIATE_SQL = 128 - SQL_SQL92_FULL_SQL = 129 SQL_SUBQUERIES = 95 SQL_EXPRESSIONS_IN_ORDERBY = 27 SQL_CORRELATION_NAME = 74 @@ -230,24 +250,24 @@ class GetInfoConstants(Enum): SQL_PROCEDURES = 21 SQL_ACCESSIBLE_TABLES = 19 SQL_ACCESSIBLE_PROCEDURES = 20 - SQL_CATALOG_NAME = 10002 + SQL_CATALOG_NAME = 10003 SQL_CATALOG_USAGE = 92 SQL_SCHEMA_USAGE = 91 SQL_COLUMN_ALIAS = 87 - SQL_DESCRIBE_PARAMETER = 10003 + SQL_DESCRIBE_PARAMETER = 10002 # Transaction support SQL_TXN_CAPABLE = 46 SQL_TXN_ISOLATION_OPTION = 72 SQL_DEFAULT_TXN_ISOLATION = 26 SQL_MULTIPLE_ACTIVE_TXN = 37 - SQL_TXN_ISOLATION_LEVEL = 108 # Data type support SQL_NUMERIC_FUNCTIONS = 49 SQL_STRING_FUNCTIONS = 50 - SQL_DATETIME_FUNCTIONS = 51 - SQL_SYSTEM_FUNCTIONS = 58 + SQL_DATETIME_FUNCTIONS = 52 + SQL_TIMEDATE_FUNCTIONS = SQL_DATETIME_FUNCTIONS + SQL_SYSTEM_FUNCTIONS = 51 SQL_CONVERT_FUNCTIONS = 48 SQL_LIKE_ESCAPE_CLAUSE = 113 @@ -271,7 +291,7 @@ class GetInfoConstants(Enum): SQL_MAX_ROW_SIZE = 104 SQL_MAX_USER_NAME_LEN = 107 - # Connection attributes + # Connection information and legacy information-type aliases SQL_ACTIVE_CONNECTIONS = 0 SQL_ACTIVE_STATEMENTS = 1 SQL_DATA_SOURCE_READ_ONLY = 25 @@ -287,16 +307,13 @@ class GetInfoConstants(Enum): SQL_DYNAMIC_CURSOR_ATTRIBUTES2 = 145 SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES1 = 146 SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES2 = 147 - SQL_STATIC_CURSOR_ATTRIBUTES1 = 150 - SQL_STATIC_CURSOR_ATTRIBUTES2 = 151 - SQL_KEYSET_CURSOR_ATTRIBUTES1 = 148 - SQL_KEYSET_CURSOR_ATTRIBUTES2 = 149 + SQL_STATIC_CURSOR_ATTRIBUTES1 = 167 + SQL_STATIC_CURSOR_ATTRIBUTES2 = 168 + SQL_KEYSET_CURSOR_ATTRIBUTES1 = 150 + SQL_KEYSET_CURSOR_ATTRIBUTES2 = 151 SQL_SCROLL_OPTIONS = 44 SQL_SCROLL_CONCURRENCY = 43 SQL_FETCH_DIRECTION = 8 - SQL_ROWSET_SIZE = 9 - SQL_CONCURRENCY = 7 - SQL_ROW_NUMBER = 14 SQL_STATIC_SENSITIVITY = 83 SQL_BATCH_SUPPORT = 121 SQL_BATCH_ROW_COUNT = 120 @@ -309,7 +326,7 @@ class GetInfoConstants(Enum): # Other constants SQL_GROUP_BY = 88 - SQL_OJ_CAPABILITIES = 65 + SQL_OJ_CAPABILITIES = 115 SQL_ORDER_BY_COLUMNS_IN_SELECT = 90 SQL_OUTER_JOINS = 38 SQL_QUOTED_IDENTIFIER_CASE = 93 @@ -324,11 +341,18 @@ class GetInfoConstants(Enum): SQL_TIMEDATE_ADD_INTERVALS = 109 SQL_TIMEDATE_DIFF_INTERVALS = 110 - # Return values for some getinfo functions - SQL_IC_UPPER = 1 - SQL_IC_LOWER = 2 - SQL_IC_SENSITIVE = 3 - SQL_IC_MIXED = 4 + # Deprecated non-information names retained for compatibility throughout 1.x. + SQL_TXN_ISOLATION_LEVEL = ConstantsDDBC.SQL_TXN_ISOLATION_LEVEL.value + SQL_CONCURRENCY = ConstantsDDBC.SQL_CONCURRENCY.value + SQL_ROWSET_SIZE = ConstantsDDBC.SQL_ROWSET_SIZE.value + SQL_ROW_NUMBER = ConstantsDDBC.SQL_ROW_NUMBER.value + SQL_IC_UPPER = ConstantsDDBC.SQL_IC_UPPER.value + SQL_IC_LOWER = ConstantsDDBC.SQL_IC_LOWER.value + SQL_IC_SENSITIVE = ConstantsDDBC.SQL_IC_SENSITIVE.value + SQL_IC_MIXED = ConstantsDDBC.SQL_IC_MIXED.value + SQL_SQL92_ENTRY_SQL = ConstantsDDBC.SQL_SQL92_ENTRY_SQL.value + SQL_SQL92_INTERMEDIATE_SQL = ConstantsDDBC.SQL_SQL92_INTERMEDIATE_SQL.value + SQL_SQL92_FULL_SQL = ConstantsDDBC.SQL_SQL92_FULL_SQL.value class AuthType(Enum): @@ -548,10 +572,12 @@ def get_attribute_set_timing(attribute): def get_info_constants() -> Dict[str, int]: """ - Returns a dictionary of all available GetInfo constants. + Return all GetInfoConstants names and values, including deprecated members. - This provides all SQLGetInfo constants that can be used with the Connection.getinfo() method - to retrieve metadata about the database server and driver. + Deprecated compatibility names remain included throughout 1.x so existing + dictionary lookups keep working. They are not valid information-type names; + their inclusion does not make them suitable inputs to Connection.getinfo(). + See CHANGELOG.md for replacements and the deprecation policy. Returns: dict: Dictionary mapping constant names to their integer values @@ -605,6 +631,23 @@ def get_info_constants() -> Dict[str, int]: "SQL_ATTR_LOGIN_TIMEOUT", "SQL_ATTR_PACKET_SIZE", "SQL_ATTR_TXN_ISOLATION", + "SQL_TXN_ISOLATION_LEVEL", + # Legacy statement options + "SQL_CONCURRENCY", + "SQL_ROWSET_SIZE", + "SQL_ROW_NUMBER", + # SQLGetInfo return values and compatibility spellings + "SQL_IC_UPPER", + "SQL_IC_LOWER", + "SQL_IC_SENSITIVE", + "SQL_IC_MIXED", + "SQL_SC_SQL92_ENTRY", + "SQL_SC_FIPS127_2_TRANSITIONAL", + "SQL_SC_SQL92_INTERMEDIATE", + "SQL_SC_SQL92_FULL", + "SQL_SQL92_ENTRY_SQL", + "SQL_SQL92_INTERMEDIATE_SQL", + "SQL_SQL92_FULL_SQL", # Transaction isolation levels "SQL_TXN_READ_UNCOMMITTED", "SQL_TXN_READ_COMMITTED", @@ -625,10 +668,11 @@ def get_info_constants() -> Dict[str, int]: _module_globals[_name] = _member.value _exported_names.append(_name) -# Export all GetInfoConstants enum members as module-level constants +# Export GetInfoConstants members not already exported from ConstantsDDBC. for _name, _member in GetInfoConstants.__members__.items(): - _module_globals[_name] = _member.value - _exported_names.append(_name) + if _name not in _DDBC_PUBLIC_API: + _module_globals[_name] = _member.value + _exported_names.append(_name) # AuthType enum is exported as a class only (not individual members) # to avoid polluting the namespace with generic names like DEFAULT diff --git a/mssql_python/mssql_python.pyi b/mssql_python/mssql_python.pyi index a69cc0ff..81222f16 100644 --- a/mssql_python/mssql_python.pyi +++ b/mssql_python/mssql_python.pyi @@ -396,6 +396,12 @@ SQL_ATTR_CURRENT_CATALOG: int SQL_ATTR_LOGIN_TIMEOUT: int SQL_ATTR_PACKET_SIZE: int SQL_ATTR_TXN_ISOLATION: int +SQL_TXN_ISOLATION_LEVEL: int + +# Legacy Statement Options (not information types) +SQL_CONCURRENCY: int +SQL_ROWSET_SIZE: int +SQL_ROW_NUMBER: int # Transaction Isolation Level Constants SQL_TXN_READ_UNCOMMITTED: int @@ -411,6 +417,9 @@ SQL_MODE_READ_ONLY: int SQL_DRIVER_NAME: int SQL_DRIVER_VER: int SQL_DRIVER_ODBC_VER: int +SQL_DRIVER_HLIB: int +SQL_DRIVER_HENV: int +SQL_DRIVER_HDBC: int SQL_DATA_SOURCE_NAME: int SQL_DATABASE_NAME: int SQL_SERVER_NAME: int @@ -428,8 +437,32 @@ SQL_DEFAULT_TXN_ISOLATION: int SQL_NUMERIC_FUNCTIONS: int SQL_STRING_FUNCTIONS: int SQL_DATETIME_FUNCTIONS: int +SQL_TIMEDATE_FUNCTIONS: int +SQL_SYSTEM_FUNCTIONS: int SQL_MAX_COLUMN_NAME_LEN: int SQL_MAX_TABLE_NAME_LEN: int SQL_MAX_SCHEMA_NAME_LEN: int SQL_MAX_CATALOG_NAME_LEN: int SQL_MAX_IDENTIFIER_LEN: int +SQL_CATALOG_NAME: int +SQL_DESCRIBE_PARAMETER: int +SQL_STATIC_CURSOR_ATTRIBUTES1: int +SQL_STATIC_CURSOR_ATTRIBUTES2: int +SQL_KEYSET_CURSOR_ATTRIBUTES1: int +SQL_KEYSET_CURSOR_ATTRIBUTES2: int +SQL_OJ_CAPABILITIES: int + +# SQLGetInfo Return Values (not information types) +SQL_IC_UPPER: int +SQL_IC_LOWER: int +SQL_IC_SENSITIVE: int +SQL_IC_MIXED: int +SQL_SC_SQL92_ENTRY: int +SQL_SC_FIPS127_2_TRANSITIONAL: int +SQL_SC_SQL92_INTERMEDIATE: int +SQL_SC_SQL92_FULL: int + +# Deprecated legacy values (127/128/129), not SQL conformance flags +SQL_SQL92_ENTRY_SQL: int +SQL_SQL92_INTERMEDIATE_SQL: int +SQL_SQL92_FULL_SQL: int diff --git a/tests/test_003_connection.py b/tests/test_003_connection.py index 13d33bc5..d4536fdb 100644 --- a/tests/test_003_connection.py +++ b/tests/test_003_connection.py @@ -3008,7 +3008,7 @@ def test_getinfo_sql_support(db_connection): # SQL conformance level sql_conformance = db_connection.getinfo(sql_const.SQL_SQL_CONFORMANCE.value) print("SQL Conformance = ", sql_conformance) - assert sql_conformance is not None, "SQL conformance should not be None" + assert type(sql_conformance) is int, "SQL conformance should be an integer" # Keywords - may return a very long string keywords = db_connection.getinfo(sql_const.SQL_KEYWORDS.value) @@ -3120,40 +3120,24 @@ def test_getinfo_type_consistency(db_connection): assert result1 == result2, f"Value inconsistency for info type {info_type}" -def test_getinfo_standard_types(db_connection): +@pytest.mark.parametrize( + "info_type,expected_type", + [ + (sql_const.SQL_ACCESSIBLE_TABLES.value, str), + (sql_const.SQL_DATA_SOURCE_NAME.value, str), + (sql_const.SQL_TABLE_TERM.value, str), + (sql_const.SQL_PROCEDURES.value, str), + (sql_const.SQL_MAX_IDENTIFIER_LEN.value, int), + (sql_const.SQL_OUTER_JOINS.value, str), + ], +) +def test_getinfo_standard_types(db_connection, info_type, expected_type): """Test a representative set of standard ODBC info types.""" - # Dictionary of common info types and their expected value types - # Avoid DBMS-specific info types - info_types = { - sql_const.SQL_ACCESSIBLE_TABLES.value: str, # "Y" or "N" - sql_const.SQL_DATA_SOURCE_NAME.value: str, # DSN - sql_const.SQL_TABLE_TERM.value: str, # Usually "table" - sql_const.SQL_PROCEDURES.value: str, # "Y" or "N" - sql_const.SQL_MAX_IDENTIFIER_LEN.value: int, # Max identifier length - sql_const.SQL_OUTER_JOINS.value: str, # "Y" or "N" - } - - for info_type, expected_type in info_types.items(): - try: - info_value = db_connection.getinfo(info_type) - print(info_type, info_value) - - # Skip None values (unsupported by driver) - if info_value is None: - continue - - # Check type, allowing empty strings for string types - if expected_type == str: - assert isinstance(info_value, str), f"Info type {info_type} should return a string" - elif expected_type == int: - assert isinstance( - info_value, int - ), f"Info type {info_type} should return an integer" - - except Exception as e: - # Log but don't fail - some drivers might not support all info types - print(f"Info type {info_type} failed: {e}") + info_value = db_connection.getinfo(info_type) + assert ( + type(info_value) is expected_type + ), f"Info type {info_type} should return {expected_type.__name__}, got {info_value!r}" def test_getinfo_numeric_limits(db_connection): diff --git a/tests/test_027_getinfo.py b/tests/test_027_getinfo.py new file mode 100644 index 00000000..2970910b --- /dev/null +++ b/tests/test_027_getinfo.py @@ -0,0 +1,621 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. +Regression coverage for SQLGetInfo IDs and ODBC return types (GH-769). +""" + +import ast +from decimal import Decimal +from enum import Enum +import logging +from pathlib import Path +import pickle +import struct +import sys +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +import mssql_python +from mssql_python import constants +from mssql_python.connection import Connection, _GETINFO_RETURN_TYPES +from mssql_python.constants import ConstantsDDBC, GetInfoConstants as G +from mssql_python.exceptions import DatabaseError, InterfaceError + +# Independent reference from ODBC 3.x sql.h/sqlext.h (Windows SDK 10.0.26100.0 +# and unixODBC 2.3.x), and the SQLGetInfo return-type descriptions: +# https://learn.microsoft.com/sql/odbc/reference/syntax/sqlgetinfo-function +# Each entry is (information-type ID, numeric byte width or 0 for text). +# Keep literal IDs here: deriving expectations from G would mask transcription +# mistakes. __members__ must be used so that erroneous Enum aliases are visible. +ODBC_INFO = { + "SQL_DRIVER_NAME": (6, 0), + "SQL_DRIVER_VER": (7, 0), + "SQL_DRIVER_ODBC_VER": (77, 0), + "SQL_DRIVER_HLIB": (76, struct.calcsize("P")), + "SQL_DRIVER_HENV": (4, struct.calcsize("P")), + "SQL_DRIVER_HDBC": (3, struct.calcsize("P")), + "SQL_DATA_SOURCE_NAME": (2, 0), + "SQL_DATABASE_NAME": (16, 0), + "SQL_SERVER_NAME": (13, 0), + "SQL_USER_NAME": (47, 0), + "SQL_SQL_CONFORMANCE": (118, 4), + "SQL_KEYWORDS": (89, 0), + "SQL_IDENTIFIER_CASE": (28, 2), + "SQL_IDENTIFIER_QUOTE_CHAR": (29, 0), + "SQL_SPECIAL_CHARACTERS": (94, 0), + "SQL_SUBQUERIES": (95, 4), + "SQL_EXPRESSIONS_IN_ORDERBY": (27, 0), + "SQL_CORRELATION_NAME": (74, 2), + "SQL_SEARCH_PATTERN_ESCAPE": (14, 0), + "SQL_CATALOG_TERM": (42, 0), + "SQL_CATALOG_NAME_SEPARATOR": (41, 0), + "SQL_SCHEMA_TERM": (39, 0), + "SQL_TABLE_TERM": (45, 0), + "SQL_PROCEDURES": (21, 0), + "SQL_ACCESSIBLE_TABLES": (19, 0), + "SQL_ACCESSIBLE_PROCEDURES": (20, 0), + "SQL_CATALOG_NAME": (10003, 0), + "SQL_CATALOG_USAGE": (92, 4), + "SQL_SCHEMA_USAGE": (91, 4), + "SQL_COLUMN_ALIAS": (87, 0), + "SQL_DESCRIBE_PARAMETER": (10002, 0), + "SQL_TXN_CAPABLE": (46, 2), + "SQL_TXN_ISOLATION_OPTION": (72, 4), + "SQL_DEFAULT_TXN_ISOLATION": (26, 4), + "SQL_MULTIPLE_ACTIVE_TXN": (37, 0), + "SQL_NUMERIC_FUNCTIONS": (49, 4), + "SQL_STRING_FUNCTIONS": (50, 4), + "SQL_TIMEDATE_FUNCTIONS": (52, 4), + "SQL_DATETIME_FUNCTIONS": (52, 4), # Deliberate Python compatibility spelling + "SQL_SYSTEM_FUNCTIONS": (51, 4), + "SQL_CONVERT_FUNCTIONS": (48, 4), + "SQL_LIKE_ESCAPE_CLAUSE": (113, 0), + "SQL_MAX_COLUMN_NAME_LEN": (30, 2), + "SQL_MAX_TABLE_NAME_LEN": (35, 2), + "SQL_MAX_SCHEMA_NAME_LEN": (32, 2), + "SQL_MAX_CATALOG_NAME_LEN": (34, 2), + "SQL_MAX_IDENTIFIER_LEN": (10005, 2), + "SQL_MAX_STATEMENT_LEN": (105, 4), + "SQL_MAX_CHAR_LITERAL_LEN": (108, 4), + "SQL_MAX_BINARY_LITERAL_LEN": (112, 4), + "SQL_MAX_COLUMNS_IN_TABLE": (101, 2), + "SQL_MAX_COLUMNS_IN_SELECT": (100, 2), + "SQL_MAX_COLUMNS_IN_GROUP_BY": (97, 2), + "SQL_MAX_COLUMNS_IN_ORDER_BY": (99, 2), + "SQL_MAX_COLUMNS_IN_INDEX": (98, 2), + "SQL_MAX_TABLES_IN_SELECT": (106, 2), + "SQL_MAX_CONCURRENT_ACTIVITIES": (1, 2), + "SQL_MAX_DRIVER_CONNECTIONS": (0, 2), + "SQL_MAX_ROW_SIZE": (104, 4), + "SQL_MAX_USER_NAME_LEN": (107, 2), + "SQL_ACTIVE_CONNECTIONS": (0, 2), + "SQL_ACTIVE_STATEMENTS": (1, 2), + "SQL_DATA_SOURCE_READ_ONLY": (25, 0), + "SQL_NEED_LONG_DATA_LEN": (111, 0), + "SQL_GETDATA_EXTENSIONS": (81, 4), + "SQL_CURSOR_COMMIT_BEHAVIOR": (23, 2), + "SQL_CURSOR_ROLLBACK_BEHAVIOR": (24, 2), + "SQL_CURSOR_SENSITIVITY": (10001, 4), + "SQL_BOOKMARK_PERSISTENCE": (82, 4), + "SQL_DYNAMIC_CURSOR_ATTRIBUTES1": (144, 4), + "SQL_DYNAMIC_CURSOR_ATTRIBUTES2": (145, 4), + "SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES1": (146, 4), + "SQL_FORWARD_ONLY_CURSOR_ATTRIBUTES2": (147, 4), + "SQL_STATIC_CURSOR_ATTRIBUTES1": (167, 4), + "SQL_STATIC_CURSOR_ATTRIBUTES2": (168, 4), + "SQL_KEYSET_CURSOR_ATTRIBUTES1": (150, 4), + "SQL_KEYSET_CURSOR_ATTRIBUTES2": (151, 4), + "SQL_SCROLL_OPTIONS": (44, 4), + "SQL_SCROLL_CONCURRENCY": (43, 4), + "SQL_FETCH_DIRECTION": (8, 4), + "SQL_STATIC_SENSITIVITY": (83, 4), + "SQL_BATCH_SUPPORT": (121, 4), + "SQL_BATCH_ROW_COUNT": (120, 4), + "SQL_PARAM_ARRAY_ROW_COUNTS": (153, 4), + "SQL_PARAM_ARRAY_SELECTS": (154, 4), + "SQL_PROCEDURE_TERM": (40, 0), + "SQL_POSITIONED_STATEMENTS": (80, 4), + "SQL_GROUP_BY": (88, 2), + "SQL_OJ_CAPABILITIES": (115, 4), + "SQL_ORDER_BY_COLUMNS_IN_SELECT": (90, 0), + "SQL_OUTER_JOINS": (38, 0), + "SQL_QUOTED_IDENTIFIER_CASE": (93, 2), + "SQL_CONCAT_NULL_BEHAVIOR": (22, 2), + "SQL_NULL_COLLATION": (85, 2), + "SQL_ALTER_TABLE": (86, 4), + "SQL_UNION": (96, 4), + "SQL_DDL_INDEX": (170, 4), + "SQL_MULT_RESULT_SETS": (36, 0), + "SQL_OWNER_USAGE": (91, 4), + "SQL_QUALIFIER_USAGE": (92, 4), + "SQL_TIMEDATE_ADD_INTERVALS": (109, 4), + "SQL_TIMEDATE_DIFF_INTERVALS": (110, 4), +} + +LEGACY_GETINFO_CONSTANTS = { + "SQL_TXN_ISOLATION_LEVEL": 108, + "SQL_CONCURRENCY": 7, + "SQL_ROWSET_SIZE": 9, + "SQL_ROW_NUMBER": 14, + "SQL_IC_UPPER": 1, + "SQL_IC_LOWER": 2, + "SQL_IC_SENSITIVE": 3, + "SQL_IC_MIXED": 4, + "SQL_SQL92_ENTRY_SQL": 127, + "SQL_SQL92_INTERMEDIATE_SQL": 128, + "SQL_SQL92_FULL_SQL": 129, +} +CONFORMANCE_VALUES = { + "SQL_SC_SQL92_ENTRY": 1, + "SQL_SC_FIPS127_2_TRANSITIONAL": 2, + "SQL_SC_SQL92_INTERMEDIATE": 4, + "SQL_SC_SQL92_FULL": 8, +} +NON_INFO_CONSTANTS = LEGACY_GETINFO_CONSTANTS | CONFORMANCE_VALUES + +UNLISTED_ODBC_INFO = { + "SQL_DBMS_NAME": (17, 0), + "SQL_DBMS_VER": (18, 0), + "SQL_XOPEN_CLI_YEAR": (10000, 0), + "SQL_ASYNC_MODE": (10021, 4), + "SQL_CREATE_ASSERTION": (127, 4), +} +ALL_ODBC_INFO = ODBC_INFO | UNLISTED_ODBC_INFO +NUMERIC_INFO = {name: spec for name, spec in ALL_ODBC_INFO.items() if spec[1]} +STRING_INFO = {name: spec for name, spec in ALL_ODBC_INFO.items() if not spec[1]} +DRIVER_MANAGER_INFO = {"SQL_DRIVER_HDBC", "SQL_DRIVER_HENV", "SQL_DRIVER_HLIB"} + + +@pytest.fixture +def mock_connection(): + # Avoid constructing a native connection for the decoder and export tests. + return SimpleNamespace(_closed=False, _conn=Mock()) + + +def test_getinfo_reference_covers_every_member_and_alias(): + assert set(G.__members__) == ODBC_INFO.keys() | LEGACY_GETINFO_CONSTANTS.keys() + assert constants.get_info_constants() == ( + {name: spec[0] for name, spec in ODBC_INFO.items()} | LEGACY_GETINFO_CONSTANTS + ) + assert set(_GETINFO_RETURN_TYPES) == {spec[0] for spec in ALL_ODBC_INFO.values()} + assert {name: member.name for name, member in G.__members__.items() if name != member.name} == { + "SQL_TIMEDATE_FUNCTIONS": "SQL_DATETIME_FUNCTIONS", + "SQL_ACTIVE_CONNECTIONS": "SQL_MAX_DRIVER_CONNECTIONS", + "SQL_ACTIVE_STATEMENTS": "SQL_MAX_CONCURRENT_ACTIVITIES", + "SQL_OWNER_USAGE": "SQL_SCHEMA_USAGE", + "SQL_QUALIFIER_USAGE": "SQL_CATALOG_USAGE", + "SQL_TXN_ISOLATION_LEVEL": "SQL_MAX_CHAR_LITERAL_LEN", + "SQL_CONCURRENCY": "SQL_DRIVER_VER", + "SQL_ROW_NUMBER": "SQL_SEARCH_PATTERN_ESCAPE", + "SQL_IC_UPPER": "SQL_MAX_CONCURRENT_ACTIVITIES", + "SQL_IC_LOWER": "SQL_DATA_SOURCE_NAME", + "SQL_IC_SENSITIVE": "SQL_DRIVER_HDBC", + "SQL_IC_MIXED": "SQL_DRIVER_HENV", + } + + +@pytest.mark.parametrize("name", ALL_ODBC_INFO) +def test_getinfo_ids_and_return_types_match_odbc(name): + info_id, size = ALL_ODBC_INFO[name] + if name in ODBC_INFO: + assert G.__members__[name].value == info_id + return_type = _GETINFO_RETURN_TYPES[info_id] + if size: + assert isinstance(return_type, struct.Struct) + assert return_type.size == size + assert return_type.format == ( + "P" if name in DRIVER_MANAGER_INFO else {2: "=H", 4: "=I"}[size] + ) + else: + assert return_type is str + + +def test_getinfo_type_registry_is_immutable(): + with pytest.raises(TypeError): + _GETINFO_RETURN_TYPES[118] = str + + +def test_getinfo_public_exports_and_stubs(): + expected = {name: spec[0] for name, spec in ODBC_INFO.items()} | NON_INFO_CONSTANTS + stub = Path(mssql_python.__file__).with_name("mssql_python.pyi") + declarations = { + node.target.id: node.annotation.id + for node in ast.parse(stub.read_text(encoding="utf-8")).body + if isinstance(node, ast.AnnAssign) + and isinstance(node.target, ast.Name) + and isinstance(node.annotation, ast.Name) + } + for name, value in expected.items(): + assert getattr(constants, name) == value + assert getattr(mssql_python, name) == value + assert constants.__all__.count(name) == 1 + assert mssql_python.__all__.count(name) == 1 + changed_info_names = { + "SQL_DRIVER_HDBC", + "SQL_DRIVER_HENV", + "SQL_CATALOG_NAME", + "SQL_DESCRIBE_PARAMETER", + "SQL_DATETIME_FUNCTIONS", + "SQL_TIMEDATE_FUNCTIONS", + "SQL_SYSTEM_FUNCTIONS", + "SQL_KEYSET_CURSOR_ATTRIBUTES1", + "SQL_KEYSET_CURSOR_ATTRIBUTES2", + "SQL_STATIC_CURSOR_ATTRIBUTES1", + "SQL_STATIC_CURSOR_ATTRIBUTES2", + "SQL_OJ_CAPABILITIES", + } + for name in changed_info_names | NON_INFO_CONSTANTS.keys(): + assert declarations[name] == "int" + + +@pytest.mark.parametrize("name,value", CONFORMANCE_VALUES.items()) +def test_conformance_values_are_not_advertised_as_information_types(name, value): + assert name not in G.__members__ + assert name not in constants.get_info_constants() + assert ConstantsDDBC.__members__[name].value == value + + +@pytest.mark.parametrize("name,value", LEGACY_GETINFO_CONSTANTS.items()) +def test_getinfo_legacy_attributes_and_imports_preserve_original_values(name, value): + assert getattr(G, name).value == value + assert G[name].value == value + assert constants.get_info_constants()[name] == value + assert getattr(ConstantsDDBC, name).value == value + assert getattr(constants, name) == value + assert getattr(mssql_python, name) == value + + +def test_getinfo_datetime_alias_preserves_existing_canonical_name(): + assert G.SQL_DATETIME_FUNCTIONS.name == "SQL_DATETIME_FUNCTIONS" + assert G.SQL_TIMEDATE_FUNCTIONS is G.SQL_DATETIME_FUNCTIONS + assert G.SQL_DATETIME_FUNCTIONS.value == 52 + + +def test_getinfo_legacy_name_does_not_change_colliding_information_type(mock_connection): + mock_connection._conn.get_info.return_value = {"data": b"\\\x00", "length": 2} + assert Connection.getinfo(mock_connection, G.SQL_ROW_NUMBER.value) == "\\" + assert Connection.getinfo(mock_connection, G.SQL_SEARCH_PATTERN_ESCAPE.value) == "\\" + assert mock_connection._conn.get_info.call_count == 2 + mock_connection._conn.get_info.assert_called_with(14) + + +@pytest.mark.parametrize("name", NUMERIC_INFO) +@pytest.mark.parametrize("case", ["zero", "one", "ascii", "high_bit", "maximum"]) +def test_getinfo_unsigned_numeric_values_and_forwarded_ids(mock_connection, name, case): + info_id, size = NUMERIC_INFO[name] + value = { + "zero": 0, + "one": 1, + "ascii": 65, + "high_bit": 1 << (size * 8 - 1), + "maximum": (1 << (size * 8)) - 1, + }[case] + mock_connection._conn.get_info.return_value = { + "data": value.to_bytes(size, sys.byteorder) + b"ignored padding", + "length": size, + } + request = G.__members__[name].value if name in ODBC_INFO else info_id + result = Connection.getinfo(mock_connection, request) + assert type(result) is int + assert result == value + mock_connection._conn.get_info.assert_called_once_with(info_id) + + +@pytest.mark.parametrize("name", STRING_INFO) +@pytest.mark.parametrize("value", ["", "Y", "N", "F", "catalog_\u03a9_\U0001f600"]) +def test_getinfo_character_values_are_preserved(mock_connection, name, value): + info_id, _ = STRING_INFO[name] + data = value.encode("utf-16-le") + mock_connection._conn.get_info.return_value = { + "data": data + "\0ignored padding".encode("utf-16-le"), + "length": len(data), + } + request = G.__members__[name].value if name in ODBC_INFO else info_id + result = Connection.getinfo(mock_connection, request) + assert type(result) is str + assert result == value + mock_connection._conn.get_info.assert_called_once_with(info_id) + + +@pytest.mark.parametrize( + "info_id,data,length", + [ + (118, b"", 0), + (118, b"\x01", 2), + (118, b"\x01\x00\x00", 4), + (118, b"\x01\x00\x00", 3), + (118, b"\x01\x00\x00\x00\x00", 5), + (118, b"\x01\x00\x00\x00", -1), + (118, b"\x01\x00\x00\x00", 4.0), + (118, b"\x01\x00\x00\x00", "4"), + (118, b"\x01\x00\x00\x00", None), + (118, b"\x01\x00\x00\x00", True), + ], +) +def test_getinfo_rejects_malformed_numeric_data(mock_connection, info_id, data, length): + mock_connection._conn.get_info.return_value = {"data": data, "length": length} + with pytest.raises(DatabaseError, match="Invalid numeric result length"): + Connection.getinfo(mock_connection, info_id) + + +@pytest.mark.parametrize("info_id", [9, 75, 58, 65, 127, 128, 129, 148, 149]) +def test_getinfo_old_colliding_ids_are_still_forwarded(mock_connection, info_id): + mock_connection._conn.get_info.return_value = 1 + result = Connection.getinfo(mock_connection, info_id) + assert type(result) is int + assert result == 1 + mock_connection._conn.get_info.assert_called_once_with(info_id) + + +@pytest.mark.parametrize( + "info_id,error", + [ + (118, RuntimeError("SQLSTATE:HY096:Invalid information type")), + (65536, TypeError("Information type out of range")), + (65536, OverflowError("Information type out of range")), + ], +) +def test_getinfo_unsupported_native_requests_keep_returning_none(mock_connection, info_id, error): + mock_connection._conn.get_info.side_effect = error + assert Connection.getinfo(mock_connection, info_id) is None + mock_connection._conn.get_info.assert_called_once_with(info_id) + + +@pytest.mark.parametrize("value", ["invalid", None, G.SQL_SQL_CONFORMANCE, 1.5]) +def test_getinfo_non_integer_input_is_rejected(mock_connection, value): + with pytest.raises(ValueError, match="info_type must be an integer"): + Connection.getinfo(mock_connection, value) + mock_connection._conn.get_info.assert_not_called() + + +def test_getinfo_closed_and_negative_requests(mock_connection): + assert Connection.getinfo(mock_connection, -1) is None + mock_connection._closed = True + with pytest.raises(InterfaceError): + Connection.getinfo(mock_connection, 118) + mock_connection._conn.get_info.assert_not_called() + + +@pytest.mark.parametrize("info_type", [-1, -65536]) +def test_getinfo_negative_id_logs_without_formatting_failure( + mock_connection, monkeypatch, capsys, info_type +): + log_sink = Mock() + monkeypatch.setattr("mssql_python.connection.logger._logger", log_sink) + monkeypatch.setattr("mssql_python.connection.logger._cached_level", logging.DEBUG) + + assert Connection.getinfo(mock_connection, info_type) is None + mock_connection._conn.get_info.assert_not_called() + log_sink.log.assert_called_once_with( + logging.DEBUG, + f"[Python] Invalid info_type: {info_type}. Must be non-negative.", + stacklevel=3, + ) + assert capsys.readouterr().err == "" + + +@pytest.mark.parametrize("result", [None, 1, "Y", True]) +def test_getinfo_already_decoded_native_results(mock_connection, result): + mock_connection._conn.get_info.return_value = result + assert Connection.getinfo(mock_connection, 118) is result + + +def test_getinfo_unknown_driver_specific_type_keeps_legacy_handling(mock_connection): + mock_connection._conn.get_info.return_value = {"data": b"vendor", "length": 6} + assert Connection.getinfo(mock_connection, 999) == "vendor" + + +@pytest.mark.parametrize("value", ["Example Driver", "\u00e9", "", "\u03a9_\U0001f600"]) +def test_getinfo_unlisted_high_ids_keep_unicode_decoding(mock_connection, value): + data = value.encode("utf-16-le") + mock_connection._conn.get_info.return_value = {"data": data, "length": len(data)} + result = Connection.getinfo(mock_connection, 65000) + assert type(result) is str + assert result == value + mock_connection._conn.get_info.assert_called_once_with(65000) + + +@pytest.mark.parametrize( + "name", + [ + ( + pytest.param( + name, + marks=pytest.mark.skip( + reason=f"{name} requires a Driver Manager; native providers are loaded directly" + ), + ) + if name in DRIVER_MANAGER_INFO + else name + ) + for name in ALL_ODBC_INFO + ], +) +def test_getinfo_matches_native_odbc_payload(db_connection, name): + info_id, size = ALL_ODBC_INFO[name] + raw = db_connection._conn.get_info(info_id) + assert isinstance(raw, dict) + assert raw["info_type"] == info_id + data = raw["data"][: raw["length"]] + if size: + assert raw["length"] == size + expected = int.from_bytes(data, sys.byteorder, signed=False) + expected_type = int + else: + expected = data.decode("utf-16-le").rstrip("\0") + expected_type = str + request = G.__members__[name].value if name in ODBC_INFO else info_id + result = db_connection.getinfo(request) + assert type(result) is expected_type + assert result == expected + + +def test_getinfo_distinguishes_swapped_ids_even_when_driver_values_match(mock_connection): + # SQL Server commonly answers "Y" to both, hiding the swapped constants in + # output-only integration tests. + payloads = {10002: "N", 10003: "Y"} + mock_connection._conn.get_info.side_effect = lambda info_id: { + "data": payloads[info_id].encode("utf-16-le"), + "length": 2, + } + assert Connection.getinfo(mock_connection, G.SQL_CATALOG_NAME.value) == "Y" + assert Connection.getinfo(mock_connection, G.SQL_DESCRIBE_PARAMETER.value) == "N" + + +@pytest.mark.parametrize( + "name,length", + [ + (name, length) + for name, (_, size) in NUMERIC_INFO.items() + for length in (1, 2, 4, 8) + if length != size + ], +) +def test_getinfo_rejects_type_specific_wrong_widths(mock_connection, name, length): + info_id, _ = NUMERIC_INFO[name] + mock_connection._conn.get_info.return_value = {"data": b"\xff" * length, "length": length} + with pytest.raises(DatabaseError, match="Invalid numeric result length"): + Connection.getinfo(mock_connection, info_id) + mock_connection._conn.get_info.assert_called_once_with(info_id) + + +@pytest.mark.parametrize( + "data", + [1, -1, True, False, 1.9, float("inf"), Decimal("1.9"), "\u00b2", "", "1.9", "text", None], +) +def test_getinfo_nonbyte_numeric_values_are_not_lossily_coerced(mock_connection, data): + mock_connection._conn.get_info.return_value = {"data": data, "length": 4} + assert Connection.getinfo(mock_connection, 118) is data + + +@pytest.mark.parametrize( + "data,expected", [("0", 0), ("000123", 123), ("123", 123), ("\u0661\u0662", 12)] +) +def test_getinfo_decimal_strings_keep_integer_compatibility(mock_connection, data, expected): + mock_connection._conn.get_info.return_value = {"data": data, "length": len(data)} + result = Connection.getinfo(mock_connection, 118) + assert type(result) is int + assert result == expected + + +def test_getinfo_oversized_decimal_string_keeps_compatibility(mock_connection): + limit = getattr(sys, "get_int_max_str_digits", lambda: 0)() + if not limit: + pytest.skip("Interpreter integer-string conversion limit is disabled or unavailable") + data = "1" * (limit + 1) + mock_connection._conn.get_info.return_value = {"data": data, "length": len(data)} + assert Connection.getinfo(mock_connection, 118) is data + + +@pytest.mark.parametrize("sqlstate", ["HY096", "HYC00", "08S01", "08003", "HYT00", "HYT01"]) +def test_getinfo_native_failures_keep_logged_none_contract(mock_connection, monkeypatch, sqlstate): + warning = Mock() + monkeypatch.setattr("mssql_python.connection.logger.warning", warning) + mock_connection._conn.get_info.side_effect = RuntimeError(f"SQLSTATE:{sqlstate}:Native failure") + assert Connection.getinfo(mock_connection, 118) is None + mock_connection._conn.get_info.assert_called_once_with(118) + warning.assert_called_once() + assert sqlstate in warning.call_args.args[0] + + +@pytest.mark.parametrize("info_id", [6, 999]) +@pytest.mark.parametrize("data", ["metadata", 1, 1.9, True, None, {"value": 1}]) +def test_getinfo_nonbyte_text_and_unknown_values_are_unchanged(mock_connection, info_id, data): + mock_connection._conn.get_info.return_value = {"data": data, "length": 4} + assert Connection.getinfo(mock_connection, info_id) is data + + +@pytest.mark.parametrize("data,expected", [(b"\xff\xff", -1), (b"\xff" * 9, b"\xff" * 9)]) +def test_getinfo_unknown_binary_fallback_is_preserved(mock_connection, data, expected): + mock_connection._conn.get_info.return_value = {"data": data, "length": len(data)} + result = Connection.getinfo(mock_connection, 999) + assert type(result) is type(expected) + assert result == expected + + +@pytest.mark.parametrize("result", [1.9, b"raw", [], {}, {"length": 4}]) +def test_getinfo_unrecognized_native_result_is_unchanged(mock_connection, result): + mock_connection._conn.get_info.return_value = result + assert Connection.getinfo(mock_connection, 118) is result + + +def test_getinfo_missing_native_length_is_not_silently_defaulted(mock_connection): + mock_connection._conn.get_info.return_value = {"data": b"\x01\x00\x00\x00"} + with pytest.raises(KeyError, match="length"): + Connection.getinfo(mock_connection, 118) + + +def test_getinfo_does_not_cache_metadata_across_connections(mock_connection): + other = SimpleNamespace(_closed=False, _conn=Mock()) + mock_connection._conn.get_info.return_value = { + "data": (1).to_bytes(4, sys.byteorder), + "length": 4, + } + other._conn.get_info.return_value = {"data": (2).to_bytes(4, sys.byteorder), "length": 4} + assert Connection.getinfo(mock_connection, 118) == 1 + assert Connection.getinfo(other, 118) == 2 + mock_connection._conn.get_info.return_value = { + "data": (4).to_bytes(4, sys.byteorder), + "length": 4, + } + assert Connection.getinfo(mock_connection, 118) == 4 + assert mock_connection._conn.get_info.call_count == 2 + other._conn.get_info.assert_called_once_with(118) + + +@pytest.mark.parametrize("name", list(ODBC_INFO) + list(LEGACY_GETINFO_CONSTANTS)) +def test_getinfo_current_enum_pickle_round_trip(name): + member = G.__members__[name] + assert pickle.loads(pickle.dumps(member)) is member + + +def _pickle_enum_by_value(member, protocol): + return member.__class__, (member.value,) + + +def _pickle_enum_by_name(member, protocol): + return getattr, (member.__class__, member.name) + + +@pytest.mark.parametrize( + "reducer", [_pickle_enum_by_value, _pickle_enum_by_name], ids=["by_value", "by_name"] +) +@pytest.mark.parametrize("name,value", LEGACY_GETINFO_CONSTANTS.items()) +def test_getinfo_legacy_attribute_pickles_preserve_values(monkeypatch, name, value, reducer): + legacy = Enum( + "GetInfoConstants", {name: value, "__reduce_ex__": reducer}, module=constants.__name__ + ) + with monkeypatch.context() as patch: + patch.setattr(constants, "GetInfoConstants", legacy) + serialized = pickle.dumps(legacy[name]) + + restored = pickle.loads(serialized) + assert restored is getattr(G, name) + assert restored.value == value + + +@pytest.mark.parametrize( + "reducer,expected_name", + [ + pytest.param(_pickle_enum_by_value, "SQL_KEYSET_CURSOR_ATTRIBUTES1", id="by_value"), + pytest.param(_pickle_enum_by_name, "SQL_STATIC_CURSOR_ATTRIBUTES1", id="by_name"), + ], +) +def test_getinfo_legacy_pickle_resolves_using_its_serialized_form( + monkeypatch, reducer, expected_name +): + # Enum's default reducer differs across Python versions; model both formats explicitly. + legacy = Enum( + "GetInfoConstants", + {"SQL_STATIC_CURSOR_ATTRIBUTES1": 150, "__reduce_ex__": reducer}, + module=constants.__name__, + ) + with monkeypatch.context() as patch: + patch.setattr(constants, "GetInfoConstants", legacy) + serialized = pickle.dumps(legacy.SQL_STATIC_CURSOR_ATTRIBUTES1) + + # Value-based pickles resolve 150 to keyset; name-based pickles keep the static identity. + assert pickle.loads(serialized) is G[expected_name] + assert G["SQL_STATIC_CURSOR_ATTRIBUTES1"].value == 167