| 207 | self.assertEqual(2013, e.args[0]) |
| 208 | |
| 209 | def test_issue_36(self): |
| 210 | # connection 0 is super user, connection 1 isn't |
| 211 | conn = self.connections[1] |
| 212 | c = conn.cursor() |
| 213 | c.execute("show processlist") |
| 214 | kill_id = None |
| 215 | for row in c.fetchall(): |
| 216 | id = row[0] |
| 217 | info = row[7] |
| 218 | if info == "show processlist": |
| 219 | kill_id = id |
| 220 | break |
| 221 | self.assertEqual(kill_id, conn.thread_id()) |
| 222 | # now nuke the connection |
| 223 | self.connections[0].kill(kill_id) |
| 224 | # make sure this connection has broken |
| 225 | with pytest.raises(pymysql.Error): |
| 226 | c.execute("show tables") |
| 227 | c.close() |
| 228 | conn.close() |
| 229 | |
| 230 | # check the process list from the other connection |
| 231 | try: |
| 232 | # Wait since Travis-CI sometimes fail this test. |
| 233 | time.sleep(0.1) |
| 234 | |
| 235 | c = self.connections[0].cursor() |
| 236 | c.execute("show processlist") |
| 237 | ids = [row[0] for row in c.fetchall()] |
| 238 | self.assertFalse(kill_id in ids) |
| 239 | finally: |
| 240 | del self.connections[1] |
| 241 | |
| 242 | def test_issue_37(self): |
| 243 | conn = self.connect() |