diff options
Diffstat (limited to 'lib/sqlalchemy/testing/fixtures.py')
| -rw-r--r-- | lib/sqlalchemy/testing/fixtures.py | 245 |
1 files changed, 146 insertions, 99 deletions
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py index ac4d3d8fa..f19b4652a 100644 --- a/lib/sqlalchemy/testing/fixtures.py +++ b/lib/sqlalchemy/testing/fixtures.py @@ -5,6 +5,7 @@ # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php +import contextlib import re import sys @@ -12,12 +13,11 @@ import sqlalchemy as sa from . import assertions from . import config from . import schema -from .engines import drop_all_tables -from .engines import testing_engine from .entities import BasicEntity from .entities import ComparableEntity from .entities import ComparableMixin # noqa from .util import adict +from .util import drop_all_tables_from_metadata from .. import event from .. import util from ..orm import declarative_base @@ -25,10 +25,8 @@ from ..orm import registry from ..orm.decl_api import DeclarativeMeta from ..schema import sort_tables_and_constraints -# whether or not we use unittest changes things dramatically, -# as far as how pytest collection works. - +@config.mark_base_test_class() class TestBase(object): # A sequence of database names to always run, regardless of the # constraints below. @@ -48,81 +46,114 @@ class TestBase(object): # skipped. __skip_if__ = None + # if True, the testing reaper will not attempt to touch connection + # state after a test is completed and before the outer teardown + # starts + __leave_connections_for_teardown__ = False + def assert_(self, val, msg=None): assert val, msg - # apparently a handful of tests are doing this....OK - def setup(self): - if hasattr(self, "setUp"): - self.setUp() - - def teardown(self): - if hasattr(self, "tearDown"): - self.tearDown() - @config.fixture() def connection(self): - eng = getattr(self, "bind", config.db) + global _connection_fixture_connection + + eng = getattr(self, "bind", None) or config.db conn = eng.connect() trans = conn.begin() - try: - yield conn - finally: - if trans.is_active: - trans.rollback() - conn.close() + + _connection_fixture_connection = conn + yield conn + + _connection_fixture_connection = None + + if trans.is_active: + trans.rollback() + # trans would not be active here if the test is using + # the legacy @provide_metadata decorator still, as it will + # run a close all connections. + conn.close() @config.fixture() - def future_connection(self): + def future_connection(self, future_engine, connection): + # integrate the future_engine and connection fixtures so + # that users of the "connection" fixture will get at the + # "future" connection + yield connection - eng = testing_engine(future=True) - conn = eng.connect() - trans = conn.begin() - try: - yield conn - finally: - if trans.is_active: - trans.rollback() - conn.close() + @config.fixture() + def future_engine(self): + eng = getattr(self, "bind", None) or config.db + with _push_future_engine(eng): + yield + + @config.fixture() + def testing_engine(self): + from . import engines + + def gen_testing_engine( + url=None, options=None, future=False, asyncio=False + ): + if options is None: + options = {} + options["scope"] = "fixture" + return engines.testing_engine( + url=url, options=options, future=future, asyncio=asyncio + ) + + yield gen_testing_engine + + engines.testing_reaper._drop_testing_engines("fixture") @config.fixture() - def metadata(self): + def metadata(self, request): """Provide bound MetaData for a single test, dropping afterwards.""" - from . import engines from ..sql import schema metadata = schema.MetaData() - try: - yield metadata - finally: - engines.drop_all_tables(metadata, config.db) + request.instance.metadata = metadata + yield metadata + del request.instance.metadata + if ( + _connection_fixture_connection + and _connection_fixture_connection.in_transaction() + ): + trans = _connection_fixture_connection.get_transaction() + trans.rollback() + with _connection_fixture_connection.begin(): + drop_all_tables_from_metadata( + metadata, _connection_fixture_connection + ) + else: + drop_all_tables_from_metadata(metadata, config.db) -class FutureEngineMixin(object): - @classmethod - def setup_class(cls): - from ..future.engine import Engine - from sqlalchemy import testing +_connection_fixture_connection = None - facade = Engine._future_facade(config.db) - config._current.push_engine(facade, testing) - super_ = super(FutureEngineMixin, cls) - if hasattr(super_, "setup_class"): - super_.setup_class() +@contextlib.contextmanager +def _push_future_engine(engine): - @classmethod - def teardown_class(cls): - super_ = super(FutureEngineMixin, cls) - if hasattr(super_, "teardown_class"): - super_.teardown_class() + from ..future.engine import Engine + from sqlalchemy import testing + + facade = Engine._future_facade(engine) + config._current.push_engine(facade, testing) + + yield facade - from sqlalchemy import testing + config._current.pop(testing) - config._current.pop(testing) + +class FutureEngineMixin(object): + @config.fixture(autouse=True, scope="class") + def _push_future_engine(self): + eng = getattr(self, "bind", None) or config.db + with _push_future_engine(eng): + yield class TablesTest(TestBase): @@ -151,18 +182,32 @@ class TablesTest(TestBase): other = None sequences = None - @property - def tables_test_metadata(self): - return self._tables_metadata - - @classmethod - def setup_class(cls): + @config.fixture(autouse=True, scope="class") + def _setup_tables_test_class(self): + cls = self.__class__ cls._init_class() cls._setup_once_tables() cls._setup_once_inserts() + yield + + cls._teardown_once_metadata_bind() + + @config.fixture(autouse=True, scope="function") + def _setup_tables_test_instance(self): + self._setup_each_tables() + self._setup_each_inserts() + + yield + + self._teardown_each_tables() + + @property + def tables_test_metadata(self): + return self._tables_metadata + @classmethod def _init_class(cls): if cls.run_define_tables == "each": @@ -213,10 +258,10 @@ class TablesTest(TestBase): if self.run_define_tables == "each": self.tables.clear() if self.run_create_tables == "each": - drop_all_tables(self._tables_metadata, self.bind) + drop_all_tables_from_metadata(self._tables_metadata, self.bind) self._tables_metadata.clear() elif self.run_create_tables == "each": - drop_all_tables(self._tables_metadata, self.bind) + drop_all_tables_from_metadata(self._tables_metadata, self.bind) # no need to run deletes if tables are recreated on setup if ( @@ -242,17 +287,10 @@ class TablesTest(TestBase): file=sys.stderr, ) - def setup(self): - self._setup_each_tables() - self._setup_each_inserts() - - def teardown(self): - self._teardown_each_tables() - @classmethod def _teardown_once_metadata_bind(cls): if cls.run_create_tables: - drop_all_tables(cls._tables_metadata, cls.bind) + drop_all_tables_from_metadata(cls._tables_metadata, cls.bind) if cls.run_dispose_bind == "once": cls.dispose_bind(cls.bind) @@ -263,10 +301,6 @@ class TablesTest(TestBase): cls.bind = None @classmethod - def teardown_class(cls): - cls._teardown_once_metadata_bind() - - @classmethod def setup_bind(cls): return config.db @@ -332,38 +366,47 @@ class RemovesEvents(object): self._event_fns.add((target, name, fn)) event.listen(target, name, fn, **kw) - def teardown(self): + @config.fixture(autouse=True, scope="function") + def _remove_events(self): + yield for key in self._event_fns: event.remove(*key) - super_ = super(RemovesEvents, self) - if hasattr(super_, "teardown"): - super_.teardown() - - -class _ORMTest(object): - @classmethod - def teardown_class(cls): - sa.orm.session.close_all_sessions() - sa.orm.clear_mappers() -def create_session(**kw): - kw.setdefault("autoflush", False) - kw.setdefault("expire_on_commit", False) - return sa.orm.Session(config.db, **kw) +_fixture_sessions = set() def fixture_session(**kw): kw.setdefault("autoflush", True) kw.setdefault("expire_on_commit", True) - return sa.orm.Session(config.db, **kw) + sess = sa.orm.Session(config.db, **kw) + _fixture_sessions.add(sess) + return sess + + +def _close_all_sessions(): + # will close all still-referenced sessions + sa.orm.session.close_all_sessions() + _fixture_sessions.clear() + + +def stop_test_class_inside_fixtures(cls): + _close_all_sessions() + sa.orm.clear_mappers() -class ORMTest(_ORMTest, TestBase): +def after_test(): + + if _fixture_sessions: + + _close_all_sessions() + + +class ORMTest(TestBase): pass -class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): +class MappedTest(TablesTest, assertions.AssertsExecutionResults): # 'once', 'each', None run_setup_classes = "once" @@ -372,8 +415,9 @@ class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): classes = None - @classmethod - def setup_class(cls): + @config.fixture(autouse=True, scope="class") + def _setup_tables_test_class(self): + cls = self.__class__ cls._init_class() if cls.classes is None: @@ -384,18 +428,20 @@ class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): cls._setup_once_mappers() cls._setup_once_inserts() - @classmethod - def teardown_class(cls): + yield + cls._teardown_once_class() cls._teardown_once_metadata_bind() - def setup(self): + @config.fixture(autouse=True, scope="function") + def _setup_tables_test_instance(self): self._setup_each_tables() self._setup_each_classes() self._setup_each_mappers() self._setup_each_inserts() - def teardown(self): + yield + sa.orm.session.close_all_sessions() self._teardown_each_mappers() self._teardown_each_classes() @@ -404,7 +450,6 @@ class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): @classmethod def _teardown_once_class(cls): cls.classes.clear() - _ORMTest.teardown_class() @classmethod def _setup_once_classes(cls): @@ -440,6 +485,8 @@ class MappedTest(_ORMTest, TablesTest, assertions.AssertsExecutionResults): """ cls_registry = cls.classes + assert cls_registry is not None + class FindFixture(type): def __init__(cls, classname, bases, dict_): cls_registry[classname] = cls |