Establish future behavior for Session cascade backrefs, bind

The behavior of the :paramref:`_orm.relationship.cascade_backrefs` flag
will be reversed in 2.0 and set to ``False`` unconditionally, such that
backrefs don't cascade save-update operations from a forwards-assignment to
a backwards assignment.   A 2.0 deprecation warning is emitted when the
parameter is left at its default of ``True`` at the point at which such a
cascade operation actually takes place.   The new behavior can be
established as always by setting the flag to ``False`` on a specific
:func:`_orm.relationship`, or more generally can be set up across the board
by setting the the :paramref:`_orm.Session.future` flag to True.

Additionally in the interests of expediency, this commit will also
move Session away from making use of bound metadata if the future=True
flag is set.   An application that sets future=True should ideally
have to change as little else as possible for full 2.0 behavior.

Fixes: #5150
Change-Id: I490d1d61f09c62ffc2de983208aeed25dfe48aec

19 files changed, 689 insertions, 328 deletions

diff --git a/doc/build/changelog/migration_14.rst b/doc/build/changelog/migration_14.rst
index ff4d58da7..08ff190b8 100644
--- a/doc/build/changelog/migration_14.rst
+++ b/doc/build/changelog/migration_14.rst
@@ -1396,6 +1396,53 @@ configured to raise an exception using the Python warnings filter.
 
 :ticket:`4662`
 
+.. _change_5150:
+
+cascade_backrefs behavior deprecated for removal in 2.0
+-------------------------------------------------------
+
+SQLAlchemy has long had a behavior of cascading objects into the
+:class:`_orm.Session` based on backref assignment.   Given ``User`` below
+already in a :class:`_orm.Session`, assigning it to the ``Address.user``
+attribute of an ``Address`` object, assuming a bidrectional relationship
+is set up, would mean that the ``Address`` also gets put into the
+:class:`_orm.Session` at that point::
+
+    u1 = User()
+    session.add(u1)
+
+    a1 = Address()
+    a1.user = u1  # <--- adds "a1" to the Session
+
+The above behavior was an unintended side effect of backref behavior, in that
+since ``a1.user`` implies ``u1.addresses.append(a1)``, ``a1`` would get
+cascaded into the :class:`_orm.Session`.  This remains the default behavior
+throughout 1.4.     At some point, a new flag :paramref:`_orm.relationship.cascade_backrefs`
+was added to disable to above behavior, as it can be surprising and also gets in
+the way of some operations where the object would be placed in the :class:`_orm.Session`
+too early and get prematurely flushed.
+
+In 2.0, the default behavior will be that "cascade_backrefs" is False, and
+additionally there will be no "True" behavior as this is not generally a desirable
+behavior.    When 2.0 deprecation warnings are enabled, a warning will be emitted
+when a "backref cascade" actually takes place.    To get the new behavior, either
+set :paramref:`_orm.relationship.cascade_backrefs` to ``False`` on the target
+relationship, as is already supported in 1.3 and earlier, or alternatively make
+use of the :paramref:`_orm.Session.future` flag to :term:`2.0-style` mode::
+
+    Session = sessionmaker(engine, future=True)
+
+    with Session() as session:
+      u1 = User()
+      session.add(u1)
+
+      a1 = Address()
+      a1.user = u1  # <--- will not add "a1" to the Session
+
+
+
+:ticket:`5150`
+
 .. _change_4994:
 
 Persistence-related cascade operations disallowed with viewonly=True
diff --git a/doc/build/changelog/unreleased_14/5150.rst b/doc/build/changelog/unreleased_14/5150.rst
new file mode 100644
index 000000000..1d72185ec
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/5150.rst
@@ -0,0 +1,17 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 5150
+
+    The behavior of the :paramref:`_orm.relationship.cascade_backrefs` flag
+    will be reversed in 2.0 and set to ``False`` unconditionally, such that
+    backrefs don't cascade save-update operations from a forwards-assignment to
+    a backwards assignment.   A 2.0 deprecation warning is emitted when the
+    parameter is left at its default of ``True`` at the point at which such a
+    cascade operation actually takes place.   The new behavior can be
+    established as always by setting the flag to ``False`` on a specific
+    :func:`_orm.relationship`, or more generally can be set up across the board
+    by setting the the :paramref:`_orm.Session.future` flag to True.
+
+    .. seealso::
+
+        :ref:`change_5150`
diff --git a/doc/build/errors.rst b/doc/build/errors.rst
index 961aa4d70..d4659101a 100644
--- a/doc/build/errors.rst
+++ b/doc/build/errors.rst
@@ -118,6 +118,55 @@ are part of SQLAlchemy 1.4 and  are there to help migrate an application to the
     the 1.x series, as well as the current goals and progress of SQLAlchemy
     2.0.
 
+.. _error_c9bf:
+
+A bind was located via legacy bound metadata, but since future=True is set on this Session, this bind is ignored.
+-------------------------------------------------------------------------------------------------------------------
+
+The concept of "bound metadata" is being removed in SQLAlchemy 2.0.  This
+refers to the :paramref:`_schema.MetaData.bind` parameter on the
+:class:`_schema.MetaData` object that in turn allows objects like the ORM
+:class:`_orm.Session` to associate a particular mapped class with an
+:class:`_orm.Engine`.   In SQLAlchemy 2.0, the :class:`_orm.Session` must be
+linked to each :class:`_orm.Engine` directly. That is, instead of instantating
+the :class:`_orm.Session` or
+:class:`_orm.sessionmaker` without any arguments, and associating the
+:class:`_engine.Engine` with the :class:`_schema.MetaData`::
+
+    engine = create_engine("sqlite://")
+    Session = sessionmaker()
+    metadata = MetaData(bind=engine)
+    Base = declarative_base(metadata=metadata)
+
+    class MyClass(Base):
+        # ...
+
+
+    session = Session()
+    session.add(MyClass())
+    session.commit()
+
+The :class:`_engine.Engine` must instead be associated directly with the
+:class:`_orm.sessionmaker` or :class:`_orm.Session`.  The
+:class:`_schema.MetaData` object should no longer be associated with any
+engine::
+
+
+    engine = create_engine("sqlite://")
+    Session = sessionmaker(engine)
+    Base = declarative_base()
+
+    class MyClass(Base):
+        # ...
+
+
+    session = Session()
+    session.add(MyClass())
+    session.commit()
+
+In SQLAlchemy 1.4, this :term:`2.x style` behavior is enabled when the
+:paramref:`_orm.Session.future` flag is set on :class:`_orm.sessionmaker`
+or :class:`_orm.Session`.
 
 Connections and Transactions
 ============================
