openstack/deb-python-eventlet
Code Issues Proposed changes

db_pool: proxy Connection.set_isolation_level()

Browse Source 
https://github.com/eventlet/eventlet/issues/389
This commit is contained in:
Sergey Shepelev
2017-02-16 14:12:00 +03:00
parent 4e775e23f3
commit 885482b576
2 changed files with 100 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
eventlet/db_pool.py
View File

@@ -327,98 +327,53 @@ class GenericConnectionWrapper(object):
def __repr__(self): def __repr__(self):
return self._base.__repr__() return self._base.__repr__()
def affected_rows(self): _proxy_funcs = (
return self._base.affected_rows() 'affected_rows',
'autocommit',
def autocommit(self, *args, **kwargs): 'begin',
return self._base.autocommit(*args, **kwargs) 'change_user',
'character_set_name',
def begin(self): 'close',
return self._base.begin() 'commit',
'cursor',
def change_user(self, *args, **kwargs): 'dump_debug_info',
return self._base.change_user(*args, **kwargs) 'errno',
'error',
def character_set_name(self, *args, **kwargs): 'errorhandler',
return self._base.character_set_name(*args, **kwargs) 'insert_id',
'literal',
def close(self, *args, **kwargs): 'ping',
return self._base.close(*args, **kwargs) 'query',
'rollback',
def commit(self, *args, **kwargs): 'select_db',
return self._base.commit(*args, **kwargs) 'server_capabilities',
'set_character_set',
def cursor(self, *args, **kwargs): 'set_isolation_level',
return self._base.cursor(*args, **kwargs) 'set_server_option',
'set_sql_mode',
def dump_debug_info(self, *args, **kwargs): 'show_warnings',
return self._base.dump_debug_info(*args, **kwargs) 'shutdown',
'sqlstate',
def errno(self, *args, **kwargs): 'stat',
return self._base.errno(*args, **kwargs) 'store_result',
'string_literal',
def error(self, *args, **kwargs): 'thread_id',
return self._base.error(*args, **kwargs) 'use_result',
'warning_count',
def errorhandler(self, *args, **kwargs): )
return self._base.errorhandler(*args, **kwargs) for _proxy_fun in GenericConnectionWrapper._proxy_funcs:
# excess wrapper for early binding (closure by value)
def insert_id(self, *args, **kwargs): def _wrapper(_proxy_fun=_proxy_fun):
return self._base.insert_id(*args, **kwargs) def _proxy_method(self, *args, **kwargs):
return getattr(self._base, _proxy_fun)(*args, **kwargs)
def literal(self, *args, **kwargs): _proxy_method.func_name = _proxy_fun
return self._base.literal(*args, **kwargs) _proxy_method.__name__ = _proxy_fun
_proxy_method.__qualname__ = 'GenericConnectionWrapper.' + _proxy_fun
def set_character_set(self, *args, **kwargs): return _proxy_method
return self._base.set_character_set(*args, **kwargs) setattr(GenericConnectionWrapper, _proxy_fun, _wrapper(_proxy_fun))
del GenericConnectionWrapper._proxy_funcs
def set_sql_mode(self, *args, **kwargs): del _proxy_fun
return self._base.set_sql_mode(*args, **kwargs) del _wrapper
def show_warnings(self):
return self._base.show_warnings()
def warning_count(self):
return self._base.warning_count()
def ping(self, *args, **kwargs):
return self._base.ping(*args, **kwargs)
def query(self, *args, **kwargs):
return self._base.query(*args, **kwargs)
def rollback(self, *args, **kwargs):
return self._base.rollback(*args, **kwargs)
def select_db(self, *args, **kwargs):
return self._base.select_db(*args, **kwargs)
def set_server_option(self, *args, **kwargs):
return self._base.set_server_option(*args, **kwargs)
def server_capabilities(self, *args, **kwargs):
return self._base.server_capabilities(*args, **kwargs)
def shutdown(self, *args, **kwargs):
return self._base.shutdown(*args, **kwargs)
def sqlstate(self, *args, **kwargs):
return self._base.sqlstate(*args, **kwargs)
def stat(self, *args, **kwargs):
return self._base.stat(*args, **kwargs)
def store_result(self, *args, **kwargs):
return self._base.store_result(*args, **kwargs)
def string_literal(self, *args, **kwargs):
return self._base.string_literal(*args, **kwargs)
def thread_id(self, *args, **kwargs):
return self._base.thread_id(*args, **kwargs)
def use_result(self, *args, **kwargs):
return self._base.use_result(*args, **kwargs)
class PooledConnectionWrapper(GenericConnectionWrapper): class PooledConnectionWrapper(GenericConnectionWrapper):

