Edit File: test_basic.py
# coding: utf-8 import datetime import json import time import warnings from unittest2 import SkipTest from pymysql import util import pymysql.cursors from pymysql.tests import base from pymysql.err import ProgrammingError __all__ = ["TestConversion", "TestCursor", "TestBulkInserts"] class TestConversion(base.PyMySQLTestCase): def test_datatypes(self): """ test every data type """ conn = self.connections[0] c = conn.cursor() c.execute("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)") try: # insert values v = (True, -3, 123456789012, 5.7, "hello'\" world", u"Espa\xc3\xb1ol", "binary\x00data".encode(conn.charset), datetime.date(1988,2,2), datetime.datetime(2014, 5, 15, 7, 45, 57), datetime.timedelta(5,6), datetime.time(16,32), time.localtime()) c.execute("insert 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)", v) c.execute("select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes") r = c.fetchone() self.assertEqual(util.int2byte(1), r[0]) self.assertEqual(v[1:10], r[1:10]) self.assertEqual(datetime.timedelta(0, 60 * (v[10].hour * 60 + v[10].minute)), r[10]) self.assertEqual(datetime.datetime(*v[-1][:6]), r[-1]) c.execute("delete from test_datatypes") # check nulls c.execute("insert 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)", [None] * 12) c.execute("select b,i,l,f,s,u,bb,d,dt,td,t,st from test_datatypes") r = c.fetchone() self.assertEqual(tuple([None] * 12), r) c.execute("delete from test_datatypes") # check sequences type for seq_type in (tuple, list, set, frozenset): c.execute("insert into test_datatypes (i, l) values (2,4), (6,8), (10,12)") seq = seq_type([2,6]) c.execute("select l from test_datatypes where i in %s order by i", (seq,)) r = c.fetchall() self.assertEqual(((4,),(8,)), r) c.execute("delete from test_datatypes") finally: c.execute("drop table test_datatypes") def test_dict(self): """ test dict escaping """ conn = self.connections[0] c = conn.cursor() c.execute("create table test_dict (a integer, b integer, c integer)") try: c.execute("insert into test_dict (a,b,c) values (%(a)s, %(b)s, %(c)s)", {"a":1,"b":2,"c":3}) c.execute("select a,b,c from test_dict") self.assertEqual((1,2,3), c.fetchone()) finally: c.execute("drop table test_dict") def test_string(self): conn = self.connections[0] c = conn.cursor() c.execute("create table test_dict (a text)") test_value = "I am a test string" try: c.execute("insert into test_dict (a) values (%s)", test_value) c.execute("select a from test_dict") self.assertEqual((test_value,), c.fetchone()) finally: c.execute("drop table test_dict") def test_integer(self): conn = self.connections[0] c = conn.cursor() c.execute("create table test_dict (a integer)") test_value = 12345 try: c.execute("insert into test_dict (a) values (%s)", test_value) c.execute("select a from test_dict") self.assertEqual((test_value,), c.fetchone()) finally: c.execute("drop table test_dict") def test_blob(self): """test binary data""" data = bytes(bytearray(range(256)) * 4) conn = self.connections[0] self.safe_create_table( conn, "test_blob", "create table test_blob (b blob)") with conn.cursor() as c: c.execute("insert into test_blob (b) values (%s)", (data,)) c.execute("select b from test_blob") self.assertEqual(data, c.fetchone()[0]) def test_untyped(self): """ test conversion of null, empty string """ conn = self.connections[0] c = conn.cursor() c.execute("select null,''") self.assertEqual((None,u''), c.fetchone()) c.execute("select '',null") self.assertEqual((u'',None), c.fetchone()) def test_timedelta(self): """ test timedelta conversion """ conn = self.connections[0] c = conn.cursor() c.execute("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')") self.assertEqual((datetime.timedelta(0, 45000), datetime.timedelta(0, 83579), datetime.timedelta(0, 83579, 51000), -datetime.timedelta(0, 45000), -datetime.timedelta(0, 83579), -datetime.timedelta(0, 83579, 51000), -datetime.timedelta(0, 1800)), c.fetchone()) def test_datetime_microseconds(self): """ test datetime conversion w microseconds""" conn = self.connections[0] if not self.mysql_server_is(conn, (5, 6, 4)): raise SkipTest("target backend does not support microseconds") c = conn.cursor() dt = datetime.datetime(2013, 11, 12, 9, 9, 9, 123450) c.execute("create table test_datetime (id int, ts datetime(6))") try: c.execute( "insert into test_datetime values (%s, %s)", (1, dt) ) c.execute("select ts from test_datetime") self.assertEqual((dt,), c.fetchone()) finally: c.execute("drop table test_datetime") class TestCursor(base.PyMySQLTestCase): # this test case does not work quite right yet, however, # we substitute in None for the erroneous field which is # compatible with the DB-API 2.0 spec and has not broken # any unit tests for anything we've tried. #def test_description(self): # """ test description attribute """ # # result is from MySQLdb module # r = (('Host', 254, 11, 60, 60, 0, 0), # ('User', 254, 16, 16, 16, 0, 0), # ('Password', 254, 41, 41, 41, 0, 0), # ('Select_priv', 254, 1, 1, 1, 0, 0), # ('Insert_priv', 254, 1, 1, 1, 0, 0), # ('Update_priv', 254, 1, 1, 1, 0, 0), # ('Delete_priv', 254, 1, 1, 1, 0, 0), # ('Create_priv', 254, 1, 1, 1, 0, 0), # ('Drop_priv', 254, 1, 1, 1, 0, 0), # ('Reload_priv', 254, 1, 1, 1, 0, 0), # ('Shutdown_priv', 254, 1, 1, 1, 0, 0), # ('Process_priv', 254, 1, 1, 1, 0, 0), # ('File_priv', 254, 1, 1, 1, 0, 0), # ('Grant_priv', 254, 1, 1, 1, 0, 0), # ('References_priv', 254, 1, 1, 1, 0, 0), # ('Index_priv', 254, 1, 1, 1, 0, 0), # ('Alter_priv', 254, 1, 1, 1, 0, 0), # ('Show_db_priv', 254, 1, 1, 1, 0, 0), # ('Super_priv', 254, 1, 1, 1, 0, 0), # ('Create_tmp_table_priv', 254, 1, 1, 1, 0, 0), # ('Lock_tables_priv', 254, 1, 1, 1, 0, 0), # ('Execute_priv', 254, 1, 1, 1, 0, 0), # ('Repl_slave_priv', 254, 1, 1, 1, 0, 0), # ('Repl_client_priv', 254, 1, 1, 1, 0, 0), # ('Create_view_priv', 254, 1, 1, 1, 0, 0), # ('Show_view_priv', 254, 1, 1, 1, 0, 0), # ('Create_routine_priv', 254, 1, 1, 1, 0, 0), # ('Alter_routine_priv', 254, 1, 1, 1, 0, 0), # ('Create_user_priv', 254, 1, 1, 1, 0, 0), # ('Event_priv', 254, 1, 1, 1, 0, 0), # ('Trigger_priv', 254, 1, 1, 1, 0, 0), # ('ssl_type', 254, 0, 9, 9, 0, 0), # ('ssl_cipher', 252, 0, 65535, 65535, 0, 0), # ('x509_issuer', 252, 0, 65535, 65535, 0, 0), # ('x509_subject', 252, 0, 65535, 65535, 0, 0), # ('max_questions', 3, 1, 11, 11, 0, 0), # ('max_updates', 3, 1, 11, 11, 0, 0), # ('max_connections', 3, 1, 11, 11, 0, 0), # ('max_user_connections', 3, 1, 11, 11, 0, 0)) # conn = self.connections[0] # c = conn.cursor() # c.execute("select * from mysql.user") # # self.assertEqual(r, c.description) def test_fetch_no_result(self): """ test a fetchone() with no rows """ conn = self.connections[0] c = conn.cursor() c.execute("create table test_nr (b varchar(32))") try: data = "pymysql" c.execute("insert into test_nr (b) values (%s)", (data,)) self.assertEqual(None, c.fetchone()) finally: c.execute("drop table test_nr") def test_aggregates(self): """ test aggregate functions """ conn = self.connections[0] c = conn.cursor() try: c.execute('create table test_aggregates (i integer)') for i in range(0, 10): c.execute('insert into test_aggregates (i) values (%s)', (i,)) c.execute('select sum(i) from test_aggregates') r, = c.fetchone() self.assertEqual(sum(range(0,10)), r) finally: c.execute('drop table test_aggregates') def test_single_tuple(self): """ test a single tuple """ conn = self.connections[0] c = conn.cursor() self.safe_create_table( conn, 'mystuff', "create table mystuff (id integer primary key)") c.execute("insert into mystuff (id) values (1)") c.execute("insert into mystuff (id) values (2)") c.execute("select id from mystuff where id in %s", ((1,),)) self.assertEqual([(1,)], list(c.fetchall())) c.close() def test_json(self): args = self.databases[0].copy() args["charset"] = "utf8mb4" conn = pymysql.connect(**args) if not self.mysql_server_is(conn, (5, 7, 0)): raise SkipTest("JSON type is not supported on MySQL <= 5.6") self.safe_create_table(conn, "test_json", """\ create table test_json ( id int not null, json JSON not null, primary key (id) );""") cur = conn.cursor() json_str = u'{"hello": "こんにちは"}' cur.execute("INSERT INTO test_json (id, `json`) values (42, %s)", (json_str,)) cur.execute("SELECT `json` from `test_json` WHERE `id`=42") res = cur.fetchone()[0] self.assertEqual(json.loads(res), json.loads(json_str)) cur.execute("SELECT CAST(%s AS JSON) AS x", (json_str,)) res = cur.fetchone()[0] self.assertEqual(json.loads(res), json.loads(json_str)) class TestBulkInserts(base.PyMySQLTestCase): cursor_type = pymysql.cursors.DictCursor def setUp(self): super(TestBulkInserts, self).setUp() self.conn = conn = self.connections[0] c = conn.cursor(self.cursor_type) # create a table ane some data to query self.safe_create_table(conn, 'bulkinsert', """\ CREATE TABLE bulkinsert ( id int(11), name char(20), age int, height int, PRIMARY KEY (id) ) """) def _verify_records(self, data): conn = self.connections[0] cursor = conn.cursor() cursor.execute("SELECT id, name, age, height from bulkinsert") result = cursor.fetchall() self.assertEqual(sorted(data), sorted(result)) def test_bulk_insert(self): conn = self.connections[0] cursor = conn.cursor() data = [(0, "bob", 21, 123), (1, "jim", 56, 45), (2, "fred", 100, 180)] cursor.executemany("insert into bulkinsert (id, name, age, height) " "values (%s,%s,%s,%s)", data) self.assertEqual( cursor._last_executed, bytearray( b"insert into bulkinsert (id, name, age, height) values " b"(0,'bob',21,123),(1,'jim',56,45),(2,'fred',100,180)")) cursor.execute('commit') self._verify_records(data) def test_bulk_insert_multiline_statement(self): conn = self.connections[0] cursor = conn.cursor() data = [(0, "bob", 21, 123), (1, "jim", 56, 45), (2, "fred", 100, 180)] cursor.executemany("""insert into bulkinsert (id, name, age, height) values (%s, %s , %s, %s ) """, data) self.assertEqual(cursor._last_executed.strip(), bytearray(b"""insert into bulkinsert (id, name, age, height) values (0, 'bob' , 21, 123 ),(1, 'jim' , 56, 45 ),(2, 'fred' , 100, 180 )""")) cursor.execute('commit') self._verify_records(data) def test_bulk_insert_single_record(self): conn = self.connections[0] cursor = conn.cursor() data = [(0, "bob", 21, 123)] cursor.executemany("insert into bulkinsert (id, name, age, height) " "values (%s,%s,%s,%s)", data) cursor.execute('commit') self._verify_records(data) def test_issue_288(self): """executemany should work with "insert ... on update" """ conn = self.connections[0] cursor = conn.cursor() data = [(0, "bob", 21, 123), (1, "jim", 56, 45), (2, "fred", 100, 180)] cursor.executemany("""insert into bulkinsert (id, name, age, height) values (%s, %s , %s, %s ) on duplicate key update age = values(age) """, data) self.assertEqual(cursor._last_executed.strip(), bytearray(b"""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)""")) cursor.execute('commit') self._verify_records(data) def test_warnings(self): con = self.connections[0] cur = con.cursor() with warnings.catch_warnings(record=True) as ws: warnings.simplefilter("always") cur.execute("drop table if exists no_exists_table") self.assertEqual(len(ws), 1) self.assertEqual(ws[0].category, pymysql.Warning) if u"no_exists_table" not in str(ws[0].message): self.fail("'no_exists_table' not in %s" % (str(ws[0].message),))