| 176 | |
| 177 | |
| 178 | class TestNewIssues(base.PyMySQLTestCase): |
| 179 | def test_issue_34(self): |
| 180 | try: |
| 181 | pymysql.connect(host="localhost", port=1237, user="root") |
| 182 | self.fail() |
| 183 | except pymysql.OperationalError as e: |
| 184 | self.assertEqual(2003, e.args[0]) |
| 185 | except Exception: |
| 186 | self.fail() |
| 187 | |
| 188 | def test_issue_33(self): |
| 189 | conn = pymysql.connect(charset="utf8", **self.databases[0]) |
| 190 | self.safe_create_table( |
| 191 | conn, "hei\xdfe", "create table hei\xdfe (name varchar(32))" |
| 192 | ) |
| 193 | c = conn.cursor() |
| 194 | c.execute("insert into hei\xdfe (name) values ('Pi\xdfata')") |
| 195 | c.execute("select name from hei\xdfe") |
| 196 | self.assertEqual("Pi\xdfata", c.fetchone()[0]) |
| 197 | |
| 198 | @pytest.mark.skip("This test requires manual intervention") |
| 199 | def test_issue_35(self): |
| 200 | conn = self.connect() |
| 201 | c = conn.cursor() |
| 202 | print("sudo killall -9 mysqld within the next 10 seconds") |
| 203 | try: |
| 204 | c.execute("select sleep(10)") |
| 205 | self.fail() |
| 206 | except pymysql.OperationalError as e: |
| 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() |
nothing calls this directly
no outgoing calls
no test coverage detected