1 from __future__ import print_function
8 from eventlet import event
12 skip_unless, using_pyevent, get_database_auth,
15 from eventlet.green import MySQLdb
20 def mysql_requirement(_f):
21 """We want to skip tests if using pyevent, MySQLdb is not installed, or if
22 there is no database running on the localhost that the auth file grants
25 This errs on the side of skipping tests if everything is not right, but
26 it's better than a million tests failing when you don't care about mysql
31 print("Skipping mysql tests, MySQLdb not importable")
34 auth = get_database_auth()['MySQLdb'].copy()
35 MySQLdb.connect(**auth)
37 except MySQLdb.OperationalError:
38 print("Skipping mysql tests, error when connecting:")
43 class TestMySQLdb(LimitedTestCase):
45 self._auth = get_database_auth()['MySQLdb']
47 self.connection = None
48 self.connection = MySQLdb.connect(**self._auth)
49 cursor = self.connection.cursor()
50 cursor.execute("""CREATE TABLE gargleblatz
54 self.connection.commit()
57 super(TestMySQLdb, self).setUp()
61 self.connection.close()
64 super(TestMySQLdb, self).tearDown()
66 @skip_unless(mysql_requirement)
68 auth = self._auth.copy()
73 dbname = 'test_%d_%d' % (os.getpid(), int(time.time() * 1000))
74 db = MySQLdb.connect(**auth).cursor()
75 db.execute("create database " + dbname)
77 self._auth['db'] = dbname
81 db = MySQLdb.connect(**self._auth).cursor()
82 db.execute("drop database " + self._auth['db'])
86 def set_up_dummy_table(self, connection=None):
87 close_connection = False
88 if connection is None:
89 close_connection = True
90 if self.connection is None:
91 connection = MySQLdb.connect(**self._auth)
93 connection = self.connection
95 cursor = connection.cursor()
96 cursor.execute(self.dummy_table_sql)
102 dummy_table_sql = """CREATE TEMPORARY TABLE test_table
104 row_id INTEGER PRIMARY KEY AUTO_INCREMENT,
107 value_string VARCHAR(200),
110 value_binary_string VARCHAR(200) BINARY,
111 value_enum ENUM('Y','N'),
115 def assert_cursor_yields(self, curs):
122 gt = eventlet.spawn(tick)
123 curs.execute("select 1")
124 rows = curs.fetchall()
125 self.assertEqual(len(rows), 1)
126 self.assertEqual(len(rows[0]), 1)
127 self.assertEqual(rows[0][0], 1)
128 assert counter[0] > 0, counter[0]
131 def assert_cursor_works(self, cursor):
132 cursor.execute("select 1")
133 rows = cursor.fetchall()
134 self.assertEqual(len(rows), 1)
135 self.assertEqual(len(rows[0]), 1)
136 self.assertEqual(rows[0][0], 1)
137 self.assert_cursor_yields(cursor)
139 def assert_connection_works(self, conn):
141 self.assert_cursor_works(curs)
143 def test_module_attributes(self):
144 import MySQLdb as orig
145 for key in dir(orig):
146 if key not in ('__author__', '__path__', '__revision__',
147 '__version__', '__loader__'):
148 assert hasattr(MySQLdb, key), "%s %s" % (key, getattr(orig, key))
150 def test_connecting(self):
151 assert self.connection is not None
153 def test_connecting_annoyingly(self):
154 self.assert_connection_works(MySQLdb.Connect(**self._auth))
155 self.assert_connection_works(MySQLdb.Connection(**self._auth))
156 self.assert_connection_works(MySQLdb.connections.Connection(**self._auth))
158 def test_create_cursor(self):
159 cursor = self.connection.cursor()
162 def test_run_query(self):
163 cursor = self.connection.cursor()
164 self.assert_cursor_works(cursor)
167 def test_run_bad_query(self):
168 cursor = self.connection.cursor()
170 cursor.execute("garbage blah blah")
172 except AssertionError:
178 def fill_up_table(self, conn):
180 for i in range(1000):
181 curs.execute('insert into test_table (value_int) values (%s)' % i)
184 def test_yields(self):
185 conn = self.connection
186 self.set_up_dummy_table(conn)
187 self.fill_up_table(conn)
190 SHORT_QUERY = "select * from test_table"
194 self.assert_cursor_works(curs)
195 curs.execute(SHORT_QUERY)
198 eventlet.spawn(a_query)
200 self.assertEqual([1], results)
202 self.assertEqual([1, 2], results)
204 def test_visibility_from_other_connections(self):
205 conn = MySQLdb.connect(**self._auth)
206 conn2 = MySQLdb.connect(**self._auth)
209 curs2 = conn2.cursor()
210 curs2.execute("insert into gargleblatz (a) values (%s)" % (314159))
211 self.assertEqual(curs2.rowcount, 1)
213 selection_query = "select * from gargleblatz"
214 curs2.execute(selection_query)
215 self.assertEqual(curs2.rowcount, 1)
217 # create a new connection, it should see the addition
218 conn3 = MySQLdb.connect(**self._auth)
219 curs3 = conn3.cursor()
220 curs3.execute(selection_query)
221 self.assertEqual(curs3.rowcount, 1)
222 # now, does the already-open connection see it?
223 curs.execute(selection_query)
224 self.assertEqual(curs.rowcount, 1)
228 curs.execute("delete from gargleblatz where a=314159")
232 class TestMonkeyPatch(LimitedTestCase):
233 @skip_unless(mysql_requirement)
234 def test_monkey_patching(self):
235 testcode_path = os.path.join(
236 os.path.dirname(os.path.abspath(__file__)),
237 'mysqldb_test_monkey_patch.py',
239 output = run_python(testcode_path)
240 lines = output.splitlines()
241 self.assertEqual(len(lines), 2, output)
242 self.assertEqual(lines[0].replace("psycopg,", ""),
243 'mysqltest MySQLdb,os,select,socket,thread,time')
244 self.assertEqual(lines[1], "connect True")