Test load local infile with a valid file
(self)
| 27 | c.close() |
| 28 | |
| 29 | def test_load_file(self): |
| 30 | """Test load local infile with a valid file""" |
| 31 | conn = self.connect() |
| 32 | c = conn.cursor() |
| 33 | c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)") |
| 34 | filename = os.path.join( |
| 35 | os.path.dirname(os.path.realpath(__file__)), "data", "load_local_data.txt" |
| 36 | ) |
| 37 | try: |
| 38 | c.execute( |
| 39 | f"LOAD DATA LOCAL INFILE '{filename}' INTO TABLE test_load_local" |
| 40 | + " FIELDS TERMINATED BY ','" |
| 41 | ) |
| 42 | c.execute("SELECT COUNT(*) FROM test_load_local") |
| 43 | self.assertEqual(22749, c.fetchone()[0]) |
| 44 | finally: |
| 45 | c.execute("DROP TABLE test_load_local") |
| 46 | |
| 47 | def test_unbuffered_load_file(self): |
| 48 | """Test unbuffered load local infile with a valid file""" |