diff --git a/doc/build/orm/cascades.rst b/doc/build/orm/cascades.rst
index 332ccc5fa..8631dedbe 100644
--- a/doc/build/orm/cascades.rst
+++ b/doc/build/orm/cascades.rst
@@ -559,9 +559,16 @@ operation should be propagated down to referred objects.
 Controlling Cascade on Backrefs
 -------------------------------
 
-The :ref:`cascade_save_update` cascade by default takes place on attribute change events
-emitted from backrefs.  This is probably a confusing statement more
-easily described through demonstration; it means that, given a mapping such as this::
+.. note:: This section applies to a behavior that is removed in SQLAlchemy 2.0.
+   By setting the :paramref:`_orm.Session.future` flag on a given
+   :class:`_orm.Session`, the 2.0 behavior will be achieved which is
+   essentially that the :paramref:`_orm.relationship.cascade_backrefs` flag is
+   ignored.   See the section :ref:`change_5150` for notes.
+
+In :term:`1.x style` ORM usage, the :ref:`cascade_save_update` cascade by
+default takes place on attribute change events emitted from backrefs.  This is
+probably a confusing statement more easily described through demonstration; it
+means that, given a mapping such as this::
 
     mapper(Order, order_table, properties={
         'items' : relationship(Item, backref='order')
diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index 082998ba8..d4680e394 100644
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -1184,7 +1184,7 @@ class ManyToManyDP(DependencyProcessor):
 
         if secondary_delete:
             associationrow = secondary_delete[0]
-            statement = self.secondary.delete(
+            statement = self.secondary.delete().where(
                 sql.and_(
                     *[
                         c == sql.bindparam(c.key, type_=c.type)
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index d991f6229..b8226dfc0 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -1291,7 +1291,9 @@ class Query(
          those being selected.
 
         """
+        return self._from_self(*entities)
 
+    def _from_self(self, *entities):
         fromclause = (
             self.with_labels()
             .enable_eagerloads(False)
@@ -2935,7 +2937,7 @@ class Query(
 
         """
         col = sql.func.count(sql.literal_column("*"))
-        return self.from_self(col).scalar()
+        return self._from_self(col).scalar()
 
     def delete(self, synchronize_session="evaluate"):
         r"""Perform a bulk delete query.
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index d4c5b4665..cb490b7d7 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -408,6 +408,10 @@ class RelationshipProperty(StrategizedProperty):
           will not cascade an incoming transient object into the session of a
           persistent parent, if the event is received via backref.
 
+          .. deprecated:: 1.4 The
+             :paramref:`_orm.relationship.cascade_backrefs`
+             flag will default to False in all cases in SQLAlchemy 2.0.
+
           .. seealso::
 
             :ref:`backref_cascade` - Full discussion and examples on how
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 01163b8d4..25aedd52d 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1937,16 +1937,34 @@ class Session(_SessionClassMethods):
 
         # now we are in legacy territory.  looking for "bind" on tables
         # that are via bound metadata.   this goes away in 2.0.
+
+        future_msg = ""
+        future_code = ""
+
         if mapper and clause is None:
             clause = mapper.persist_selectable
 
         if clause is not None:
             if clause.bind:
-                return clause.bind
+                if self.future:
+                    future_msg = (
+                        " A bind was located via legacy bound metadata, but "
+                        "since future=True is set on this Session, this "
+                        "bind is ignored."
+                    )
+                else:
+                    return clause.bind
 
         if mapper:
             if mapper.persist_selectable.bind:
-                return mapper.persist_selectable.bind
+                if self.future:
+                    future_msg = (
+                        " A bind was located via legacy bound metadata, but "
+                        "since future=True is set on this Session, this "
+                        "bind is ignored."
+                    )
+                else:
+                    return mapper.persist_selectable.bind
 
         context = []
         if mapper is not None:
@@ -1955,8 +1973,9 @@ class Session(_SessionClassMethods):
             context.append("SQL expression")
 
         raise sa_exc.UnboundExecutionError(
-            "Could not locate a bind configured on %s or this Session"
-            % (", ".join(context))
+            "Could not locate a bind configured on %s or this Session.%s"
+            % (", ".join(context), future_msg),
+            code=future_code,
         )
 
     def query(self, *entities, **kwargs):
diff --git a/lib/sqlalchemy/orm/unitofwork.py b/lib/sqlalchemy/orm/unitofwork.py
index 97eea4864..9c67130ce 100644
--- a/lib/sqlalchemy/orm/unitofwork.py
+++ b/lib/sqlalchemy/orm/unitofwork.py
@@ -21,6 +21,16 @@ from .. import util
 from ..util import topological
 
 
+def _warn_for_cascade_backrefs(state, prop):
+    util.warn_deprecated_20(
+        '"%s" object is being merged into a Session along the backref '
+        'cascade path for relationship "%s"; in SQLAlchemy 2.0, this '
+        "reverse cascade will not take place.  Set cascade_backrefs to "
+        "False for the 2.0 behavior; or to set globally for the whole "
+        "Session, set the future=True flag" % (state.class_.__name__, prop)
+    )
+
+
 def track_cascade_events(descriptor, prop):
     """Establish event listeners on object attributes which handle
     cascade-on-set/append.
@@ -42,11 +52,17 @@ def track_cascade_events(descriptor, prop):
 
             prop = state.manager.mapper._props[key]
             item_state = attributes.instance_state(item)
+
             if (
                 prop._cascade.save_update
-                and (prop.cascade_backrefs or key == initiator.key)
+                and (
+                    (prop.cascade_backrefs and not sess.future)
+                    or key == initiator.key
+                )
                 and not sess._contains_state(item_state)
             ):
+                if key != initiator.key:
+                    _warn_for_cascade_backrefs(item_state, prop)
                 sess._save_or_update_state(item_state)
         return item
 
@@ -101,9 +117,14 @@ def track_cascade_events(descriptor, prop):
                 newvalue_state = attributes.instance_state(newvalue)
                 if (
                     prop._cascade.save_update
-                    and (prop.cascade_backrefs or key == initiator.key)
+                    and (
+                        (prop.cascade_backrefs and not sess.future)
+                        or key == initiator.key
+                    )
                     and not sess._contains_state(newvalue_state)
                 ):
+                    if key != initiator.key:
+                        _warn_for_cascade_backrefs(newvalue_state, prop)
                     sess._save_or_update_state(newvalue_state)
 
             if (
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py
index 1eac76598..1583147d4 100644
--- a/lib/sqlalchemy/testing/fixtures.py
+++ b/lib/sqlalchemy/testing/fixtures.py
@@ -293,13 +293,14 @@ class TablesTest(TestBase):
                 continue
             if table not in headers:
                 continue
-            cls.bind.execute(
-                table.insert(),
-                [
-                    dict(zip(headers[table], column_values))
-                    for column_values in rows[table]
-                ],
-            )
+            with cls.bind.begin() as conn:
+                conn.execute(
+                    table.insert(),
+                    [
+                        dict(zip(headers[table], column_values))
+                        for column_values in rows[table]
+                    ],
+                )
 
 
 class RemovesEvents(object):
diff --git a/test/orm/inheritance/test_polymorphic_rel.py b/test/orm/inheritance/test_polymorphic_rel.py
index d13b2f945..d8214465a 100644
--- a/test/orm/inheritance/test_polymorphic_rel.py
+++ b/test/orm/inheritance/test_polymorphic_rel.py
@@ -230,7 +230,7 @@ class _PolymorphicTestBase(object):
         )
 
     def test_multi_join_future(self):
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
         e = aliased(Person)
         c = aliased(Company)
 
@@ -283,7 +283,7 @@ class _PolymorphicTestBase(object):
         eq_(sess.query(Engineer).all()[0], Engineer(name="dilbert"))
 
     def test_filter_on_subclass_one_future(self):
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
         eq_(
             sess.execute(select(Engineer)).scalar(), Engineer(name="dilbert"),
         )
@@ -337,7 +337,7 @@ class _PolymorphicTestBase(object):
         )
 
     def test_join_from_polymorphic_nonaliased_one_future(self):
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
         eq_(
             sess.execute(
                 select(Person)
@@ -396,7 +396,7 @@ class _PolymorphicTestBase(object):
         )
 
     def test_join_from_polymorphic_flag_aliased_one_future(self):
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
 
         pa = aliased(Paperwork)
         eq_(
@@ -496,7 +496,7 @@ class _PolymorphicTestBase(object):
         )
 
     def test_join_from_with_polymorphic_nonaliased_one_future(self):
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
 
         pm = with_polymorphic(Person, [Manager])
         eq_(
@@ -1544,7 +1544,7 @@ class _PolymorphicTestBase(object):
         # TODO: this is the first test *EVER* of an aliased class of
         # an aliased class.  we should add many more tests for this.
         # new case added in Id810f485c5f7ed971529489b84694e02a3356d6d
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
         expected = [(m1, e1), (m1, e2), (m1, b1)]
 
         p1 = aliased(Person)
diff --git a/test/orm/test_cache_key.py b/test/orm/test_cache_key.py
index 4156d606c..02b1b9fbf 100644
--- a/test/orm/test_cache_key.py
+++ b/test/orm/test_cache_key.py
@@ -472,7 +472,7 @@ class RoundTripTest(QueryTest, AssertsCompiledSQL):
         # query.
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         def query(names):
             stmt = (
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index 7e6db3b89..8a21297f1 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -12,7 +12,6 @@ from sqlalchemy.orm import attributes
 from sqlalchemy.orm import backref
 from sqlalchemy.orm import class_mapper
 from sqlalchemy.orm import configure_mappers
-from sqlalchemy.orm import create_session
 from sqlalchemy.orm import exc as orm_exc
 from sqlalchemy.orm import foreign
 from sqlalchemy.orm import mapper
@@ -271,68 +270,71 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
     def test_list_assignment_new(self):
         User, Order = self.classes.User, self.classes.Order
 
-        sess = Session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="order 1"),
-                Order(description="order 2"),
-            ],
-        )
-        sess.add(u)
-        sess.commit()
-
-        eq_(
-            u,
-            User(
+        with Session() as sess:
+            u = User(
                 name="jack",
                 orders=[
                     Order(description="order 1"),
                     Order(description="order 2"),
                 ],
-            ),
-        )
+            )
+            sess.add(u)
+            sess.commit()
+
+            eq_(
+                u,
+                User(
+                    name="jack",
+                    orders=[
+                        Order(description="order 1"),
+                        Order(description="order 2"),
+                    ],
+                ),
+            )
 
     def test_list_assignment_replace(self):
         User, Order = self.classes.User, self.classes.Order
 
-        sess = Session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="someorder"),
-                Order(description="someotherorder"),
-            ],
-        )
-        sess.add(u)
-
-        u.orders = [Order(description="order 3"), Order(description="order 4")]
-        sess.commit()
-
-        eq_(
-            u,
-            User(
+        with Session() as sess:
+            u = User(
                 name="jack",
                 orders=[
-                    Order(description="order 3"),
-                    Order(description="order 4"),
+                    Order(description="someorder"),
+                    Order(description="someotherorder"),
                 ],
-            ),
-        )
+            )
+            sess.add(u)
 
-        # order 1, order 2 have been deleted
-        eq_(
-            sess.query(Order).order_by(Order.id).all(),
-            [Order(description="order 3"), Order(description="order 4")],
-        )
+            u.orders = [
+                Order(description="order 3"),
+                Order(description="order 4"),
+            ]
+            sess.commit()
+
+            eq_(
+                u,
+                User(
+                    name="jack",
+                    orders=[
+                        Order(description="order 3"),
+                        Order(description="order 4"),
+                    ],
+                ),
+            )
+
+            # order 1, order 2 have been deleted
+            eq_(
+                sess.query(Order).order_by(Order.id).all(),
+                [Order(description="order 3"), Order(description="order 4")],
+            )
 
     def test_standalone_orphan(self):
         Order = self.classes.Order
 
-        sess = Session()
-        o5 = Order(description="order 5")
-        sess.add(o5)
-        assert_raises(sa_exc.DBAPIError, sess.flush)
+        with Session() as sess:
+            o5 = Order(description="order 5")
+            sess.add(o5)
+            assert_raises(sa_exc.DBAPIError, sess.flush)
 
     def test_save_update_sends_pending(self):
         """test that newly added and deleted collection items are
@@ -361,41 +363,41 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
     def test_remove_pending_from_collection(self):
         User, Order = self.classes.User, self.classes.Order
 
-        sess = Session()
+        with Session() as sess:
 
-        u = User(name="jack")
-        sess.add(u)
-        sess.commit()
+            u = User(name="jack")
+            sess.add(u)
+            sess.commit()
 
-        o1 = Order()
-        u.orders.append(o1)
-        assert o1 in sess
-        u.orders.remove(o1)
-        assert o1 not in sess
+            o1 = Order()
+            u.orders.append(o1)
+            assert o1 in sess
+            u.orders.remove(o1)
+            assert o1 not in sess
 
     def test_remove_pending_from_pending_parent(self):
         # test issue #4040
 
         User, Order = self.classes.User, self.classes.Order
 
-        sess = Session()
+        with Session() as sess:
 
-        u = User(name="jack")
+            u = User(name="jack")
 
-        o1 = Order()
-        sess.add(o1)
+            o1 = Order()
+            sess.add(o1)
 
-        # object becomes an orphan, but parent is not in session
-        u.orders.append(o1)
-        u.orders.remove(o1)
+            # object becomes an orphan, but parent is not in session
+            u.orders.append(o1)
+            u.orders.remove(o1)
 
-        sess.add(u)
+            sess.add(u)
 
-        assert o1 in sess
+            assert o1 in sess
 
-        sess.flush()
+            sess.flush()
 
-        assert o1 not in sess
+            assert o1 not in sess
 
     def test_delete(self):
         User, users, orders, Order = (
@@ -405,21 +407,31 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.classes.Order,
         )
 
-        sess = create_session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="someorder"),
-                Order(description="someotherorder"),
-            ],
-        )
-        sess.add(u)
-        sess.flush()
-
-        sess.delete(u)
-        sess.flush()
-        eq_(select(func.count("*")).select_from(users).scalar(), 0)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+        with Session() as sess:
+            u = User(
+                name="jack",
+                orders=[
+                    Order(description="someorder"),
+                    Order(description="someotherorder"),
+                ],
+            )
+            sess.add(u)
+            sess.flush()
+
+            sess.delete(u)
+            sess.flush()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                0,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                0,
+            )
 
     def test_delete_unloaded_collections(self):
         """Unloaded collections are still included in a delete-cascade
@@ -432,27 +444,47 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.classes.Address,
         )
 
-        sess = create_session()
-        u = User(
-            name="jack",
-            addresses=[
-                Address(email_address="address1"),
-                Address(email_address="address2"),
-            ],
-        )
-        sess.add(u)
-        sess.flush()
-        sess.expunge_all()
-        eq_(select(func.count("*")).select_from(addresses).scalar(), 2)
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
+        with Session() as sess:
+            u = User(
+                name="jack",
+                addresses=[
+                    Address(email_address="address1"),
+                    Address(email_address="address2"),
+                ],
+            )
+            sess.add(u)
+            sess.flush()
+            sess.expunge_all()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(addresses)
+                ).scalar(),
+                2,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                1,
+            )
 
