3 (h @sdddlmZmZmZddlmZddlZddlZdgZGdddej Z e dkr`ddl Z e j dS))cursorsOperationalErrorWarning)baseN TestLoadLocalc@s,eZdZddZddZddZddZd S) rc CsJ|jd}|j}|jdz|jt|jdWd|jd|jXdS)z3Test load local infile when the file does not existrz3CREATE TABLE test_load_local (a INTEGER, b INTEGER)zXLOAD DATA LOCAL INFILE 'no_data.txt' INTO TABLE test_load_local fields terminated by ','NzDROP TABLE test_load_local) connectionscursorexecuteZ assertRaisesrclose)selfconncrB/tmp/pip-install-q3hcpn_q/PyMySQL/pymysql/tests/test_load_local.py test_no_file s    zTestLoadLocal.test_no_filec Cs|jd}|j}|jdtjjtjjtjjtdd}z2|jd j ||jd|j d|j dWd |jd Xd S) z(Test load local infile with a valid filerz3CREATE TABLE test_load_local (a INTEGER, b INTEGER)datazload_local_data.txtz(LOAD DATA LOCAL INFILE '{0}' INTO TABLE z(test_load_local FIELDS TERMINATED BY ','z$SELECT COUNT(*) FROM test_load_localiXNzDROP TABLE test_load_localzPLOAD DATA LOCAL INFILE '{0}' INTO TABLE test_load_local FIELDS TERMINATED BY ',') rrr ospathjoindirnamerealpath__file__format assertEqualfetchone)r r r filenamerrrtest_load_files    zTestLoadLocal.test_load_filec Cs|jd}|jtj}|jdtjjtjjtjj t dd}z2|jd j ||jd|j d|j dWd |j|j|j|j}|jd Xd S) z3Test unbuffered load local infile with a valid filerz3CREATE TABLE test_load_local (a INTEGER, b INTEGER)rzload_local_data.txtz(LOAD DATA LOCAL INFILE '{0}' INTO TABLE z(test_load_local FIELDS TERMINATED BY ','z$SELECT COUNT(*) FROM test_load_localiXNzDROP TABLE test_load_localzPLOAD DATA LOCAL INFILE '{0}' INTO TABLE test_load_local FIELDS TERMINATED BY ',')rrrZSSCursorr rrrrrrrrrr connect)r r r rrrrtest_unbuffered_load_file-s"     z'TestLoadLocal.test_unbuffered_load_filecCs|jd}|j}|jdtjjtjjtjjtdd}zvt j dd`}t j d|jdj ||j |djtd }|t|djkr|jd ||djfWd QRXWd |jd|jXd S)z8Test load local infile produces the appropriate warningsrz3CREATE TABLE test_load_local (a INTEGER, b INTEGER)rzload_local_warn_data.txtT)recordalwaysz(LOAD DATA LOCAL INFILE '{0}' INTO TABLE z(test_load_local FIELDS TERMINATED BY ','zIncorrect integer valuez %r not in %rNzDROP TABLE test_load_localzPLOAD DATA LOCAL INFILE '{0}' INTO TABLE test_load_local FIELDS TERMINATED BY ','r")rrr rrrrrrwarningscatch_warnings simplefilterrrcategoryrstrmessagefailr )r r r rwZexpected_messagerrrtest_load_warningsCs$    & z TestLoadLocal.test_load_warningsN)__name__ __module__ __qualname__rrrr+rrrrr s__main__)ZpymysqlrrrZ pymysql.testsrrr#__all__ZPyMySQLTestCaserr,Zunittestmainrrrrs Q