ER_BAD_NULL should be IntegrityError. (#579)

Fixes #535

Author: methane

MySQLdb/_mysql.c

@@ -180,6 +180,7 @@ _mysql_Exception(_mysql_ConnectionObject *c)
 #ifdef ER_NO_DEFAULT_FOR_FIELD
     case ER_NO_DEFAULT_FOR_FIELD:
 #endif
+    case ER_BAD_NULL_ERROR:
         e = _mysql_IntegrityError;
         break;
 #ifdef ER_WARNING_NOT_COMPLETE_ROLLBACK

tests/capabilities.py

@@ -11,7 +11,6 @@
 
 
 class DatabaseTest(unittest.TestCase):
-
     db_module = None
     connect_args = ()
     connect_kwargs = dict()
@@ -20,7 +19,6 @@ class DatabaseTest(unittest.TestCase):
     debug = False
 
     def setUp(self):
-
         db = connection_factory(**self.connect_kwargs)
         self.connection = db
         self.cursor = db.cursor()
@@ -67,7 +65,6 @@ def new_table_name(self):
             i = i + 1
 
     def create_table(self, columndefs):
-
         """Create a table using a list of column definitions given in
         columndefs.
 

tests/test_MySQLdb_capabilities.py

@@ -12,7 +12,6 @@
 
 
 class test_MySQLdb(capabilities.DatabaseTest):
-
     db_module = MySQLdb
     connect_args = ()
     connect_kwargs = dict(

tests/test_errors.py

@@ -0,0 +1,56 @@
+import pytest
+import MySQLdb.cursors
+from configdb import connection_factory
+
+
+_conns = []
+_tables = []
+
+
+def connect(**kwargs):
+    conn = connection_factory(**kwargs)
+    _conns.append(conn)
+    return conn
+
+
+def teardown_function(function):
+    if _tables:
+        c = _conns[0]
+        cur = c.cursor()
+        for t in _tables:
+            cur.execute("DROP TABLE {}".format(t))
+        cur.close()
+        del _tables[:]
+
+    for c in _conns:
+        c.close()
+    del _conns[:]
+
+
+def test_null():
+    """Inserting NULL into non NULLABLE column"""
+    # https://github.com/PyMySQL/mysqlclient/issues/535
+    table_name = "test_null"
+    conn = connect()
+    cursor = conn.cursor()
+
+    cursor.execute(f"create table {table_name} (c1 int primary key)")
+    _tables.append(table_name)
+
+    with pytest.raises(MySQLdb.IntegrityError):
+        cursor.execute(f"insert into {table_name} values (null)")
+
+
+def test_duplicated_pk():
+    """Inserting row with duplicated PK"""
+    # https://github.com/PyMySQL/mysqlclient/issues/535
+    table_name = "test_duplicated_pk"
+    conn = connect()
+    cursor = conn.cursor()
+
+    cursor.execute(f"create table {table_name} (c1 int primary key)")
+    _tables.append(table_name)
+
+    cursor.execute(f"insert into {table_name} values (1)")
+    with pytest.raises(MySQLdb.IntegrityError):
+        cursor.execute(f"insert into {table_name} values (1)")