| 438 | |
| 439 | |
| 440 | class TestConnection(base.PyMySQLTestCase): |
| 441 | def test_utf8mb4(self): |
| 442 | """This test requires MySQL >= 5.5""" |
| 443 | arg = self.databases[0].copy() |
| 444 | arg["charset"] = "utf8mb4" |
| 445 | pymysql.connect(**arg) |
| 446 | |
| 447 | def test_set_character_set(self): |
| 448 | con = self.connect() |
| 449 | cur = con.cursor() |
| 450 | |
| 451 | con.set_character_set("latin1") |
| 452 | cur.execute("SELECT @@character_set_connection") |
| 453 | self.assertEqual(cur.fetchone(), ("latin1",)) |
| 454 | self.assertEqual(con.encoding, "cp1252") |
| 455 | |
| 456 | con.set_character_set("utf8mb4", "utf8mb4_general_ci") |
| 457 | cur.execute("SELECT @@character_set_connection, @@collation_connection") |
| 458 | self.assertEqual(cur.fetchone(), ("utf8mb4", "utf8mb4_general_ci")) |
| 459 | self.assertEqual(con.encoding, "utf8") |
| 460 | |
| 461 | def test_largedata(self): |
| 462 | """Large query and response (>=16MB)""" |
| 463 | cur = self.connect().cursor() |
| 464 | cur.execute("SELECT @@max_allowed_packet") |
| 465 | if cur.fetchone()[0] < 16 * 1024 * 1024 + 10: |
| 466 | print("Set max_allowed_packet to bigger than 17MB") |
| 467 | return |
| 468 | t = "a" * (16 * 1024 * 1024) |
| 469 | cur.execute("SELECT '" + t + "'") |
| 470 | assert cur.fetchone()[0] == t |
| 471 | |
| 472 | def test_autocommit(self): |
| 473 | con = self.connect() |
| 474 | self.assertFalse(con.get_autocommit()) |
| 475 | |
| 476 | cur = con.cursor() |
| 477 | cur.execute("SET AUTOCOMMIT=1") |
| 478 | self.assertTrue(con.get_autocommit()) |
| 479 | |
| 480 | con.autocommit(False) |
| 481 | self.assertFalse(con.get_autocommit()) |
| 482 | cur.execute("SELECT @@AUTOCOMMIT") |
| 483 | self.assertEqual(cur.fetchone()[0], 0) |
| 484 | |
| 485 | def test_select_db(self): |
| 486 | con = self.connect() |
| 487 | current_db = self.databases[0]["database"] |
| 488 | other_db = self.databases[1]["database"] |
| 489 | |
| 490 | cur = con.cursor() |
| 491 | cur.execute("SELECT database()") |
| 492 | self.assertEqual(cur.fetchone()[0], current_db) |
| 493 | |
| 494 | con.select_db(other_db) |
| 495 | cur.execute("SELECT database()") |
| 496 | self.assertEqual(cur.fetchone()[0], other_db) |
| 497 |
nothing calls this directly
no outgoing calls
no test coverage detected