Merge branch 'BUG19282158' into develop

Author: peeyushgupta

lib/mysql/connector/protocol.py

@@ -620,6 +620,8 @@ def make_stmt_execute(self, statement_id, data=(), parameters=(),
                 flags = 0
                 if value is None:
                     null_bitmap[(pos // 8)] |= 1 << (pos % 8)
+                    types.append(utils.int1store(FieldType.NULL) +
+                                 utils.int1store(flags))
                     continue
                 elif pos in long_data_used:
                     if long_data_used[pos][0]:

tests/test_bugs.py

@@ -2755,3 +2755,42 @@ def test_duplicate_optionfiles(self):
         ]
         self.assertRaises(ValueError, mysql.connector.connect,
                           option_files=files)
+
+
+class BugOra19282158(tests.MySQLConnectorTests):
+    """BUG#19282158: NULL values with prepared statements
+    """
+    def setUp(self):
+        config = tests.get_mysql_config()
+        self.cnx = connection.MySQLConnection(**config)
+        self.cursor = self.cnx.cursor()
+
+        self.tbl = 'Bug19282158'
+        self.cursor.execute("DROP TABLE IF EXISTS %s" % self.tbl)
+
+        create = ('CREATE TABLE {0}(col1 INT NOT NULL, col2 INT NULL, '
+                  'col3 VARCHAR(10), col4 DECIMAL(4,2) NULL, '
+                  'col5 DATETIME NULL, col6 INT NOT NULL, col7 VARCHAR(10), '
+                  'PRIMARY KEY(col1))'.format(self.tbl))
+
+        self.cursor.execute(create)
+
+    def tearDown(self):
+        self.cursor.execute("DROP TABLE IF EXISTS %s" % self.tbl)
+        self.cursor.close()
+        self.cnx.close()
+
+    def test_null(self):
+        cur = self.cnx.cursor(prepared=True)
+        sql = ("INSERT INTO {0}(col1, col2, col3, col4, col5, col6, col7) "
+               "VALUES (?, ?, ?, ?, ?, ?, ?)".format(self.tbl))
+        params = (100, None, 'foo', None, datetime(2014, 8, 4, 9, 11, 14),
+                  10, 'bar')
+        exp = (100, None, bytearray(b'foo'), None,
+               datetime(2014, 8, 4, 9, 11, 14), 10, bytearray(b'bar'))
+        cur.execute(sql, params)
+
+        sql = "SELECT * FROM {0}".format(self.tbl)
+        cur.execute(sql)
+        self.assertEqual(exp, cur.fetchone())
+        cur.close()

tests/test_protocol.py

@@ -535,13 +535,14 @@ def test_make_stmt_execute(self):
 
         # Testing null bitmap
         data = (None, None)
-        exp = bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x03\x01')
+        exp = bytearray(b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x03\x01\x06'
+                        b'\x00\x06\x00')
         res = self._protocol.make_stmt_execute(statement_id, data, (1, 2))
         self.assertEqual(exp, res)
 
         data = (None, 'Ham')
         exp = bytearray(
-            b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x01\x01\x0f\x00'
+            b'\x01\x00\x00\x00\x00\x01\x00\x00\x00\x01\x01\x06\x00\x0f\x00'
             b'\x03\x48\x61\x6d'
         )
         res = self._protocol.make_stmt_execute(statement_id, data, (1, 2))