105
tests/db_pool_test.py
View File

@@ -1,17 +1,27 @@
'''Test cases for db_pool
'''
from __future__ import print_function from __future__ import print_function
import sys
import os import os
import sys
import traceback import traceback
from unittest import TestCase, main
from tests import mock, skip_unless, skip_with_pyevent, get_database_auth
from eventlet import event
from eventlet import db_pool from eventlet import db_pool
from eventlet.support import six from eventlet.support import six
import eventlet import eventlet
import eventlet.tpool
import tests
import tests.mock
psycopg2 = None
try:
import psycopg2
import psycopg2.extensions
except ImportError:
pass
MySQLdb = None
try:
import MySQLdb
except ImportError:
pass
class DBTester(object): class DBTester(object):
@@ -141,7 +151,7 @@ class DBConnectionPool(DBTester):
curs = conn.cursor() curs = conn.cursor()
results = [] results = []
SHORT_QUERY = "select * from test_table" SHORT_QUERY = "select * from test_table"
evt = event.Event() evt = eventlet.Event()
def a_query(): def a_query():
self.assert_cursor_works(curs) self.assert_cursor_works(curs)
@@ -282,7 +292,7 @@ class DBConnectionPool(DBTester):
self.connection = self.pool.get() self.connection = self.pool.get()
self.assertEqual(self.pool.free(), 0) self.assertEqual(self.pool.free(), 0)
self.assertEqual(self.pool.waiting(), 0) self.assertEqual(self.pool.waiting(), 0)
e = event.Event() e = eventlet.Event()
def retrieve(pool, ev): def retrieve(pool, ev):
c = pool.get() c = pool.get()
@@ -337,14 +347,13 @@ class TpoolConnectionPool(DBConnectionPool):
connect_timeout=connect_timeout, connect_timeout=connect_timeout,
**self._auth) **self._auth)
@skip_with_pyevent @tests.skip_with_pyevent
def setUp(self): def setUp(self):
super(TpoolConnectionPool, self).setUp() super(TpoolConnectionPool, self).setUp()
def tearDown(self): def tearDown(self):
super(TpoolConnectionPool, self).tearDown() super(TpoolConnectionPool, self).tearDown()
from eventlet import tpool eventlet.tpool.killall()
tpool.killall()
class RawConnectionPool(DBConnectionPool): class RawConnectionPool(DBConnectionPool):
@@ -373,7 +382,7 @@ def test_raw_pool_issue_125():
def test_raw_pool_custom_cleanup_ok(): def test_raw_pool_custom_cleanup_ok():
cleanup_mock = mock.Mock() cleanup_mock = tests.mock.Mock()
pool = db_pool.RawConnectionPool(DummyDBModule(), cleanup=cleanup_mock) pool = db_pool.RawConnectionPool(DummyDBModule(), cleanup=cleanup_mock)
conn = pool.get() conn = pool.get()
pool.put(conn) pool.put(conn)
@@ -385,7 +394,7 @@ def test_raw_pool_custom_cleanup_ok():
def test_raw_pool_custom_cleanup_arg_error(): def test_raw_pool_custom_cleanup_arg_error():
cleanup_mock = mock.Mock(side_effect=NotImplementedError) cleanup_mock = tests.mock.Mock(side_effect=NotImplementedError)
pool = db_pool.RawConnectionPool(DummyDBModule()) pool = db_pool.RawConnectionPool(DummyDBModule())
conn = pool.get() conn = pool.get()
pool.put(conn, cleanup=cleanup_mock) pool.put(conn, cleanup=cleanup_mock)
@@ -427,27 +436,23 @@ def test_raw_pool_clear_update_current_size():
assert len(pool.free_items) == 0 assert len(pool.free_items) == 0
get_auth = get_database_auth
def mysql_requirement(_f): def mysql_requirement(_f):
verbose = os.environ.get('eventlet_test_mysql_verbose') verbose = os.environ.get('eventlet_test_mysql_verbose')
try: if MySQLdb is None:
import MySQLdb
try:
auth = get_auth()['MySQLdb'].copy()
MySQLdb.connect(**auth)
return True
except MySQLdb.OperationalError:
if verbose:
print(">> Skipping mysql tests, error when connecting:", file=sys.stderr)
traceback.print_exc()
return False
except ImportError:
if verbose: if verbose:
print(">> Skipping mysql tests, MySQLdb not importable", file=sys.stderr) print(">> Skipping mysql tests, MySQLdb not importable", file=sys.stderr)
return False return False
try:
auth = tests.get_database_auth()['MySQLdb'].copy()
MySQLdb.connect(**auth)
return True
except MySQLdb.OperationalError:
if verbose:
print(">> Skipping mysql tests, error when connecting:", file=sys.stderr)
traceback.print_exc()
return False
class MysqlConnectionPool(object): class MysqlConnectionPool(object):
dummy_table_sql = """CREATE TEMPORARY TABLE test_table dummy_table_sql = """CREATE TEMPORARY TABLE test_table
@@ -463,11 +468,10 @@ class MysqlConnectionPool(object):
created TIMESTAMP created TIMESTAMP
) ENGINE=InnoDB;""" ) ENGINE=InnoDB;"""
@skip_unless(mysql_requirement) @tests.skip_unless(mysql_requirement)
def setUp(self): def setUp(self):
import MySQLdb
self._dbmodule = MySQLdb self._dbmodule = MySQLdb
self._auth = get_auth()['MySQLdb'] self._auth = tests.get_database_auth()['MySQLdb']
super(MysqlConnectionPool, self).setUp() super(MysqlConnectionPool, self).setUp()
def tearDown(self): def tearDown(self):
@@ -493,28 +497,27 @@ class MysqlConnectionPool(object):
del db del db
class Test01MysqlTpool(MysqlConnectionPool, TpoolConnectionPool, TestCase): class Test01MysqlTpool(MysqlConnectionPool, TpoolConnectionPool, tests.LimitedTestCase):
__test__ = True __test__ = True
class Test02MysqlRaw(MysqlConnectionPool, RawConnectionPool, TestCase): class Test02MysqlRaw(MysqlConnectionPool, RawConnectionPool, tests.LimitedTestCase):
__test__ = True __test__ = True
def postgres_requirement(_f): def postgres_requirement(_f):
try: if psycopg2 is None:
import psycopg2
try:
auth = get_auth()['psycopg2'].copy()
psycopg2.connect(**auth)
return True
except psycopg2.OperationalError:
print("Skipping postgres tests, error when connecting")
return False
except ImportError:
print("Skipping postgres tests, psycopg2 not importable") print("Skipping postgres tests, psycopg2 not importable")
return False return False
try:
auth = tests.get_database_auth()['psycopg2'].copy()
psycopg2.connect(**auth)
return True
except psycopg2.OperationalError:
print("Skipping postgres tests, error when connecting")
return False
class Psycopg2ConnectionPool(object): class Psycopg2ConnectionPool(object):
dummy_table_sql = """CREATE TEMPORARY TABLE test_table dummy_table_sql = """CREATE TEMPORARY TABLE test_table
@@ -529,11 +532,10 @@ class Psycopg2ConnectionPool(object):
created TIMESTAMP created TIMESTAMP
);""" );"""
@skip_unless(postgres_requirement) @tests.skip_unless(postgres_requirement)
def setUp(self): def setUp(self):
import psycopg2
self._dbmodule = psycopg2 self._dbmodule = psycopg2
self._auth = get_auth()['psycopg2'] self._auth = tests.get_database_auth()['psycopg2']
super(Psycopg2ConnectionPool, self).setUp() super(Psycopg2ConnectionPool, self).setUp()
def tearDown(self): def tearDown(self):
@@ -566,7 +568,7 @@ class Psycopg2ConnectionPool(object):
conn.close() conn.close()
class TestPsycopg2Base(TestCase): class TestPsycopg2Base(tests.LimitedTestCase):
__test__ = False __test__ = False
def test_cursor_works_as_context_manager(self): def test_cursor_works_as_context_manager(self):
@@ -575,6 +577,9 @@ class TestPsycopg2Base(TestCase):
row = c.fetchone() row = c.fetchone()
assert row == (1,) assert row == (1,)
def test_set_isolation_level(self):
self.connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
class Test01Psycopg2Tpool(Psycopg2ConnectionPool, TpoolConnectionPool, TestPsycopg2Base): class Test01Psycopg2Tpool(Psycopg2ConnectionPool, TpoolConnectionPool, TestPsycopg2Base):
__test__ = True __test__ = True
@@ -582,7 +587,3 @@ class Test01Psycopg2Tpool(Psycopg2ConnectionPool, TpoolConnectionPool, TestPsyco
class Test02Psycopg2Raw(Psycopg2ConnectionPool, RawConnectionPool, TestPsycopg2Base): class Test02Psycopg2Raw(Psycopg2ConnectionPool, RawConnectionPool, TestPsycopg2Base):
__test__ = True __test__ = True
if __name__ == '__main__':
main()