-        u = sess.query(User).get(u.id)
+            u = sess.get(User, u.id)
 
-        assert "addresses" not in u.__dict__
-        sess.delete(u)
-        sess.flush()
-        eq_(select(func.count("*")).select_from(addresses).scalar(), 0)
-        eq_(select(func.count("*")).select_from(users).scalar(), 0)
+            assert "addresses" not in u.__dict__
+            sess.delete(u)
+            sess.flush()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(addresses)
+                ).scalar(),
+                0,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                0,
+            )
 
     def test_cascades_onlycollection(self):
         """Cascade only reaches instances that are still part of the
@@ -465,34 +497,48 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.tables.orders,
         )
 
-        sess = create_session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="someorder"),
-                Order(description="someotherorder"),
-            ],
-        )
-        sess.add(u)
-        sess.flush()
-
-        o = u.orders[0]
-        del u.orders[0]
-        sess.delete(u)
-        assert u in sess.deleted
-        assert o not in sess.deleted
-        assert o in sess
-
-        u2 = User(name="newuser", orders=[o])
-        sess.add(u2)
-        sess.flush()
-        sess.expunge_all()
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 1)
-        eq_(
-            sess.query(User).all(),
-            [User(name="newuser", orders=[Order(description="someorder")])],
-        )
+        with Session(autoflush=False) as sess:
+            u = User(
+                name="jack",
+                orders=[
+                    Order(description="someorder"),
+                    Order(description="someotherorder"),
+                ],
+            )
+            sess.add(u)
+            sess.flush()
+
+            o = u.orders[0]
+            del u.orders[0]
+            sess.delete(u)
+            assert u in sess.deleted
+            assert o not in sess.deleted
+            assert o in sess
+
+            u2 = User(name="newuser", orders=[o])
+            sess.add(u2)
+            sess.flush()
+            sess.expunge_all()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                1,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                1,
+            )
+            eq_(
+                sess.query(User).all(),
+                [
+                    User(
+                        name="newuser", orders=[Order(description="someorder")]
+                    )
+                ],
+            )
 
     def test_cascade_nosideeffects(self):
         """test that cascade leaves the state of unloaded
