summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
authorFederico Caselli <cfederico87@gmail.com>2020-04-08 20:26:21 +0200
committerFederico Caselli <cfederico87@gmail.com>2020-04-09 00:33:22 +0200
commita9b068ae564e5e775e312373088545b75aeaa1b0 (patch)
tree3a20c79acfefe49b62ee4bca360bb11001f9eec9 /test
parentecca4fe3f8aebc5b42c2acda2e5d28d6a90a821e (diff)
downloadsqlalchemy-a9b068ae564e5e775e312373088545b75aeaa1b0.tar.gz
Remove code deprecated before version 1.1
- Remove deprecated method ``get_primary_keys` in the :class:`.Dialect` and :class:`.Inspector` classes. - Remove deprecated event ``dbapi_error`` and the method ``ConnectionEvents.dbapi_error`. - Remove support for deprecated engine URLs of the form ``postgres://``. - Remove deprecated dialect ``mysql+gaerdbms``. - Remove deprecated parameter ``quoting`` from :class:`.mysql.ENUM` and :class:`.mysql.SET` in the ``mysql`` dialect. - Remove deprecated function ``comparable_property``. and function ``comparable_using`` in the declarative extension. - Remove deprecated function ``compile_mappers``. - Remove deprecated method ``collection.linker``. - Remove deprecated method ``Session.prune`` and parameter ``Session.weak_identity_map``. This change also removes the class ``StrongInstanceDict``. - Remove deprecated parameter ``mapper.order_by``. - Remove deprecated parameter ``Session._enable_transaction_accounting`. - Remove deprecated parameter ``Session.is_modified.passive``. - Remove deprecated class ``Binary``. Please use :class:`.LargeBinary`. - Remove deprecated methods ``Compiled.compile``, ``ClauseElement.__and__`` and ``ClauseElement.__or__`` and attribute ``Over.func``. - Remove deprecated ``FromClause.count`` method. - Remove deprecated parameter ``Table.useexisting``. - Remove deprecated parameters ``text.bindparams`` and ``text.typemap``. - Remove boolean support for the ``passive`` parameter in ``get_history``. - Remove deprecated ``adapt_operator`` in ``UserDefinedType.Comparator``. Fixes: #4643 Change-Id: Idcd390c77bf7b0e9957907716993bdaa3f1a1763
Diffstat (limited to 'test')
-rw-r--r--test/dialect/mysql/test_deprecations.py20
-rw-r--r--test/dialect/mysql/test_reflection.py3
-rw-r--r--test/dialect/mysql/test_types.py74
-rw-r--r--test/dialect/postgresql/test_dialect.py9
-rw-r--r--test/engine/test_deprecations.py82
-rw-r--r--test/engine/test_execute.py2
-rw-r--r--test/orm/test_attributes.py25
-rw-r--r--test/orm/test_deprecations.py760
-rw-r--r--test/orm/test_session.py9
-rw-r--r--test/sql/test_deprecations.py60
-rw-r--r--test/sql/test_types.py1
-rw-r--r--test/test_deprecations.py3
12 files changed, 61 insertions, 987 deletions
diff --git a/test/dialect/mysql/test_deprecations.py b/test/dialect/mysql/test_deprecations.py
new file mode 100644
index 000000000..544284a21
--- /dev/null
+++ b/test/dialect/mysql/test_deprecations.py
@@ -0,0 +1,20 @@
+from sqlalchemy.dialects.mysql import ENUM
+from sqlalchemy.dialects.mysql import SET
+from sqlalchemy.testing import expect_deprecated_20
+from sqlalchemy.testing import fixtures
+
+
+class DeprecateQuoting(fixtures.TestBase):
+ def test_enum_warning(self):
+ ENUM("a", "b")
+ with expect_deprecated_20(
+ "The 'quoting' parameter to :class:`.mysql.ENUM` is deprecated."
+ ):
+ ENUM("a", quoting="foo")
+
+ def test_set_warning(self):
+ SET("a", "b")
+ with expect_deprecated_20(
+ "The 'quoting' parameter to :class:`.mysql.SET` is deprecated.*"
+ ):
+ SET("a", quoting="foo")
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index d5ff1314b..2de530607 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -209,10 +209,9 @@ class TypeReflectionTest(fixtures.TestBase):
]
self._run_test(specs, [])
- @testing.uses_deprecated("Manually quoting ENUM value literals")
def test_legacy_enum_types(self):
- specs = [(mysql.ENUM("''", "'fleem'"), mysql.ENUM("''", "'fleem'"))]
+ specs = [(mysql.ENUM("", "fleem"), mysql.ENUM("", "fleem"))]
self._run_test(specs, ["enums"])
diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py
index 010165ac2..74f743852 100644
--- a/test/dialect/mysql/test_types.py
+++ b/test/dialect/mysql/test_types.py
@@ -820,10 +820,10 @@ class EnumSetTest(
def test_enum(self):
"""Exercise the ENUM type."""
- with testing.expect_deprecated("Manually quoting ENUM value literals"):
- e1, e2 = mysql.ENUM("'a'", "'b'"), mysql.ENUM("'a'", "'b'")
- e3 = mysql.ENUM("'a'", "'b'", strict=True)
- e4 = mysql.ENUM("'a'", "'b'", strict=True)
+ e1 = mysql.ENUM("a", "b")
+ e2 = mysql.ENUM("a", "b")
+ e3 = mysql.ENUM("a", "b", strict=True)
+ e4 = mysql.ENUM("a", "b", strict=True)
enum_table = Table(
"mysql_enum",
@@ -951,10 +951,10 @@ class EnumSetTest(
eq_(res, expected)
def _set_fixture_one(self):
- with testing.expect_deprecated("Manually quoting SET value literals"):
- e1, e2 = mysql.SET("'a'", "'b'"), mysql.SET("'a'", "'b'")
- e4 = mysql.SET("'a'", "b")
- e5 = mysql.SET("'a'", "'b'", quoting="quoted")
+ e1 = mysql.SET("a", "b")
+ e2 = mysql.SET("a", "b")
+ e4 = mysql.SET("'a'", "b")
+ e5 = mysql.SET("a", "b")
set_table = Table(
"mysql_set",
@@ -1246,18 +1246,17 @@ class EnumSetTest(
@testing.exclude("mysql", "<", (4,), "3.23 can't handle an ENUM of ''")
def test_enum_parse(self):
- with testing.expect_deprecated("Manually quoting ENUM value literals"):
- enum_table = Table(
- "mysql_enum",
- self.metadata,
- Column("e1", mysql.ENUM("'a'")),
- Column("e2", mysql.ENUM("''")),
- Column("e3", mysql.ENUM("a")),
- Column("e4", mysql.ENUM("")),
- Column("e5", mysql.ENUM("'a'", "''")),
- Column("e6", mysql.ENUM("''", "'a'")),
- Column("e7", mysql.ENUM("''", "'''a'''", "'b''b'", "''''")),
- )
+ enum_table = Table(
+ "mysql_enum",
+ self.metadata,
+ Column("e1", mysql.ENUM("a")),
+ Column("e2", mysql.ENUM("")),
+ Column("e3", mysql.ENUM("a")),
+ Column("e4", mysql.ENUM("")),
+ Column("e5", mysql.ENUM("a", "")),
+ Column("e6", mysql.ENUM("", "a")),
+ Column("e7", mysql.ENUM("", "'a'", "b'b", "'")),
+ )
for col in enum_table.c:
self.assert_(repr(col))
@@ -1276,27 +1275,20 @@ class EnumSetTest(
@testing.provide_metadata
@testing.exclude("mysql", "<", (5,))
def test_set_parse(self):
- with testing.expect_deprecated("Manually quoting SET value literals"):
- set_table = Table(
- "mysql_set",
- self.metadata,
- Column("e1", mysql.SET("'a'")),
- Column("e2", mysql.SET("''", retrieve_as_bitwise=True)),
- Column("e3", mysql.SET("a")),
- Column("e4", mysql.SET("", retrieve_as_bitwise=True)),
- Column("e5", mysql.SET("'a'", "''", retrieve_as_bitwise=True)),
- Column("e6", mysql.SET("''", "'a'", retrieve_as_bitwise=True)),
- Column(
- "e7",
- mysql.SET(
- "''",
- "'''a'''",
- "'b''b'",
- "''''",
- retrieve_as_bitwise=True,
- ),
- ),
- )
+ set_table = Table(
+ "mysql_set",
+ self.metadata,
+ Column("e1", mysql.SET("a")),
+ Column("e2", mysql.SET("", retrieve_as_bitwise=True)),
+ Column("e3", mysql.SET("a")),
+ Column("e4", mysql.SET("", retrieve_as_bitwise=True)),
+ Column("e5", mysql.SET("a", "", retrieve_as_bitwise=True)),
+ Column("e6", mysql.SET("", "a", retrieve_as_bitwise=True)),
+ Column(
+ "e7",
+ mysql.SET("", "'a'", "b'b", "'", retrieve_as_bitwise=True,),
+ ),
+ )
for col in set_table.c:
self.assert_(repr(col))
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index d92559ac3..d381d83fa 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -9,7 +9,6 @@ from sqlalchemy import bindparam
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import DateTime
-from sqlalchemy import dialects
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import extract
@@ -98,14 +97,6 @@ class DialectTest(fixtures.TestBase):
]:
eq_(dialect._get_server_version_info(mock_conn(string)), version)
- def test_deprecated_dialect_name_still_loads(self):
- dialects.registry.clear()
- with expect_deprecated(
- "The 'postgres' dialect name " "has been renamed to 'postgresql'"
- ):
- dialect = url.URL("postgres").get_dialect()
- is_(dialect, postgresql.dialect)
-
@testing.requires.psycopg2_compatibility
def test_pg_dialect_use_native_unicode_from_config(self):
config = {
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index edf1503d5..c9bf83f31 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -1,22 +1,17 @@
import sqlalchemy as tsa
-from sqlalchemy import column
from sqlalchemy import create_engine
-from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import INT
from sqlalchemy import Integer
-from sqlalchemy import literal
from sqlalchemy import MetaData
from sqlalchemy import pool
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
-from sqlalchemy import TypeDecorator
from sqlalchemy import VARCHAR
from sqlalchemy.engine import reflection
from sqlalchemy.engine.base import Connection
-from sqlalchemy.engine.base import Engine
from sqlalchemy.engine.mock import MockConnection
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
@@ -255,83 +250,6 @@ class HandleInvalidatedOnConnectTest(fixtures.TestBase):
self.ProgrammingError = sqlite3.ProgrammingError
-class HandleErrorTest(fixtures.TestBase):
- __requires__ = ("ad_hoc_engines",)
- __backend__ = True
-
- def tearDown(self):
- Engine.dispatch._clear()
- Engine._has_events = False
-
- def test_legacy_dbapi_error(self):
- engine = engines.testing_engine()
- canary = Mock()
-
- with testing.expect_deprecated(
- r"The ConnectionEvents.dbapi_error\(\) event is deprecated"
- ):
- event.listen(engine, "dbapi_error", canary)
-
- with engine.connect() as conn:
- try:
- conn.exec_driver_sql("SELECT FOO FROM I_DONT_EXIST")
- assert False
- except tsa.exc.DBAPIError as e:
- eq_(canary.mock_calls[0][1][5], e.orig)
- eq_(canary.mock_calls[0][1][2], "SELECT FOO FROM I_DONT_EXIST")
-
- def test_legacy_dbapi_error_no_ad_hoc_context(self):
- engine = engines.testing_engine()
-
- listener = Mock(return_value=None)
- with testing.expect_deprecated(
- r"The ConnectionEvents.dbapi_error\(\) event is deprecated"
- ):
- event.listen(engine, "dbapi_error", listener)
-
- nope = SomeException("nope")
-
- class MyType(TypeDecorator):
- impl = Integer
-
- def process_bind_param(self, value, dialect):
- raise nope
-
- with engine.connect() as conn:
- assert_raises_message(
- tsa.exc.StatementError,
- r"\(.*SomeException\) " r"nope\n\[SQL\: u?SELECT 1 ",
- conn.execute,
- select([1]).where(column("foo") == literal("bar", MyType())),
- )
- # no legacy event
- eq_(listener.mock_calls, [])
-
- def test_legacy_dbapi_error_non_dbapi_error(self):
- engine = engines.testing_engine()
-
- listener = Mock(return_value=None)
- with testing.expect_deprecated(
- r"The ConnectionEvents.dbapi_error\(\) event is deprecated"
- ):
- event.listen(engine, "dbapi_error", listener)
-
- nope = TypeError("I'm not a DBAPI error")
- with engine.connect() as c:
- c.connection.cursor = Mock(
- return_value=Mock(execute=Mock(side_effect=nope))
- )
-
- assert_raises_message(
- TypeError,
- "I'm not a DBAPI error",
- c.exec_driver_sql,
- "select ",
- )
- # no legacy event
- eq_(listener.mock_calls, [])
-
-
def MockDBAPI(): # noqa
def cursor():
return Mock()
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 566cf0654..e2c009e2d 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2120,7 +2120,7 @@ class HandleErrorTest(fixtures.TestBase):
is_(ctx.original_exception, nope)
def test_exception_event_non_dbapi_error(self):
- """test that dbapi_error is called with a context in
+ """test that handle_error is called with a context in
cases where DBAPI raises an exception that is not a DBAPI
exception, e.g. internal errors or encoding problems.
diff --git a/test/orm/test_attributes.py b/test/orm/test_attributes.py
index bb3399a5a..43a7d8dc8 100644
--- a/test/orm/test_attributes.py
+++ b/test/orm/test_attributes.py
@@ -1636,7 +1636,9 @@ class PendingBackrefTest(fixtures.ORMTest):
# then this would fail.
eq_(
- Blog.posts.impl.get_history(state, dict_, passive=True),
+ Blog.posts.impl.get_history(
+ state, dict_, passive=attributes.PASSIVE_NO_INITIALIZE
+ ),
([p2], (), ()),
)
@@ -2903,27 +2905,6 @@ class HistoryTest(fixtures.TestBase):
([f1], (), ()),
)
- def test_deprecated_flags(self):
- assert_raises_message(
- sa_exc.SADeprecationWarning,
- "Passing True for 'passive' is deprecated. "
- "Use attributes.PASSIVE_NO_INITIALIZE",
- attributes.get_history,
- object(),
- "foo",
- True,
- )
-
- assert_raises_message(
- sa_exc.SADeprecationWarning,
- "Passing False for 'passive' is deprecated. "
- "Use attributes.PASSIVE_OFF",
- attributes.get_history,
- object(),
- "foo",
- False,
- )
-
class LazyloadHistoryTest(fixtures.TestBase):
def test_lazy_backref_collections(self):
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index 475e1c4ec..c63379504 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -1,7 +1,6 @@
import sqlalchemy as sa
from sqlalchemy import and_
from sqlalchemy import desc
-from sqlalchemy import event
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import select
@@ -9,13 +8,10 @@ from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import true
-from sqlalchemy.ext.declarative import comparable_using
-from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import aliased
from sqlalchemy.orm import attributes
from sqlalchemy.orm import collections
from sqlalchemy.orm import column_property
-from sqlalchemy.orm import comparable_property
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import contains_alias
from sqlalchemy.orm import contains_eager
@@ -24,15 +20,12 @@ from sqlalchemy.orm import defer
from sqlalchemy.orm import deferred
from sqlalchemy.orm import eagerload
from sqlalchemy.orm import foreign
-from sqlalchemy.orm import identity
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import mapper
-from sqlalchemy.orm import PropComparator
from sqlalchemy.orm import relation
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
-from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import synonym
from sqlalchemy.orm import undefer
from sqlalchemy.orm import with_polymorphic
@@ -47,355 +40,10 @@ from sqlalchemy.testing import is_
from sqlalchemy.testing import is_true
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
-from sqlalchemy.testing.util import gc_collect
from . import _fixtures
from .inheritance import _poly_fixtures
from .test_options import PathTest as OptionsPathTest
from .test_query import QueryTest
-from .test_transaction import _LocalFixture
-
-
-class DeprecationWarningsTest(fixtures.DeclarativeMappedTest):
- run_setup_classes = "each"
- run_setup_mappers = "each"
- run_define_tables = "each"
- run_create_tables = None
-
- def test_session_weak_identity_map(self):
- with testing.expect_deprecated(
- ".*Session.weak_identity_map parameter as well as the"
- ):
- s = Session(weak_identity_map=True)
-
- is_(s._identity_cls, identity.WeakInstanceDict)
-
- with assertions.expect_deprecated(
- "The Session.weak_identity_map parameter as well as"
- ):
- s = Session(weak_identity_map=False)
-
- is_(s._identity_cls, identity.StrongInstanceDict)
-
- s = Session()
- is_(s._identity_cls, identity.WeakInstanceDict)
-
- def test_session_prune(self):
- s = Session()
-
- with assertions.expect_deprecated(
- r"The Session.prune\(\) method is deprecated along with "
- "Session.weak_identity_map"
- ):
- s.prune()
-
- def test_session_enable_transaction_accounting(self):
- with assertions.expect_deprecated(
- "the Session._enable_transaction_accounting parameter is "
- "deprecated"
- ):
- Session(_enable_transaction_accounting=False)
-
- def test_session_is_modified(self):
- class Foo(self.DeclarativeBasic):
- __tablename__ = "foo"
-
- id = Column(Integer, primary_key=True)
-
- f1 = Foo()
- s = Session()
- with assertions.expect_deprecated(
- "The Session.is_modified.passive flag is deprecated"
- ):
- # this flag was for a long time documented as requiring
- # that it be set to True, so we've changed the default here
- # so that the warning emits
- s.is_modified(f1, passive=True)
-
-
-class DeprecatedAccountingFlagsTest(_LocalFixture):
- def test_rollback_no_accounting(self):
- User, users = self.classes.User, self.tables.users
-
- with testing.expect_deprecated(
- "The Session._enable_transaction_accounting parameter"
- ):
- sess = sessionmaker(_enable_transaction_accounting=False)()
- u1 = User(name="ed")
- sess.add(u1)
- sess.commit()
-
- u1.name = "edwardo"
- sess.rollback()
-
- testing.db.execute(
- users.update(users.c.name == "ed").values(name="edward")
- )
-
- assert u1.name == "edwardo"
- sess.expire_all()
- assert u1.name == "edward"
-
- def test_commit_no_accounting(self):
- User, users = self.classes.User, self.tables.users
-
- with testing.expect_deprecated(
- "The Session._enable_transaction_accounting parameter"
- ):
- sess = sessionmaker(_enable_transaction_accounting=False)()
- u1 = User(name="ed")
- sess.add(u1)
- sess.commit()
-
- u1.name = "edwardo"
- sess.rollback()
-
- testing.db.execute(
- users.update(users.c.name == "ed").values(name="edward")
- )
-
- assert u1.name == "edwardo"
- sess.commit()
-
- assert testing.db.execute(select([users.c.name])).fetchall() == [
- ("edwardo",)
- ]
- assert u1.name == "edwardo"
-
- sess.delete(u1)
- sess.commit()
-
- def test_preflush_no_accounting(self):
- User, users = self.classes.User, self.tables.users
-
- with testing.expect_deprecated(
- "The Session._enable_transaction_accounting parameter"
- ):
- sess = Session(
- _enable_transaction_accounting=False,
- autocommit=True,
- autoflush=False,
- )
- u1 = User(name="ed")
- sess.add(u1)
- sess.flush()
-
- sess.begin()
- u1.name = "edwardo"
- u2 = User(name="some other user")
- sess.add(u2)
-
- sess.rollback()
-
- sess.begin()
- assert testing.db.execute(select([users.c.name])).fetchall() == [
- ("ed",)
- ]
-
-
-class DeprecatedSessionFeatureTest(_fixtures.FixtureTest):
- run_inserts = None
-
- def test_fast_discard_race(self):
- # test issue #4068
- users, User = self.tables.users, self.classes.User
-
- mapper(User, users)
-
- with testing.expect_deprecated(".*identity map are deprecated"):
- sess = Session(weak_identity_map=False)
-
- u1 = User(name="u1")
- sess.add(u1)
- sess.commit()
-
- u1_state = u1._sa_instance_state
- sess.identity_map._dict.pop(u1_state.key)
- ref = u1_state.obj
- u1_state.obj = lambda: None
-
- u2 = sess.query(User).first()
- u1_state._cleanup(ref)
-
- u3 = sess.query(User).first()
-
- is_(u2, u3)
-
- u2_state = u2._sa_instance_state
- assert sess.identity_map.contains_state(u2._sa_instance_state)
- ref = u2_state.obj
- u2_state.obj = lambda: None
- u2_state._cleanup(ref)
- assert not sess.identity_map.contains_state(u2._sa_instance_state)
-
- def test_is_modified_passive_on(self):
- User, Address = self.classes.User, self.classes.Address
- users, addresses = self.tables.users, self.tables.addresses
- mapper(User, users, properties={"addresses": relationship(Address)})
- mapper(Address, addresses)
-
- s = Session()
- u = User(name="fred", addresses=[Address(email_address="foo")])
- s.add(u)
- s.commit()
-
- u.id
-
- def go():
- assert not s.is_modified(u, passive=True)
-
- with testing.expect_deprecated(
- ".*Session.is_modified.passive flag is deprecated "
- ):
- self.assert_sql_count(testing.db, go, 0)
-
- u.name = "newname"
-
- def go():
- assert s.is_modified(u, passive=True)
-
- with testing.expect_deprecated(
- ".*Session.is_modified.passive flag is deprecated "
- ):
- self.assert_sql_count(testing.db, go, 0)
-
-
-class StrongIdentityMapTest(_fixtures.FixtureTest):
- run_inserts = None
-
- def _strong_ident_fixture(self):
- with testing.expect_deprecated(
- ".*Session.weak_identity_map parameter as well as the"
- ):
- sess = create_session(weak_identity_map=False)
-
- def prune():
- with testing.expect_deprecated(".*Session.prune"):
- return sess.prune()
-
- return sess, prune
-
- def _event_fixture(self):
- session = create_session()
-
- @event.listens_for(session, "pending_to_persistent")
- @event.listens_for(session, "deleted_to_persistent")
- @event.listens_for(session, "detached_to_persistent")
- @event.listens_for(session, "loaded_as_persistent")
- def strong_ref_object(sess, instance):
- if "refs" not in sess.info:
- sess.info["refs"] = refs = set()
- else:
- refs = sess.info["refs"]
-
- refs.add(instance)
-
- @event.listens_for(session, "persistent_to_detached")
- @event.listens_for(session, "persistent_to_deleted")
- @event.listens_for(session, "persistent_to_transient")
- def deref_object(sess, instance):
- sess.info["refs"].discard(instance)
-
- def prune():
- if "refs" not in session.info:
- return 0
-
- sess_size = len(session.identity_map)
- session.info["refs"].clear()
- gc_collect()
- session.info["refs"] = set(
- s.obj() for s in session.identity_map.all_states()
- )
- return sess_size - len(session.identity_map)
-
- return session, prune
-
- def test_strong_ref_imap(self):
- self._test_strong_ref(self._strong_ident_fixture)
-
- def test_strong_ref_events(self):
- self._test_strong_ref(self._event_fixture)
-
- def _test_strong_ref(self, fixture):
- s, prune = fixture()
-
- users, User = self.tables.users, self.classes.User
-
- mapper(User, users)
-
- # save user
- s.add(User(name="u1"))
- s.flush()
- user = s.query(User).one()
- user = None
- print(s.identity_map)
- gc_collect()
- assert len(s.identity_map) == 1
-
- user = s.query(User).one()
- assert not s.identity_map._modified
- user.name = "u2"
- assert s.identity_map._modified
- s.flush()
- eq_(users.select().execute().fetchall(), [(user.id, "u2")])
-
- def test_prune_imap(self):
- self._test_prune(self._strong_ident_fixture)
-
- def test_prune_events(self):
- self._test_prune(self._event_fixture)
-
- @testing.requires.cpython
- def _test_prune(self, fixture):
- s, prune = fixture()
-
- users, User = self.tables.users, self.classes.User
-
- mapper(User, users)
-
- for o in [User(name="u%s" % x) for x in range(10)]:
- s.add(o)
- # o is still live after this loop...
-
- self.assert_(len(s.identity_map) == 0)
- eq_(prune(), 0)
- s.flush()
- gc_collect()
- eq_(prune(), 9)
- # o is still in local scope here, so still present
- self.assert_(len(s.identity_map) == 1)
-
- id_ = o.id
- del o
- eq_(prune(), 1)
- self.assert_(len(s.identity_map) == 0)
-
- u = s.query(User).get(id_)
- eq_(prune(), 0)
- self.assert_(len(s.identity_map) == 1)
- u.name = "squiznart"
- del u
- eq_(prune(), 0)
- self.assert_(len(s.identity_map) == 1)
- s.flush()
- eq_(prune(), 1)
- self.assert_(len(s.identity_map) == 0)
-
- s.add(User(name="x"))
- eq_(prune(), 0)
- self.assert_(len(s.identity_map) == 0)
- s.flush()
- self.assert_(len(s.identity_map) == 1)
- eq_(prune(), 1)
- self.assert_(len(s.identity_map) == 0)
-
- u = s.query(User).get(id_)
- s.delete(u)
- del u
- eq_(prune(), 0)
- self.assert_(len(s.identity_map) == 1)
- s.flush()
- eq_(prune(), 0)
- self.assert_(len(s.identity_map) == 0)
class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
@@ -952,152 +600,6 @@ class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
)
is_true(dep.compare(subq_version))
- def test_cancel_order_by(self):
- users, User = self.tables.users, self.classes.User
-
- with testing.expect_deprecated(
- "The Mapper.order_by parameter is deprecated, and will be "
- "removed in a future release."
- ):
- mapper(User, users, order_by=users.c.name.desc())
-
- assert (
- "order by users.name desc"
- in str(create_session().query(User).statement).lower()
- )
- assert (
- "order by"
- not in str(
- create_session().query(User).order_by(None).statement
- ).lower()
- )
- assert (
- "order by users.name asc"
- in str(
- create_session()
- .query(User)
- .order_by(User.name.asc())
- .statement
- ).lower()
- )
-
- eq_(
- create_session().query(User).all(),
- [
- User(id=7, name="jack"),
- User(id=9, name="fred"),
- User(id=8, name="ed"),
- User(id=10, name="chuck"),
- ],
- )
-
- eq_(
- create_session().query(User).order_by(User.name).all(),
- [
- User(id=10, name="chuck"),
- User(id=8, name="ed"),
- User(id=9, name="fred"),
- User(id=7, name="jack"),
- ],
- )
-
- def test_comparable(self):
- users = self.tables.users
-
- class extendedproperty(property):
- attribute = 123
-
- def method1(self):
- return "method1"
-
- from sqlalchemy.orm.properties import ColumnProperty
-
- class UCComparator(ColumnProperty.Comparator):
- __hash__ = None
-
- def method1(self):
- return "uccmethod1"
-
- def method2(self, other):
- return "method2"
-
- def __eq__(self, other):
- cls = self.prop.parent.class_
- col = getattr(cls, "name")
- if other is None:
- return col is None
- else:
- return sa.func.upper(col) == sa.func.upper(other)
-
- def map_(with_explicit_property):
- class User(object):
- @extendedproperty
- def uc_name(self):
- if self.name is None:
- return None
- return self.name.upper()
-
- if with_explicit_property:
- args = (UCComparator, User.uc_name)
- else:
- args = (UCComparator,)
-
- with assertions.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
- mapper(
- User,
- users,
- properties=dict(uc_name=sa.orm.comparable_property(*args)),
- )
- return User
-
- for User in (map_(True), map_(False)):
- sess = create_session()
- sess.begin()
- q = sess.query(User)
-
- assert hasattr(User, "name")
- assert hasattr(User, "uc_name")
-
- eq_(User.uc_name.method1(), "method1")
- eq_(User.uc_name.method2("x"), "method2")
-
- assert_raises_message(
- AttributeError,
- "Neither 'extendedproperty' object nor 'UCComparator' "
- "object associated with User.uc_name has an attribute "
- "'nonexistent'",
- getattr,
- User.uc_name,
- "nonexistent",
- )
-
- # test compile
- assert not isinstance(User.uc_name == "jack", bool)
- u = q.filter(User.uc_name == "JACK").one()
-
- assert u.uc_name == "JACK"
- assert u not in sess.dirty
-
- u.name = "some user name"
- eq_(u.name, "some user name")
- assert u in sess.dirty
- eq_(u.uc_name, "SOME USER NAME")
-
- sess.flush()
- sess.expunge_all()
-
- q = sess.query(User)
- u2 = q.filter(User.name == "some user name").one()
- u3 = q.filter(User.uc_name == "SOME USER NAME").one()
-
- assert u2 is u3
-
- eq_(User.uc_name.attribute, 123)
- sess.rollback()
-
def test_comparable_column(self):
users, User = self.tables.users, self.classes.User
@@ -1151,25 +653,6 @@ class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
"users.name &= :name_1",
)
- def test_info(self):
- class MyComposite(object):
- pass
-
- with assertions.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
- for constructor, args in [(comparable_property, "foo")]:
- obj = constructor(info={"x": "y"}, *args)
- eq_(obj.info, {"x": "y"})
- obj.info["q"] = "p"
- eq_(obj.info, {"x": "y", "q": "p"})
-
- obj = constructor(*args)
- eq_(obj.info, {})
- obj.info["q"] = "p"
- eq_(obj.info, {"q": "p"})
-
def test_add_property(self):
users = self.tables.users
@@ -1186,37 +669,10 @@ class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
name = property(_get_name, _set_name)
- def _uc_name(self):
- if self._name is None:
- return None
- return self._name.upper()
-
- uc_name = property(_uc_name)
- uc_name2 = property(_uc_name)
-
m = mapper(User, users)
- class UCComparator(PropComparator):
- __hash__ = None
-
- def __eq__(self, other):
- cls = self.prop.parent.class_
- col = getattr(cls, "name")
- if other is None:
- return col is None
- else:
- return func.upper(col) == func.upper(other)
-
m.add_property("_name", deferred(users.c.name))
m.add_property("name", synonym("_name"))
- with assertions.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
- m.add_property("uc_name", comparable_property(UCComparator))
- m.add_property(
- "uc_name2", comparable_property(UCComparator, User.uc_name2)
- )
sess = create_session(autocommit=False)
assert sess.query(User).get(7)
@@ -1225,116 +681,10 @@ class DeprecatedMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
def go():
eq_(u.name, "jack")
- eq_(u.uc_name, "JACK")
- eq_(u.uc_name2, "JACK")
eq_(assert_col, [("get", "jack")], str(assert_col))
self.sql_count_(1, go)
- def test_kwarg_accepted(self):
- class DummyComposite(object):
- def __init__(self, x, y):
- pass
-
- class MyFactory(PropComparator):
- pass
-
- with assertions.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
- for args in ((comparable_property,),):
- fn = args[0]
- args = args[1:]
- fn(comparator_factory=MyFactory, *args)
-
- def test_merge_synonym_comparable(self):
- users = self.tables.users
-
- class User(object):
- class Comparator(PropComparator):
- pass
-
- def _getValue(self):
- return self._value
-
- def _setValue(self, value):
- setattr(self, "_value", value)
-
- value = property(_getValue, _setValue)
-
- with assertions.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
- mapper(
- User,
- users,
- properties={
- "uid": synonym("id"),
- "foobar": comparable_property(User.Comparator, User.value),
- },
- )
-
- sess = create_session()
- u = User()
- u.name = "ed"
- sess.add(u)
- sess.flush()
- sess.expunge(u)
- sess.merge(u)
-
-
-class DeprecatedDeclTest(fixtures.TestBase):
- @testing.provide_metadata
- def test_comparable_using(self):
- class NameComparator(sa.orm.PropComparator):
- @property
- def upperself(self):
- cls = self.prop.parent.class_
- col = getattr(cls, "name")
- return sa.func.upper(col)
-
- def operate(self, op, other, **kw):
- return op(self.upperself, other, **kw)
-
- Base = declarative_base(metadata=self.metadata)
-
- with testing.expect_deprecated(
- r"comparable_property\(\) is deprecated and will be "
- "removed in a future release."
- ):
-
- class User(Base, fixtures.ComparableEntity):
-
- __tablename__ = "users"
- id = Column(
- "id",
- Integer,
- primary_key=True,
- test_needs_autoincrement=True,
- )
- name = Column("name", String(50))
-
- @comparable_using(NameComparator)
- @property
- def uc_name(self):
- return self.name is not None and self.name.upper() or None
-
- Base.metadata.create_all()
- sess = create_session()
- u1 = User(name="someuser")
- eq_(u1.name, "someuser", u1.name)
- eq_(u1.uc_name, "SOMEUSER", u1.uc_name)
- sess.add(u1)
- sess.flush()
- sess.expunge_all()
- rt = sess.query(User).filter(User.uc_name == "SOMEUSER").one()
- eq_(rt, u1)
- sess.expunge_all()
- rt = sess.query(User).filter(User.uc_name.startswith("SOMEUSE")).one()
- eq_(rt, u1)
-
class DeprecatedOptionAllTest(OptionsPathTest, _fixtures.FixtureTest):
run_inserts = "once"
@@ -1397,84 +747,6 @@ class DeprecatedOptionAllTest(OptionsPathTest, _fixtures.FixtureTest):
*options
)
- def test_subqueryload_mapper_order_by(self):
- users, User, Address, addresses = (
- self.tables.users,
- self.classes.User,
- self.classes.Address,
- self.tables.addresses,
- )
-
- mapper(Address, addresses)
-
- with testing.expect_deprecated(
- ".*Mapper.order_by parameter is deprecated"
- ):
- mapper(
- User,
- users,
- properties={
- "addresses": relationship(
- Address, lazy="subquery", order_by=addresses.c.id
- )
- },
- order_by=users.c.id.desc(),
- )
-
- sess = create_session()
- q = sess.query(User)
-
- result = q.limit(2).all()
- eq_(result, list(reversed(self.static.user_address_result[2:4])))
-
- def test_selectinload_mapper_order_by(self):
- users, User, Address, addresses = (
- self.tables.users,
- self.classes.User,
- self.classes.Address,
- self.tables.addresses,
- )
-
- mapper(Address, addresses)
- with testing.expect_deprecated(
- ".*Mapper.order_by parameter is deprecated"
- ):
- mapper(
- User,
- users,
- properties={
- "addresses": relationship(
- Address, lazy="selectin", order_by=addresses.c.id
- )
- },
- order_by=users.c.id.desc(),
- )
-
- sess = create_session()
- q = sess.query(User)
-
- result = q.limit(2).all()
- eq_(result, list(reversed(self.static.user_address_result[2:4])))
-
- def test_join_mapper_order_by(self):
- """test that mapper-level order_by is adapted to a selectable."""
-
- User, users = self.classes.User, self.tables.users
-
- with testing.expect_deprecated(
- ".*Mapper.order_by parameter is deprecated"
- ):
- mapper(User, users, order_by=users.c.id)
-
- sel = users.select(users.c.id.in_([7, 8]))
- sess = create_session()
-
- with DeprecatedQueryTest._expect_implicit_subquery():
- eq_(
- sess.query(User).select_entity_from(sel).all(),
- [User(name="jack", id=7), User(name="ed", id=8)],
- )
-
def test_defer_addtl_attrs(self):
users, User, Address, addresses = (
self.tables.users,
@@ -1595,38 +867,6 @@ class InstrumentationTest(fixtures.ORMTest):
eq_(Sub._sa_iterator(Sub(), 5), "base_iterate")
eq_(Sub._sa_converter(Sub(), 5), "sub_convert")
- def test_link_event(self):
- canary = []
-
- with testing.expect_deprecated(
- r"The collection.linker\(\) handler is deprecated and will "
- "be removed in a future release. Please refer to the "
- "AttributeEvents"
- ):
-
- class Collection(list):
- @collection.linker
- def _on_link(self, obj):
- canary.append(obj)
-
- class Foo(object):
- pass
-
- instrumentation.register_class(Foo)
- attributes.register_attribute(
- Foo, "attr", uselist=True, typecallable=Collection, useobject=True
- )
-
- f1 = Foo()
- f1.attr.append(3)
-
- eq_(canary, [f1.attr._sa_adapter])
- adapter_1 = f1.attr._sa_adapter
-
- l2 = Collection()
- f1.attr = l2
- eq_(canary, [adapter_1, f1.attr._sa_adapter, None])
-
class NonPrimaryRelationshipLoaderTest(_fixtures.FixtureTest):
run_inserts = "once"
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 8dab797ed..fa4ffb0e6 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -752,12 +752,6 @@ class SessionStateTest(_fixtures.FixtureTest):
)
def test_internal_identity_conflict_warning_weak(self):
- self._test_internal_identity_conflict_warning(True)
-
- def test_internal_identity_conflict_warning_strong(self):
- self._test_internal_identity_conflict_warning(False)
-
- def _test_internal_identity_conflict_warning(self, weak_identity_map):
# test for issue #4890
# see also test_naturalpks::ReversePKsTest::test_reverse
users, User = self.tables.users, self.classes.User
@@ -770,8 +764,7 @@ class SessionStateTest(_fixtures.FixtureTest):
)
mapper(Address, addresses)
- with testing.expect_deprecated():
- session = Session(weak_identity_map=weak_identity_map)
+ session = Session()
@event.listens_for(session, "after_flush")
def load_collections(session, flush_context):
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index d641c18e2..7c115789b 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -101,17 +101,6 @@ class DeprecationWarningsTest(fixtures.TestBase, AssertsCompiledSQL):
):
self.assert_compile(or_(and_()), "")
- def test_fromclause_count(self):
- with testing.expect_deprecated(
- r"The FromClause.count\(\) method is deprecated, and will be "
- r"removed in a future release."
- ):
- self.assert_compile(
- table("q", column("x")).count(),
- "SELECT count(q.x) AS tbl_row_count FROM q",
- dialect="default",
- )
-
class ConvertUnicodeDeprecationTest(fixtures.TestBase):
@@ -468,55 +457,6 @@ class LateralSubqueryCoercionsTest(fixtures.TablesTest, AssertsCompiledSQL):
)
-class TextTest(fixtures.TestBase, AssertsCompiledSQL):
- __dialect__ = "default"
-
- def test_legacy_bindparam(self):
- with testing.expect_deprecated(
- "The text.bindparams parameter is deprecated"
- ):
- t = text(
- "select * from foo where lala=:bar and hoho=:whee",
- bindparams=[bindparam("bar", 4), bindparam("whee", 7)],
- )
-
- self.assert_compile(
- t,
- "select * from foo where lala=:bar and hoho=:whee",
- checkparams={"bar": 4, "whee": 7},
- )
-
- def test_legacy_typemap(self):
- table1 = table(
- "mytable",
- column("myid", Integer),
- column("name", String),
- column("description", String),
- )
- with testing.expect_deprecated(
- "The text.typemap parameter is deprecated"
- ):
- t = text(
- "select id, name from user",
- typemap=dict(id=Integer, name=String),
- ).subquery()
-
- stmt = select([table1.c.myid]).select_from(
- table1.join(t, table1.c.myid == t.c.id)
- )
- compiled = stmt.compile()
- eq_(
- compiled._create_result_map(),
- {
- "myid": (
- "myid",
- (table1.c.myid, "myid", "myid", "mytable_myid"),
- table1.c.myid.type,
- )
- },
- )
-
-
class SelectableTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 4b1fb6a72..2aef6f2a5 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -214,7 +214,6 @@ class AdaptTest(fixtures.TestBase):
subcl.__name__,
), True, subcl, [typ]
- @testing.uses_deprecated(".*Binary.*")
@testing.combinations(_adaptions(), id_="iaaa")
def test_adapt_method(self, is_down_adaption, typ, target_adaptions):
"""ensure all types have a working adapt() method,
diff --git a/test/test_deprecations.py b/test/test_deprecations.py
index 19a8c54cc..5f8e79cba 100644
--- a/test/test_deprecations.py
+++ b/test/test_deprecations.py
@@ -1,4 +1,5 @@
-from sqlalchemy.testing import expect_deprecated_20, fixtures
+from sqlalchemy.testing import expect_deprecated_20
+from sqlalchemy.testing import fixtures
from sqlalchemy.util.compat import import_