| 329 | |
| 330 | |
| 331 | class TestBulkInserts(base.PyMySQLTestCase): |
| 332 | cursor_type = pymysql.cursors.DictCursor |
| 333 | |
| 334 | def setUp(self): |
| 335 | super().setUp() |
| 336 | self.conn = conn = self.connect() |
| 337 | |
| 338 | # create a table and some data to query |
| 339 | self.safe_create_table( |
| 340 | conn, |
| 341 | "bulkinsert", |
| 342 | """\ |
| 343 | CREATE TABLE bulkinsert |
| 344 | ( |
| 345 | id int, |
| 346 | name char(20), |
| 347 | age int, |
| 348 | height int, |
| 349 | PRIMARY KEY (id) |
| 350 | ) |
| 351 | """, |
| 352 | ) |
| 353 | |
| 354 | def _verify_records(self, data): |
| 355 | conn = self.connect() |
| 356 | cursor = conn.cursor() |
| 357 | cursor.execute("SELECT id, name, age, height from bulkinsert") |
| 358 | result = cursor.fetchall() |
| 359 | self.assertEqual(sorted(data), sorted(result)) |
| 360 | |
| 361 | def test_bulk_insert(self): |
| 362 | conn = self.connect() |
| 363 | cursor = conn.cursor() |
| 364 | |
| 365 | data = [(0, "bob", 21, 123), (1, "jim", 56, 45), (2, "fred", 100, 180)] |
| 366 | cursor.executemany( |
| 367 | "insert into bulkinsert (id, name, age, height) values (%s,%s,%s,%s)", |
| 368 | data, |
| 369 | ) |
| 370 | self.assertEqual( |
| 371 | cursor._executed, |
| 372 | bytearray( |
| 373 | b"insert into bulkinsert (id, name, age, height) values " |
| 374 | b"(0,'bob',21,123),(1,'jim',56,45),(2,'fred',100,180)" |
| 375 | ), |
| 376 | ) |
| 377 | cursor.execute("commit") |
| 378 | self._verify_records(data) |
| 379 | |
| 380 | def test_bulk_insert_multiline_statement(self): |
| 381 | conn = self.connect() |
| 382 | cursor = conn.cursor() |
| 383 | data = [(0, "bob", 21, 123), (1, "jim", 56, 45), (2, "fred", 100, 180)] |
| 384 | cursor.executemany( |
| 385 | """insert |
| 386 | into bulkinsert (id, name, |
| 387 | age, height) |
| 388 | values (%s, |
nothing calls this directly
no outgoing calls
no test coverage detected