Duplicate field overwrites the previous one in the result of DictCursor
(self)
| 299 | c.execute("drop table issue66") |
| 300 | |
| 301 | def test_issue_79(self): |
| 302 | """Duplicate field overwrites the previous one in the result of DictCursor""" |
| 303 | conn = self.connect() |
| 304 | c = conn.cursor(pymysql.cursors.DictCursor) |
| 305 | |
| 306 | with warnings.catch_warnings(): |
| 307 | warnings.filterwarnings("ignore") |
| 308 | c.execute("drop table if exists a") |
| 309 | c.execute("drop table if exists b") |
| 310 | c.execute("""CREATE TABLE a (id int, value int)""") |
| 311 | c.execute("""CREATE TABLE b (id int, value int)""") |
| 312 | |
| 313 | a = (1, 11) |
| 314 | b = (1, 22) |
| 315 | try: |
| 316 | c.execute("insert into a values (%s, %s)", a) |
| 317 | c.execute("insert into b values (%s, %s)", b) |
| 318 | |
| 319 | c.execute("SELECT * FROM a inner join b on a.id = b.id") |
| 320 | r = c.fetchall()[0] |
| 321 | self.assertEqual(r["id"], 1) |
| 322 | self.assertEqual(r["value"], 11) |
| 323 | self.assertEqual(r["b.value"], 22) |
| 324 | finally: |
| 325 | c.execute("drop table a") |
| 326 | c.execute("drop table b") |
| 327 | |
| 328 | def test_issue_95(self): |
| 329 | """Leftover trailing OK packet for "CALL my_sp" queries""" |