/usr/lib/python3/dist-packages/pymysql/tests/test_load_local.py is in python3-pymysql 0.8.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | from pymysql import cursors, OperationalError, Warning
from pymysql.tests import base
import os
import warnings
import unittest2
__all__ = ["TestLoadLocal"]
@unittest2.skip("Not working on Debian")
class TestLoadLocal(base.PyMySQLTestCase):
def test_no_file(self):
"""Test load local infile when the file does not exist"""
conn = self.connections[0]
c = conn.cursor()
c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
try:
self.assertRaises(
OperationalError,
c.execute,
("LOAD DATA LOCAL INFILE 'no_data.txt' INTO TABLE "
"test_load_local fields terminated by ','")
)
finally:
c.execute("DROP TABLE test_load_local")
c.close()
def test_load_file(self):
"""Test load local infile with a valid file"""
conn = self.connections[0]
c = conn.cursor()
c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
'data',
'load_local_data.txt')
try:
c.execute(
("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
"test_load_local FIELDS TERMINATED BY ','").format(filename)
)
c.execute("SELECT COUNT(*) FROM test_load_local")
self.assertEqual(22749, c.fetchone()[0])
finally:
c.execute("DROP TABLE test_load_local")
def test_unbuffered_load_file(self):
"""Test unbuffered load local infile with a valid file"""
conn = self.connections[0]
c = conn.cursor(cursors.SSCursor)
c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
'data',
'load_local_data.txt')
try:
c.execute(
("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
"test_load_local FIELDS TERMINATED BY ','").format(filename)
)
c.execute("SELECT COUNT(*) FROM test_load_local")
self.assertEqual(22749, c.fetchone()[0])
finally:
c.close()
conn.close()
conn.connect()
c = conn.cursor()
c.execute("DROP TABLE test_load_local")
def test_load_warnings(self):
"""Test load local infile produces the appropriate warnings"""
conn = self.connections[0]
c = conn.cursor()
c.execute("CREATE TABLE test_load_local (a INTEGER, b INTEGER)")
filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
'data',
'load_local_warn_data.txt')
try:
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('always')
c.execute(
("LOAD DATA LOCAL INFILE '{0}' INTO TABLE " +
"test_load_local FIELDS TERMINATED BY ','").format(filename)
)
self.assertEqual(w[0].category, Warning)
expected_message = "Incorrect integer value"
if expected_message not in str(w[-1].message):
self.fail("%r not in %r" % (expected_message, w[-1].message))
finally:
c.execute("DROP TABLE test_load_local")
c.close()
if __name__ == "__main__":
import unittest
unittest.main()
|