@@ -504,7 +550,7 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.classes.Address,
         )
 
-        sess = create_session()
+        sess = Session()
         u = User(name="jack")
         sess.add(u)
         assert "orders" not in u.__dict__
@@ -534,7 +580,7 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.classes.Order,
         )
 
-        sess = create_session()
+        sess = Session()
         u = User(
             name="jack",
             orders=[
@@ -544,14 +590,26 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
         )
         sess.add(u)
         sess.flush()
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(users)).scalar(),
+            1,
+        )
+        eq_(
+            sess.execute(select(func.count("*")).select_from(orders)).scalar(),
+            2,
+        )
 
         del u.orders[0]
         sess.delete(u)
         sess.flush()
-        eq_(select(func.count("*")).select_from(users).scalar(), 0)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(users)).scalar(),
+            0,
+        )
+        eq_(
+            sess.execute(select(func.count("*")).select_from(orders)).scalar(),
+            0,
+        )
 
     def test_collection_orphans(self):
         User, users, orders, Order = (
@@ -561,26 +619,46 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
             self.classes.Order,
         )
 
-        sess = create_session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="someorder"),
-                Order(description="someotherorder"),
-            ],
-        )
-        sess.add(u)
-        sess.flush()
-
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+        with Session() as sess:
+            u = User(
+                name="jack",
+                orders=[
+                    Order(description="someorder"),
+                    Order(description="someotherorder"),
+                ],
+            )
+            sess.add(u)
+            sess.flush()
+
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                1,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                2,
+            )
 
-        u.orders[:] = []
+            u.orders[:] = []
 
-        sess.flush()
+            sess.flush()
 
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                1,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                0,
+            )
 
 
 class O2MCascadeTest(fixtures.MappedTest):
@@ -715,24 +793,44 @@ class O2MCascadeDeleteNoOrphanTest(fixtures.MappedTest):
             self.tables.users,
         )
 
-        sess = create_session()
-        u = User(
-            name="jack",
-            orders=[
-                Order(description="someorder"),
-                Order(description="someotherorder"),
-            ],
-        )
-        sess.add(u)
-        sess.flush()
-        eq_(select(func.count("*")).select_from(users).scalar(), 1)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+        with Session() as sess:
+            u = User(
+                name="jack",
+                orders=[
+                    Order(description="someorder"),
+                    Order(description="someotherorder"),
+                ],
+            )
+            sess.add(u)
+            sess.flush()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                1,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                2,
+            )
 
