BUG#34223015: Invalidate the usage of non-compatible cursor types

Connector/Python c-ext and pure Python cursors types are not
interchangeable when using connection objects that aren't of the same
type.
Therefore, c-ext connection objects can't deal with pure Python cursor
objects and vice-versa.

This patch fixes this issue by checking if the provided cursor class
uses CMySQLCursor as the base class.

Change-Id: I8941360bc816961c5d5cd5540a5398dfea7bb3d8

Author: nmariz

CHANGES.txt

@@ -18,6 +18,7 @@ v8.0.30
 - WL#14822: Refactor the authentication plugin mechanism
 - WL#14815: Support OpenSSL 3.0
 - BUG#34228442: Fix NO_BACKSLASH_ESCAPES SQL mode support in c-ext
+- BUG#34223015: Invalidate the usage of non-compatible cursor types
 - BUG#34127959: Add isolation level support in Django backend
 - BUG#33923516: Allow tuple of dictionaries as "failover" argument
 - BUG#28821983: Fix rounding errors for decimal values

lib/mysql/connector/connection_cext.py

@@ -621,7 +621,7 @@ def cursor(
         if not self.is_connected():
             raise OperationalError("MySQL Connection not available.")
         if cursor_class is not None:
-            if not issubclass(cursor_class, MySQLCursorAbstract):
+            if not issubclass(cursor_class, CMySQLCursor):
                 raise ProgrammingError(
                     "Cursor class needs be to subclass of cursor_cext.CMySQLCursor"
                 )

tests/cext/test_cext_connection.py

@@ -35,7 +35,7 @@
 
 import tests
 
-from mysql.connector import errors
+from mysql.connector import cursor, cursor_cext, errors
 from mysql.connector.connection import MySQLConnection
 from mysql.connector.connection_cext import CMySQLConnection
 from mysql.connector.constants import ClientFlag, flag_is_set
@@ -158,3 +158,56 @@ def test_connection_id(self):
         self.cnx.close()
         self.assertIsNone(self.cnx.connection_id)
         self.cnx.connect()
+
+    def test_cursor(self):
+        """Test CEXT cursors."""
+
+        class FalseCursor:
+            ...
+
+        class TrueCursor(cursor_cext.CMySQLCursor):
+            ...
+
+        self.assertRaises(
+            errors.ProgrammingError, self.cnx.cursor, cursor_class=FalseCursor
+        )
+        self.assertTrue(
+            isinstance(self.cnx.cursor(cursor_class=TrueCursor), TrueCursor)
+        )
+
+        self.assertRaises(
+            errors.ProgrammingError, self.cnx.cursor, cursor_class=cursor.MySQLCursor
+        )
+        self.assertRaises(
+            errors.ProgrammingError,
+            self.cnx.cursor,
+            cursor_class=cursor.MySQLCursorBuffered,
+        )
+
+        cases = [
+            ({}, cursor_cext.CMySQLCursor),
+            ({"buffered": True}, cursor_cext.CMySQLCursorBuffered),
+            ({"raw": True}, cursor_cext.CMySQLCursorRaw),
+            ({"buffered": True, "raw": True}, cursor_cext.CMySQLCursorBufferedRaw),
+            ({"prepared": True}, cursor_cext.CMySQLCursorPrepared),
+            ({"dictionary": True}, cursor_cext.CMySQLCursorDict),
+            ({"named_tuple": True}, cursor_cext.CMySQLCursorNamedTuple),
+            (
+                {"dictionary": True, "buffered": True},
+                cursor_cext.CMySQLCursorBufferedDict,
+            ),
+            (
+                {"named_tuple": True, "buffered": True},
+                cursor_cext.CMySQLCursorBufferedNamedTuple,
+            ),
+        ]
+        for kwargs, exp in cases:
+            self.assertTrue(isinstance(self.cnx.cursor(**kwargs), exp))
+
+        self.assertRaises(ValueError, self.cnx.cursor, prepared=True, buffered=True)
+        self.assertRaises(ValueError, self.cnx.cursor, dictionary=True, raw=True)
+        self.assertRaises(ValueError, self.cnx.cursor, named_tuple=True, raw=True)
+
+        # Test when connection is closed
+        self.cnx.close()
+        self.assertRaises(errors.OperationalError, self.cnx.cursor)

tests/test_connection.py

@@ -1694,6 +1694,18 @@ def __init__(self, cnx=None):
             isinstance(self.cnx.cursor(cursor_class=TrueCursor), TrueCursor)
         )
 
+        if HAVE_CMYSQL:
+            self.assertRaises(
+                errors.ProgrammingError,
+                self.cnx.cursor,
+                cursor_class=cursor_cext.CMySQLCursor,
+            )
+            self.assertRaises(
+                errors.ProgrammingError,
+                self.cnx.cursor,
+                cursor_class=cursor_cext.CMySQLCursorBufferedRaw,
+            )
+
         cases = [
             ({}, cursor.MySQLCursor),
             ({"buffered": True}, cursor.MySQLCursorBuffered),