| 199 | self.assertEqual(cur.fetchone(), (1,)) |
| 200 | |
| 201 | def test_warnings(self): |
| 202 | con = self.connect() |
| 203 | cur = con.cursor(pymysql.cursors.SSCursor) |
| 204 | cur.execute("DROP TABLE IF EXISTS `no_exists_table`") |
| 205 | self.assertEqual(cur.warning_count, 1) |
| 206 | |
| 207 | cur.execute("SHOW WARNINGS") |
| 208 | w = cur.fetchone() |
| 209 | self.assertEqual(w[1], ER.BAD_TABLE_ERROR) |
| 210 | self.assertIn( |
| 211 | "no_exists_table", |
| 212 | w[2], |
| 213 | ) |
| 214 | |
| 215 | # ensure unbuffered result is finished |
| 216 | self.assertIsNone(cur.fetchone()) |
| 217 | |
| 218 | cur.execute("SELECT 1") |
| 219 | self.assertEqual(cur.fetchone(), (1,)) |
| 220 | self.assertIsNone(cur.fetchone()) |
| 221 | |
| 222 | self.assertEqual(cur.warning_count, 0) |
| 223 | |
| 224 | cur.execute("SELECT CAST('abc' AS SIGNED)") |
| 225 | # this ensures fully retrieving the unbuffered result |
| 226 | rows = cur.fetchmany(2) |
| 227 | self.assertEqual(len(rows), 1) |
| 228 | self.assertEqual(cur.warning_count, 1) |
| 229 | |
| 230 | |
| 231 | __all__ = ["TestSSCursor"] |