-        del u.orders[0]
-        sess.delete(u)
-        sess.flush()
-        eq_(select(func.count("*")).select_from(users).scalar(), 0)
-        eq_(select(func.count("*")).select_from(orders).scalar(), 1)
+            del u.orders[0]
+            sess.delete(u)
+            sess.flush()
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(users)
+                ).scalar(),
+                0,
+            )
+            eq_(
+                sess.execute(
+                    select(func.count("*")).select_from(orders)
+                ).scalar(),
+                1,
+            )
 
 
 class O2OSingleParentTest(_fixtures.FixtureTest):
@@ -1183,18 +1281,44 @@ class NoSaveCascadeFlushTest(_fixtures.FixtureTest):
         User, Address = self.classes.User, self.classes.Address
 
         self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
-        sess = Session()
-        u1 = User(name="u1")
-        sess.add(u1)
-        sess.flush()
+        with Session() as sess:
+            u1 = User(name="u1")
+            sess.add(u1)
+            sess.flush()
+
+            a1 = Address(email_address="a1")
+            with testing.expect_deprecated(
+                '"Address" object is being merged into a Session along '
+                'the backref cascade path for relationship "User.addresses"'
+            ):
+                a1.user = u1
+            sess.add(a1)
+            sess.expunge(u1)
+            assert u1 not in sess
+            assert a1 in sess
+            assert_raises_message(
+                sa_exc.SAWarning, "not in session", sess.flush
+            )
 
-        a1 = Address(email_address="a1")
-        a1.user = u1
-        sess.add(a1)
-        sess.expunge(u1)
-        assert u1 not in sess
-        assert a1 in sess
-        assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+    def test_m2o_backref_future_child_expunged(self):
+        User, Address = self.classes.User, self.classes.Address
+
+        self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
+        with Session(testing.db, future=True) as sess:
+            u1 = User(name="u1")
+            sess.add(u1)
+            sess.flush()
+
+            a1 = Address(email_address="a1")
+            a1.user = u1
+            assert a1 not in sess
+            sess.add(a1)
+            sess.expunge(u1)
+            assert u1 not in sess
+            assert a1 in sess
+            assert_raises_message(
+                sa_exc.SAWarning, "not in session", sess.flush
+            )
 
     def test_m2o_backref_child_pending_nochange(self):
         User, Address = self.classes.User, self.classes.Address
@@ -1221,25 +1345,56 @@ class NoSaveCascadeFlushTest(_fixtures.FixtureTest):
         User, Address = self.classes.User, self.classes.Address
 
         self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
-        sess = Session()
-        u1 = User(name="u1")
-        sess.add(u1)
-        sess.flush()
 
-        a1 = Address(email_address="a1")
-        a1.user = u1
-        sess.add(a1)
-        sess.expunge(u1)
-        assert u1 not in sess
-        assert a1 in sess
+        with Session() as sess:
+            u1 = User(name="u1")
+            sess.add(u1)
+            sess.flush()
+
+            a1 = Address(email_address="a1")
+            with testing.expect_deprecated(
+                '"Address" object is being merged into a Session along the '
+                'backref cascade path for relationship "User.addresses"'
+            ):
+                a1.user = u1
+            sess.add(a1)
+            sess.expunge(u1)
+            assert u1 not in sess
+            assert a1 in sess
+
+            @testing.emits_warning(r".*not in session")
+            def go():
+                sess.commit()
+
+            go()
+            # didn't get flushed
+            assert a1.user is None
+
+    def test_m2o_backref_future_child_expunged_nochange(self):
+        User, Address = self.classes.User, self.classes.Address
 
-        @testing.emits_warning(r".*not in session")
-        def go():
-            sess.commit()
+        self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
 
-        go()
-        # didn't get flushed
-        assert a1.user is None
+        with Session(testing.db, future=True) as sess:
+            u1 = User(name="u1")
+            sess.add(u1)
+            sess.flush()
+
+            a1 = Address(email_address="a1")
+            a1.user = u1
+            assert a1 not in sess
+            sess.add(a1)
+            sess.expunge(u1)
+            assert u1 not in sess
+            assert a1 in sess
+
+            @testing.emits_warning(r".*not in session")
+            def go():
+                sess.commit()
+
+            go()
+            # didn't get flushed
+            assert a1.user is None
 
     def test_m2m_only_child_pending(self):
         Item, Keyword = self.classes.Item, self.classes.Keyword
@@ -1394,7 +1549,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
             ),
         )
 
-        sess = create_session()
+        sess = Session()
 
         o1 = Order()
         sess.add(o1)
@@ -1429,7 +1584,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
         )
         mapper(User, users)
 
-        sess = create_session()
+        sess = Session()
 
         u1 = User()
         sess.add(u1)
@@ -1470,7 +1625,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
         )
         mapper(Keyword, keywords)
 
-        sess = create_session()
+        sess = Session()
 
         i1 = Item()
         k1 = Keyword()
@@ -1586,7 +1741,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
         u1 = User(name="ed", pref=Pref(data="pref 1", extra=[Extra()]))
         u2 = User(name="jack", pref=Pref(data="pref 2", extra=[Extra()]))
         u3 = User(name="foo", pref=Pref(data="pref 3", extra=[Extra()]))
-        sess = create_session(connection)
+        sess = Session(connection)
         sess.add_all((u1, u2, u3))
         sess.flush()
         sess.close()
@@ -1598,14 +1753,26 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
             self.tables.extra,
         )
 
-        sess = create_session()
-        eq_(select(func.count("*")).select_from(prefs).scalar(), 3)
-        eq_(select(func.count("*")).select_from(extra).scalar(), 3)
+        sess = Session()
+        eq_(
+            sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+            3,
+        )
+        eq_(
+            sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+            3,
+        )
         jack = sess.query(User).filter_by(name="jack").one()
         jack.pref = None
         sess.flush()
-        eq_(select(func.count("*")).select_from(prefs).scalar(), 2)
-        eq_(select(func.count("*")).select_from(extra).scalar(), 2)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+            2,
+        )
+        eq_(
+            sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+            2,
+        )
 
     def test_cascade_on_deleted(self):
         """test a bug introduced by r6711"""
@@ -1657,7 +1824,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
             self.tables.extra,
         )
 
-        sess = create_session()
+        sess = Session()
         jack = sess.query(User).filter_by(name="jack").one()
         p = jack.pref
         e = jack.pref.extra[0]
@@ -1670,13 +1837,19 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
         assert p in sess
         assert e in sess
         sess.flush()
