From 793fb4c173fbbbfcfdc89664384a217d92dae7b4 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Tue, 14 Apr 2026 14:02:35 +0530 Subject: [PATCH 1/2] FIX: ODBC Catalog method fetchone() issue --- mssql_python/cursor.py | 3 + tests/test_004_cursor.py | 145 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 148 insertions(+) diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index ba5065d56..4f9ff533b 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -1638,6 +1638,9 @@ def fetchall_with_mapping(): self.fetchmany = fetchmany_with_mapping self.fetchall = fetchall_with_mapping + # Initialize rownumber tracking so fetchone() and iteration work + self._reset_rownumber() + # Return the cursor itself for method chaining return self diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index 6ac157389..83d2ed643 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -15961,3 +15961,148 @@ def reader(reader_id): finally: stop_event.set() mssql_python.native_uuid = original + + +def test_catalog_fetchone_iteration_setup(cursor, db_connection): + """Create test objects for catalog fetchone/iteration testing""" + try: + cursor.execute( + "IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = 'pytest_cat_fetch') " + "EXEC('CREATE SCHEMA pytest_cat_fetch')" + ) + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") + + cursor.execute(""" + CREATE TABLE pytest_cat_fetch.fetch_test ( + id INT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + value DECIMAL(10,2), + ts DATETIME DEFAULT GETDATE() + ) + """) + cursor.execute(""" + CREATE TABLE pytest_cat_fetch.fetch_test_child ( + child_id INT PRIMARY KEY, + parent_id INT NOT NULL, + CONSTRAINT fk_parent FOREIGN KEY (parent_id) + REFERENCES pytest_cat_fetch.fetch_test(id) + ) + """) + db_connection.commit() + except Exception as e: + pytest.fail(f"Catalog fetchone/iteration setup failed: {e}") + + +def test_tables_fetchone(cursor, db_connection): + """Test that fetchone() works on tables() result set (GH-505)""" + cursor.tables(table="fetch_test", schema="pytest_cat_fetch") + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row" + assert row.table_name.lower() == "fetch_test" + assert row.table_schem.lower() == "pytest_cat_fetch" + assert cursor.fetchone() is None + + +def test_tables_iteration(cursor, db_connection): + """Test that 'for row in cursor.tables()' works (GH-505)""" + rows = list(cursor.tables(table="fetch_test", schema="pytest_cat_fetch")) + assert len(rows) == 1, "Iteration should yield 1 row" + assert rows[0].table_name.lower() == "fetch_test" + + +def test_columns_fetchone(cursor, db_connection): + """Test that fetchone() works on columns() result set (GH-505)""" + cursor.columns(table="fetch_test", schema="pytest_cat_fetch") + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from columns()" + assert hasattr(row, "column_name") + assert row.table_name.lower() == "fetch_test" + + +def test_primarykeys_fetchone(cursor, db_connection): + """Test that fetchone() works on primaryKeys() result set (GH-505)""" + cursor.primaryKeys(table="fetch_test", schema="pytest_cat_fetch") + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from primaryKeys()" + assert row.column_name.lower() == "id" + assert cursor.fetchone() is None + + +def test_foreignkeys_fetchone(cursor, db_connection): + """Test that fetchone() works on foreignKeys() result set (GH-505)""" + cursor.foreignKeys( + table="fetch_test_child", + schema="pytest_cat_fetch", + ) + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from foreignKeys()" + assert row.pkcolumn_name.lower() == "id" + assert row.fkcolumn_name.lower() == "parent_id" + assert cursor.fetchone() is None + + +def test_statistics_fetchone(cursor, db_connection): + """Test that fetchone() works on statistics() result set (GH-505)""" + cursor.statistics(table="fetch_test", schema="pytest_cat_fetch") + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from statistics()" + assert row.table_name.lower() == "fetch_test" + + +def test_procedures_fetchone(cursor, db_connection): + """Test that fetchone() works on procedures() result set (GH-505)""" + cursor.procedures() + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from procedures()" + assert hasattr(row, "procedure_name") + + +def test_rowid_columns_fetchone(cursor, db_connection): + """Test that fetchone() works on rowIdColumns() result set (GH-505)""" + cursor.rowIdColumns(table="fetch_test", schema="pytest_cat_fetch") + # May or may not have rowid columns; just verify no InterfaceError + row = cursor.fetchone() + if row is not None: + assert hasattr(row, "column_name") + + +def test_rowver_columns_fetchone(cursor, db_connection): + """Test that fetchone() works on rowVerColumns() result set (GH-505)""" + cursor.rowVerColumns(table="fetch_test", schema="pytest_cat_fetch") + # May or may not have rowver columns; just verify no InterfaceError + row = cursor.fetchone() + if row is not None: + assert hasattr(row, "column_name") + + +def test_gettypeinfo_fetchone(cursor, db_connection): + """Test that fetchone() works on getTypeInfo() result set (GH-505)""" + cursor.getTypeInfo() + row = cursor.fetchone() + assert row is not None, "fetchone() should return a row from getTypeInfo()" + assert hasattr(row, "type_name") + + +def test_catalog_rownumber_increments_correctly(cursor, db_connection): + """Test that rownumber increments correctly during fetchone() on catalog results (GH-505)""" + cursor.columns(table="fetch_test", schema="pytest_cat_fetch") + assert cursor.rownumber == -1 + + for expected_idx in range(4): + row = cursor.fetchone() + assert row is not None, f"Expected row at index {expected_idx}" + assert cursor.rownumber == expected_idx + + assert cursor.fetchone() is None + + +def test_catalog_fetchone_iteration_cleanup(cursor, db_connection): + """Clean up test objects for catalog fetchone/iteration testing""" + try: + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") + cursor.execute("DROP SCHEMA IF EXISTS pytest_cat_fetch") + db_connection.commit() + except Exception as e: + pytest.fail(f"Catalog fetchone/iteration cleanup failed: {e}") From 9f644b1d6dc2c268fd93ee23fed93cd14016a9ce Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Tue, 21 Apr 2026 14:32:00 +0530 Subject: [PATCH 2/2] Resoling comments --- mssql_python/cursor.py | 5 ++ tests/test_004_cursor.py | 106 ++++++++++++++++++++------------------- 2 files changed, 60 insertions(+), 51 deletions(-) diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index 4f9ff533b..9841a6738 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -1641,6 +1641,11 @@ def fetchall_with_mapping(): # Initialize rownumber tracking so fetchone() and iteration work self._reset_rownumber() + # Metadata methods produce a new result set, so rowcount is unknown + # until rows are fetched. Reset it to avoid exposing a stale value + # from a previous statement. + self.rowcount = -1 + # Return the cursor itself for method chaining return self diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index 83d2ed643..fe3436d1e 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -15963,38 +15963,52 @@ def reader(reader_id): mssql_python.native_uuid = original -def test_catalog_fetchone_iteration_setup(cursor, db_connection): - """Create test objects for catalog fetchone/iteration testing""" - try: - cursor.execute( - "IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = 'pytest_cat_fetch') " - "EXEC('CREATE SCHEMA pytest_cat_fetch')" +@pytest.fixture(scope="module") +def catalog_fetch_schema(cursor, db_connection): + """Create and tear down test objects for catalog fetchone/iteration testing.""" + cursor.execute( + "IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = 'pytest_cat_fetch') " + "EXEC('CREATE SCHEMA pytest_cat_fetch')" + ) + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") + cursor.execute("DROP PROCEDURE IF EXISTS pytest_cat_fetch.fetch_test_proc") + + cursor.execute(""" + CREATE TABLE pytest_cat_fetch.fetch_test ( + id INT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + value DECIMAL(10,2), + ts DATETIME DEFAULT GETDATE() + ) + """) + cursor.execute(""" + CREATE TABLE pytest_cat_fetch.fetch_test_child ( + child_id INT PRIMARY KEY, + parent_id INT NOT NULL, + CONSTRAINT fk_parent FOREIGN KEY (parent_id) + REFERENCES pytest_cat_fetch.fetch_test(id) ) - cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") - cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") + """) + cursor.execute(""" + CREATE PROCEDURE pytest_cat_fetch.fetch_test_proc + AS + BEGIN + SELECT 1 AS result + END + """) + db_connection.commit() - cursor.execute(""" - CREATE TABLE pytest_cat_fetch.fetch_test ( - id INT PRIMARY KEY, - name VARCHAR(100) NOT NULL, - value DECIMAL(10,2), - ts DATETIME DEFAULT GETDATE() - ) - """) - cursor.execute(""" - CREATE TABLE pytest_cat_fetch.fetch_test_child ( - child_id INT PRIMARY KEY, - parent_id INT NOT NULL, - CONSTRAINT fk_parent FOREIGN KEY (parent_id) - REFERENCES pytest_cat_fetch.fetch_test(id) - ) - """) - db_connection.commit() - except Exception as e: - pytest.fail(f"Catalog fetchone/iteration setup failed: {e}") + yield + + cursor.execute("DROP PROCEDURE IF EXISTS pytest_cat_fetch.fetch_test_proc") + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") + cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") + cursor.execute("DROP SCHEMA IF EXISTS pytest_cat_fetch") + db_connection.commit() -def test_tables_fetchone(cursor, db_connection): +def test_tables_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on tables() result set (GH-505)""" cursor.tables(table="fetch_test", schema="pytest_cat_fetch") row = cursor.fetchone() @@ -16004,14 +16018,14 @@ def test_tables_fetchone(cursor, db_connection): assert cursor.fetchone() is None -def test_tables_iteration(cursor, db_connection): +def test_tables_iteration(cursor, db_connection, catalog_fetch_schema): """Test that 'for row in cursor.tables()' works (GH-505)""" rows = list(cursor.tables(table="fetch_test", schema="pytest_cat_fetch")) assert len(rows) == 1, "Iteration should yield 1 row" assert rows[0].table_name.lower() == "fetch_test" -def test_columns_fetchone(cursor, db_connection): +def test_columns_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on columns() result set (GH-505)""" cursor.columns(table="fetch_test", schema="pytest_cat_fetch") row = cursor.fetchone() @@ -16020,7 +16034,7 @@ def test_columns_fetchone(cursor, db_connection): assert row.table_name.lower() == "fetch_test" -def test_primarykeys_fetchone(cursor, db_connection): +def test_primarykeys_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on primaryKeys() result set (GH-505)""" cursor.primaryKeys(table="fetch_test", schema="pytest_cat_fetch") row = cursor.fetchone() @@ -16029,7 +16043,7 @@ def test_primarykeys_fetchone(cursor, db_connection): assert cursor.fetchone() is None -def test_foreignkeys_fetchone(cursor, db_connection): +def test_foreignkeys_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on foreignKeys() result set (GH-505)""" cursor.foreignKeys( table="fetch_test_child", @@ -16042,7 +16056,7 @@ def test_foreignkeys_fetchone(cursor, db_connection): assert cursor.fetchone() is None -def test_statistics_fetchone(cursor, db_connection): +def test_statistics_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on statistics() result set (GH-505)""" cursor.statistics(table="fetch_test", schema="pytest_cat_fetch") row = cursor.fetchone() @@ -16050,15 +16064,16 @@ def test_statistics_fetchone(cursor, db_connection): assert row.table_name.lower() == "fetch_test" -def test_procedures_fetchone(cursor, db_connection): +def test_procedures_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on procedures() result set (GH-505)""" - cursor.procedures() + cursor.procedures(procedure="fetch_test_proc", schema="pytest_cat_fetch") row = cursor.fetchone() assert row is not None, "fetchone() should return a row from procedures()" - assert hasattr(row, "procedure_name") + assert row.procedure_name.lower().startswith("fetch_test_proc") + assert cursor.fetchone() is None -def test_rowid_columns_fetchone(cursor, db_connection): +def test_rowid_columns_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on rowIdColumns() result set (GH-505)""" cursor.rowIdColumns(table="fetch_test", schema="pytest_cat_fetch") # May or may not have rowid columns; just verify no InterfaceError @@ -16067,7 +16082,7 @@ def test_rowid_columns_fetchone(cursor, db_connection): assert hasattr(row, "column_name") -def test_rowver_columns_fetchone(cursor, db_connection): +def test_rowver_columns_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on rowVerColumns() result set (GH-505)""" cursor.rowVerColumns(table="fetch_test", schema="pytest_cat_fetch") # May or may not have rowver columns; just verify no InterfaceError @@ -16076,7 +16091,7 @@ def test_rowver_columns_fetchone(cursor, db_connection): assert hasattr(row, "column_name") -def test_gettypeinfo_fetchone(cursor, db_connection): +def test_gettypeinfo_fetchone(cursor, db_connection, catalog_fetch_schema): """Test that fetchone() works on getTypeInfo() result set (GH-505)""" cursor.getTypeInfo() row = cursor.fetchone() @@ -16084,7 +16099,7 @@ def test_gettypeinfo_fetchone(cursor, db_connection): assert hasattr(row, "type_name") -def test_catalog_rownumber_increments_correctly(cursor, db_connection): +def test_catalog_rownumber_increments_correctly(cursor, db_connection, catalog_fetch_schema): """Test that rownumber increments correctly during fetchone() on catalog results (GH-505)""" cursor.columns(table="fetch_test", schema="pytest_cat_fetch") assert cursor.rownumber == -1 @@ -16095,14 +16110,3 @@ def test_catalog_rownumber_increments_correctly(cursor, db_connection): assert cursor.rownumber == expected_idx assert cursor.fetchone() is None - - -def test_catalog_fetchone_iteration_cleanup(cursor, db_connection): - """Clean up test objects for catalog fetchone/iteration testing""" - try: - cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test_child") - cursor.execute("DROP TABLE IF EXISTS pytest_cat_fetch.fetch_test") - cursor.execute("DROP SCHEMA IF EXISTS pytest_cat_fetch") - db_connection.commit() - except Exception as e: - pytest.fail(f"Catalog fetchone/iteration cleanup failed: {e}")