File: //opt/alt/python38/lib/python3.8/site-packages/pymysql/tests/__pycache__/test_basic.cpython-38.pyc
U
s_�We8 � @ s� d dl Z d dlZd dlZd dlZd dlmZ d dlmZ d dlZd dl m
Z
d dl m Z dddgZ
G d d� de
j�ZG d
d� de
j�ZG d d� de
j�ZdS ) � N)�SkipTest)�util)�base)�ProgrammingError�TestConversion�
TestCursor�TestBulkInsertsc @ sL e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dS )r c C s� | j d }|�� }|�d� �z�dddddd d
�|j�t�d d d �t�d
ddddd�t�dd�t�dd�t� � f }|�d|� |�d� |�
� }| � t �
d�|d � | � |dd� |dd� � | � t�dd|d jd |d j �|d � | � tj|d d d� � |d � |�d � |�dd gd � |�d� |�
� }| � td gd �|� |�d � ttttfD ]H}|�d � |d dg�}|�d |f� |�� }| � d!|� |�d � �qnW 5 |�d� X d S )"z test every data type r z�create 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)zdrop table test_datatypesT���l 2}r g������@z
hello'" worldu
Españolz binary datai� � i� � � � �- �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 i))� )� )� connections�cursor�execute�encode�charset�datetime�date� timedelta�time� localtime�fetchone� assertEqualr Zint2byteZhourZminute�tuple�list�set� frozenset�fetchall)�self�conn�c�v�rZseq_type�seq� r1 �/test_basic.py�test_datatypes s4
R
0"
z TestConversion.test_datatypesc C s` | j d }|�� }|�d� z2|�ddddd�� |�d � | �d
|�� � W 5 |�d� X d S ) z test dict escaping r z8create table test_dict (a integer, b integer, c integer)�drop table test_dictz:insert into test_dict (a,b,c) values (%(a)s, %(b)s, %(c)s)r r
� )�a�br- zselect a,b,c from test_dict)r r
r5 N�r r r r% r$ �r+ r, r- r1 r1 r2 � test_dict: s
zTestConversion.test_dictc C s^ | j d }|�� }|�d� d}z,|�d|� |�d� | �|f|�� � W 5 |�d� X d S )Nr z create table test_dict (a text)zI am a test stringr4 �%insert into test_dict (a) values (%s)�select a from test_dictr8 �r+ r, r- Z
test_valuer1 r1 r2 � test_stringF s
zTestConversion.test_stringc C s^ | j d }|�� }|�d� d}z,|�d|� |�d� | �|f|�� � W 5 |�d� X d S )Nr z"create table test_dict (a integer)i90 r4 r; r< r8 r= r1 r1 r2 � test_integerR s
zTestConversion.test_integerc C sp t ttd��d �}| jd }| �|dd� |�� �2}|�d|f� |�d� | �||�� d � W 5 Q R X dS ) ztest binary data� r r � test_blobz create table test_blob (b blob)z%insert into test_blob (b) values (%s)zselect b from test_blobN) �bytes� bytearray�ranger �safe_create_tabler r r% r$ )r+ �datar, r- r1 r1 r2 rA ^ s
�
zTestConversion.test_blobc C sJ | j d }|�� }|�d� | �d|�� � |�d� | �d|�� � dS )z' test conversion of null, empty string r zselect null,'')N� zselect '',null)rG NNr8 r9 r1 r1 r2 � test_untypedj s
zTestConversion.test_untypedc C s� | j d }|�� }|�d� | �t�dd�t�dd�t�ddd�t�dd� t�dd� t�ddd� t�dd� f|�� � dS )z test timedelta conversion r z�select 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{F i8� i N)r r r r% r r! r$ r9 r1 r1 r2 �test_timedeltas s
��z TestConversion.test_timedeltac C s� | j d }| �|d�s td��|�� }t�ddddddd�}|�d � z0|�d d |f� |�d
� | �|f|�� � W 5 |�d
� X dS )z( test datetime conversion w microsecondsr )r r r z,target backend does not support microsecondsi� � r � i:� z3create table test_datetime (id int, ts datetime(6))zdrop table test_datetimez)insert into test_datetime values (%s, %s)r z select ts from test_datetimeN)r �mysql_server_isr r r r r% r$ )r+ r, r- Zdtr1 r1 r2 �test_datetime_microseconds� s
�
z)TestConversion.test_datetime_microsecondsN) �__name__�
__module__� __qualname__r3 r: r> r? rA rH rI rM r1 r1 r1 r2 r s ' c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S )
r c C sT | j d }|�� }|�d� z&d}|�d|f� | �d|�� � W 5 |�d� X dS )z test a fetchone() with no rows r z$create table test_nr (b varchar(32))zdrop table test_nr�pymysqlz#insert into test_nr (b) values (%s)Nr8 )r+ r, r- rF r1 r1 r2 �test_fetch_no_result� s
z TestCursor.test_fetch_no_resultc C sz | j d }|�� }zV|�d� tdd�D ]}|�d|f� q(|�d� |�� \}| �ttdd��|� W 5 |�d� X dS )z test aggregate functions r zdrop table test_aggregatesz(create table test_aggregates (i integer)r z+insert into test_aggregates (i) values (%s)z"select sum(i) from test_aggregatesN)r r r rD r$ r% �sum)r+ r, r- �ir/ r1 r1 r2 �test_aggregates� s
zTestCursor.test_aggregatesc C sb | j d }|�� }| �|dd� |�d� |�d� |�dd� | �dgt|�� �� |�� d S )
z test a single tuple r Zmystuffz-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 %s)�r rV N)r r rE r r% r' r* �closer9 r1 r1 r2 �test_single_tuple� s
�
z TestCursor.test_single_tuplec C s� | j d �� }d|d< tjf |�}| �|d�s6td��| �|dd� |�� }d}|�d |f� |�d
� |� � d }| �
t � |�t � |�� |�d |f� |� � d }| �
t � |�t � |�� d S ) Nr Zutf8mb4r )r 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`=42z SELECT CAST(%s AS JSON) AS x)
Z databases�copyrQ ZconnectrL r rE r r r$ r% �json�loads)r+ �argsr, �curZjson_str�resr1 r1 r2 rY � s
zTestCursor.test_jsonN)rN rO rP rR rU rX rY r1 r1 r1 r2 r � s 6
c sT e Zd ZejjZ� fdd�Zdd� Zdd� Z dd� Z
d d
� Z d d � Z d
d� Z
� ZS )r c s< t t| ��� | jd | _}|�| j�}| �|dd� d S )Nr Z
bulkinsertz]CREATE TABLE bulkinsert
(
id int(11),
name char(20),
age int,
height int,
PRIMARY KEY (id)
)
)�superr �setUpr r, r � cursor_typerE r9 �� __class__r1 r2 ra s zTestBulkInserts.setUpc C s< | j d }|�� }|�d� |�� }| �t|�t|�� d S )Nr z,SELECT id, name, age, height from bulkinsert)r r r r* r% �sorted)r+ rF r, r �resultr1 r1 r2 �_verify_records! s
z TestBulkInserts._verify_recordsc C sR | j d }|�� }dddg}|�d|� | �|jtd�� |�d� | �|� d S )Nr �r Zbob� �{ �r Zjim�8 r �r
Zfred�d � �Cinsert into bulkinsert (id, name, age, height) values (%s,%s,%s,%s)si insert into bulkinsert (id, name, age, height) values (0,'bob',21,123),(1,'jim',56,45),(2,'fred',100,180)�commit)r r � executemanyr% �_last_executedrC r rg �r+ r, r rF r1 r1 r2 �test_bulk_insert( s
� ��
z TestBulkInserts.test_bulk_insertc C sV | j d }|�� }dddg}|�d|� | �|j�� td�� |�d� | �|� d S )Nr rh rk rm zJinsert
into bulkinsert (id, name,
age, height)
values (%s,
%s , %s,
%s )
sx insert
into bulkinsert (id, name,
age, height)
values (0,
'bob' , 21,
123 ),(1,
'jim' , 56,
45 ),(2,
'fred' , 100,
180 )rq � r r rr r% rs �striprC r rg rt r1 r1 r2 �$test_bulk_insert_multiline_statement6 s
�
z4TestBulkInserts.test_bulk_insert_multiline_statementc C s< | j d }|�� }dg}|�d|� |�d� | �|� d S )Nr rh rp rq )r r rr r rg rt r1 r1 r2 � test_bulk_insert_single_recordN s
�
z.TestBulkInserts.test_bulk_insert_single_recordc C sV | j d }|�� }dddg}|�d|� | �|j�� td�� |�d� | �|� dS ) z4executemany should work with "insert ... on update" r rh rk rm ztinsert
into bulkinsert (id, name,
age, height)
values (%s,
%s , %s,
%s ) on duplicate key update
age = values(age)
s� insert
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)rq Nrv rt r1 r1 r2 �test_issue_288W s
�
z TestBulkInserts.test_issue_288c C s� | j d }|�� }tjdd��}t�d� |�d� W 5 Q R X | �t|�d� | �|d jt j
� dt |d j �kr�| �
dt |d j �f � d S ) Nr T)�record�alwaysz$drop table if exists no_exists_tabler Zno_exists_tablez'no_exists_table' not in %s)r r �warnings�catch_warnings� simplefilterr r% �len�categoryrQ �Warning�str�messageZfail)r+ Zconr^ Zwsr1 r1 r2 �
test_warningsr s
z TestBulkInserts.test_warnings)rN rO rP rQ ZcursorsZ
DictCursorrb ra rg ru rx ry rz r� �
__classcell__r1 r1 rc r2 r s )r r[ r" r} Z unittest2r rQ r Zpymysql.cursorsZ
pymysql.testsr Z pymysql.errr �__all__ZPyMySQLTestCaser r r r1 r1 r1 r2 �<module> s
w