-        eq_(select(func.count("*")).select_from(prefs).scalar(), 2)
-        eq_(select(func.count("*")).select_from(extra).scalar(), 2)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+            2,
+        )
+        eq_(
+            sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+            2,
+        )
 
     def test_pending_expunge(self):
         Pref, User = self.classes.Pref, self.classes.User
 
-        sess = create_session()
+        sess = Session()
         someuser = User(name="someuser")
         sess.add(someuser)
         sess.flush()
@@ -1695,7 +1868,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
 
         Pref, User = self.classes.Pref, self.classes.User
 
-        sess = create_session()
+        sess = Session()
         jack = sess.query(User).filter_by(name="jack").one()
 
         newpref = Pref(data="newpref")
@@ -1788,7 +1961,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
     def test_cascade_delete(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
         sess.add(x)
         sess.flush()
@@ -1802,7 +1975,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
     def test_deletes_orphans_onelevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
         sess.add(x2)
         sess.flush()
@@ -1817,7 +1990,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
     def test_deletes_orphans_twolevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
         sess.add(x)
         sess.flush()
@@ -1832,7 +2005,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
     def test_finds_orphans_twolevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
         sess.add(x)
         sess.flush()
@@ -1929,7 +2102,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_cascade_delete(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
         sess.add(x)
         sess.flush()
@@ -1943,7 +2116,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_cascade_delete_postappend_onelevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x1 = T1(data="t1")
         x2 = T2(data="t2")
         x3 = T3(data="t3")
@@ -1961,7 +2134,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_cascade_delete_postappend_twolevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x1 = T1(data="t1", t2=T2(data="t2"))
         x3 = T3(data="t3")
         sess.add_all((x1, x3))
@@ -1977,7 +2150,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_preserves_orphans_onelevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
         sess.add(x2)
         sess.flush()
@@ -1993,7 +2166,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_preserves_orphans_onelevel_postremove(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
         sess.add(x2)
         sess.flush()
@@ -2008,7 +2181,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
     def test_preserves_orphans_twolevel(self):
         T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
 
-        sess = create_session()
+        sess = Session()
         x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
         sess.add(x)
         sess.flush()
@@ -2097,7 +2270,7 @@ class M2MCascadeTest(fixtures.MappedTest):
         )
         mapper(B, b)
 
-        sess = create_session()
+        sess = Session()
         b1 = B(data="b1")
         a1 = A(data="a1", bs=[b1])
         sess.add(a1)
@@ -2105,9 +2278,11 @@ class M2MCascadeTest(fixtures.MappedTest):
 
         a1.bs.remove(b1)
         sess.flush()
-        eq_(select(func.count("*")).select_from(atob).scalar(), 0)
-        eq_(select(func.count("*")).select_from(b).scalar(), 0)
-        eq_(select(func.count("*")).select_from(a).scalar(), 1)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+        )
+        eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+        eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
 
     def test_delete_orphan_dynamic(self):
         a, A, B, b, atob = (
@@ -2135,7 +2310,7 @@ class M2MCascadeTest(fixtures.MappedTest):
         # failed until [ticket:427] was fixed
         mapper(B, b)
 
-        sess = create_session()
+        sess = Session()
         b1 = B(data="b1")
         a1 = A(data="a1", bs=[b1])
         sess.add(a1)
@@ -2143,9 +2318,11 @@ class M2MCascadeTest(fixtures.MappedTest):
 
         a1.bs.remove(b1)
         sess.flush()
-        eq_(select(func.count("*")).select_from(atob).scalar(), 0)
-        eq_(select(func.count("*")).select_from(b).scalar(), 0)
-        eq_(select(func.count("*")).select_from(a).scalar(), 1)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+        )
+        eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+        eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
 
     def test_delete_orphan_cascades(self):
         a, A, c, b, C, B, atob = (
@@ -2179,7 +2356,7 @@ class M2MCascadeTest(fixtures.MappedTest):
         )
         mapper(C, c)
 
-        sess = create_session()
+        sess = Session()
         b1 = B(data="b1", cs=[C(data="c1")])
         a1 = A(data="a1", bs=[b1])
         sess.add(a1)
@@ -2187,10 +2364,12 @@ class M2MCascadeTest(fixtures.MappedTest):
 
         a1.bs.remove(b1)
         sess.flush()
-        eq_(select(func.count("*")).select_from(atob).scalar(), 0)
-        eq_(select(func.count("*")).select_from(b).scalar(), 0)
-        eq_(select(func.count("*")).select_from(a).scalar(), 1)
-        eq_(select(func.count("*")).select_from(c).scalar(), 0)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+        )
+        eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+        eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
+        eq_(sess.execute(select(func.count("*")).select_from(c)).scalar(), 0)
 
     def test_cascade_delete(self):
         a, A, B, b, atob = (
@@ -2215,16 +2394,18 @@ class M2MCascadeTest(fixtures.MappedTest):
         )
         mapper(B, b)
 
-        sess = create_session()
+        sess = Session()
         a1 = A(data="a1", bs=[B(data="b1")])
         sess.add(a1)
         sess.flush()
 
         sess.delete(a1)
         sess.flush()
-        eq_(select(func.count("*")).select_from(atob).scalar(), 0)
-        eq_(select(func.count("*")).select_from(b).scalar(), 0)
-        eq_(select(func.count("*")).select_from(a).scalar(), 0)
+        eq_(
+            sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+        )
+        eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+        eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 0)
 
     def test_single_parent_error(self):
         a, A, B, b, atob = (
@@ -2492,7 +2673,11 @@ class NoBackrefCascadeTest(_fixtures.FixtureTest):
         sess.add(a1)
 
         d1 = Dingaling()
-        d1.address = a1
+        with testing.expect_deprecated(
+            '"Dingaling" object is being merged into a Session along the '
+            'backref cascade path for relationship "Address.dingalings"'
+        ):
+            d1.address = a1
         assert d1 in a1.dingalings
         assert d1 in sess
 
@@ -2519,7 +2704,11 @@ class NoBackrefCascadeTest(_fixtures.FixtureTest):
         sess.add(a1)
 
         u1 = User(name="u1")
-        u1.addresses.append(a1)
+        with testing.expect_deprecated(
+            '"User" object is being merged into a Session along the backref '
+            'cascade path for relationship "Address.user"'
+        ):
+            u1.addresses.append(a1)
         assert u1 in sess
 
     def test_m2o_commit_warns(self):
@@ -2666,7 +2855,7 @@ class PendingOrphanTestSingleLevel(fixtures.MappedTest):
                 )
             ),
         )
-        s = create_session()
+        s = Session()
 
         u = User()
         s.add(u)
@@ -2702,7 +2891,7 @@ class PendingOrphanTestSingleLevel(fixtures.MappedTest):
                 )
             ),
         )
