+++ /dev/null
-'''Test cases for db_pool
-'''
-from __future__ import print_function
-
-import sys
-import os
-import traceback
-from unittest import TestCase, main
-
-from tests import mock, skipped, skip_unless, skip_with_pyevent, get_database_auth
-from eventlet import event
-from eventlet import db_pool
-from eventlet.support import six
-import eventlet
-
-
-class DBTester(object):
- __test__ = False # so that nose doesn't try to execute this directly
-
- def setUp(self):
- self.create_db()
- self.connection = None
- connection = self._dbmodule.connect(**self._auth)
- cursor = connection.cursor()
- cursor.execute("""CREATE TABLE gargleblatz
- (
- a INTEGER
- );""")
- connection.commit()
- cursor.close()
- connection.close()
-
- def tearDown(self):
- if self.connection:
- self.connection.close()
- self.drop_db()
-
- def set_up_dummy_table(self, connection=None):
- close_connection = False
- if connection is None:
- close_connection = True
- if self.connection is None:
- connection = self._dbmodule.connect(**self._auth)
- else:
- connection = self.connection
-
- cursor = connection.cursor()
- cursor.execute(self.dummy_table_sql)
- connection.commit()
- cursor.close()
- if close_connection:
- connection.close()
-
-
-# silly mock class
-class Mock(object):
- pass
-
-
-class DBConnectionPool(DBTester):
- __test__ = False # so that nose doesn't try to execute this directly
-
- def setUp(self):
- super(DBConnectionPool, self).setUp()
- self.pool = self.create_pool()
- self.connection = self.pool.get()
-
- def tearDown(self):
- if self.connection:
- self.pool.put(self.connection)
- self.pool.clear()
- super(DBConnectionPool, self).tearDown()
-
- def assert_cursor_works(self, cursor):
- cursor.execute("select 1")
- rows = cursor.fetchall()
- assert rows
-
- def test_connecting(self):
- assert self.connection is not None
-
- def test_create_cursor(self):
- cursor = self.connection.cursor()
- cursor.close()
-
- def test_run_query(self):
- cursor = self.connection.cursor()
- self.assert_cursor_works(cursor)
- cursor.close()
-
- def test_run_bad_query(self):
- cursor = self.connection.cursor()
- try:
- cursor.execute("garbage blah blah")
- assert False
- except AssertionError:
- raise
- except Exception:
- pass
- cursor.close()
-
- def test_put_none(self):
- # the pool is of size 1, and its only connection is out
- assert self.pool.free() == 0
- self.pool.put(None)
- # ha ha we fooled it into thinking that we had a dead process
- assert self.pool.free() == 1
- conn2 = self.pool.get()
- assert conn2 is not None
- assert conn2.cursor
- self.pool.put(conn2)
-
- def test_close_does_a_put(self):
- assert self.pool.free() == 0
- self.connection.close()
- assert self.pool.free() == 1
- self.assertRaises(AttributeError, self.connection.cursor)
-
- @skipped
- def test_deletion_does_a_put(self):
- # doing a put on del causes some issues if __del__ is called in the
- # main coroutine, so, not doing that for now
- assert self.pool.free() == 0
- self.connection = None
- assert self.pool.free() == 1
-
- def test_put_doesnt_double_wrap(self):
- self.pool.put(self.connection)
- conn = self.pool.get()
- assert not isinstance(conn._base, db_pool.PooledConnectionWrapper)
- self.pool.put(conn)
-
- def test_bool(self):
- assert self.connection
- self.connection.close()
- assert not self.connection
-
- def fill_up_table(self, conn):
- curs = conn.cursor()
- for i in six.moves.range(1000):
- curs.execute('insert into test_table (value_int) values (%s)' % i)
- conn.commit()
-
- def test_returns_immediately(self):
- self.pool = self.create_pool()
- conn = self.pool.get()
- self.set_up_dummy_table(conn)
- self.fill_up_table(conn)
- curs = conn.cursor()
- results = []
- SHORT_QUERY = "select * from test_table"
- evt = event.Event()
-
- def a_query():
- self.assert_cursor_works(curs)
- curs.execute(SHORT_QUERY)
- results.append(2)
- evt.send()
- eventlet.spawn(a_query)
- results.append(1)
- self.assertEqual([1], results)
- evt.wait()
- self.assertEqual([1, 2], results)
- self.pool.put(conn)
-
- def test_connection_is_clean_after_put(self):
- self.pool = self.create_pool()
- conn = self.pool.get()
- self.set_up_dummy_table(conn)
- curs = conn.cursor()
- for i in range(10):
- curs.execute('insert into test_table (value_int) values (%s)' % i)
- # do not commit :-)
- self.pool.put(conn)
- del conn
- conn2 = self.pool.get()
- curs2 = conn2.cursor()
- for i in range(10):
- curs2.execute('insert into test_table (value_int) values (%s)' % i)
- conn2.commit()
- curs2.execute("select * from test_table")
- # we should have only inserted them once
- self.assertEqual(10, curs2.rowcount)
- self.pool.put(conn2)
-
- def test_visibility_from_other_connections(self):
- self.pool = self.create_pool(max_size=3)
- conn = self.pool.get()
- conn2 = self.pool.get()
- curs = conn.cursor()
- try:
- curs2 = conn2.cursor()
- curs2.execute("insert into gargleblatz (a) values (%s)" % (314159))
- self.assertEqual(curs2.rowcount, 1)
- conn2.commit()
- selection_query = "select * from gargleblatz"
- curs2.execute(selection_query)
- self.assertEqual(curs2.rowcount, 1)
- del curs2
- self.pool.put(conn2)
- # create a new connection, it should see the addition
- conn3 = self.pool.get()
- curs3 = conn3.cursor()
- curs3.execute(selection_query)
- self.assertEqual(curs3.rowcount, 1)
- # now, does the already-open connection see it?
- curs.execute(selection_query)
- self.assertEqual(curs.rowcount, 1)
- self.pool.put(conn3)
- finally:
- # clean up my litter
- curs.execute("delete from gargleblatz where a=314159")
- conn.commit()
- self.pool.put(conn)
-
- @skipped
- def test_two_simultaneous_connections(self):
- # timing-sensitive test, disabled until we come up with a better
- # way to do this
- self.pool = self.create_pool(max_size=2)
- conn = self.pool.get()
- self.set_up_dummy_table(conn)
- self.fill_up_table(conn)
- curs = conn.cursor()
- conn2 = self.pool.get()
- self.set_up_dummy_table(conn2)
- self.fill_up_table(conn2)
- curs2 = conn2.cursor()
- results = []
- LONG_QUERY = "select * from test_table"
- SHORT_QUERY = "select * from test_table where row_id <= 20"
-
- evt = event.Event()
-
- def long_running_query():
- self.assert_cursor_works(curs)
- curs.execute(LONG_QUERY)
- results.append(1)
- evt.send()
- evt2 = event.Event()
-
- def short_running_query():
- self.assert_cursor_works(curs2)
- curs2.execute(SHORT_QUERY)
- results.append(2)
- evt2.send()
-
- eventlet.spawn(long_running_query)
- eventlet.spawn(short_running_query)
- evt.wait()
- evt2.wait()
- results.sort()
- self.assertEqual([1, 2], results)
-
- def test_clear(self):
- self.pool = self.create_pool()
- self.pool.put(self.connection)
- self.pool.clear()
- self.assertEqual(len(self.pool.free_items), 0)
-
- def test_clear_warmup(self):
- """Clear implicitly created connections (min_size > 0)"""
- self.pool = self.create_pool(min_size=1)
- self.pool.clear()
- self.assertEqual(len(self.pool.free_items), 0)
-
- def test_unwrap_connection(self):
- self.assert_(isinstance(self.connection,
- db_pool.GenericConnectionWrapper))
- conn = self.pool._unwrap_connection(self.connection)
- assert not isinstance(conn, db_pool.GenericConnectionWrapper)
-
- self.assertEqual(None, self.pool._unwrap_connection(None))
- self.assertEqual(None, self.pool._unwrap_connection(1))
-
- # testing duck typing here -- as long as the connection has a
- # _base attribute, it should be unwrappable
- x = Mock()
- x._base = 'hi'
- self.assertEqual('hi', self.pool._unwrap_connection(x))
- conn.close()
-
- def test_safe_close(self):
- self.pool._safe_close(self.connection, quiet=True)
- self.assertEqual(len(self.pool.free_items), 1)
-
- self.pool._safe_close(None)
- self.pool._safe_close(1)
-
- # now we're really going for 100% coverage
- x = Mock()
-
- def fail():
- raise KeyboardInterrupt()
- x.close = fail
- self.assertRaises(KeyboardInterrupt, self.pool._safe_close, x)
-
- x = Mock()
-
- def fail2():
- raise RuntimeError("if this line has been printed, the test succeeded")
- x.close = fail2
- self.pool._safe_close(x, quiet=False)
-
- def test_zero_max_idle(self):
- self.pool.put(self.connection)
- self.pool.clear()
- self.pool = self.create_pool(max_size=2, max_idle=0)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 0)
-
- def test_zero_max_age(self):
- self.pool.put(self.connection)
- self.pool.clear()
- self.pool = self.create_pool(max_size=2, max_age=0)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 0)
-
- @skipped
- def test_max_idle(self):
- # This test is timing-sensitive. Rename the function without
- # the "dont" to run it, but beware that it could fail or take
- # a while.
-
- self.pool = self.create_pool(max_size=2, max_idle=0.02)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.01) # not long enough to trigger the idle timeout
- self.assertEqual(len(self.pool.free_items), 1)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.01) # idle timeout should have fired but done nothing
- self.assertEqual(len(self.pool.free_items), 1)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.03) # long enough to trigger idle timeout for real
- self.assertEqual(len(self.pool.free_items), 0)
-
- @skipped
- def test_max_idle_many(self):
- # This test is timing-sensitive. Rename the function without
- # the "dont" to run it, but beware that it could fail or take
- # a while.
-
- self.pool = self.create_pool(max_size=2, max_idle=0.02)
- self.connection, conn2 = self.pool.get(), self.pool.get()
- self.connection.close()
- eventlet.sleep(0.01)
- self.assertEqual(len(self.pool.free_items), 1)
- conn2.close()
- self.assertEqual(len(self.pool.free_items), 2)
- eventlet.sleep(0.02) # trigger cleanup of conn1 but not conn2
- self.assertEqual(len(self.pool.free_items), 1)
-
- @skipped
- def test_max_age(self):
- # This test is timing-sensitive. Rename the function without
- # the "dont" to run it, but beware that it could fail or take
- # a while.
-
- self.pool = self.create_pool(max_size=2, max_age=0.05)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.01) # not long enough to trigger the age timeout
- self.assertEqual(len(self.pool.free_items), 1)
- self.connection = self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.05) # long enough to trigger age timeout
- self.assertEqual(len(self.pool.free_items), 0)
-
- @skipped
- def test_max_age_many(self):
- # This test is timing-sensitive. Rename the function without
- # the "dont" to run it, but beware that it could fail or take
- # a while.
-
- self.pool = self.create_pool(max_size=2, max_age=0.15)
- self.connection, conn2 = self.pool.get(), self.pool.get()
- self.connection.close()
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0) # not long enough to trigger the age timeout
- self.assertEqual(len(self.pool.free_items), 1)
- eventlet.sleep(0.2) # long enough to trigger age timeout
- self.assertEqual(len(self.pool.free_items), 0)
- conn2.close() # should not be added to the free items
- self.assertEqual(len(self.pool.free_items), 0)
-
- def test_waiters_get_woken(self):
- # verify that when there's someone waiting on an empty pool
- # and someone puts an immediately-closed connection back in
- # the pool that the waiter gets woken
- self.pool.put(self.connection)
- self.pool.clear()
- self.pool = self.create_pool(max_size=1, max_age=0)
-
- self.connection = self.pool.get()
- self.assertEqual(self.pool.free(), 0)
- self.assertEqual(self.pool.waiting(), 0)
- e = event.Event()
-
- def retrieve(pool, ev):
- c = pool.get()
- ev.send(c)
- eventlet.spawn(retrieve, self.pool, e)
- eventlet.sleep(0) # these two sleeps should advance the retrieve
- eventlet.sleep(0) # coroutine until it's waiting in get()
- self.assertEqual(self.pool.free(), 0)
- self.assertEqual(self.pool.waiting(), 1)
- self.pool.put(self.connection)
- timer = eventlet.Timeout(1)
- conn = e.wait()
- timer.cancel()
- self.assertEqual(self.pool.free(), 0)
- self.assertEqual(self.pool.waiting(), 0)
- self.pool.put(conn)
-
- @skipped
- def test_0_straight_benchmark(self):
- """ Benchmark; don't run unless you want to wait a while."""
- import time
- iterations = 20000
- c = self.connection.cursor()
- self.connection.commit()
-
- def bench(c):
- for i in six.moves.range(iterations):
- c.execute('select 1')
-
- bench(c) # warm-up
- results = []
- for i in range(3):
- start = time.time()
- bench(c)
- end = time.time()
- results.append(end - start)
-
- print("\n%u iterations took an average of %f seconds, (%s) in %s\n" % (
- iterations, sum(results) / len(results), results, type(self)))
-
- def test_raising_create(self):
- # if the create() method raises an exception the pool should
- # not lose any connections
- self.pool = self.create_pool(max_size=1, module=RaisingDBModule())
- self.assertRaises(RuntimeError, self.pool.get)
- self.assertEqual(self.pool.free(), 1)
-
-
-class DummyConnection(object):
- def rollback(self):
- pass
-
-
-class DummyDBModule(object):
- def connect(self, *args, **kwargs):
- return DummyConnection()
-
-
-class RaisingDBModule(object):
- def connect(self, *args, **kw):
- raise RuntimeError()
-
-
-class TpoolConnectionPool(DBConnectionPool):
- __test__ = False # so that nose doesn't try to execute this directly
-
- def create_pool(self, min_size=0, max_size=1, max_idle=10, max_age=10,
- connect_timeout=0.5, module=None):
- if module is None:
- module = self._dbmodule
- return db_pool.TpooledConnectionPool(
- module,
- min_size=min_size, max_size=max_size,
- max_idle=max_idle, max_age=max_age,
- connect_timeout=connect_timeout,
- **self._auth)
-
- @skip_with_pyevent
- def setUp(self):
- super(TpoolConnectionPool, self).setUp()
-
- def tearDown(self):
- super(TpoolConnectionPool, self).tearDown()
- from eventlet import tpool
- tpool.killall()
-
-
-class RawConnectionPool(DBConnectionPool):
- __test__ = False # so that nose doesn't try to execute this directly
-
- def create_pool(self, min_size=0, max_size=1, max_idle=10, max_age=10,
- connect_timeout=0.5, module=None):
- if module is None:
- module = self._dbmodule
- return db_pool.RawConnectionPool(
- module,
- min_size=min_size, max_size=max_size,
- max_idle=max_idle, max_age=max_age,
- connect_timeout=connect_timeout,
- **self._auth)
-
-
-def test_raw_pool_issue_125():
- # pool = self.create_pool(min_size=3, max_size=5)
- pool = db_pool.RawConnectionPool(
- DummyDBModule(),
- dsn="dbname=test user=jessica port=5433",
- min_size=3, max_size=5)
- conn = pool.get()
- pool.put(conn)
-
-
-def test_raw_pool_custom_cleanup_ok():
- cleanup_mock = mock.Mock()
- pool = db_pool.RawConnectionPool(DummyDBModule(), cleanup=cleanup_mock)
- conn = pool.get()
- pool.put(conn)
- assert cleanup_mock.call_count == 1
-
- with pool.item() as conn:
- pass
- assert cleanup_mock.call_count == 2
-
-
-def test_raw_pool_custom_cleanup_arg_error():
- cleanup_mock = mock.Mock(side_effect=NotImplementedError)
- pool = db_pool.RawConnectionPool(DummyDBModule())
- conn = pool.get()
- pool.put(conn, cleanup=cleanup_mock)
- assert cleanup_mock.call_count == 1
-
- with pool.item(cleanup=cleanup_mock):
- pass
- assert cleanup_mock.call_count == 2
-
-
-def test_raw_pool_custom_cleanup_fatal():
- state = [0]
-
- def cleanup(conn):
- state[0] += 1
- raise KeyboardInterrupt
-
- pool = db_pool.RawConnectionPool(DummyDBModule(), cleanup=cleanup)
- conn = pool.get()
- try:
- pool.put(conn)
- except KeyboardInterrupt:
- pass
- else:
- assert False, 'Expected KeyboardInterrupt'
- assert state[0] == 1
-
-
-def test_raw_pool_clear_update_current_size():
- # https://github.com/eventlet/eventlet/issues/139
- # BaseConnectionPool.clear does not update .current_size.
- # That leads to situation when new connections could not be created.
- pool = db_pool.RawConnectionPool(DummyDBModule())
- pool.get().close()
- assert pool.current_size == 1
- assert len(pool.free_items) == 1
- pool.clear()
- assert pool.current_size == 0
- assert len(pool.free_items) == 0
-
-
-get_auth = get_database_auth
-
-
-def mysql_requirement(_f):
- verbose = os.environ.get('eventlet_test_mysql_verbose')
- try:
- 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:
- print(">> Skipping mysql tests, MySQLdb not importable", file=sys.stderr)
- return False
-
-
-class MysqlConnectionPool(object):
- dummy_table_sql = """CREATE TEMPORARY TABLE test_table
- (
- row_id INTEGER PRIMARY KEY AUTO_INCREMENT,
- value_int INTEGER,
- value_float FLOAT,
- value_string VARCHAR(200),
- value_uuid CHAR(36),
- value_binary BLOB,
- value_binary_string VARCHAR(200) BINARY,
- value_enum ENUM('Y','N'),
- created TIMESTAMP
- ) ENGINE=InnoDB;"""
-
- @skip_unless(mysql_requirement)
- def setUp(self):
- import MySQLdb
- self._dbmodule = MySQLdb
- self._auth = get_auth()['MySQLdb']
- super(MysqlConnectionPool, self).setUp()
-
- def tearDown(self):
- super(MysqlConnectionPool, self).tearDown()
-
- def create_db(self):
- auth = self._auth.copy()
- try:
- self.drop_db()
- except Exception:
- pass
- dbname = 'test%s' % os.getpid()
- db = self._dbmodule.connect(**auth).cursor()
- db.execute("create database " + dbname)
- db.close()
- self._auth['db'] = dbname
- del db
-
- def drop_db(self):
- db = self._dbmodule.connect(**self._auth).cursor()
- db.execute("drop database " + self._auth['db'])
- db.close()
- del db
-
-
-class Test01MysqlTpool(MysqlConnectionPool, TpoolConnectionPool, TestCase):
- __test__ = True
-
-
-class Test02MysqlRaw(MysqlConnectionPool, RawConnectionPool, TestCase):
- __test__ = True
-
-
-def postgres_requirement(_f):
- try:
- 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")
- return False
-
-
-class Psycopg2ConnectionPool(object):
- dummy_table_sql = """CREATE TEMPORARY TABLE test_table
- (
- row_id SERIAL PRIMARY KEY,
- value_int INTEGER,
- value_float FLOAT,
- value_string VARCHAR(200),
- value_uuid CHAR(36),
- value_binary BYTEA,
- value_binary_string BYTEA,
- created TIMESTAMP
- );"""
-
- @skip_unless(postgres_requirement)
- def setUp(self):
- import psycopg2
- self._dbmodule = psycopg2
- self._auth = get_auth()['psycopg2']
- super(Psycopg2ConnectionPool, self).setUp()
-
- def tearDown(self):
- super(Psycopg2ConnectionPool, self).tearDown()
-
- def create_db(self):
- dbname = 'test%s' % os.getpid()
- self._auth['database'] = dbname
- try:
- self.drop_db()
- except Exception:
- pass
- auth = self._auth.copy()
- auth.pop('database') # can't create if you're connecting to it
- conn = self._dbmodule.connect(**auth)
- conn.set_isolation_level(0)
- db = conn.cursor()
- db.execute("create database " + dbname)
- db.close()
- conn.close()
-
- def drop_db(self):
- auth = self._auth.copy()
- auth.pop('database') # can't drop database we connected to
- conn = self._dbmodule.connect(**auth)
- conn.set_isolation_level(0)
- db = conn.cursor()
- db.execute("drop database " + self._auth['database'])
- db.close()
- conn.close()
-
-
-class TestPsycopg2Base(TestCase):
- __test__ = False
-
- def test_cursor_works_as_context_manager(self):
- with self.connection.cursor() as c:
- c.execute('select 1')
- row = c.fetchone()
- assert row == (1,)
-
-
-class Test01Psycopg2Tpool(Psycopg2ConnectionPool, TpoolConnectionPool, TestPsycopg2Base):
- __test__ = True
-
-
-class Test02Psycopg2Raw(Psycopg2ConnectionPool, RawConnectionPool, TestPsycopg2Base):
- __test__ = True
-
-
-if __name__ == '__main__':
- main()
Code Review / packages / trusty / python-eventlet.git / blobdiff