3 (he8@sddlZddlZddlZddlZddlmZddlmZddlZddl m Z ddl m Z dddgZ Gd dde jZGd dde jZGd dde jZdS) N)SkipTest)util)base)ProgrammingErrorTestConversion TestCursorTestBulkInsertsc@sLeZdZddZddZddZddZd d Zd d Zd dZ ddZ dS)rcCs|jd}|j}|jdzdd"ddddd j|jtjd d d tjd d ddddtjd dtjddtj f }|jd||jd|j }|j t j d|d|j |dd|dd|j tjdd|djd|dj|d|j tj|d#dd|d$|jd|jddgd|jd|j }|j tdgd||jdxXttttfD]H}|jd|d dg}|jd|f|j}|j d'||jdqpWWd|jd!XdS)(z test every data type rzcreate table test_datatypes (b bit, i int, l bigint, f real, s varchar(32), u varchar(32), bb blob, d date, dt datetime, ts timestamp, td time, t time, st datetime)Tl2}rg@z hello'" worldu Españolz binarydataii-9 zeinsert into test_datatypes (b,i,l,f,s,u,bb,d,dt,td,t,st) values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)z6select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes <Nzdelete from test_datatypes z>insert into test_datatypes (i, l) values (2,4), (6,8), (10,12)z5select l from test_datatypes where i in %s order by izdrop table test_datatypesrrr)rr) connectionscursorexecuteencodecharsetdatetimedate timedeltatime localtimefetchone assertEqualrint2bytehourminutetuplelistset frozensetfetchall)selfconncvrZseq_typeseqr7=/tmp/pip-install-q3hcpn_q/PyMySQL/pymysql/tests/test_basic.pytest_datatypess4  R  0"      zTestConversion.test_datatypesc Cs`|jd}|j}|jdz2|jddddd|jd|jd |jWd |jd Xd S) z test dict escaping rz8create table test_dict (a integer, b integer, c integer)z:insert into test_dict (a,b,c) values (%(a)s, %(b)s, %(c)s)rr r )abr3zselect a,b,c from test_dictNzdrop table test_dict)rr r )rrrr(r')r1r2r3r7r7r8 test_dict:s   zTestConversion.test_dictc Cs^|jd}|j}|jdd}z,|jd||jd|j|f|jWd|jdXdS)Nrzcreate table test_dict (a text)zI am a test stringz%insert into test_dict (a) values (%s)zselect a from test_dictzdrop table test_dict)rrrr(r')r1r2r3 test_valuer7r7r8 test_stringFs    zTestConversion.test_stringc Cs^|jd}|j}|jdd}z,|jd||jd|j|f|jWd|jdXdS)Nrz"create table test_dict (a integer)i90z%insert into test_dict (a) values (%s)zselect a from test_dictzdrop table test_dict)rrrr(r')r1r2r3r=r7r7r8 test_integerRs    zTestConversion.test_integerc Csptttdd}|jd}|j|dd|j2}|jd|f|jd|j||jdWdQRXdS) ztest binary datarr test_blobzcreate table test_blob (b blob)z%insert into test_blob (b) values (%s)zselect b from test_blobN) bytes bytearrayrangersafe_create_tablerrr(r')r1datar2r3r7r7r8rA^s    zTestConversion.test_blobcCsJ|jd}|j}|jd|jd|j|jd|jd|jdS)z' test conversion of null, empty string rzselect null,''Nzselect '',null)NrG)rGN)rrrr(r')r1r2r3r7r7r8 test_untypedjs    zTestConversion.test_untypedc Cs|jd}|j}|jd|jtjddtjddtjdddtjdd tjdd tjddd tjdd f|jdS)z test timedelta conversion rzselect time('12:30'), time('23:12:59'), time('23:12:59.05100'), time('-12:30'), time('-23:12:59'), time('-23:12:59.05100'), time('-00:30')iȯi{Fi8iN)rrrr(r"r$r')r1r2r3r7r7r8test_timedeltass      zTestConversion.test_timedeltac Cs|jd}|j|dstd|j}tjdddd d d d }|jd z0|jd d |f|jd|j|f|jWd|jdXdS)z( test datetime conversion w microsecondsrr rrz,target backend does not support microsecondsi r i:z3create table test_datetime (id int, ts datetime(6))z)insert into test_datetime values (%s, %s)rzselect ts from test_datetimeNzdrop table test_datetime)r rr)rmysql_server_isrrr"rr(r')r1r2r3dtr7r7r8test_datetime_microsecondss     z)TestConversion.test_datetime_microsecondsN) __name__ __module__ __qualname__r9r<r>r?rArHrIrNr7r7r7r8rs'     c@s,eZdZddZddZddZddZd S) rc CsT|jd}|j}|jdz&d}|jd|f|jd|jWd|jdXdS)z test a fetchone() with no rows rz$create table test_nr (b varchar(32))pymysqlz#insert into test_nr (b) values (%s)Nzdrop table test_nr)rrrr(r')r1r2r3rFr7r7r8test_fetch_no_results  zTestCursor.test_fetch_no_resultc Cs~|jd}|j}zZ|jdx tddD]}|jd|fq*W|jd|j\}|jttdd|Wd|jdXdS)z test aggregate functions rz(create table test_aggregates (i integer)rz+insert into test_aggregates (i) values (%s)z"select sum(i) from test_aggregatesNzdrop table test_aggregates)rrrrDr'r(sum)r1r2r3ir5r7r7r8test_aggregatess    zTestCursor.test_aggregatescCsb|jd}|j}|j|dd|jd|jd|jdd |jd gt|j|jdS) z test a single tuple rZmystuffz-create table mystuff (id integer primary key)z#insert into mystuff (id) values (1)z#insert into mystuff (id) values (2)z%select id from mystuff where id in %srNr)rW)r)rrrErr(r-r0close)r1r2r3r7r7r8test_single_tuples    zTestCursor.test_single_tuplecCs|jdj}d|d<tjf|}|j|d s6td|j|dd|j}d }|jd |f|jd |j d}|j t j |t j ||jd |f|j d}|j t j |t j |dS)NrZutf8mb4r!r r z*JSON type is not supported on MySQL <= 5.6 test_jsonz]create table test_json ( id int not null, json JSON not null, primary key (id) );u{"hello": "こんにちは"}z2INSERT INTO test_json (id, `json`) values (42, %s)z,SELECT `json` from `test_json` WHERE `id`=42zSELECT CAST(%s AS JSON) AS x)r r r) Z databasescopyrRconnectrLrrErrr'r(jsonloads)r1argsr2curZjson_strresr7r7r8rZs      zTestCursor.test_jsonN)rOrPrQrSrVrYrZr7r7r7r8rs6  csTeZdZejjZfddZddZddZ ddZ d d Z d d Z d dZ ZS)rcs<tt|j|jd|_}|j|j}|j|dddS)NrZ bulkinsertz]CREATE TABLE bulkinsert ( id int(11), name char(20), age int, height int, PRIMARY KEY (id) ) )superrsetUprr2r cursor_typerE)r1r2r3) __class__r7r8rcs   zTestBulkInserts.setUpcCs<|jd}|j}|jd|j}|jt|t|dS)Nrz,SELECT id, name, age, height from bulkinsert)rrrr0r(sorted)r1rFr2rresultr7r7r8_verify_records!s   zTestBulkInserts._verify_recordscCsR|jd}|j}dddg}|jd ||j|jtd|jd|j|dS)Nrbob{rjim8rr freddzCinsert into bulkinsert (id, name, age, height) values (%s,%s,%s,%s)siinsert into bulkinsert (id, name, age, height) values (0,'bob',21,123),(1,'jim',56,45),(2,'fred',100,180)commit)rrirjrk)rrlrmr)r rnrorp)rr executemanyr(_last_executedrCrrh)r1r2rrFr7r7r8test_bulk_insert(s   z TestBulkInserts.test_bulk_insertcCsV|jd}|j}dddg}|jd ||j|jjtd|jd|j|dS)Nrrirjrkrrlrmrr rnrorpzJinsert into bulkinsert (id, name, age, height) values (%s, %s , %s, %s ) sxinsert into bulkinsert (id, name, age, height) values (0, 'bob' , 21, 123 ),(1, 'jim' , 56, 45 ),(2, 'fred' , 100, 180 )rq)rrirjrk)rrlrmr)r rnrorp) rrrrr(rsstriprCrrh)r1r2rrFr7r7r8$test_bulk_insert_multiline_statement6s    z4TestBulkInserts.test_bulk_insert_multiline_statementcCs<|jd}|j}dg}|jd||jd|j|dS)NrrirjrkzCinsert into bulkinsert (id, name, age, height) values (%s,%s,%s,%s)rq)rrirjrk)rrrrrrh)r1r2rrFr7r7r8test_bulk_insert_single_recordNs  z.TestBulkInserts.test_bulk_insert_single_recordcCsV|jd}|j}dddg}|jd ||j|jjtd|jd|j|dS)z4executemany should work with "insert ... on update" rrirjrkrrlrmrr rnrorpztinsert into bulkinsert (id, name, age, height) values (%s, %s , %s, %s ) on duplicate key update age = values(age) sinsert into bulkinsert (id, name, age, height) values (0, 'bob' , 21, 123 ),(1, 'jim' , 56, 45 ),(2, 'fred' , 100, 180 ) on duplicate key update age = values(age)rqN)rrirjrk)rrlrmr)r rnrorp) rrrrr(rsrurCrrh)r1r2rrFr7r7r8test_issue_288Ws    zTestBulkInserts.test_issue_288c Cs|jd}|j}tjdd}tjd|jdWdQRX|jt|d|j|djt j dt |dj kr|j dt |dj fdS) NrT)recordalwaysz$drop table if exists no_exists_tablerZno_exists_tablez'no_exists_table' not in %s)rrwarningscatch_warnings simplefilterrr(lencategoryrRWarningstrmessagefail)r1conr`wsr7r7r8 test_warningsrs  zTestBulkInserts.test_warnings)rOrPrQrRZcursorsZ DictCursorrdrcrhrtrvrwrxr __classcell__r7r7)rer8r s  )r"r]r%r{Z unittest2rrRrZpymysql.cursorsZ pymysql.testsrZ pymysql.errr__all__ZPyMySQLTestCaserrrr7r7r7r8s     w