Fixes to make tests pass on python2.4. The tricks to make with-statement stuff work even when imported by 2.4 are especially tricky, but there doesn't seem to be a better alternative.
This commit is contained in:
@@ -14,6 +14,33 @@ class SomeFailed(FanFailed):
|
||||
class AllFailed(FanFailed):
|
||||
pass
|
||||
|
||||
# have to stick this in an exec so it works in 2.4
|
||||
try:
|
||||
from contextlib import contextmanager
|
||||
exec('''
|
||||
@contextmanager
|
||||
def item_impl(self):
|
||||
""" Get an object out of the pool, for use with with statement.
|
||||
|
||||
>>> from eventlet import pools
|
||||
>>> pool = pools.TokenPool(max_size=4)
|
||||
>>> with pool.item() as obj:
|
||||
... print "got token"
|
||||
...
|
||||
got token
|
||||
>>> pool.free()
|
||||
4
|
||||
"""
|
||||
obj = self.get()
|
||||
try:
|
||||
yield obj
|
||||
finally:
|
||||
self.put(obj)
|
||||
''')
|
||||
except ImportError:
|
||||
item_impl = None
|
||||
|
||||
|
||||
|
||||
class Pool(object):
|
||||
"""
|
||||
@@ -70,29 +97,8 @@ class Pool(object):
|
||||
return created
|
||||
return self.channel.wait()
|
||||
|
||||
try:
|
||||
from contextlib import contextmanager
|
||||
@contextmanager
|
||||
def item(self):
|
||||
""" Get an object out of the pool, for use with with statement.
|
||||
|
||||
>>> from eventlet import pools
|
||||
>>> pool = pools.TokenPool(max_size=4)
|
||||
>>> with pool.item() as obj:
|
||||
... print "got token"
|
||||
...
|
||||
got token
|
||||
>>> pool.free()
|
||||
4
|
||||
"""
|
||||
obj = self.get()
|
||||
try:
|
||||
yield obj
|
||||
finally:
|
||||
self.put(obj)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
if item_impl is not None:
|
||||
item = item_impl
|
||||
|
||||
def put(self, item):
|
||||
"""Put an item back into the pool, when done
|
||||
|
@@ -132,7 +132,7 @@ class TestDBConnectionPool(DBTester):
|
||||
self.connection.close()
|
||||
self.assert_(not self.connection)
|
||||
|
||||
def fill_test_table(self, conn):
|
||||
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)
|
||||
@@ -142,7 +142,7 @@ class TestDBConnectionPool(DBTester):
|
||||
self.pool = self.create_pool()
|
||||
conn = self.pool.get()
|
||||
self.set_up_test_table(conn)
|
||||
self.fill_test_table(conn)
|
||||
self.fill_up_table(conn)
|
||||
curs = conn.cursor()
|
||||
results = []
|
||||
SHORT_QUERY = "select * from test_table"
|
||||
@@ -213,11 +213,11 @@ class TestDBConnectionPool(DBTester):
|
||||
self.pool = self.create_pool(2)
|
||||
conn = self.pool.get()
|
||||
self.set_up_test_table(conn)
|
||||
self.fill_test_table(conn)
|
||||
self.fill_up_table(conn)
|
||||
curs = conn.cursor()
|
||||
conn2 = self.pool.get()
|
||||
self.set_up_test_table(conn2)
|
||||
self.fill_test_table(conn2)
|
||||
self.fill_up_table(conn2)
|
||||
curs2 = conn2.cursor()
|
||||
results = []
|
||||
LONG_QUERY = "select * from test_table"
|
||||
|
@@ -8,5 +8,11 @@ parent_dir = dirname(dirname(realpath(abspath(__file__))))
|
||||
if parent_dir not in sys.path:
|
||||
sys.path.insert(0, parent_dir)
|
||||
|
||||
# hacky hacks: skip test__api_timeout when under 2.4 because otherwise it SyntaxErrors
|
||||
if sys.version_info < (2,5):
|
||||
argv = sys.argv + ["--exclude=.*test__api_timeout.*"]
|
||||
else:
|
||||
argv = sys.argv
|
||||
|
||||
from tests import eventlethub
|
||||
nose.main(addplugins=[eventlethub.EventletHub()])
|
||||
nose.main(addplugins=[eventlethub.EventletHub()], argv=argv)
|
||||
|
Reference in New Issue
Block a user