--- /dev/null
+.. change::
+ :tags: bug, tests
+ :tickets: 7599
+
+ Improvements to the test suite's integration with pytest such that the
+ "warnings" plugin, if manually enabled, will not interfere with the test
+ suite, such that third parties can enable the warnings plugin or make use
+ of the ``-W`` parameter and SQLAlchemy's test suite will continue to pass.
+ Additionally, modernized the detection of the "pytest-xdist" plugin so that
+ plugins can be globally disabled using PYTEST_DISABLE_PLUGIN_AUTOLOAD=1
+ without breaking the test suite if xdist were still installed. Warning
+ filters that promote deprecation warnings to errors are now localized to
+ SQLAlchemy-specific warnings, or within SQLAlchemy-specific sources for
+ general Python deprecation warnings, so that non-SQLAlchemy deprecation
+ warnings emitted from pytest plugins should also not impact the test suite.
+
template_db = cfg.options.postgresql_templatedb
with eng.execution_options(isolation_level="AUTOCOMMIT").begin() as conn:
- try:
- _pg_drop_db(cfg, conn, ident)
- except Exception:
- pass
+
if not template_db:
template_db = conn.exec_driver_sql(
"select current_database()"
from .assertions import assert_raises_context_ok
from .assertions import assert_raises_message
from .assertions import assert_raises_message_context_ok
+from .assertions import assert_warns
+from .assertions import assert_warns_message
from .assertions import AssertsCompiledSQL
from .assertions import AssertsExecutionResults
from .assertions import ComparesTables
exc_cls,
messages,
regex=True,
+ search_msg=False,
assert_=True,
py2konly=False,
raise_on_any_unexpected=False,
+ squelch_other_warnings=False,
):
global _FILTERS, _SEEN, _EXC_CLS
- if regex:
+ if regex or search_msg:
filters = [re.compile(msg, re.I | re.S) for msg in messages]
else:
filters = list(messages)
exception = None
if not exception or not issubclass(exception, _EXC_CLS):
- return real_warn(msg, *arg, **kw)
+ if not squelch_other_warnings:
+ return real_warn(msg, *arg, **kw)
if not filters and not raise_on_any_unexpected:
return
for filter_ in filters:
- if (regex and filter_.match(msg)) or (
- not regex and filter_ == msg
+ if (
+ (search_msg and filter_.search(msg))
+ or (regex and filter_.match(msg))
+ or (not regex and filter_ == msg)
):
seen.discard(filter_)
break
else:
- real_warn(msg, *arg, **kw)
+ if not squelch_other_warnings:
+ real_warn(msg, *arg, **kw)
with mock.patch("warnings.warn", our_warn), mock.patch(
"sqlalchemy.util.SQLALCHEMY_WARN_20", True
)
+def assert_warns(except_cls, callable_, *args, **kwargs):
+ """legacy adapter function for functions that were previously using
+ assert_raises with SAWarning or similar.
+
+ has some workarounds to accommodate the fact that the callable completes
+ with this approach rather than stopping at the exception raise.
+
+
+ """
+ with _expect_warnings(except_cls, [".*"], squelch_other_warnings=True):
+ return callable_(*args, **kwargs)
+
+
+def assert_warns_message(except_cls, msg, callable_, *args, **kwargs):
+ """legacy adapter function for functions that were previously using
+ assert_raises with SAWarning or similar.
+
+ has some workarounds to accommodate the fact that the callable completes
+ with this approach rather than stopping at the exception raise.
+
+ Also uses regex.search() to match the given message to the error string
+ rather than regex.match().
+
+ """
+ with _expect_warnings(
+ except_cls,
+ [msg],
+ search_msg=True,
+ regex=False,
+ squelch_other_warnings=True,
+ ):
+ return callable_(*args, **kwargs)
+
+
def assert_raises_message_context_ok(
except_cls, msg, callable_, *args, **kwargs
):
@contextlib.contextmanager
def _expect_raises(except_cls, msg=None, check_context=False):
+ if (
+ isinstance(except_cls, type)
+ and issubclass(except_cls, Warning)
+ or isinstance(except_cls, Warning)
+ ):
+ raise TypeError(
+ "Use expect_warnings for warnings, not "
+ "expect_raises / assert_raises"
+ )
ec = _ErrorContainer()
if check_context:
are_we_already_in_a_traceback = sys.exc_info()[0]
import os
import re
import sys
+import uuid
import pytest
-try:
- import xdist # noqa
-
- has_xdist = True
-except ImportError:
- has_xdist = False
-
py2k = sys.version_info < (3, 0)
if py2k:
def pytest_configure(config):
+ if config.pluginmanager.hasplugin("xdist"):
+ config.pluginmanager.register(XDistHooks())
+
if hasattr(config, "workerinput"):
plugin_base.restore_important_follower_config(config.workerinput)
plugin_base.configure_follower(config.workerinput["follower_ident"])
collect_types.init_types_collection(filter_filename=_filter)
-if has_xdist:
- import uuid
-
- def pytest_configure_node(node):
+class XDistHooks(object):
+ def pytest_configure_node(self, node):
from sqlalchemy.testing import provision
from sqlalchemy.testing import asyncio
provision.create_follower_db, node.workerinput["follower_ident"]
)
- def pytest_testnodedown(node, error):
+ def pytest_testnodedown(self, node, error):
from sqlalchemy.testing import provision
from sqlalchemy.testing import asyncio
from ..util.langhelpers import _warnings_warn
-class SATestSuiteWarning(sa_exc.SAWarning):
- """warning for a condition detected during tests that is non-fatal"""
+class SATestSuiteWarning(Warning):
+ """warning for a condition detected during tests that is non-fatal
+
+ Currently outside of SAWarning so that we can work around tools like
+ Alembic doing the wrong thing with warnings.
+
+ """
def warn_test_suite(message):
def setup_filters():
"""Set global warning behavior for the test suite."""
+ # TODO: at this point we can use the normal pytest warnings plugin,
+ # if we decide the test suite can be linked to pytest only
+
+ origin = r"^(?:test|sqlalchemy)\..*"
+
warnings.filterwarnings(
"ignore", category=sa_exc.SAPendingDeprecationWarning
)
warnings.filterwarnings("error", category=sa_exc.SADeprecationWarning)
warnings.filterwarnings("error", category=sa_exc.SAWarning)
- warnings.filterwarnings("always", category=SATestSuiteWarning)
- # some selected deprecations...
- warnings.filterwarnings("error", category=DeprecationWarning)
- warnings.filterwarnings(
- "ignore", category=DeprecationWarning, message=r".*StopIteration"
- )
- warnings.filterwarnings(
- "ignore",
- category=DeprecationWarning,
- message=r".*inspect.get.*argspec",
- )
+ warnings.filterwarnings("always", category=SATestSuiteWarning)
warnings.filterwarnings(
- "ignore",
- category=DeprecationWarning,
- message="The loop argument is deprecated",
+ "error", category=DeprecationWarning, module=origin
)
# ignore things that are deprecated *as of* 2.0 :)
pass
else:
warnings.filterwarnings(
- "once", category=pytest.PytestDeprecationWarning
+ "once", category=pytest.PytestDeprecationWarning, module=origin
)
from sqlalchemy.dialects.oracle.base import BINARY_FLOAT
from sqlalchemy.dialects.oracle.base import DOUBLE_PRECISION
from sqlalchemy.dialects.oracle.base import NUMBER
-from sqlalchemy.testing import assert_raises
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
"foo_id", Integer, ForeignKey("foo.id", onupdate="CASCADE")
),
)
- assert_raises(exc.SAWarning, bar.create, connection)
+ assert_warns(exc.SAWarning, bar.create, connection)
bat = Table(
"bat",
Column("foo_id", Integer),
ForeignKeyConstraint(["foo_id"], ["foo.id"], onupdate="CASCADE"),
)
- assert_raises(exc.SAWarning, bat.create, connection)
+ assert_warns(exc.SAWarning, bat.create, connection)
def test_reflect_check_include_all(self, connection):
insp = inspect(connection)
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import assert_raises
+from sqlalchemy.testing.assertions import assert_warns
from sqlalchemy.testing.assertions import AssertsExecutionResults
from sqlalchemy.testing.assertions import eq_
from sqlalchemy.testing.assertions import is_
base.PGDialect.ischema_names = {}
try:
m2 = MetaData()
- assert_raises(
+ assert_warns(
exc.SAWarning, Table, "testtable", m2, autoload_with=connection
)
from sqlalchemy.pool.base import _ConnDialect
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_context_ok
-from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import fixtures
c1 = p.connect()
rec = c1._connection_record
c1.close()
- assert_raises_message(
+ assert_warns_message(
Warning, "Double checkin attempted on %s" % rec, rec.checkin
)
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assert_raises_message_context_ok
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
util.warn("Exception attempting to detect")
eng.dialect._get_default_schema_name = get_default_schema_name
+ eng.dialect._check_unicode_description = mock.Mock()
return eng
def test_cursor_explode(self):
def test_cursor_shutdown_in_initialize(self):
db = self._fixture(True, True)
- assert_raises_message_context_ok(
+ assert_warns_message(
exc.SAWarning, "Exception attempting to detect", db.connect
)
+ # there's legacy py2k stuff happening here making this
+ # less smooth and probably buggy
eq_(
- db.pool.logger.error.mock_calls,
+ db.pool.logger.error.mock_calls[0:1],
[call("Error closing cursor", exc_info=True)],
)
id = Column(Integer, primary_key=True)
def test_column_named_twice(self):
- def go():
+ with assertions.expect_deprecated(
+ "A column with name 'x' is already present in table 'foo'"
+ ), expect_warnings(
+ "On class 'Foo', Column object 'x' named directly multiple times, "
+ "only one will be used: x, y",
+ ):
+
class Foo(Base):
__tablename__ = "foo"
x = Column("x", Integer)
y = Column("x", Integer)
- assert_raises_message(
- sa.exc.SAWarning,
+ def test_column_repeated_under_prop(self):
+ with assertions.expect_deprecated(
+ "A column with name 'x' is already present in table 'foo'"
+ ), expect_warnings(
"On class 'Foo', Column object 'x' named directly multiple times, "
- "only one will be used: x, y",
- go,
- )
+ "only one will be used: x, y, z",
+ ):
- def test_column_repeated_under_prop(self):
- def go():
class Foo(Base):
__tablename__ = "foo"
y = column_property(x)
z = Column("x", Integer)
- assert_raises_message(
- sa.exc.SAWarning,
- "On class 'Foo', Column object 'x' named directly multiple times, "
- "only one will be used: x, y, z",
- go,
- )
-
def test_using_explicit_prop_in_schema_objects(self):
class Foo(Base):
__tablename__ = "foo"
__tablename__ = "a"
id = Column(Integer, primary_key=True)
- assert_raises_message(
- sa.exc.SAWarning,
+ with expect_warnings(
"This declarative base already contains a class with ",
- lambda: type(Base)(
+ ):
+ type(Base)(
"Test",
(Base,),
dict(__tablename__="b", id=Column(Integer, primary_key=True)),
- ),
- )
+ )
@testing.teardown_events(MapperEvents)
@testing.teardown_events(InstrumentationEvents)
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.util import gc_collect
clsregistry.add_class("Foo", f1, base._class_registry)
gc_collect()
- assert_raises_message(
- exc.SAWarning,
+ with expect_warnings(
"This declarative base already contains a class with the "
"same class name and module name as foo.bar.Foo, and "
- "will be replaced in the string-lookup table.",
- clsregistry.add_class,
- "Foo",
- f2,
- base._class_registry,
- )
+ "will be replaced in the string-lookup table."
+ ):
+ clsregistry.add_class(
+ "Foo",
+ f2,
+ base._class_registry,
+ )
def test_resolve(self):
base = registry()
def my_prop(cls):
return Column("x", Integer)
- assert_raises_message(
- sa.exc.SAWarning,
+ with expect_warnings(
"Unmanaged access of declarative attribute my_prop "
- "from non-mapped class Mixin",
- getattr,
- Mixin,
- "my_prop",
- )
+ "from non-mapped class Mixin"
+ ):
+ Mixin.my_prop
def test_can_we_access_the_mixin_straight_special_names(self):
class Mixin(object):
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import assert_warns_message
from sqlalchemy.testing.assertsql import AllOf
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.assertsql import Conditional
c1 = C()
c1.class_name = "b"
sess.add(c1)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Flushing object %s with incompatible "
"polymorphic identity 'b'; the object may not "
b1 = B()
b1.class_name = "c"
sess.add(b1)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Flushing object %s with incompatible "
"polymorphic identity 'c'; the object may not "
b1 = B()
b1.class_name = "xyz"
sess.add(b1)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Flushing object %s with incompatible "
"polymorphic identity 'xyz'; the object may not "
sess.expire(c1)
c1.class_name = "b"
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Flushing object %s with incompatible "
"polymorphic identity 'b'; the object may not "
properties=dict(id=[employee_table.c.eid, person_table.c.id]),
primary_key=[person_table.c.id, employee_table.c.eid],
)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
r"On mapper mapped class Employee->employees, "
"primary key column 'persons.id' is being "
Sub, subtable_two, inherits=Base
)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Implicitly combining column base.base_id with "
"column subtable_two.base_id under attribute 'base_id'",
pass
self.mapper_registry.map_imperatively(P, parent)
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"Could not assemble any primary keys for locally mapped "
"table 'child' - no rows will be persisted in this Table.",
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
from sqlalchemy.testing import not_in
+from sqlalchemy.testing.assertions import assert_warns
from sqlalchemy.testing.mock import call
from sqlalchemy.testing.mock import Mock
from sqlalchemy.testing.util import all_partial_orderings
a1.bs.append(B())
state = attributes.instance_state(a1)
state._expire(state.dict, set())
- assert_raises(Warning, coll.append, B())
+ assert_warns(Warning, coll.append, B())
def test_replaced(self):
A, B = self.A, self.B
a1.bs.append(B())
state = attributes.instance_state(a1)
state._reset(state.dict, "bs")
- assert_raises(Warning, coll.append, B())
+ assert_warns(Warning, coll.append, B())
def test_ad_hoc_lazy(self):
A, B = self.A, self.B
a1.bs.append(B())
state = attributes.instance_state(a1)
_set_callable(state, state.dict, "bs", lambda: B())
- assert_raises(Warning, coll.append, B())
+ assert_warns(Warning, coll.append, B())
from sqlalchemy.orm.decl_api import declarative_base
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
def test_delete_orphan_without_delete(self):
Address = self.classes.Address
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
"The 'delete-orphan' cascade option requires 'delete'.",
relationship,
sess.add(u1)
assert u1 in sess
assert a1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_o2m_only_child_persistent(self):
User, Address = self.classes.User, self.classes.Address
sess.add(u1)
assert u1 in sess
assert a1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_o2m_backref_child_pending(self):
User, Address = self.classes.User, self.classes.Address
sess.add(u1)
assert u1 in sess
assert a1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_o2m_backref_child_transient_nochange(self):
User, Address = self.classes.User, self.classes.Address
sess.expunge(a1)
assert u1 in sess
assert a1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_o2m_backref_child_expunged_nochange(self):
User, Address = self.classes.User, self.classes.Address
sess.add(a1)
assert u1 not in sess
assert a1 in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2o_only_child_expunged(self):
User, Address = self.classes.User, self.classes.Address
sess.expunge(u1)
assert u1 not in sess
assert a1 in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2o_backref_child_pending(self):
User, Address = self.classes.User, self.classes.Address
sess.add(a1)
assert u1 not in sess
assert a1 in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2o_backref_child_expunged(self):
User, Address = self.classes.User, self.classes.Address
sess.expunge(u1)
assert u1 not in sess
assert a1 in sess
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning, "not in session", sess.flush
)
sess.expunge(u1)
assert u1 not in sess
assert a1 in sess
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning, "not in session", sess.flush
)
sess.add(i1)
assert i1 in sess
assert k1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2m_only_child_persistent(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
sess.add(i1)
assert i1 in sess
assert k1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2m_backref_child_pending(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
sess.add(i1)
assert i1 in sess
assert k1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2m_backref_child_transient_nochange(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
sess.expunge(k1)
assert i1 in sess
assert k1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.flush)
def test_m2m_backref_child_expunged_nochange(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
a1 = Address(email_address="a1")
a1.user = u1
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.commit)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.commit)
assert a1 not in sess
a1.dingalings.append(d1)
assert a1 not in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.commit)
+ assert_warns_message(sa_exc.SAWarning, "not in session", sess.commit)
class PendingOrphanTestSingleLevel(fixtures.MappedTest):
from sqlalchemy.orm import relationship
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
},
)
self.mapper_registry.map_imperatively(User, users)
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
"On relationship Address.user, 'dynamic' loaders cannot be "
"used with many-to-one/one-to-one relationships and/or "
from sqlalchemy.orm import undefer
from sqlalchemy.sql import operators
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
-from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
)
self.mapper_registry.map_imperatively(Order, orders)
s = fixture_session()
- assert_raises(
+ assert_warns(
sa.exc.SAWarning, s.query(User).options(joinedload(User.order)).all
)
from sqlalchemy.sql.traversals import NO_CACHE
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_not
m1 = Mock()
self.mapper_registry.map_imperatively(User, users)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
r"before_configured' and 'after_configured' ORM events only "
r"invoke with the mapper\(\) function or Mapper class as "
m1,
)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
r"before_configured' and 'after_configured' ORM events only "
r"invoke with the mapper\(\) function or Mapper class as "
u2 = User(name="u1", id=1)
sess.add(u2)
- assert_raises(sa.exc.SAWarning, sess.commit)
+
+ with expect_raises(sa.exc.IntegrityError), expect_warnings(
+ "New instance"
+ ):
+ sess.commit()
+
sess.rollback()
eq_(
canary,
u2 = User(name="u1", id=1)
sess.add(u2)
- assert_raises(sa.exc.SAWarning, sess.commit)
+ with expect_raises(sa.exc.IntegrityError), expect_warnings(
+ "New instance"
+ ):
+ sess.commit()
+
sess.rollback()
eq_(assertions, [True, True])
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import relationship
from sqlalchemy.testing import assert_raises
-from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import ne_
def __del__(self):
pass
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
r"__del__\(\) method on class "
r"<class '.*\.A'> will cause "
from sqlalchemy.orm import Session
from sqlalchemy.orm import with_parent
from sqlalchemy.testing import assert_raises
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
self.mapper_registry.map_imperatively(Order, orders)
s = fixture_session()
u1 = s.query(User).filter(User.id == 7).one()
- assert_raises(sa.exc.SAWarning, getattr, u1, "order")
+ assert_warns(sa.exc.SAWarning, getattr, u1, "order")
def test_callable_bind(self):
Address, addresses, users, User = (
from sqlalchemy.orm.persistence import _sort_states
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_deprecated_20
)
self.mapper(Address, addresses)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
"Property User.addresses on Mapper|User|users being replaced "
"with new property User.addresses; the old property will "
polymorphic_on=users.c.name,
polymorphic_identity="user",
)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
"Reassigning polymorphic association for identity 'user'",
self.mapper_registry.map_imperatively,
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import assert_raises
from sqlalchemy.testing.assertions import assert_raises_message
+from sqlalchemy.testing.assertions import assert_warns_message
from sqlalchemy.testing.assertions import eq_
from sqlalchemy.testing.assertions import expect_raises
from sqlalchemy.testing.assertions import expect_warnings
User = self.classes.User
s = fixture_session()
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
r"fully NULL primary key identity cannot load any object. "
"This condition may raise an error in a future release.",
s = fixture_session()
- assert_raises_message(
+ assert_warns_message(
sa_exc.SAWarning,
r"fully NULL primary key identity cannot load any object. "
"This condition may raise an error in a future release.",
from sqlalchemy.orm.interfaces import MANYTOONE
from sqlalchemy.orm.interfaces import ONETOMANY
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_raises_message
class _JoinFixtures(object):
)
def _assert_non_simple_warning(self, fn):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
"Non-simple column elements in "
"primary join condition for property "
self._join_fixture_o2m_composite_selfref_func_remote_side()
def test_determine_local_remote_pairs_o2m_overlap_func_warning(self):
- self._assert_non_simple_warning(
- self._join_fixture_m2o_sub_to_joined_sub_func
- )
+ with expect_raises_message(
+ exc.ArgumentError, "Could not locate any relevant"
+ ):
+ self._assert_non_simple_warning(
+ self._join_fixture_m2o_sub_to_joined_sub_func
+ )
def test_determine_local_remote_pairs_o2m_composite_selfref_func_annotated(
self,
from sqlalchemy.orm.interfaces import ONETOMANY
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
-from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.assertsql import assert_engine
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.fixtures import fixture_session
@testing.provide_metadata
def test_simple_warn(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship '(?:Child.parent|Parent.children)' will copy "
r"column parent.id to column child.parent_id, which conflicts "
@testing.provide_metadata
def test_double_rel_same_mapper_warns(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship 'Parent.child[12]' will copy column parent.id to "
r"column child.parent_id, which conflicts with relationship\(s\): "
@testing.provide_metadata
def test_warn_one(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship '(?:BSub1.a|BSub2.a_member|B.a)' will copy column "
r"(?:a.id|a_member.a_id) to column b.a_id",
@testing.provide_metadata
def test_warn_two(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship '(?:BSub1.a|B.a_member)' will copy column "
r"(?:a.id|a_member.a_id) to column b.a_id",
@testing.provide_metadata
def test_warn_three(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship '(?:BSub1.a|B.a_member|B.a)' will copy column "
r"(?:a.id|a_member.a_id) to column b.a_id",
@testing.provide_metadata
def test_warn_four(self):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship '(?:B.a|BSub2.a_member|B.a)' will copy column "
r"(?:a.id|a_member.a_id) to column b.a_id",
},
)
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"relationship .* will copy column .* to column "
r"employee_t.company_id, which conflicts with relationship\(s\)",
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
bind=testing.db,
)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
"At least one scoped session is already present. ",
Session.configure,
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import undefer
from sqlalchemy.orm import with_polymorphic
-from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
)
self.mapper_registry.map_imperatively(Order, orders)
s = fixture_session()
- assert_raises(
+ assert_warns(
sa.exc.SAWarning,
s.query(User).options(selectinload(User.order)).all,
)
from sqlalchemy.orm import was_deleted
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import assertions
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
def test_extra_dirty_state_post_flush_warning(self):
s, a1, a2 = self._test_extra_dirty_state()
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
"Attribute history events accumulated on 1 previously "
"clean instances",
def evt(mapper, conn, instance):
instance.addresses[0].user = User(name="u2")
- self._test(evt, "related attribute set")
+ with expect_raises_message(orm_exc.FlushError, ".*Over 100"):
+ self._test(evt, "related attribute set")
def test_m2o_cascade_remove(self):
def evt(mapper, conn, instance):
def evt(mapper, conn, instance):
object_session(instance).delete(Address(email="x1"))
- self._test(evt, r"Session.delete\(\)")
+ with expect_raises_message(
+ sa.exc.InvalidRequestError, ".*is not persisted"
+ ):
+ self._test(evt, r"Session.delete\(\)")
def _test(self, fn, method):
User = self.classes.User
u1 = User(name="u1", addresses=[Address(name="a1")])
s.add(u1)
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning, "Usage of the '%s'" % method, s.commit
)
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import undefer
from sqlalchemy.orm import with_polymorphic
-from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
)
self.mapper_registry.map_imperatively(Order, orders)
s = fixture_session()
- assert_raises(
+ assert_warns(
sa.exc.SAWarning,
s.query(User).options(subqueryload(User.order)).all,
)
from sqlalchemy.orm import Session
from sqlalchemy.orm import unitofwork
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
sess.delete(p1)
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"DELETE statement on table 'parent' expected to "
r"delete 1 row\(s\); 0 were matched.",
sess.delete(p1)
sess.delete(p2)
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"DELETE statement on table 'parent' expected to "
r"delete 2 row\(s\); 0 were matched.",
with patch.object(
config.db.dialect, "supports_sane_multi_rowcount", False
):
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
r"DELETE statement on table 'parent' expected to "
r"delete 1 row\(s\); 0 were matched.",
from sqlalchemy.orm import Session
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
s1.commit()
f1.value = "f1rev2"
- assert_raises(sa.exc.SAWarning, s1.commit)
+ assert_warns(sa.exc.SAWarning, s1.commit)
finally:
testing.db.dialect.supports_sane_rowcount = save
Base, base, version_id_col=base.c.version_id
)
- assert_raises_message(
+ assert_warns_message(
exc.SAWarning,
"Inheriting version_id_col 'version_id' does not "
"match inherited version_id_col 'version_id' and will not "
from sqlalchemy.sql import select
from sqlalchemy.sql import text
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
def test_nonunicode_default(self):
default = b("foo")
- assert_raises_message(
+ assert_warns_message(
sa.exc.SAWarning,
"Unicode column 'foobar' has non-unicode "
"default value b?'foo' specified.",
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_warnings
class MetaDataTest(fixtures.TestBase, ComparesTables):
def test_pk_col_mismatch_one(self):
m = MetaData()
- assert_raises_message(
- exc.SAWarning,
+
+ with expect_warnings(
"Table 't' specifies columns 'x' as primary_key=True, "
- "not matching locally specified columns 'q'",
- Table,
- "t",
- m,
- Column("x", Integer, primary_key=True),
- Column("q", Integer),
- PrimaryKeyConstraint("q"),
- )
+ "not matching locally specified columns 'q'"
+ ):
+ Table(
+ "t",
+ m,
+ Column("x", Integer, primary_key=True),
+ Column("q", Integer),
+ PrimaryKeyConstraint("q"),
+ )
def test_pk_col_mismatch_two(self):
m = MetaData()
- assert_raises_message(
- exc.SAWarning,
+
+ with expect_warnings(
"Table 't' specifies columns 'a', 'b', 'c' as primary_key=True, "
- "not matching locally specified columns 'b', 'c'",
- Table,
- "t",
- m,
- Column("a", Integer, primary_key=True),
- Column("b", Integer, primary_key=True),
- Column("c", Integer, primary_key=True),
- PrimaryKeyConstraint("b", "c"),
- )
+ "not matching locally specified columns 'b', 'c'"
+ ):
+
+ Table(
+ "t",
+ m,
+ Column("a", Integer, primary_key=True),
+ Column("b", Integer, primary_key=True),
+ Column("c", Integer, primary_key=True),
+ PrimaryKeyConstraint("b", "c"),
+ )
@testing.emits_warning("Table 't'")
def test_pk_col_mismatch_three(self):
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_not
+from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.types import ARRAY
from sqlalchemy.types import Boolean
from sqlalchemy.types import Concatenable
# these warning classes will change to ArgumentError when the
# deprecated behavior is disabled
- assert_raises_message(
- exc.SADeprecationWarning,
+ with expect_deprecated(
r"Invoking %(str_op)s\(\) without arguments is deprecated, and "
r"will be disallowed in a future release. For an empty "
r"%(str_op)s\(\) construct, use "
r"%(str_op)s\(%(str_continue)s, \*args\)\."
- % {"str_op": str_op, "str_continue": str_continue},
- op,
- )
+ % {"str_op": str_op, "str_continue": str_continue}
+ ):
+ op()
def test_empty_and_raw(self):
self.assert_compile(
from sqlalchemy.sql.sqltypes import TypeEngine
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import assert_warns
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import AssertsExecutionResults
from sqlalchemy.testing import engines
dialect.supports_unicode_binds = True
uni = u.dialect_impl(dialect).bind_processor(dialect)
if util.py3k:
- assert_raises(exc.SAWarning, uni, b"x")
+ assert_warns(exc.SAWarning, uni, b"x")
assert isinstance(uni(unicodedata), str)
else:
- assert_raises(exc.SAWarning, uni, "x")
+ assert_warns(exc.SAWarning, uni, "x")
assert isinstance(uni(unicodedata), unicode) # noqa
def test_unicode_warnings_typelevel_sqla_unicode(self):
dialect = default.DefaultDialect()
dialect.supports_unicode_binds = False
uni = u.dialect_impl(dialect).bind_processor(dialect)
- assert_raises(exc.SAWarning, uni, util.b("x"))
+ assert_warns(exc.SAWarning, uni, util.b("x"))
assert isinstance(uni(unicodedata), util.binary_type)
eq_(uni(unicodedata), unicodedata.encode("utf-8"))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
