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