| 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 | try: |
| 226 | c.execute("show tables") |
| 227 | self.fail() |
| 228 | except Exception: |
| 229 | pass |
| 230 | c.close() |
| 231 | conn.close() |
| 232 | |
| 233 | # check the process list from the other connection |
| 234 | try: |
| 235 | # Wait since Travis-CI sometimes fail this test. |
| 236 | time.sleep(0.1) |
| 237 | |
| 238 | c = self.connections[0].cursor() |
| 239 | c.execute("show processlist") |
| 240 | ids = [row[0] for row in c.fetchall()] |
| 241 | self.assertFalse(kill_id in ids) |
| 242 | finally: |
| 243 | del self.connections[1] |
| 244 | |
| 245 | def test_issue_37(self): |
| 246 | conn = self.connect() |