3316c6e9d1
API:
- write main test body into tests/isolated/filename.py
- it must write 'pass' to stdout or 'skip:[optional reason]'
- write a test with single line: tests.run_isolated('filename.py')
FIXME: autorun all files in tests/isolated
This deprecates tests.run_python and ProcessBase.
TODO: rewrite old multiline string test bodies to isolated files
TODO: add timeout to p.communicate() in run_python()
233 lines
7.1 KiB
Python
233 lines
7.1 KiB
Python
from __future__ import print_function
|
|
|
|
import os
|
|
import time
|
|
import traceback
|
|
|
|
import eventlet
|
|
from eventlet import event
|
|
try:
|
|
from eventlet.green import MySQLdb
|
|
except ImportError:
|
|
MySQLdb = False
|
|
import tests
|
|
from tests import skip_unless, using_pyevent, get_database_auth
|
|
|
|
|
|
def mysql_requirement(_f):
|
|
"""We want to skip tests if using pyevent, MySQLdb is not installed, or if
|
|
there is no database running on the localhost that the auth file grants
|
|
us access to.
|
|
|
|
This errs on the side of skipping tests if everything is not right, but
|
|
it's better than a million tests failing when you don't care about mysql
|
|
support."""
|
|
if using_pyevent(_f):
|
|
return False
|
|
if MySQLdb is False:
|
|
print("Skipping mysql tests, MySQLdb not importable")
|
|
return False
|
|
try:
|
|
auth = get_database_auth()['MySQLdb'].copy()
|
|
MySQLdb.connect(**auth)
|
|
return True
|
|
except MySQLdb.OperationalError:
|
|
print("Skipping mysql tests, error when connecting:")
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
class TestMySQLdb(tests.LimitedTestCase):
|
|
def setUp(self):
|
|
self._auth = get_database_auth()['MySQLdb']
|
|
self.create_db()
|
|
self.connection = None
|
|
self.connection = MySQLdb.connect(**self._auth)
|
|
cursor = self.connection.cursor()
|
|
cursor.execute("""CREATE TABLE gargleblatz
|
|
(
|
|
a INTEGER
|
|
);""")
|
|
self.connection.commit()
|
|
cursor.close()
|
|
|
|
super(TestMySQLdb, self).setUp()
|
|
|
|
def tearDown(self):
|
|
if self.connection:
|
|
self.connection.close()
|
|
self.drop_db()
|
|
|
|
super(TestMySQLdb, self).tearDown()
|
|
|
|
@skip_unless(mysql_requirement)
|
|
def create_db(self):
|
|
auth = self._auth.copy()
|
|
try:
|
|
self.drop_db()
|
|
except Exception:
|
|
pass
|
|
dbname = 'test_%d_%d' % (os.getpid(), int(time.time() * 1000))
|
|
db = MySQLdb.connect(**auth).cursor()
|
|
db.execute("create database " + dbname)
|
|
db.close()
|
|
self._auth['db'] = dbname
|
|
del db
|
|
|
|
def drop_db(self):
|
|
db = MySQLdb.connect(**self._auth).cursor()
|
|
db.execute("drop database " + self._auth['db'])
|
|
db.close()
|
|
del db
|
|
|
|
def set_up_dummy_table(self, connection=None):
|
|
close_connection = False
|
|
if connection is None:
|
|
close_connection = True
|
|
if self.connection is None:
|
|
connection = MySQLdb.connect(**self._auth)
|
|
else:
|
|
connection = self.connection
|
|
|
|
cursor = connection.cursor()
|
|
cursor.execute(self.dummy_table_sql)
|
|
connection.commit()
|
|
cursor.close()
|
|
if close_connection:
|
|
connection.close()
|
|
|
|
dummy_table_sql = """CREATE TEMPORARY TABLE test_table
|
|
(
|
|
row_id INTEGER PRIMARY KEY AUTO_INCREMENT,
|
|
value_int INTEGER,
|
|
value_float FLOAT,
|
|
value_string VARCHAR(200),
|
|
value_uuid CHAR(36),
|
|
value_binary BLOB,
|
|
value_binary_string VARCHAR(200) BINARY,
|
|
value_enum ENUM('Y','N'),
|
|
created TIMESTAMP
|
|
) ENGINE=InnoDB;"""
|
|
|
|
def assert_cursor_yields(self, curs):
|
|
counter = [0]
|
|
|
|
def tick():
|
|
while True:
|
|
counter[0] += 1
|
|
eventlet.sleep()
|
|
gt = eventlet.spawn(tick)
|
|
curs.execute("select 1")
|
|
rows = curs.fetchall()
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(len(rows[0]), 1)
|
|
self.assertEqual(rows[0][0], 1)
|
|
assert counter[0] > 0, counter[0]
|
|
gt.kill()
|
|
|
|
def assert_cursor_works(self, cursor):
|
|
cursor.execute("select 1")
|
|
rows = cursor.fetchall()
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(len(rows[0]), 1)
|
|
self.assertEqual(rows[0][0], 1)
|
|
self.assert_cursor_yields(cursor)
|
|
|
|
def assert_connection_works(self, conn):
|
|
curs = conn.cursor()
|
|
self.assert_cursor_works(curs)
|
|
|
|
def test_module_attributes(self):
|
|
import MySQLdb as orig
|
|
for key in dir(orig):
|
|
if key not in ('__author__', '__path__', '__revision__',
|
|
'__version__', '__loader__'):
|
|
assert hasattr(MySQLdb, key), "%s %s" % (key, getattr(orig, key))
|
|
|
|
def test_connecting(self):
|
|
assert self.connection is not None
|
|
|
|
def test_connecting_annoyingly(self):
|
|
self.assert_connection_works(MySQLdb.Connect(**self._auth))
|
|
self.assert_connection_works(MySQLdb.Connection(**self._auth))
|
|
self.assert_connection_works(MySQLdb.connections.Connection(**self._auth))
|
|
|
|
def test_create_cursor(self):
|
|
cursor = self.connection.cursor()
|
|
cursor.close()
|
|
|
|
def test_run_query(self):
|
|
cursor = self.connection.cursor()
|
|
self.assert_cursor_works(cursor)
|
|
cursor.close()
|
|
|
|
def test_run_bad_query(self):
|
|
cursor = self.connection.cursor()
|
|
try:
|
|
cursor.execute("garbage blah blah")
|
|
assert False
|
|
except AssertionError:
|
|
raise
|
|
except Exception:
|
|
pass
|
|
cursor.close()
|
|
|
|
def fill_up_table(self, conn):
|
|
curs = conn.cursor()
|
|
for i in range(1000):
|
|
curs.execute('insert into test_table (value_int) values (%s)' % i)
|
|
conn.commit()
|
|
|
|
def test_yields(self):
|
|
conn = self.connection
|
|
self.set_up_dummy_table(conn)
|
|
self.fill_up_table(conn)
|
|
curs = conn.cursor()
|
|
results = []
|
|
SHORT_QUERY = "select * from test_table"
|
|
evt = event.Event()
|
|
|
|
def a_query():
|
|
self.assert_cursor_works(curs)
|
|
curs.execute(SHORT_QUERY)
|
|
results.append(2)
|
|
evt.send()
|
|
eventlet.spawn(a_query)
|
|
results.append(1)
|
|
self.assertEqual([1], results)
|
|
evt.wait()
|
|
self.assertEqual([1, 2], results)
|
|
|
|
def test_visibility_from_other_connections(self):
|
|
conn = MySQLdb.connect(**self._auth)
|
|
conn2 = MySQLdb.connect(**self._auth)
|
|
curs = conn.cursor()
|
|
try:
|
|
curs2 = conn2.cursor()
|
|
curs2.execute("insert into gargleblatz (a) values (%s)" % (314159))
|
|
self.assertEqual(curs2.rowcount, 1)
|
|
conn2.commit()
|
|
selection_query = "select * from gargleblatz"
|
|
curs2.execute(selection_query)
|
|
self.assertEqual(curs2.rowcount, 1)
|
|
del curs2, conn2
|
|
# create a new connection, it should see the addition
|
|
conn3 = MySQLdb.connect(**self._auth)
|
|
curs3 = conn3.cursor()
|
|
curs3.execute(selection_query)
|
|
self.assertEqual(curs3.rowcount, 1)
|
|
# now, does the already-open connection see it?
|
|
curs.execute(selection_query)
|
|
self.assertEqual(curs.rowcount, 1)
|
|
del curs3, conn3
|
|
finally:
|
|
# clean up my litter
|
|
curs.execute("delete from gargleblatz where a=314159")
|
|
conn.commit()
|
|
|
|
|
|
class TestMonkeyPatch(tests.LimitedTestCase):
|
|
@skip_unless(mysql_requirement)
|
|
def test_monkey_patching(self):
|
|
tests.run_isolated('mysqldb_monkey_patch.py')
|