← Back to branch summary

~ubuntu-branches/ubuntu/utopic/python-eventlet/utopic

« back to all changes in this revision

Viewing changes to .pc/debian-changes-0.9.15-0ubuntu1/tests/mysqldb_test.py

  • Committer: Bazaar Package Importer
  • Author(s): Soren Hansen
  • Date: 2011-05-25 22:02:42 UTC
  • Revision ID: james.westby@ubuntu.com-20110525220242-kpc3wbq83gxo4too
Tags: 0.9.15-0ubuntu2
* retry-on-timeout.patch: This addresses a problem where a timeout fires
  even though the connection was actually correctly established. (LP:
  #771512)
* reuseaddr.patch: Make sure SO_REUSEADDR is set.
* mysql-import-traceback.patch: Add missing import in the mysql tests.
* Add python-greenlet dependency.

Show diffs side-by-side

added added

removed removed

Lines of Context:
.pc/debian-changes-0.9.15-0ubuntu1/tests/mysqldb_test.py
1
 
import os
2
 
import sys
3
 
import time
4
 
from tests import skipped, skip_unless, using_pyevent, get_database_auth, LimitedTestCase
5
 
import eventlet
6
 
from eventlet import event
7
 
try:
8
 
    from eventlet.green import MySQLdb
9
 
except ImportError:
10
 
    MySQLdb = False
11
 
 
12
 
def mysql_requirement(_f):
13
 
    """We want to skip tests if using pyevent, MySQLdb is not installed, or if
14
 
    there is no database running on the localhost that the auth file grants
15
 
    us access to.
16
 
    
17
 
    This errs on the side of skipping tests if everything is not right, but
18
 
    it's better than a million tests failing when you don't care about mysql
19
 
    support."""
20
 
    if using_pyevent(_f):
21
 
        return False
22
 
    if MySQLdb is False:
23
 
        print "Skipping mysql tests, MySQLdb not importable"
24
 
        return False
25
 
    try:
26
 
        auth = get_database_auth()['MySQLdb'].copy()
27
 
        MySQLdb.connect(**auth)
28
 
        return True
29
 
    except MySQLdb.OperationalError:
30
 
        print "Skipping mysql tests, error when connecting:"
31
 
        traceback.print_exc()
32
 
        return False
33
 
 
34
 
class MySQLdbTester(LimitedTestCase):
35
 
    def setUp(self):
36
 
        self._auth = get_database_auth()['MySQLdb']
37
 
        self.create_db()
38
 
        self.connection = None
39
 
        self.connection = MySQLdb.connect(**self._auth)
40
 
        cursor = self.connection.cursor()
41
 
        cursor.execute("""CREATE TABLE gargleblatz
42
 
        (
43
 
        a INTEGER
44
 
        );""")
45
 
        self.connection.commit()
46
 
        cursor.close()
47
 
 
48
 
    def tearDown(self):
49
 
        if self.connection:
50
 
            self.connection.close()
51
 
        self.drop_db()
52
 
 
53
 
    @skip_unless(mysql_requirement)    
54
 
    def create_db(self):
55
 
        auth = self._auth.copy()
56
 
        try:
57
 
            self.drop_db()
58
 
        except Exception:
59
 
            pass
60
 
        dbname = 'test_%d_%d' % (os.getpid(), int(time.time()*1000))
61
 
        db = MySQLdb.connect(**auth).cursor()
62
 
        db.execute("create database "+dbname)
63
 
        db.close()
64
 
        self._auth['db'] = dbname
65
 
        del db
66
 
 
67
 
    def drop_db(self):
68
 
        db = MySQLdb.connect(**self._auth).cursor()
69
 
        db.execute("drop database "+self._auth['db'])
70
 
        db.close()
71
 
        del db
72
 
 
73
 
    def set_up_dummy_table(self, connection=None):
74
 
        close_connection = False
75
 
        if connection is None:
76
 
            close_connection = True
77
 
            if self.connection is None:
78
 
                connection = MySQLdb.connect(**self._auth)
79
 
            else:
80
 
                connection = self.connection
81
 
 
82
 
        cursor = connection.cursor()
83
 
        cursor.execute(self.dummy_table_sql)
84
 
        connection.commit()
85
 
        cursor.close()
86
 
        if close_connection:
87
 
            connection.close()
88
 
 
89
 
    dummy_table_sql = """CREATE TEMPORARY TABLE test_table
90
 
        (
91
 
        row_id INTEGER PRIMARY KEY AUTO_INCREMENT,
92
 
        value_int INTEGER,
93
 
        value_float FLOAT,
94
 
        value_string VARCHAR(200),
95
 
        value_uuid CHAR(36),
96
 
        value_binary BLOB,
97
 
        value_binary_string VARCHAR(200) BINARY,
98
 
        value_enum ENUM('Y','N'),
99
 
        created TIMESTAMP
100
 
        ) ENGINE=InnoDB;"""
101
 
 
102
 
    def assert_cursor_yields(self, curs):
103
 
        counter = [0]
104
 
        def tick():
105
 
            while True:
106
 
                counter[0] += 1
107
 
                eventlet.sleep()
108
 
        gt = eventlet.spawn(tick)
109
 
        curs.execute("select 1")
110
 
        rows = curs.fetchall()
111
 
        self.assertEqual(rows, ((1L,),))
112
 
        self.assert_(counter[0] > 0, counter[0])
113
 
        gt.kill()
114
 
 
115
 
    def assert_cursor_works(self, cursor):
116
 
        cursor.execute("select 1")
117
 
        rows = cursor.fetchall()
118
 
        self.assertEqual(rows, ((1L,),))
119
 
        self.assert_cursor_yields(cursor)
120
 
        
121
 
    def assert_connection_works(self, conn):
122
 
        curs = conn.cursor()
123
 
        self.assert_cursor_works(curs)
124
 
 
125
 
    def test_module_attributes(self):
126
 
        import MySQLdb as orig
127
 
        for key in dir(orig):
128
 
            if key not in ('__author__', '__path__', '__revision__',
129
 
                           '__version__', '__loader__'):
130
 
                self.assert_(hasattr(MySQLdb, key), "%s %s" % (key, getattr(orig, key)))
131
 
 
132
 
    def test_connecting(self):
133
 
        self.assert_(self.connection is not None)
134
 
        
135
 
    def test_connecting_annoyingly(self):
136
 
        self.assert_connection_works(MySQLdb.Connect(**self._auth))
137
 
        self.assert_connection_works(MySQLdb.Connection(**self._auth))
138
 
        self.assert_connection_works(MySQLdb.connections.Connection(**self._auth))
139
 
 
140
 
    def test_create_cursor(self):
141
 
        cursor = self.connection.cursor()
142
 
        cursor.close()
143
 
 
144
 
    def test_run_query(self):
145
 
        cursor = self.connection.cursor()
146
 
        self.assert_cursor_works(cursor)
147
 
        cursor.close()
148
 
 
149
 
    def test_run_bad_query(self):
150
 
        cursor = self.connection.cursor()
151
 
        try:
152
 
            cursor.execute("garbage blah blah")
153
 
            self.assert_(False)
154
 
        except AssertionError:
155
 
            raise
156
 
        except Exception:
157
 
            pass
158
 
        cursor.close()
159
 
 
160
 
    def fill_up_table(self, conn):
161
 
        curs = conn.cursor()
162
 
        for i in range(1000):
163
 
            curs.execute('insert into test_table (value_int) values (%s)' % i)
164
 
        conn.commit()
165
 
 
166
 
    def test_yields(self):
167
 
        conn = self.connection
168
 
        self.set_up_dummy_table(conn)
169
 
        self.fill_up_table(conn)
170
 
        curs = conn.cursor()
171
 
        results = []
172
 
        SHORT_QUERY = "select * from test_table"
173
 
        evt = event.Event()
174
 
        def a_query():
175
 
            self.assert_cursor_works(curs)
176
 
            curs.execute(SHORT_QUERY)
177
 
            results.append(2)
178
 
            evt.send()
179
 
        eventlet.spawn(a_query)
180
 
        results.append(1)
181
 
        self.assertEqual([1], results)
182
 
        evt.wait()
183
 
        self.assertEqual([1, 2], results)
184
 
 
185
 
    def test_visibility_from_other_connections(self):
186
 
        conn = MySQLdb.connect(**self._auth)
187
 
        conn2 = MySQLdb.connect(**self._auth)
188
 
        curs = conn.cursor()
189
 
        try:
190
 
            curs2 = conn2.cursor()
191
 
            curs2.execute("insert into gargleblatz (a) values (%s)" % (314159))
192
 
            self.assertEqual(curs2.rowcount, 1)
193
 
            conn2.commit()
194
 
            selection_query = "select * from gargleblatz"
195
 
            curs2.execute(selection_query)
196
 
            self.assertEqual(curs2.rowcount, 1)
197
 
            del curs2, conn2
198
 
            # create a new connection, it should see the addition
199
 
            conn3 = MySQLdb.connect(**self._auth)
200
 
            curs3 = conn3.cursor()
201
 
            curs3.execute(selection_query)
202
 
            self.assertEqual(curs3.rowcount, 1)
203
 
            # now, does the already-open connection see it?
204
 
            curs.execute(selection_query)
205
 
            self.assertEqual(curs.rowcount, 1)
206
 
            del curs3, conn3
207
 
        finally:
208
 
            # clean up my litter
209
 
            curs.execute("delete from gargleblatz where a=314159")
210
 
            conn.commit()
211
 
 
212
 
from tests import patcher_test
213
 
 
214
 
class MonkeyPatchTester(patcher_test.ProcessBase):
215
 
    @skip_unless(mysql_requirement)    
216
 
    def test_monkey_patching(self):
217
 
        output, lines = self.run_script("""
218
 
from eventlet import patcher
219
 
import MySQLdb as m
220
 
from eventlet.green import MySQLdb as gm
221
 
patcher.monkey_patch(all=True, MySQLdb=True)
222
 
print "mysqltest", ",".join(sorted(patcher.already_patched.keys()))
223
 
print "connect", m.connect == gm.connect
224
 
""")
225
 
        self.assertEqual(len(lines), 3)
226
 
        self.assertEqual(lines[0].replace("psycopg,", ""),
227
 
                         'mysqltest MySQLdb,os,select,socket,thread,time')
228
 
        self.assertEqual(lines[1], "connect True")