-        s = create_session()
+        s = Session()
         u = User(name="u1", addresses=[Address(email_address="ad1")])
         s.add(u)
         a1 = u.addresses[0]
@@ -2928,12 +3117,12 @@ class DoubleParentO2MOrphanTest(fixtures.MappedTest):
                 )
             ),
         )
-        s = create_session()
+        s = Session(expire_on_commit=False, autoflush=False)
 
         a = Account(balance=0)
         sr = SalesRep(name="John")
         s.add_all((a, sr))
-        s.flush()
+        s.commit()
 
         c = Customer(name="Jane")
 
@@ -3093,7 +3282,7 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
             },
         )
 
-        session = create_session()
+        session = Session()
         h1 = Home(description="home1", address=Address(street="address1"))
         b1 = Business(
             description="business1", address=Address(street="address2")
@@ -3103,11 +3292,11 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
         session.expunge_all()
 
         eq_(
-            session.query(Home).get(h1.id),
+            session.get(Home, h1.id),
             Home(description="home1", address=Address(street="address1")),
         )
         eq_(
-            session.query(Business).get(b1.id),
+            session.get(Business, b1.id),
             Business(
                 description="business1", address=Address(street="address2")
             ),
@@ -3152,7 +3341,7 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
                 )
             },
         )
-        session = create_session()
+        session = Session()
         a1 = Address()
         session.add(a1)
         session.flush()
@@ -3197,18 +3386,18 @@ class CollectionAssignmentOrphanTest(fixtures.MappedTest):
 
         a1 = A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")])
 
-        sess = create_session()
+        sess = Session()
         sess.add(a1)
         sess.flush()
 
         sess.expunge_all()
 
         eq_(
-            sess.query(A).get(a1.id),
+            sess.get(A, a1.id),
             A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")]),
         )
 
-        a1 = sess.query(A).get(a1.id)
+        a1 = sess.get(A, a1.id)
         assert not class_mapper(B)._is_orphan(
             attributes.instance_state(a1.bs[0])
         )
@@ -3218,7 +3407,7 @@ class CollectionAssignmentOrphanTest(fixtures.MappedTest):
 
         sess.expunge_all()
         eq_(
-            sess.query(A).get(a1.id),
+            sess.get(A, a1.id),
             A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")]),
         )
 
@@ -3440,27 +3629,26 @@ class O2MConflictTest(fixtures.MappedTest):
     def _do_move_test(self, delete_old):
         Parent, Child = self.classes.Parent, self.classes.Child
 
-        sess = create_session()
-
-        p1, p2, c1 = Parent(), Parent(), Child()
-        if Parent.child.property.uselist:
-            p1.child.append(c1)
-        else:
-            p1.child = c1
-        sess.add_all([p1, c1])
-        sess.flush()
+        with Session(autoflush=False) as sess:
+            p1, p2, c1 = Parent(), Parent(), Child()
+            if Parent.child.property.uselist:
+                p1.child.append(c1)
+            else:
+                p1.child = c1
+            sess.add_all([p1, c1])
+            sess.flush()
 
-        if delete_old:
-            sess.delete(p1)
+            if delete_old:
+                sess.delete(p1)
 
-        if Parent.child.property.uselist:
-            p2.child.append(c1)
-        else:
-            p2.child = c1
-        sess.add(p2)
+            if Parent.child.property.uselist:
+                p2.child.append(c1)
+            else:
+                p2.child = c1
+            sess.add(p2)
 
-        sess.flush()
-        eq_(sess.query(Child).filter(Child.parent_id == p2.id).all(), [c1])
+            sess.flush()
+            eq_(sess.query(Child).filter(Child.parent_id == p2.id).all(), [c1])
 
     def test_o2o_delete_old(self):
         Child, Parent, parent, child = (
@@ -3508,7 +3696,11 @@ class O2MConflictTest(fixtures.MappedTest):
             Parent,
             parent,
             properties={
-                "child": relationship(Child, uselist=False, backref="parent")
+                "child": relationship(
+                    Child,
+                    uselist=False,
+                    backref=backref("parent", cascade_backrefs=False),
+                )
             },
         )
         mapper(Child, child)
@@ -3573,7 +3765,7 @@ class O2MConflictTest(fixtures.MappedTest):
                     Child,
                     uselist=False,
                     cascade="all, delete, delete-orphan",
-                    backref="parent",
+                    backref=backref("parent", cascade_backrefs=False),
                 )
             },
         )
@@ -3600,6 +3792,7 @@ class O2MConflictTest(fixtures.MappedTest):
                     single_parent=True,
                     backref=backref("child", uselist=False),
                     cascade="all,delete,delete-orphan",
+                    cascade_backrefs=False,
                 )
             },
         )
@@ -3625,6 +3818,7 @@ class O2MConflictTest(fixtures.MappedTest):
                     single_parent=True,
                     backref=backref("child", uselist=True),
                     cascade="all,delete,delete-orphan",
+                    cascade_backrefs=False,
                 )
             },
         )
@@ -3684,7 +3878,7 @@ class PartialFlushTest(fixtures.MappedTest):
         )
         mapper(Child, noninh_child)
 
-        sess = create_session()
+        sess = Session()
 
         c1, c2 = Child(), Child()
         b1 = Base(descr="b1", children=[c1, c2])
@@ -3701,7 +3895,7 @@ class PartialFlushTest(fixtures.MappedTest):
         assert c2 in sess and c2 not in sess.new
         assert b1 in sess and b1 not in sess.new
 
-        sess = create_session()
+        sess = Session()
         c1, c2 = Child(), Child()
         b1 = Base(descr="b1", children=[c1, c2])
         sess.add(b1)
@@ -3711,7 +3905,7 @@ class PartialFlushTest(fixtures.MappedTest):
         assert c2 in sess and c2 in sess.new
         assert b1 in sess and b1 in sess.new
 
-        sess = create_session()
+        sess = Session()
         c1, c2 = Child(), Child()
         b1 = Base(descr="b1", children=[c1, c2])
         sess.add(b1)
@@ -3756,7 +3950,7 @@ class PartialFlushTest(fixtures.MappedTest):
 
         mapper(Parent, parent, inherits=Base)
 
-        sess = create_session()
+        sess = Session()
         p1 = Parent()
 
         c1, c2, c3 = Child(), Child(), Child()
diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py
index b4c0c2dc1..046a6acb9 100644
--- a/test/orm/test_composites.py
+++ b/test/orm/test_composites.py
@@ -95,7 +95,7 @@ class PointTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
             self.classes.Point,
         )
 
