From 3c487812a6934df28a7677bc0626d5f82a3ab3dd Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Thu, 24 Sep 2026 10:07:23 +0000 Subject: [PATCH 1/8] FEAT: Add preview TVP support for async queries Expose native TVP and SQL type-hint helpers lazily. Support iterable executemany batches and preserve result state on iteration failure. Add cursor closed-state reporting, document async result contracts, and cover these behaviors with async integration tests. --- mssql_python/async_query/__init__.py | 42 +++++ mssql_python/async_query/async_cursor.py | 32 +++- mssql_python/async_query/async_execute.py | 29 +++- .../AsyncTest/test_001_async_query_native.py | 50 ++++++ tests/AsyncTest/test_004_async_logging.py | 2 +- tests/AsyncTest/test_005_async_cursor.py | 67 +++++++ tests/AsyncTest/test_006_async_execute.py | 164 +++++++++++++++++- 7 files changed, 373 insertions(+), 13 deletions(-) diff --git a/mssql_python/async_query/__init__.py b/mssql_python/async_query/__init__.py index 95f8f0ecf..774fa136a 100644 --- a/mssql_python/async_query/__init__.py +++ b/mssql_python/async_query/__init__.py @@ -2,6 +2,9 @@ The internal entry points are ``_AsyncConnection`` and ``_AsyncCursor``. Their methods retain DB-API names, but the classes are not stable public API. +``_TableValuedParameter`` lazily exposes the native TVP constructor. The +``SQL_*`` tokens are native setinputsizes() hints; server support is required +for the corresponding SQL types, particularly JSON and VECTOR. Warning: Async query execution APIs are under active development and are not intended @@ -25,6 +28,40 @@ Warning, ) +SQL_MONEY: int +SQL_SMALLMONEY: int +SQL_XML: int +SQL_JSON: int +SQL_VECTOR: int + +_NATIVE_EXPORTS = { + "_TableValuedParameter": "TableValuedParameter", + "SQL_MONEY": "SQL_MONEY", + "SQL_SMALLMONEY": "SQL_SMALLMONEY", + "SQL_XML": "SQL_XML", + "SQL_JSON": "SQL_JSON", + "SQL_VECTOR": "SQL_VECTOR", +} + + +def __getattr__(name: str): + native_name = _NATIVE_EXPORTS.get(name) + if native_name is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + py_core = load_py_core() + try: + return getattr(py_core, native_name) + except AttributeError as error: + raise ImportError( + f"The installed mssql-python-rs dependency does not provide {native_name}; " + "install a version with this async feature." + ) from error + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_NATIVE_EXPORTS)) + + __all__ = [ "DataError", "DatabaseError", @@ -35,6 +72,11 @@ "NotSupportedError", "OperationalError", "ProgrammingError", + "SQL_MONEY", + "SQL_SMALLMONEY", + "SQL_XML", + "SQL_JSON", + "SQL_VECTOR", "Warning", "load_py_core", ] diff --git a/mssql_python/async_query/async_cursor.py b/mssql_python/async_query/async_cursor.py index 2547e2fae..89e005864 100644 --- a/mssql_python/async_query/async_cursor.py +++ b/mssql_python/async_query/async_cursor.py @@ -7,7 +7,7 @@ """ import asyncio -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from contextlib import asynccontextmanager from typing import Any, Optional import uuid @@ -142,7 +142,7 @@ async def execute( async def executemany( self, operation: str, - seq_of_parameters: Sequence[Sequence[Any]] | Sequence[Mapping[str, Any]], + seq_of_parameters: Iterable[Sequence[Any] | Mapping[str, Any]], *, use_prepare: bool = True, ) -> None: @@ -164,6 +164,11 @@ async def fetchall(self) -> list[Row]: return await async_fetch.fetchall(self) async def nextset(self) -> bool: + """Discard remaining rows and advance to the next statement result. + + Return True for another result, including DML results, or False at + batch end. Fetch counts are reset rather than carried between results. + """ async with self._result_transition(): try: with translate_py_core_exceptions(): @@ -178,6 +183,13 @@ async def nextset(self) -> bool: return has_next async def close(self) -> None: + """Close this cursor; repeated calls are harmless. + + Successful close clears description and resets rowcount to -1, unlike + the synchronous cursor, which retains its last metadata and count. + Rejected close preserves the cursor; accepted but interrupted cleanup + retires the wrapper without guaranteeing native cleanup has completed. + """ logger.debug("AsyncCursor.close: starting") async with self._result_transition(): with translate_py_core_exceptions(): @@ -203,6 +215,15 @@ def setinputsizes(self, sizes: Any) -> None: with translate_py_core_exceptions(): self._py_core_async_cursor.setinputsizes(sizes) + @property + def closed(self) -> bool: + """Whether the wrapper is retired or its parent connection is closed. + + This does not indicate completion of native cleanup or an in-flight + close operation. A rejected close leaves this value unchanged. + """ + return self._closed or (self._connection is not None and self._connection.closed) + @property def timeout(self) -> int: with translate_py_core_exceptions(): @@ -210,10 +231,17 @@ def timeout(self) -> int: @property def description(self) -> Any: + """Seven-item column descriptors, or None when no row result is published.""" return self._description @property def rowcount(self) -> int: + """Fetched rows for the current SELECT, or the native affected-row count. + + SELECT counts start at -1 and accumulate as rows are fetched; fetching + an empty result sets the count to zero. Unlike the synchronous cursor, + nextset() does not retain fetched counts from the previous result. + """ if self._fetch_rowcount is not None: return self._fetch_rowcount with translate_py_core_exceptions(): diff --git a/mssql_python/async_query/async_execute.py b/mssql_python/async_query/async_execute.py index 6f9bb7b05..8445f8263 100644 --- a/mssql_python/async_query/async_execute.py +++ b/mssql_python/async_query/async_execute.py @@ -1,7 +1,7 @@ """Asynchronous statement execution through mssql-py-core.""" import asyncio -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from typing import TYPE_CHECKING, Any from ..exceptions import OperationalError, ProgrammingError @@ -85,29 +85,42 @@ async def execute( async def executemany( cursor: "_AsyncCursor", operation: str, - seq_of_parameters: Sequence[Sequence[Any]] | Sequence[Mapping[str, Any]], + seq_of_parameters: Iterable[Sequence[Any] | Mapping[str, Any]], *, use_prepare: bool = True, ) -> None: - """Execute a statement for every parameter row using the py-core async cursor.""" + """Execute parameter rows from a synchronous iterable through py-core. + + Py-core consumes and validates the iterable before dispatch; this is not + streaming execution. Asynchronous iterables are not supported. + """ cursor._check_closed() # pyright: ignore[reportPrivateUsage] - batch_count = len(seq_of_parameters) + iteration_failed = False + + def parameter_rows(): + nonlocal iteration_failed + try: + yield from seq_of_parameters + except BaseException: + iteration_failed = True + raise + logger.debug( - "AsyncCursor.executemany: starting; batch_count=%d; use_prepare=%s", - batch_count, + "AsyncCursor.executemany: starting; use_prepare=%s", use_prepare, ) with translate_py_core_exceptions(): executemany_awaitable = _get_py_core_async_cursor(cursor).executemany( operation, - seq_of_parameters, + parameter_rows(), use_prepare=use_prepare, ) try: with translate_py_core_exceptions(): await executemany_awaitable except (Exception, asyncio.CancelledError) as error: - _reconcile_failed_execution(cursor, error) + if not iteration_failed: + _reconcile_failed_execution(cursor, error) raise cursor._reset_fetch_tracking() # pyright: ignore[reportPrivateUsage] cursor._clear_result_metadata() # pyright: ignore[reportPrivateUsage] diff --git a/tests/AsyncTest/test_001_async_query_native.py b/tests/AsyncTest/test_001_async_query_native.py index 9685bbecc..65229246e 100644 --- a/tests/AsyncTest/test_001_async_query_native.py +++ b/tests/AsyncTest/test_001_async_query_native.py @@ -46,3 +46,53 @@ def test_load_py_core_reports_missing_async_types(monkeypatch): with pytest.raises(ImportError, match="PyAsyncConnection, PyAsyncCursor"): native.load_py_core() + + +@pytest.mark.parametrize( + "name", ("SQL_MONEY", "SQL_SMALLMONEY", "SQL_XML", "SQL_JSON", "SQL_VECTOR") +) +def test_async_type_hints_match_native_exports(name): + py_core = native.load_py_core() + assert getattr(async_query, name) == getattr(py_core, name) + assert isinstance(getattr(async_query, name), int) + assert name in async_query.__all__ + assert name in dir(async_query) + + +def test_async_tvp_is_internal_native_constructor(): + from mssql_python.async_query import _TableValuedParameter + + assert _TableValuedParameter is native.load_py_core().TableValuedParameter + assert "_TableValuedParameter" not in async_query.__all__ + assert "_TableValuedParameter" in dir(async_query) + assert not hasattr(async_query, "TableValuedParameter") + value = _TableValuedParameter("TestType", [(4, 0, 0)], [(1,)], schema="dbo") + assert (value.type_name, value.schema) == ("TestType", "dbo") + assert (value.column_count, value.row_count, value.is_null) == (1, 1, False) + assert _TableValuedParameter("dbo.TestType").is_null is True + with pytest.raises(ValueError, match="requires column definitions"): + _TableValuedParameter("dbo.TestType", rows=[(1,)]) + + +def test_async_native_exports_load_only_when_requested(monkeypatch): + calls = [] + py_core = ModuleType("mssql_py_core") + setattr(py_core, "SQL_JSON", 244) + + def load(): + calls.append(True) + return py_core + + monkeypatch.setattr(async_query, "load_py_core", load) + assert "SQL_JSON" in dir(async_query) + assert not hasattr(async_query, "unknown_export") + assert calls == [] + assert async_query.SQL_JSON == 244 + assert calls == [True] + + +@pytest.mark.parametrize("name", ("_TableValuedParameter", "SQL_VECTOR")) +def test_async_native_exports_report_missing_feature(monkeypatch, name): + monkeypatch.setattr(async_query, "load_py_core", lambda: ModuleType("mssql_py_core")) + with pytest.raises(ImportError, match="installed mssql-python-rs dependency"): + getattr(async_query, name) diff --git a/tests/AsyncTest/test_004_async_logging.py b/tests/AsyncTest/test_004_async_logging.py index 7d8e48ce4..183836c61 100644 --- a/tests/AsyncTest/test_004_async_logging.py +++ b/tests/AsyncTest/test_004_async_logging.py @@ -94,7 +94,7 @@ async def test_default_logger_combines_python_and_py_core_operation_logs( "AsyncCursor.fetchmany: starting; requested_size=1", "AsyncCursor.fetchmany: completed; row_count=1; rowcount=2", "AsyncCursor.fetchall: completed; row_count=1; rowcount=1", - "AsyncCursor.executemany: starting; batch_count=2", + "AsyncCursor.executemany: starting; use_prepare=True", "AsyncCursor.executemany: completed; rowcount=-1", ) for expected in expected_python_messages: diff --git a/tests/AsyncTest/test_005_async_cursor.py b/tests/AsyncTest/test_005_async_cursor.py index ad610277e..551545c72 100644 --- a/tests/AsyncTest/test_005_async_cursor.py +++ b/tests/AsyncTest/test_005_async_cursor.py @@ -10,6 +10,9 @@ async def test_properties_and_setinputsizes_use_py_core_async_cursor(async_connection): cursor = async_connection.cursor() try: + assert cursor.closed is False + with pytest.raises(AttributeError): + cursor.closed = True assert cursor.timeout == async_connection.timeout assert cursor.description is None assert cursor.rowcount == -1 @@ -33,7 +36,67 @@ async def test_close_is_idempotent(async_connection): cursor = async_connection.cursor() assert await cursor.close() is None + assert cursor.closed is True assert await cursor.close() is None + assert cursor.closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_context_manager", (False, True)) +async def test_cursor_closed_reflects_parent_connection(async_connection, use_context_manager): + cursor = async_connection.cursor() + assert cursor.closed is False + + if use_context_manager: + async with async_connection: + assert cursor.closed is False + else: + await async_connection.close() + + assert cursor.closed is True + assert await cursor.close() is None + assert cursor.closed is True + + +@pytest.mark.asyncio +async def test_nextset_tracks_rowcount_and_description_per_result(async_connection): + cursor = async_connection.cursor() + try: + await cursor.execute( + "SET NOCOUNT OFF; DECLARE @values TABLE (value INT); " + "INSERT INTO @values VALUES (1), (2); " + "SELECT value FROM @values ORDER BY value; " + "UPDATE @values SET value = value + 1; " + "SELECT value AS empty_value FROM @values WHERE 1 = 0;" + ) + assert cursor.description is None + assert cursor.rowcount == 2 + + assert await cursor.nextset() is True + assert cursor.description is not None + assert cursor.description[0][0] == "value" + assert cursor.rowcount == -1 + assert await cursor.fetchone() == [1] + assert cursor.rowcount == 1 + + assert await cursor.nextset() is True + assert cursor.description is None + assert cursor.rowcount == 2 + + assert await cursor.nextset() is True + assert cursor.description is not None + assert cursor.description[0][0] == "empty_value" + assert cursor.rowcount == -1 + assert await cursor.fetchall() == [] + assert cursor.rowcount == 0 + + assert await cursor.nextset() is False + assert cursor.description is None + assert cursor.rowcount == -1 + assert await cursor.nextset() is False + assert cursor.closed is False + finally: + await cursor.close() @pytest.mark.asyncio @@ -45,6 +108,8 @@ async def test_close_clears_cached_fetch_rowcount(async_connection): await cursor.close() + assert cursor.closed is True + assert cursor.description is None assert cursor.rowcount == -1 @@ -93,6 +158,7 @@ async def close(self): await close_task assert caught.value is close_error + assert cursor.closed is True assert cursor.description is None assert cursor.rowcount == native_cursor.rowcount assert getattr(cursor, "_result_generation") > previous_generation @@ -137,6 +203,7 @@ def close(self): with pytest.raises(public_error, match=message): await cursor.close() + assert cursor.closed is False assert cursor.description is previous_description assert getattr(cursor, "_result_generation") == previous_generation assert cursor.rowcount == 1 diff --git a/tests/AsyncTest/test_006_async_execute.py b/tests/AsyncTest/test_006_async_execute.py index f028a9120..07d03be51 100644 --- a/tests/AsyncTest/test_006_async_execute.py +++ b/tests/AsyncTest/test_006_async_execute.py @@ -9,6 +9,7 @@ mssql_py_core = pytest.importorskip("mssql_py_core", exc_type=ImportError) import mssql_python +import mssql_python.async_query as async_query from mssql_python.async_query import _AsyncConnection # pyright: ignore[reportPrivateUsage] from mssql_python.async_query import _AsyncCursor # pyright: ignore[reportPrivateUsage] from mssql_python.async_query import async_execute @@ -397,9 +398,10 @@ async def test_executemany_matches_sync_contract( @pytest.mark.asyncio -async def test_executemany_rejects_non_sequence_like_sync(async_cursor): +@pytest.mark.parametrize("parameters", (None, 42)) +async def test_executemany_rejects_non_iterable(async_cursor, parameters): with pytest.raises(TypeError): - await async_cursor.executemany("SELECT CAST(? AS INT)", iter([(1,), (2,)])) + await async_cursor.executemany("SELECT CAST(? AS INT)", parameters) @pytest.mark.asyncio @@ -513,6 +515,62 @@ async def test_executemany_empty_sequence_sets_rowcount_zero(async_cursor): assert async_cursor.rowcount == 0 +@pytest.mark.asyncio +@pytest.mark.parametrize("mapped", (False, True)) +@pytest.mark.parametrize("use_prepare", (False, True)) +async def test_executemany_consumes_iterable_once(async_cursor, mapped, use_prepare): + consumed = [] + + class ParameterRows: + def __iter__(self): + assert not consumed + for value in (1, 2): + consumed.append(value) + yield {"value": value} if mapped else (value,) + + def __len__(self): + raise AssertionError("Parameter iterable must not require a length") + + operation = "SELECT CAST(%(value)s AS INT)" if mapped else "SELECT CAST(? AS INT)" + assert ( + await async_cursor.executemany(operation, ParameterRows(), use_prepare=use_prepare) is None + ) + assert consumed == [1, 2] + assert await async_cursor.fetchall() == [[1]] + assert await async_cursor.nextset() is True + assert await async_cursor.fetchall() == [[2]] + assert await async_cursor.nextset() is False + + +@pytest.mark.asyncio +async def test_executemany_empty_generator(async_cursor): + assert await async_cursor.executemany("SELECT ?", (row for row in ())) is None + assert async_cursor.rowcount == 0 + assert async_cursor.description is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "failure", (ValueError("iteration failed"), RuntimeError("iteration failed")) +) +async def test_executemany_iterator_failure_preserves_result(async_cursor, failure): + await async_cursor.execute("SELECT 1 AS value UNION ALL SELECT 2 ORDER BY value") + assert await async_cursor.fetchone() == [1] + description = async_cursor.description + + def rows(): + yield (3,) + raise failure + + with pytest.raises(type(failure)) as caught: + await async_cursor.executemany("SELECT ?", rows()) + assert caught.value is failure + assert async_cursor.description is description + assert async_cursor.rowcount == 1 + assert await async_cursor.fetchone() == [2] + assert async_cursor.rowcount == 2 + + @pytest.mark.asyncio async def test_executemany_handles_sync_edge_value_batches(async_cursor): table_name = f"async_many_values_{uuid4().hex}" @@ -562,6 +620,108 @@ async def test_executemany_handles_sync_edge_value_batches(async_cursor): await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("batch", (False, True)) +@pytest.mark.parametrize("rows", ([(1, "first"), (2, None)], [], None)) +async def test_async_tvp_execute_and_executemany( + async_cursor, async_connection, use_prepare, batch, rows +): + type_name = f"dbo.AsyncTvp_{uuid4().hex}" + await async_cursor.execute(f"CREATE TYPE {type_name} AS TABLE (id INT, value NVARCHAR(50))") + try: + await async_connection.commit() + value = async_query._TableValuedParameter( + type_name, + [(4, 0, 0), (-9, 50, 0)] if rows is not None else None, + rows, + ) + operation = "SELECT id, value FROM ? ORDER BY id" + if batch: + assert ( + await async_cursor.executemany( + operation, ((value,) for _ in range(2)), use_prepare=use_prepare + ) + is None + ) + else: + assert ( + await async_cursor.execute(operation, value, use_prepare=use_prepare) + is async_cursor + ) + assert [tuple(row) for row in await async_cursor.fetchall()] == (rows or []) + if batch: + assert await async_cursor.nextset() is True + assert [tuple(row) for row in await async_cursor.fetchall()] == (rows or []) + assert await async_cursor.nextset() is False + finally: + await async_cursor.execute(f"DROP TYPE IF EXISTS {type_name}") + await async_connection.commit() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize( + ("token", "sql_type", "value", "operation", "expected"), + ( + ( + "SQL_MONEY", + "money", + Decimal("123.45"), + "SELECT SQL_VARIANT_PROPERTY(CAST(? AS sql_variant), 'BaseType')", + "money", + ), + ( + "SQL_SMALLMONEY", + "smallmoney", + Decimal("123.45"), + "SELECT SQL_VARIANT_PROPERTY(CAST(? AS sql_variant), 'BaseType')", + "smallmoney", + ), + ( + "SQL_XML", + "xml", + "", + "DECLARE @value xml = ?; SELECT @value.exist('/root')", + 1, + ), + ( + "SQL_JSON", + "json", + {"answer": 42}, + "SELECT JSON_VALUE(%(payload)s, '$.answer')", + "42", + ), + ( + "SQL_VECTOR", + "vector", + [1.0, 2.0, 3.0], + "SELECT VECTOR_DISTANCE('euclidean', ?, CAST('[1,2,3]' AS VECTOR(3)))", + 0.0, + ), + ), +) +async def test_async_exported_type_hints( + async_cursor, use_prepare, token, sql_type, value, operation, expected +): + if sql_type in ("json", "vector"): + await async_cursor.execute(f"SELECT TYPE_ID('{sql_type}')") + row = await async_cursor.fetchone() + assert row is not None + if row[0] is None: + pytest.skip(f"SQL Server does not expose the {sql_type} type") + hint = getattr(async_query, token) + async_cursor.setinputsizes([(hint, 3, 0)] if sql_type == "vector" else [hint]) + try: + parameters = {"payload": value} if sql_type == "json" else (value,) + await async_cursor.execute(operation, parameters, use_prepare=use_prepare) + row = await async_cursor.fetchone() + assert row is not None + assert row[0] == expected + finally: + async_cursor.setinputsizes(None) + + @pytest.mark.asyncio async def test_executemany_handles_multiple_all_null_columns(async_cursor): table_name = f"async_many_nulls_{uuid4().hex}" From d3fd82d5907d4a07979a3b6ec1ecbb271427b1b4 Mon Sep 17 00:00:00 2001 From: Subrata <141804867+subrata-ms@users.noreply.github.com> Date: Thu, 24 Sep 2026 15:44:23 +0530 Subject: [PATCH 2/8] Change loading of py_core to use pytest importorskip Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- tests/AsyncTest/test_001_async_query_native.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/AsyncTest/test_001_async_query_native.py b/tests/AsyncTest/test_001_async_query_native.py index 65229246e..15d202dfe 100644 --- a/tests/AsyncTest/test_001_async_query_native.py +++ b/tests/AsyncTest/test_001_async_query_native.py @@ -52,7 +52,7 @@ def test_load_py_core_reports_missing_async_types(monkeypatch): "name", ("SQL_MONEY", "SQL_SMALLMONEY", "SQL_XML", "SQL_JSON", "SQL_VECTOR") ) def test_async_type_hints_match_native_exports(name): - py_core = native.load_py_core() +py_core = pytest.importorskip("mssql_py_core", exc_type=ImportError) assert getattr(async_query, name) == getattr(py_core, name) assert isinstance(getattr(async_query, name), int) assert name in async_query.__all__ From 23dd23c199b3d458f64dc284856f21774437338f Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Thu, 24 Sep 2026 10:59:55 +0000 Subject: [PATCH 3/8] enhancing String test cases and coverage --- tests/AsyncTest/test_003_async_exceptions.py | 57 ++++ tests/AsyncTest/test_006_async_execute.py | 291 ++++++++++++++++++- tests/AsyncTest/test_007_async_fetch.py | 75 +++++ 3 files changed, 421 insertions(+), 2 deletions(-) diff --git a/tests/AsyncTest/test_003_async_exceptions.py b/tests/AsyncTest/test_003_async_exceptions.py index 0b36a593c..e65f0a15b 100644 --- a/tests/AsyncTest/test_003_async_exceptions.py +++ b/tests/AsyncTest/test_003_async_exceptions.py @@ -304,6 +304,63 @@ async def test_executemany_integrity_error_reports_row_and_preserves_partial_pro await connection.close() +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("batch", (False, True)) +@pytest.mark.parametrize("supplementary", (False, True)) +async def test_string_truncation_diagnostics_partial_progress_and_recovery( + async_cursor, use_prepare, batch, supplementary +): + value = "\U0001f600" * 3 if supplementary else "abcdef" + await async_cursor.execute("DBCC TRACEON(460) WITH NO_INFOMSGS", use_prepare=False) + await async_cursor.execute( + "CREATE TABLE #async_truncation (id INT, value NVARCHAR(5))", use_prepare=False + ) + try: + with pytest.raises( + (public_exceptions.DataError, public_exceptions.OperationalError) + ) as raised: + if batch: + await async_cursor.executemany( + "INSERT INTO #async_truncation VALUES (?, ?)", + [(1, "ok"), (2, value), (3, "later")], + use_prepare=use_prepare, + ) + else: + await async_cursor.execute( + "INSERT INTO #async_truncation VALUES (?, ?)", + 2, + value, + use_prepare=use_prepare, + ) + error = cast(Any, raised.value) + native_decode_failure = ( + supplementary + and isinstance(error, public_exceptions.OperationalError) + and "invalid utf-16: lone surrogate found" in str(error) + and not getattr(error, "sql_errors", None) + ) + if not native_decode_failure: + assert isinstance(error, public_exceptions.DataError) + diagnostics = getattr(error, "sql_errors", []) + assert diagnostics + assert any(item["number"] in (8152, 2628) for item in diagnostics) + assert error.__cause__ is not None + await async_cursor.execute("SELECT id, value FROM #async_truncation ORDER BY id") + assert [tuple(row) for row in await async_cursor.fetchall()] == ( + [(1, "ok")] if batch else [] + ) + await async_cursor.execute("INSERT INTO #async_truncation VALUES (4, N'new')") + assert async_cursor.rowcount == 1 + await async_cursor.execute("SELECT value FROM #async_truncation WHERE id = 4") + assert tuple(await async_cursor.fetchone()) == ("new",) + if native_decode_failure: + pytest.xfail("py-core truncation diagnostic decoder rejects split UTF-16 surrogate") + finally: + await async_cursor.execute("DROP TABLE IF EXISTS #async_truncation", use_prepare=False) + await async_cursor.execute("DBCC TRACEOFF(460) WITH NO_INFOMSGS", use_prepare=False) + + @pytest.mark.asyncio async def test_timeout_is_operational_error_and_cursor_is_reusable(async_connection_string): connection = await _AsyncConnection.connect(async_connection_string) diff --git a/tests/AsyncTest/test_006_async_execute.py b/tests/AsyncTest/test_006_async_execute.py index 07d03be51..efd843a99 100644 --- a/tests/AsyncTest/test_006_async_execute.py +++ b/tests/AsyncTest/test_006_async_execute.py @@ -13,7 +13,7 @@ from mssql_python.async_query import _AsyncConnection # pyright: ignore[reportPrivateUsage] from mssql_python.async_query import _AsyncCursor # pyright: ignore[reportPrivateUsage] from mssql_python.async_query import async_execute -from mssql_python import DatabaseError, OperationalError, ProgrammingError +from mssql_python import DataError, DatabaseError, OperationalError, ProgrammingError from mssql_python.row import Row @@ -392,6 +392,8 @@ async def test_executemany_matches_sync_contract( ) assert result is None assert cursor.rowcount == 2 + await cursor.execute(f"SELECT id, value FROM {table_name} ORDER BY id") + assert [tuple(row) for row in await cursor.fetchall()] == [(1, "one"), (2, "two")] finally: await cursor.execute(f"DROP TABLE IF EXISTS {table_name}") await cursor.close() @@ -467,6 +469,225 @@ async def test_execute_reset_cursor_false_supports_repeated_execution(async_curs assert result is async_cursor +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("reset_cursor", (False, True)) +async def test_string_execution_reuse_preserves_values_and_lengths( + async_cursor, use_prepare, reset_cursor +): + values = ["", None, "a\x00b", "caf\u00e9", "\u4e2d\u6587", "e\u0301"] + values.extend("x" * size for size in (3999, 4000, 4001, 7999, 8000, 8001)) + values.extend( + ( + "\U0001f600" * 1999 + "x", + "\U0001f600" * 2000, + "\U0001f600" * 2000 + "x", + "short again ", + None, + "", + ) + ) + for value in values: + await async_cursor.execute( + "SELECT CAST(? AS NVARCHAR(MAX)) AS value, " + "DATALENGTH(CAST(? AS NVARCHAR(MAX))) AS byte_length", + value, + value, + use_prepare=use_prepare, + reset_cursor=reset_cursor, + ) + row = await async_cursor.fetchone() + assert row is not None + assert row[0] == value + assert row[1] == (None if value is None else len(value.encode("utf-16-le"))) + assert async_cursor.description is not None + assert async_cursor.description[0][1] is str + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("mapped", (False, True)) +async def test_string_batches_match_individual_execution(async_cursor, use_prepare, mapped): + values = [ + None, + "", + " ", + "\t\n", + "a\x00b", + "caf\u00e9", + "\u4e2d\u6587", + "\U0001f600", + "\u03a9" * 4001, + "x" * 8001, + "end ", + ] + operation = ( + "SELECT CAST(%(value)s AS NVARCHAR(MAX)) AS value" + if mapped + else "SELECT CAST(? AS NVARCHAR(MAX)) AS value" + ) + individual = [] + for value in values: + parameters = {"value": value} if mapped else (value,) + await async_cursor.execute(operation, parameters, use_prepare=use_prepare) + individual.append(tuple(await async_cursor.fetchone())) + + rows = ({"value": value} if mapped else (value,) for value in values) + assert await async_cursor.executemany(operation, rows, use_prepare=use_prepare) is None + for index, value in enumerate(values): + assert async_cursor.rowcount == -1 + assert [tuple(row) for row in await async_cursor.fetchall()] == [(value,)] + assert individual[index] == (value,) + assert async_cursor.rowcount == 1 + assert await async_cursor.nextset() is (index + 1 < len(values)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize( + ("hint", "expected_type", "values"), + ( + ((-9, 100, 0), "nvarchar", (None, "", "\u4e2d\u6587\U0001f600 ")), + ((12, 100, 0), "varchar", (None, "", "caf\u00e9 ")), + ((-8, 100, 0), "nchar", (None, "", "\u4e2d\u6587\U0001f600 ")), + ((1, 100, 0), "char", (None, "", "caf\u00e9 ")), + ), +) +async def test_string_setinputsizes_type_null_and_consumption( + async_cursor, use_prepare, hint, expected_type, values +): + for value in values: + async_cursor.setinputsizes([hint, hint]) + await async_cursor.execute( + "SELECT ?, CONVERT(VARCHAR(30), SQL_VARIANT_PROPERTY(?, 'BaseType'))", + value, + value, + use_prepare=use_prepare, + ) + row = await async_cursor.fetchone() + expected_value = value + if value is not None and expected_type in ("char", "nchar"): + units = len(value.encode("utf-16-le")) // 2 if expected_type == "nchar" else len(value) + expected_value = value + " " * (100 - units) + assert tuple(row) == (expected_value, None if value is None else expected_type) + assert async_cursor.description[0][1] is str + await async_cursor.execute( + "SELECT CONVERT(VARCHAR(30), SQL_VARIANT_PROPERTY(?, 'BaseType'))", + 42, + use_prepare=use_prepare, + ) + assert (await async_cursor.fetchone())[0] == "tinyint" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +async def test_string_setinputsizes_preserved_after_local_rejection(async_cursor, use_prepare): + async_cursor.setinputsizes([(12, 100, 0)]) + with pytest.raises(TypeError, match="Failed to convert parameter to string"): + await async_cursor.execute("SELECT ?", object(), use_prepare=use_prepare) + await async_cursor.execute( + "SELECT CONVERT(VARCHAR(30), SQL_VARIANT_PROPERTY(?, 'BaseType'))", + "retained", + use_prepare=use_prepare, + ) + assert (await async_cursor.fetchone())[0] == "varchar" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("batch", (False, True)) +@pytest.mark.parametrize("value_kind", ("null", "empty", "large")) +@pytest.mark.parametrize( + ("hint", "sql_type", "encoding"), + ( + (-1, "VARCHAR(MAX)", "ascii"), + (-10, "NVARCHAR(MAX)", "utf-16-le"), + ), +) +async def test_string_long_hints_preserve_null_and_payload( + async_cursor, use_prepare, batch, value_kind, hint, sql_type, encoding +): + value = None if value_kind == "null" else "" + if value_kind == "large": + value = "x" * 8001 + " " if encoding == "ascii" else "\U0001f600" * 2001 + " " + await async_cursor.execute( + f"CREATE TABLE #async_long_hint (id INT, value {sql_type})", use_prepare=False + ) + try: + async_cursor.setinputsizes([4, hint]) + try: + if batch: + await async_cursor.executemany( + "INSERT INTO #async_long_hint VALUES (?, ?)", + [(1, value)], + use_prepare=use_prepare, + ) + else: + await async_cursor.execute( + "INSERT INTO #async_long_hint VALUES (?, ?)", + 1, + value, + use_prepare=use_prepare, + ) + except DatabaseError as error: + if not ( + hint == -10 + and isinstance(error.__cause__, mssql_py_core.DatabaseError) + and [item["number"] for item in getattr(error, "sql_errors", [])] == [4002] + ): + raise + async_cursor.setinputsizes(None) + await async_cursor.execute("SELECT 1") + assert tuple(await async_cursor.fetchone()) == (1,) + pytest.xfail("py-core SQL_WLONGVARCHAR emits invalid TDS (SQL 4002)") + assert async_cursor.rowcount == 1 + await async_cursor.execute( + "SELECT value, DATALENGTH(value) FROM #async_long_hint ORDER BY id" + ) + assert [tuple(row) for row in await async_cursor.fetchall()] == [ + (value, None if value is None else len(value.encode(encoding))), + ] + finally: + async_cursor.setinputsizes(None) + await async_cursor.execute("DROP TABLE IF EXISTS #async_long_hint", use_prepare=False) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize( + ("sql_type", "capacity", "encoding"), + (("VARCHAR(8000)", 8000, "ascii"), ("NVARCHAR(4000)", 4000, "utf-16-le")), +) +async def test_string_column_capacity_and_overflow( + async_cursor, use_prepare, sql_type, capacity, encoding +): + values = ["x" * (capacity - 1), "x" * capacity] + if encoding == "utf-16-le": + values.append("\U0001f600" * (capacity // 2)) + await async_cursor.execute( + f"CREATE TABLE #async_capacity (id INT, value {sql_type})", use_prepare=False + ) + try: + await async_cursor.executemany( + "INSERT INTO #async_capacity VALUES (?, ?)", enumerate(values), use_prepare=use_prepare + ) + with pytest.raises(DataError): + await async_cursor.execute( + "INSERT INTO #async_capacity VALUES (?, ?)", + len(values), + "x" * (capacity + 1), + use_prepare=use_prepare, + ) + await async_cursor.execute( + "SELECT value, DATALENGTH(value) FROM #async_capacity ORDER BY id" + ) + assert [tuple(row) for row in await async_cursor.fetchall()] == [ + (value, len(value.encode(encoding))) for value in values + ] + finally: + await async_cursor.execute("DROP TABLE IF EXISTS #async_capacity", use_prepare=False) + + @pytest.mark.asyncio async def test_execute_updates_rowcount_and_description(async_cursor): table_name = f"async_execute_state_{uuid4().hex}" @@ -623,7 +844,15 @@ async def test_executemany_handles_sync_edge_value_batches(async_cursor): @pytest.mark.asyncio @pytest.mark.parametrize("use_prepare", (False, True)) @pytest.mark.parametrize("batch", (False, True)) -@pytest.mark.parametrize("rows", ([(1, "first"), (2, None)], [], None)) +@pytest.mark.parametrize( + "rows", + ( + [(1, "first"), (2, None)], + [], + None, + [(1, ""), (2, "\u4e2d\u6587 "), (3, "\U0001f600" * 25), (4, "a\x00b")], + ), +) async def test_async_tvp_execute_and_executemany( async_cursor, async_connection, use_prepare, batch, rows ): @@ -741,3 +970,61 @@ async def test_executemany_handles_multiple_all_null_columns(async_cursor): ) finally: await async_cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("batch", (False, True)) +@pytest.mark.parametrize("named", (False, True)) +@pytest.mark.parametrize( + ("sql_type", "unicode_values"), + ( + ("VARCHAR(128)", False), + ("NVARCHAR(128)", True), + ("VARCHAR(MAX)", False), + ("NVARCHAR(MAX)", True), + ("TEXT", False), + ("NTEXT", True), + ), +) +async def test_string_roundtrip_matrix( + async_cursor, use_prepare, batch, named, sql_type, unicode_values +): + class StringValue(str): + pass + + values = [None, "", " ", "\t\r\n", "a\x00b", "trailing ", StringValue("subclass")] + if unicode_values: + values.extend(["caf\u00e9", "\u4e2d\u6587", "\U0001f600\U0001f680", "e\u0301"]) + rows = list(enumerate(values)) + parameters = ( + [{"id": identifier, "value": value} for identifier, value in rows] if named else rows + ) + markers = "%(id)s, %(value)s" if named else "?, ?" + await async_cursor.execute( + f"CREATE TABLE #async_strings (id INT, value {sql_type})", use_prepare=False + ) + try: + operation = f"INSERT INTO #async_strings VALUES ({markers})" + if batch: + assert ( + await async_cursor.executemany(operation, iter(parameters), use_prepare=use_prepare) + is None + ) + assert async_cursor.rowcount == len(rows) + else: + for parameter in parameters: + await async_cursor.execute(operation, parameter, use_prepare=use_prepare) + assert async_cursor.rowcount == 1 + await async_cursor.execute( + "SELECT id, value, DATALENGTH(value) FROM #async_strings ORDER BY id" + ) + fetched = await async_cursor.fetchall() + assert [tuple(row)[:2] for row in fetched] == rows + encoding = "utf-16-le" if unicode_values else "ascii" + assert [row[2] for row in fetched] == [ + None if value is None else len(value.encode(encoding)) for value in values + ] + assert all(row[1] is None or isinstance(row[1], str) for row in fetched) + finally: + await async_cursor.execute("DROP TABLE IF EXISTS #async_strings", use_prepare=False) diff --git a/tests/AsyncTest/test_007_async_fetch.py b/tests/AsyncTest/test_007_async_fetch.py index 4d7d62966..d04c7ec84 100644 --- a/tests/AsyncTest/test_007_async_fetch.py +++ b/tests/AsyncTest/test_007_async_fetch.py @@ -917,6 +917,81 @@ async def test_fetchmany_handles_mixed_large_lob_sizes(async_cursor): assert await async_cursor.fetchmany(3) == [] +@pytest.mark.asyncio +@pytest.mark.parametrize("fetch_method", ("fetchone", "fetchmany", "fetchall")) +@pytest.mark.parametrize("use_prepare", (False, True)) +async def test_string_lob_fetch_and_nextset(async_cursor, fetch_method, use_prepare): + large = "\u6771\u4eac\U0001f600" * 12500 + " " + await async_cursor.execute( + "SELECT CAST('x' AS CHAR(4)) AS fixed_ascii, CAST(N'\u6771' AS NCHAR(3)) AS fixed_unicode, " + "CAST(? AS NVARCHAR(MAX)) AS large_value; " + "SELECT CAST(NULL AS NVARCHAR(MAX)) AS empty_value UNION ALL SELECT N''", + large, + use_prepare=use_prepare, + ) + result = await getattr(async_cursor, fetch_method)() + row = result if fetch_method == "fetchone" else result[0] + assert tuple(row) == ("x ", "\u6771 ", large) + assert row.large_value == large + assert async_cursor.rowcount == 1 + assert await async_cursor.nextset() is True + assert async_cursor.description is not None + assert async_cursor.description[0][0] == "empty_value" + assert [tuple(row) for row in await async_cursor.fetchall()] == [(None,), ("",)] + assert await async_cursor.nextset() is False + assert row.large_value == large + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fetch_method", ("fetchone", "fetchmany", "fetchall")) +@pytest.mark.parametrize( + ("collation", "capacity", "value", "encoding"), + ( + ("SQL_Latin1_General_CP1_CI_AS", 10, "caf\u00e9 Ren\u00e9!", "cp1252"), + ("Latin1_General_100_CI_AS_SC_UTF8", 12, "\u4e2d\u6587\U0001f600 ", "utf-8"), + ), +) +async def test_string_varchar_collation_exact_capacity( + async_cursor, fetch_method, collation, capacity, value, encoding +): + assert len(value.encode(encoding)) == capacity + await async_cursor.execute( + f"CREATE TABLE #async_collation (value VARCHAR({capacity}) COLLATE {collation})", + use_prepare=False, + ) + try: + await async_cursor.execute("INSERT INTO #async_collation VALUES (?)", value) + await async_cursor.execute("SELECT value, DATALENGTH(value) FROM #async_collation") + result = await getattr(async_cursor, fetch_method)() + row = result if fetch_method == "fetchone" else result[0] + assert tuple(row) == (value, capacity) + assert isinstance(row[0], str) + finally: + await async_cursor.execute("DROP TABLE IF EXISTS #async_collation", use_prepare=False) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +async def test_string_nextset_discards_unread_lob_rows(async_cursor, use_prepare): + large = "\u4e2d\U0001f600" * 20000 + await async_cursor.execute( + "SELECT CAST(? AS NVARCHAR(MAX)) AS large_value FROM (VALUES (1), (2), (3)) AS ids(id); " + "SELECT N'after' AS small_value", + large, + use_prepare=use_prepare, + ) + first = await async_cursor.fetchone() + assert first.large_value == large + assert async_cursor.rowcount == 1 + assert await async_cursor.nextset() is True + assert async_cursor.rowcount == -1 + assert async_cursor.description[0][0] == "small_value" + assert [tuple(row) for row in await async_cursor.fetchall()] == [("after",)] + assert async_cursor.rowcount == 1 + assert await async_cursor.nextset() is False + assert first.large_value == large + + @pytest.mark.asyncio async def test_fetch_roundtrips_representative_sync_result_types(async_cursor): expected = ( From 94ba027b201a34b3f5efc872370356d1c903777f Mon Sep 17 00:00:00 2001 From: Subrata <141804867+subrata-ms@users.noreply.github.com> Date: Thu, 24 Sep 2026 16:35:59 +0530 Subject: [PATCH 4/8] Fix indentation in async type hints test Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- tests/AsyncTest/test_001_async_query_native.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/AsyncTest/test_001_async_query_native.py b/tests/AsyncTest/test_001_async_query_native.py index 15d202dfe..f8de8673b 100644 --- a/tests/AsyncTest/test_001_async_query_native.py +++ b/tests/AsyncTest/test_001_async_query_native.py @@ -52,7 +52,7 @@ def test_load_py_core_reports_missing_async_types(monkeypatch): "name", ("SQL_MONEY", "SQL_SMALLMONEY", "SQL_XML", "SQL_JSON", "SQL_VECTOR") ) def test_async_type_hints_match_native_exports(name): -py_core = pytest.importorskip("mssql_py_core", exc_type=ImportError) + py_core = pytest.importorskip("mssql_py_core", exc_type=ImportError) assert getattr(async_query, name) == getattr(py_core, name) assert isinstance(getattr(async_query, name), int) assert name in async_query.__all__ From 5dd4a19cab5e7da9f7ada28d3bf6f6c32067eb5e Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Thu, 24 Sep 2026 11:34:50 +0000 Subject: [PATCH 5/8] CHORE: Expand async iterator failure integration coverage --- tests/AsyncTest/test_006_async_execute.py | 45 ++++++++++++++++------- 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/tests/AsyncTest/test_006_async_execute.py b/tests/AsyncTest/test_006_async_execute.py index efd843a99..4cebd2f3b 100644 --- a/tests/AsyncTest/test_006_async_execute.py +++ b/tests/AsyncTest/test_006_async_execute.py @@ -771,25 +771,44 @@ async def test_executemany_empty_generator(async_cursor): @pytest.mark.asyncio +@pytest.mark.parametrize("use_prepare", (False, True)) +@pytest.mark.parametrize("yield_before_failure", (False, True)) @pytest.mark.parametrize( "failure", (ValueError("iteration failed"), RuntimeError("iteration failed")) ) -async def test_executemany_iterator_failure_preserves_result(async_cursor, failure): - await async_cursor.execute("SELECT 1 AS value UNION ALL SELECT 2 ORDER BY value") - assert await async_cursor.fetchone() == [1] - description = async_cursor.description - +async def test_executemany_iterator_failure_preserves_result( + async_cursor, failure, yield_before_failure, use_prepare +): def rows(): - yield (3,) + if yield_before_failure: + yield (3,) raise failure - with pytest.raises(type(failure)) as caught: - await async_cursor.executemany("SELECT ?", rows()) - assert caught.value is failure - assert async_cursor.description is description - assert async_cursor.rowcount == 1 - assert await async_cursor.fetchone() == [2] - assert async_cursor.rowcount == 2 + await async_cursor.execute( + "CREATE TABLE #async_iterator_failure (value INT)", use_prepare=False + ) + try: + await async_cursor.execute("SELECT 1 AS value UNION ALL SELECT 2 ORDER BY value") + assert await async_cursor.fetchone() == [1] + description = async_cursor.description + + with pytest.raises(type(failure)) as caught: + await async_cursor.executemany( + "INSERT INTO #async_iterator_failure VALUES (?)", + rows(), + use_prepare=use_prepare, + ) + assert caught.value is failure + assert async_cursor.description is description + assert async_cursor.rowcount == 1 + assert await async_cursor.fetchone() == [2] + assert async_cursor.rowcount == 2 + await async_cursor.execute("SELECT COUNT(*) FROM #async_iterator_failure") + assert await async_cursor.fetchone() == [0] + finally: + await async_cursor.execute( + "DROP TABLE IF EXISTS #async_iterator_failure", use_prepare=False + ) @pytest.mark.asyncio From 59aeacf41c2b0480b24d87746a3d98a22e57b758 Mon Sep 17 00:00:00 2001 From: Subrata <141804867+subrata-ms@users.noreply.github.com> Date: Thu, 24 Sep 2026 17:22:06 +0530 Subject: [PATCH 6/8] Add collation support check in async fetch test Skip test if SQL Server does not support the specified collation. Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- tests/AsyncTest/test_007_async_fetch.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/AsyncTest/test_007_async_fetch.py b/tests/AsyncTest/test_007_async_fetch.py index d04c7ec84..8aa3b3a34 100644 --- a/tests/AsyncTest/test_007_async_fetch.py +++ b/tests/AsyncTest/test_007_async_fetch.py @@ -955,6 +955,12 @@ async def test_string_varchar_collation_exact_capacity( async_cursor, fetch_method, collation, capacity, value, encoding ): assert len(value.encode(encoding)) == capacity + if encoding == "utf-8": + await async_cursor.execute( + "SELECT 1 FROM sys.fn_helpcollations() WHERE name = ?", collation + ) + if await async_cursor.fetchone() is None: + pytest.skip(f"SQL Server does not support the {collation} collation") await async_cursor.execute( f"CREATE TABLE #async_collation (value VARCHAR({capacity}) COLLATE {collation})", use_prepare=False, From 7e0d70cf7c10c0d1347ec169ff7575c0785722e1 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Thu, 24 Sep 2026 13:55:36 +0000 Subject: [PATCH 7/8] FIX: Remove privileged DBCC calls from async truncation test --- tests/AsyncTest/test_003_async_exceptions.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/AsyncTest/test_003_async_exceptions.py b/tests/AsyncTest/test_003_async_exceptions.py index e65f0a15b..e8987d759 100644 --- a/tests/AsyncTest/test_003_async_exceptions.py +++ b/tests/AsyncTest/test_003_async_exceptions.py @@ -312,7 +312,6 @@ async def test_string_truncation_diagnostics_partial_progress_and_recovery( async_cursor, use_prepare, batch, supplementary ): value = "\U0001f600" * 3 if supplementary else "abcdef" - await async_cursor.execute("DBCC TRACEON(460) WITH NO_INFOMSGS", use_prepare=False) await async_cursor.execute( "CREATE TABLE #async_truncation (id INT, value NVARCHAR(5))", use_prepare=False ) @@ -358,7 +357,6 @@ async def test_string_truncation_diagnostics_partial_progress_and_recovery( pytest.xfail("py-core truncation diagnostic decoder rejects split UTF-16 surrogate") finally: await async_cursor.execute("DROP TABLE IF EXISTS #async_truncation", use_prepare=False) - await async_cursor.execute("DBCC TRACEOFF(460) WITH NO_INFOMSGS", use_prepare=False) @pytest.mark.asyncio From 0579d4b55365745d13360c0bd1421679c4bd223f Mon Sep 17 00:00:00 2001 From: Subrata <141804867+subrata-ms@users.noreply.github.com> Date: Thu, 24 Sep 2026 21:24:28 +0530 Subject: [PATCH 8/8] Refactor SQL query to handle empty values Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- tests/AsyncTest/test_007_async_fetch.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/AsyncTest/test_007_async_fetch.py b/tests/AsyncTest/test_007_async_fetch.py index 8aa3b3a34..569854670 100644 --- a/tests/AsyncTest/test_007_async_fetch.py +++ b/tests/AsyncTest/test_007_async_fetch.py @@ -925,7 +925,9 @@ async def test_string_lob_fetch_and_nextset(async_cursor, fetch_method, use_prep await async_cursor.execute( "SELECT CAST('x' AS CHAR(4)) AS fixed_ascii, CAST(N'\u6771' AS NCHAR(3)) AS fixed_unicode, " "CAST(? AS NVARCHAR(MAX)) AS large_value; " - "SELECT CAST(NULL AS NVARCHAR(MAX)) AS empty_value UNION ALL SELECT N''", + "SELECT empty_value FROM " + "(VALUES (0, CAST(NULL AS NVARCHAR(MAX))), (1, N'')) AS values_(ordinal, empty_value) " + "ORDER BY ordinal", large, use_prepare=use_prepare, )