FIX: ODBC Catalog method fetchone() issue

Author: jahnvi480

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
 

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}")