-        sess = Session(future=future)
+        sess = Session(testing.db, future=future)
         g = Graph(
             id=1,
             edges=[
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 4b4c2bf73..e5a53df4b 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -1219,7 +1219,7 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
             self.classes.User,
         )
 
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
 
         selectquery = users.outerjoin(addresses).select(
             users.c.id < 10,
@@ -2154,7 +2154,7 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
             (user10, None),
         ]
 
-        sess = create_session(future=True)
+        sess = create_session(testing.db, future=True)
 
         selectquery = users.outerjoin(addresses).select(
             use_labels=True, order_by=[users.c.id, addresses.c.id]
diff --git a/test/orm/test_lambdas.py b/test/orm/test_lambdas.py
index 2aac956ca..d4fae7f6f 100644
--- a/test/orm/test_lambdas.py
+++ b/test/orm/test_lambdas.py
@@ -92,7 +92,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_cols_round_trip(self, plain_fixture):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         # note this does a traversal + _clone of the InstrumentedAttribute
         # for the first time ever
@@ -135,7 +135,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_entity_round_trip(self, plain_fixture):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         def query(names):
             stmt = lambda_stmt(
@@ -182,7 +182,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_subqueryload_internal_lambda(self, plain_fixture):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         def query(names):
             stmt = (
@@ -220,7 +220,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_subqueryload_external_lambda_caveats(self, plain_fixture):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         def query(names):
             stmt = lambda_stmt(
@@ -263,7 +263,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_does_filter_aliasing_work(self, plain_fixture):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         # aliased=True is to be deprecated, other filter lambdas
         # that go into effect include polymorphic filtering.
@@ -314,7 +314,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
     def test_join_entity_arg(self, plain_fixture, test_case):
         User, Address = plain_fixture
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         stmt = testing.resolve_lambda(test_case, **locals())
         self.assert_compile(
@@ -411,7 +411,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
     def test_update(self):
         User, Address = self.classes("User", "Address")
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         def go(ids, values):
             stmt = lambda_stmt(lambda: update(User).where(User.id.in_(ids)))
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index b6d014936..4663789c4 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -187,7 +187,7 @@ class RowTupleTest(QueryTest):
 
         mapper(User, users)
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         q = testing.resolve_lambda(test_case, **locals())
 
@@ -876,7 +876,7 @@ class GetTest(QueryTest):
     def test_populate_existing_future(self):
         User, Address = self.classes.User, self.classes.Address
 
-        s = Session(future=True, autoflush=False)
+        s = Session(testing.db, future=True, autoflush=False)
 
         userlist = s.query(User).all()
 
@@ -4533,7 +4533,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
     def test_select_star_future(self):
         User = self.classes.User
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
         eq_(
             sess.execute(
                 select(User).from_statement(
@@ -4581,7 +4581,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
         # ordering doesn't matter
         User = self.classes.User
 
-        s = create_session(future=True)
+        s = create_session(testing.db, future=True)
         q = select(User).from_statement(
             text(
                 "select name, 27 as foo, id as users_id from users order by id"
@@ -4628,7 +4628,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
         User = self.classes.User
         Address = self.classes.Address
 
-        s = create_session(future=True)
+        s = create_session(testing.db, future=True)
         q = select(User, Address).from_statement(
             text(
                 "select users.name AS users_name, users.id AS users_id, "
@@ -4679,7 +4679,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
         User = self.classes.User
         Address = self.classes.Address
 
-        s = create_session(future=True)
+        s = create_session(testing.db, future=True)
         q = (
             select(User)
             .from_statement(
@@ -4731,7 +4731,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
         User = self.classes.User
         Address = self.classes.Address
 
-        s = create_session(future=True)
+        s = create_session(testing.db, future=True)
         q = (
             select(User)
             .from_statement(
@@ -4838,7 +4838,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
     def test_whereclause_future(self):
         User = self.classes.User
 
-        s = create_session(future=True)
+        s = create_session(testing.db, future=True)
         eq_(
             s.execute(select(User).filter(text("id in (8, 9)")))
             .scalars()
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 660bc7a5d..6095d8642 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -483,7 +483,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         User, users = self.classes.User, self.tables.users
 
         mapper(User, users)
-        sess = create_session(autocommit=False, future=True)
+        sess = create_session(testing.db, autocommit=False, future=True)
         u = User(name="u1")
         sess.add(u)
         sess.flush()
@@ -1032,7 +1032,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
         User, users = self.classes.User, self.tables.users
 
         mapper(User, users)
-        session = create_session(autocommit=False, future=True)
+        session = create_session(testing.db, autocommit=False, future=True)
         session.add(User(name="ed"))
         session.transaction.commit()
 
diff --git a/test/orm/test_update_delete.py b/test/orm/test_update_delete.py
index e4396d919..75dca1c99 100644
--- a/test/orm/test_update_delete.py
+++ b/test/orm/test_update_delete.py
@@ -154,7 +154,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
 
         User = self.classes.User
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         jill = s.query(User).filter(User.name == "jill").one()
 
@@ -179,7 +179,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
 
         User = self.classes.User
 
-        s = Session(future=True)
+        s = Session(testing.db, future=True)
 
         jill = s.query(User).filter(User.name == "jill").one()
 
@@ -435,7 +435,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
     def test_update_future(self):
         User, users = self.classes.User, self.tables.users
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
 
         john, jack, jill, jane = (
             sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -487,7 +487,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
     def test_update_future_lambda(self):
         User, users = self.classes.User, self.tables.users
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
 
         john, jack, jill, jane = (
             sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -623,7 +623,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
     def test_update_fetch_returning_lambda(self):
         User = self.classes.User
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
 
         john, jack, jill, jane = (
             sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -711,7 +711,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
     def test_delete_fetch_returning_lambda(self):
         User = self.classes.User
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
 
         john, jack, jill, jane = (
             sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -1068,7 +1068,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
 
     def test_update_multi_values_error_future(self):
         User = self.classes.User
-        session = Session(future=True)
+        session = Session(testing.db, future=True)
 
         # Do update using a tuple and check that order is preserved
 
@@ -1087,7 +1087,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
 
     def test_update_preserve_parameter_order_future(self):
         User = self.classes.User
-        session = Session(future=True)
+        session = Session(testing.db, future=True)
 
         # Do update using a tuple and check that order is preserved
 
@@ -1585,7 +1585,7 @@ class InheritTest(fixtures.DeclarativeMappedTest):
         person = self.classes.Person.__table__
         engineer = self.classes.Engineer.__table__
 
-        sess = Session(future=True)
+        sess = Session(testing.db, future=True)
         sess.query(person.join(engineer)).filter(person.c.name == "e2").update(
             {person.c.name: "updated", engineer.c.engineer_name: "e2a"},
         )