Test load local infile produces the appropriate warnings
(self)
| 67 | c.execute("DROP TABLE test_load_local") |
| 68 | |
| 69 | def test_load_warnings(self): |
| 70 | """Test load local infile produces the appropriate warnings""" |
| 71 | conn = self.connect() |
| 72 | c = conn.cursor() |
| 73 | c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)") |
| 74 | filename = os.path.join( |
| 75 | os.path.dirname(os.path.realpath(__file__)), |
| 76 | "data", |
| 77 | "load_local_warn_data.txt", |
| 78 | ) |
| 79 | try: |
| 80 | c.execute( |
| 81 | ( |
| 82 | "LOAD DATA LOCAL INFILE '{0}' INTO TABLE " |
| 83 | + "test_load_local FIELDS TERMINATED BY ','" |
| 84 | ).format(filename) |
| 85 | ) |
| 86 | self.assertEqual(1, c.warning_count) |
| 87 | |
| 88 | c.execute("SHOW WARNINGS") |
| 89 | w = c.fetchone() |
| 90 | |
| 91 | self.assertEqual(ER.TRUNCATED_WRONG_VALUE_FOR_FIELD, w[1]) |
| 92 | self.assertIn( |
| 93 | "incorrect integer value", |
| 94 | w[2].lower(), |
| 95 | ) |
| 96 | finally: |
| 97 | c.execute("DROP TABLE test_load_local") |
| 98 | c.close() |
| 99 | |
| 100 | |
| 101 | if __name__ == "__main__": |