1 from __future__ import print_function
8 from eventlet import event
10 from eventlet.green import MySQLdb
14 from tests import skip_unless, using_pyevent, get_database_auth
17 def mysql_requirement(_f):
18 """We want to skip tests if using pyevent, MySQLdb is not installed, or if
19 there is no database running on the localhost that the auth file grants
22 This errs on the side of skipping tests if everything is not right, but
23 it's better than a million tests failing when you don't care about mysql
28 print("Skipping mysql tests, MySQLdb not importable")
31 auth = get_database_auth()['MySQLdb'].copy()
32 MySQLdb.connect(**auth)
34 except MySQLdb.OperationalError:
35 print("Skipping mysql tests, error when connecting:")
40 class TestMySQLdb(tests.LimitedTestCase):
42 self._auth = get_database_auth()['MySQLdb']
44 self.connection = None
45 self.connection = MySQLdb.connect(**self._auth)
46 cursor = self.connection.cursor()
47 cursor.execute("""CREATE TABLE gargleblatz
51 self.connection.commit()
54 super(TestMySQLdb, self).setUp()
58 self.connection.close()
61 super(TestMySQLdb, self).tearDown()
63 @skip_unless(mysql_requirement)
65 auth = self._auth.copy()
70 dbname = 'test_%d_%d' % (os.getpid(), int(time.time() * 1000))
71 db = MySQLdb.connect(**auth).cursor()
72 db.execute("create database " + dbname)
74 self._auth['db'] = dbname
78 db = MySQLdb.connect(**self._auth).cursor()
79 db.execute("drop database " + self._auth['db'])
83 def set_up_dummy_table(self, connection=None):
84 close_connection = False
85 if connection is None:
86 close_connection = True
87 if self.connection is None:
88 connection = MySQLdb.connect(**self._auth)
90 connection = self.connection
92 cursor = connection.cursor()
93 cursor.execute(self.dummy_table_sql)
99 dummy_table_sql = """CREATE TEMPORARY TABLE test_table
101 row_id INTEGER PRIMARY KEY AUTO_INCREMENT,
104 value_string VARCHAR(200),
107 value_binary_string VARCHAR(200) BINARY,
108 value_enum ENUM('Y','N'),
112 def assert_cursor_yields(self, curs):
119 gt = eventlet.spawn(tick)
120 curs.execute("select 1")
121 rows = curs.fetchall()
122 self.assertEqual(len(rows), 1)
123 self.assertEqual(len(rows[0]), 1)
124 self.assertEqual(rows[0][0], 1)
125 assert counter[0] > 0, counter[0]
128 def assert_cursor_works(self, cursor):
129 cursor.execute("select 1")
130 rows = cursor.fetchall()
131 self.assertEqual(len(rows), 1)
132 self.assertEqual(len(rows[0]), 1)
133 self.assertEqual(rows[0][0], 1)
134 self.assert_cursor_yields(cursor)
136 def assert_connection_works(self, conn):
138 self.assert_cursor_works(curs)
140 def test_module_attributes(self):
141 import MySQLdb as orig
142 for key in dir(orig):
143 if key not in ('__author__', '__path__', '__revision__',
144 '__version__', '__loader__'):
145 assert hasattr(MySQLdb, key), "%s %s" % (key, getattr(orig, key))
147 def test_connecting(self):
148 assert self.connection is not None
150 def test_connecting_annoyingly(self):
151 self.assert_connection_works(MySQLdb.Connect(**self._auth))
152 self.assert_connection_works(MySQLdb.Connection(**self._auth))
153 self.assert_connection_works(MySQLdb.connections.Connection(**self._auth))
155 def test_create_cursor(self):
156 cursor = self.connection.cursor()
159 def test_run_query(self):
160 cursor = self.connection.cursor()
161 self.assert_cursor_works(cursor)
164 def test_run_bad_query(self):
165 cursor = self.connection.cursor()
167 cursor.execute("garbage blah blah")
169 except AssertionError:
175 def fill_up_table(self, conn):
177 for i in range(1000):
178 curs.execute('insert into test_table (value_int) values (%s)' % i)
181 def test_yields(self):
182 conn = self.connection
183 self.set_up_dummy_table(conn)
184 self.fill_up_table(conn)
187 SHORT_QUERY = "select * from test_table"
191 self.assert_cursor_works(curs)
192 curs.execute(SHORT_QUERY)
195 eventlet.spawn(a_query)
197 self.assertEqual([1], results)
199 self.assertEqual([1, 2], results)
201 def test_visibility_from_other_connections(self):
202 conn = MySQLdb.connect(**self._auth)
203 conn2 = MySQLdb.connect(**self._auth)
206 curs2 = conn2.cursor()
207 curs2.execute("insert into gargleblatz (a) values (%s)" % (314159))
208 self.assertEqual(curs2.rowcount, 1)
210 selection_query = "select * from gargleblatz"
211 curs2.execute(selection_query)
212 self.assertEqual(curs2.rowcount, 1)
214 # create a new connection, it should see the addition
215 conn3 = MySQLdb.connect(**self._auth)
216 curs3 = conn3.cursor()
217 curs3.execute(selection_query)
218 self.assertEqual(curs3.rowcount, 1)
219 # now, does the already-open connection see it?
220 curs.execute(selection_query)
221 self.assertEqual(curs.rowcount, 1)
225 curs.execute("delete from gargleblatz where a=314159")
229 class TestMonkeyPatch(tests.LimitedTestCase):
230 @skip_unless(mysql_requirement)
231 def test_monkey_patching(self):
232 tests.run_isolated('mysqldb_monkey_patch.py')