openstack/deb-python-pymysql
Code Issues Proposed changes

Make safe_create_table() is usable from many tests.

Browse Source
This commit is contained in:
INADA Naoki
2015-02-04 15:59:52 +09:00
parent 1a6cddb6ec
commit a4a460c051
3 changed files with 16 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pymysql/tests/base.py
View File

@@ -50,23 +50,20 @@ class PyMySQLTestCase(unittest2.TestCase):
for connection in self.connections: for connection in self.connections:
connection.close() connection.close()
def safe_create_table(self, connection, tablename, ddl, cleanup=False): def safe_create_table(self, connection, tablename, ddl, cleanup=True):
"""create a table. """create a table.
Ensures any existing version of that table Ensures any existing version of that table is first dropped.
is first dropped.
Also adds a cleanup rule to drop the table after the test Also adds a cleanup rule to drop the table after the test
completes. completes.
""" """
cursor = connection.cursor() cursor = connection.cursor()
with warnings.catch_warnings(): with warnings.catch_warnings():
warnings.simplefilter("ignore") warnings.simplefilter("ignore")
cursor.execute("drop table if exists test") cursor.execute("drop table if exists `%s`" % (tablename,))
cursor.execute("create table test (data varchar(10))") cursor.execute(ddl)
cursor.close() cursor.close()
if cleanup: if cleanup:
self.addCleanup(self.drop_table, connection, tablename) self.addCleanup(self.drop_table, connection, tablename)
@@ -75,7 +72,7 @@ class PyMySQLTestCase(unittest2.TestCase):
cursor = connection.cursor() cursor = connection.cursor()
with warnings.catch_warnings(): with warnings.catch_warnings():
warnings.simplefilter("ignore") warnings.simplefilter("ignore")
cursor.execute("drop table if exists %s" % tablename) cursor.execute("drop table if exists `%s`" % (tablename,))
cursor.close() cursor.close()
def safe_gc_collect(self): def safe_gc_collect(self):

23
pymysql/tests/test_basic.py
View File

@@ -227,14 +227,14 @@ class TestCursor(base.PyMySQLTestCase):
""" test a single tuple """ """ test a single tuple """
conn = self.connections[0] conn = self.connections[0]
c = conn.cursor() c = conn.cursor()
try: self.safe_create_table(
c.execute("create table mystuff (id integer primary key)") conn, 'mystuff',
c.execute("insert into mystuff (id) values (1)") "create table mystuff (id integer primary key)")
c.execute("insert into mystuff (id) values (2)") c.execute("insert into mystuff (id) values (1)")
c.execute("select id from mystuff where id in %s", ((1,),)) c.execute("insert into mystuff (id) values (2)")
self.assertEqual([(1,)], list(c.fetchall())) c.execute("select id from mystuff where id in %s", ((1,),))
finally: self.assertEqual([(1,)], list(c.fetchall()))
c.execute("drop table mystuff") c.close()
class TestBulkInserts(base.PyMySQLTestCase): class TestBulkInserts(base.PyMySQLTestCase):
@@ -247,11 +247,8 @@ class TestBulkInserts(base.PyMySQLTestCase):
c = conn.cursor(self.cursor_type) c = conn.cursor(self.cursor_type)
# create a table ane some data to query # create a table ane some data to query
with warnings.catch_warnings(): self.safe_create_table(conn, 'bulkinsert', """\
warnings.simplefilter("ignore") CREATE TABLE bulkinsert
c.execute("drop table if exists bulkinsert")
c.execute(
"""CREATE TABLE bulkinsert
( (
id int(11), id int(11),
name char(20), name char(20),

2
pymysql/tests/test_cursor.py
View File

@@ -11,7 +11,7 @@ class CursorTest(base.PyMySQLTestCase):
self.safe_create_table( self.safe_create_table(
conn, conn,
"test", "create table test (data varchar(10))", "test", "create table test (data varchar(10))",
cleanup=True) )
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute( cursor.execute(
"insert into test (data) values " "insert into test (data) values "