fix test suite warnings

fix a handful of warnings that were emitting but not raising,
usually because they were inside an "expect_warnings" block.

modify "expect_warnings" to always use "raise_on_any_unexpected"
behavior; remove this parameter.

Fixed issue in semi-private ``await_only()`` and ``await_fallback()``
concurrency functions where the given awaitable would remain un-awaited if
the function threw a ``GreenletError``, which could cause "was not awaited"
warnings later on if the program continued. In this case, the given
awaitable is now cancelled before the exception is thrown.

Change-Id: I33668c5e8c670454a3d879e559096fb873b57244

28 files changed, 203 insertions, 139 deletions

diff --git a/doc/build/changelog/unreleased_20/await_cancel.rst b/doc/build/changelog/unreleased_20/await_cancel.rst
new file mode 100644
index 000000000..5d50a4fcc
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/await_cancel.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, asyncio
+
+    Fixed issue in semi-private ``await_only()`` and ``await_fallback()``
+    concurrency functions where the given awaitable would remain un-awaited if
+    the function threw a ``GreenletError``, which could cause "was not awaited"
+    warnings later on if the program continued. In this case, the given
+    awaitable is now cancelled before the exception is thrown.
diff --git a/doc/build/orm/queryguide/columns.rst b/doc/build/orm/queryguide/columns.rst
index 255c1f902..93d0919ba 100644
--- a/doc/build/orm/queryguide/columns.rst
+++ b/doc/build/orm/queryguide/columns.rst
@@ -869,7 +869,7 @@ onto newly loaded instances of ``A``::
     >>> orm_stmt = (
     ...     select(User)
     ...     .from_statement(union_stmt)
-    ...     .options(with_expression(User.book_count, union_stmt.c.book_count))
+    ...     .options(with_expression(User.book_count, union_stmt.selected_columns.book_count))
     ... )
     >>> for user in session.scalars(orm_stmt):
     ...     print(f"Username: {user.name}  Number of books: {user.book_count}")
diff --git a/examples/versioned_history/test_versioning.py b/examples/versioned_history/test_versioning.py
index 9caadc043..392a415ff 100644
--- a/examples/versioned_history/test_versioning.py
+++ b/examples/versioned_history/test_versioning.py
@@ -13,9 +13,9 @@ from sqlalchemy import Integer
 from sqlalchemy import join
 from sqlalchemy import select
 from sqlalchemy import String
-from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import clear_mappers
 from sqlalchemy.orm import column_property
+from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.orm import deferred
 from sqlalchemy.orm import exc as orm_exc
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index c879205e4..d198620d3 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -850,6 +850,7 @@ class AsyncAdapt_asyncpg_connection(AdaptedConnection):
 
     def terminate(self):
         self._connection.terminate()
+        self._started = False
 
     @staticmethod
     def _default_name_func():
diff --git a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
index f9c60efc1..2981976ac 100644
--- a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
@@ -239,6 +239,16 @@ class AsyncAdapt_aiosqlite_connection(AdaptedConnection):
     def close(self):
         try:
             self.await_(self._connection.close())
+        except ValueError:
+            # this is undocumented for aiosqlite, that ValueError
+            # was raised if .close() was called more than once, which is
+            # both not customary for DBAPI and is also not a DBAPI.Error
+            # exception. This is now fixed in aiosqlite via my PR
+            # https://github.com/omnilib/aiosqlite/pull/238, so we can be
+            # assured this will not become some other kind of exception,
+            # since it doesn't raise anymore.
+
+            pass
         except Exception as error:
             self._handle_exception(error)
 
diff --git a/lib/sqlalchemy/orm/_orm_constructors.py b/lib/sqlalchemy/orm/_orm_constructors.py
index 563fef3c5..fab93f682 100644
--- a/lib/sqlalchemy/orm/_orm_constructors.py
+++ b/lib/sqlalchemy/orm/_orm_constructors.py
@@ -83,6 +83,7 @@ _T = typing.TypeVar("_T")
     "The :class:`.AliasOption` object is not necessary "
     "for entities to be matched up to a query that is established "
     "via :meth:`.Query.from_statement` and now does nothing.",
+    enable_warnings=False,  # AliasOption itself warns
 )
 def contains_alias(alias: Union[Alias, Subquery]) -> AliasOption:
     r"""Return a :class:`.MapperOption` that will indicate to the
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 5cd7cc117..e6381bee1 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -1436,7 +1436,13 @@ class Query(
         to the given list of columns
 
         """
+        return self._values_no_warn(*columns)
 
+    _values = values
+
+    def _values_no_warn(
+        self, *columns: _ColumnsClauseArgument[Any]
+    ) -> Iterable[Any]:
         if not columns:
             return iter(())
         q = self._clone().enable_eagerloads(False)
@@ -1445,8 +1451,6 @@ class Query(
             q.load_options += {"_yield_per": 10}
         return iter(q)  # type: ignore
 
-    _values = values
-
     @util.deprecated(
         "1.4",
         ":meth:`_query.Query.value` "
@@ -1460,7 +1464,7 @@ class Query(
 
         """
         try:
-            return next(self.values(column))[0]  # type: ignore
+            return next(self._values_no_warn(column))[0]  # type: ignore
         except StopIteration:
             return None
 
@@ -3332,7 +3336,7 @@ class AliasOption(interfaces.LoaderOption):
 
     @util.deprecated(
         "1.4",
-        "The :class:`.AliasOption` is not necessary "
+        "The :class:`.AliasOption` object is not necessary "
         "for entities to be matched up to a query that is established "
         "via :meth:`.Query.from_statement` and now does nothing.",
     )
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index a51d831a9..e7b416167 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -47,7 +47,7 @@ def expect_warnings(*messages, **kw):
     Note that the test suite sets SAWarning warnings to raise exceptions.
 
     """  # noqa
-    return _expect_warnings(sa_exc.SAWarning, messages, **kw)
+    return _expect_warnings_sqla_only(sa_exc.SAWarning, messages, **kw)
 
 
 @contextlib.contextmanager
@@ -84,11 +84,15 @@ def emits_warning(*messages):
 
 
 def expect_deprecated(*messages, **kw):
-    return _expect_warnings(sa_exc.SADeprecationWarning, messages, **kw)
+    return _expect_warnings_sqla_only(
+        sa_exc.SADeprecationWarning, messages, **kw
+    )
 
 
 def expect_deprecated_20(*messages, **kw):
-    return _expect_warnings(sa_exc.Base20DeprecationWarning, messages, **kw)
+    return _expect_warnings_sqla_only(
+        sa_exc.Base20DeprecationWarning, messages, **kw
+    )
 
 
 def emits_warning_on(db, *messages):
@@ -140,6 +144,29 @@ _SEEN = None
 _EXC_CLS = None
 
 
+def _expect_warnings_sqla_only(
+    exc_cls,
+    messages,
+    regex=True,
+    search_msg=False,
+    assert_=True,
+):
+    """SQLAlchemy internal use only _expect_warnings().
+
+    Alembic is using _expect_warnings() directly, and should be updated
+    to use this new interface.
+
+    """
+    return _expect_warnings(
+        exc_cls,
+        messages,
+        regex=regex,
+        search_msg=search_msg,
+        assert_=assert_,
+        raise_on_any_unexpected=True,
+    )
+
+
 @contextlib.contextmanager
 def _expect_warnings(
     exc_cls,
@@ -150,7 +177,6 @@ def _expect_warnings(
     raise_on_any_unexpected=False,
     squelch_other_warnings=False,
 ):
-
     global _FILTERS, _SEEN, _EXC_CLS
 
     if regex or search_msg:
@@ -181,7 +207,6 @@ def _expect_warnings(
             real_warn = warnings.warn
 
         def our_warn(msg, *arg, **kw):
-
             if isinstance(msg, _EXC_CLS):
                 exception = type(msg)
                 msg = str(msg)
@@ -379,7 +404,7 @@ def assert_warns(except_cls, callable_, *args, **kwargs):
 
 
     """
-    with _expect_warnings(except_cls, [".*"], squelch_other_warnings=True):
+    with _expect_warnings_sqla_only(except_cls, [".*"]):
         return callable_(*args, **kwargs)
 
 
@@ -394,12 +419,11 @@ def assert_warns_message(except_cls, msg, callable_, *args, **kwargs):
     rather than regex.match().
 
     """
-    with _expect_warnings(
+    with _expect_warnings_sqla_only(
         except_cls,
         [msg],
         search_msg=True,
         regex=False,
-        squelch_other_warnings=True,
     ):
         return callable_(*args, **kwargs)
 
@@ -413,7 +437,6 @@ def assert_raises_message_context_ok(
 def _assert_raises(
     except_cls, callable_, args, kwargs, msg=None, check_context=False
 ):
-
     with _expect_raises(except_cls, msg, check_context) as ec:
         callable_(*args, **kwargs)
     return ec.error
@@ -892,7 +915,6 @@ class AssertsExecutionResults:
         return result
 
     def assert_sql(self, db, callable_, rules):
-
         newrules = []
         for rule in rules:
             if isinstance(rule, dict):
diff --git a/lib/sqlalchemy/util/_concurrency_py3k.py b/lib/sqlalchemy/util/_concurrency_py3k.py
index b29cc2119..0e26425b2 100644
--- a/lib/sqlalchemy/util/_concurrency_py3k.py
+++ b/lib/sqlalchemy/util/_concurrency_py3k.py
@@ -17,11 +17,13 @@ from typing import Awaitable
 from typing import Callable
 from typing import Coroutine
 from typing import Optional
+from typing import TYPE_CHECKING
 from typing import TypeVar
 
 from .langhelpers import memoized_property
 from .. import exc
 from ..util.typing import Protocol
+from ..util.typing import TypeGuard
 
 _T = TypeVar("_T")
 
@@ -78,6 +80,26 @@ class _AsyncIoGreenlet(greenlet):  # type: ignore
             self.gr_context = driver.gr_context
 
 
+_T_co = TypeVar("_T_co", covariant=True)
+
+if TYPE_CHECKING:
+
+    def iscoroutine(
+        awaitable: Awaitable[_T_co],
+    ) -> TypeGuard[Coroutine[Any, Any, _T_co]]:
+        ...
+
+else:
+    iscoroutine = asyncio.iscoroutine
+
+
+def _safe_cancel_awaitable(awaitable: Awaitable[Any]) -> None:
+    # https://docs.python.org/3/reference/datamodel.html#coroutine.close
+
+    if iscoroutine(awaitable):
+        awaitable.close()
+
+
 def await_only(awaitable: Awaitable[_T]) -> _T:
     """Awaits an async function in a sync method.
 
@@ -90,6 +112,8 @@ def await_only(awaitable: Awaitable[_T]) -> _T:
     # this is called in the context greenlet while running fn
     current = getcurrent()
     if not isinstance(current, _AsyncIoGreenlet):
+        _safe_cancel_awaitable(awaitable)
+
         raise exc.MissingGreenlet(
             "greenlet_spawn has not been called; can't call await_only() "
             "here. Was IO attempted in an unexpected place?"
@@ -117,6 +141,9 @@ def await_fallback(awaitable: Awaitable[_T]) -> _T:
     if not isinstance(current, _AsyncIoGreenlet):
         loop = get_event_loop()
         if loop.is_running():
+
+            _safe_cancel_awaitable(awaitable)
+
             raise exc.MissingGreenlet(
                 "greenlet_spawn has not been called and asyncio event "
                 "loop is already running; can't call await_fallback() here. "
diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index 9b2cb31be..dc5a39910 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -296,7 +296,6 @@ class MemUsageTest(EnsureZeroed):
 
     @testing.requires.cextensions
     def test_cycles_in_row(self):
-
         tup = result.result_tuple(["a", "b", "c"])
 
         @profile_memory()
@@ -695,7 +694,6 @@ class MemUsageWBackendTest(fixtures.MappedTest, EnsureZeroed):
         @testing.emits_warning()
         @profile_memory()
         def go():
-
             # execute with a non-unicode object. a warning is emitted,
             # this warning shouldn't clog up memory.
 
@@ -1066,7 +1064,9 @@ class MemUsageWBackendTest(fixtures.MappedTest, EnsureZeroed):
 
         t1_mapper = self.mapper_registry.map_imperatively(T1, t1)
 
-        @testing.emits_warning()
+        @testing.emits_warning(
+            r"This declarative base", r"Property .* being replaced"
+        )
         @profile_memory()
         def go():
             class T2:
@@ -1128,7 +1128,9 @@ class MemUsageWBackendTest(fixtures.MappedTest, EnsureZeroed):
             s = table2.select()
             sess = session()
             with testing.expect_deprecated(
-                "Implicit coercion of SELECT and " "textual SELECT constructs"
+                "Implicit coercion of SELECT and textual SELECT constructs",
+                "An alias is being generated automatically",
+                assert_=False,
             ):
                 sess.query(Foo).join(s, Foo.bars).all()
             sess.rollback()
@@ -1637,7 +1639,6 @@ class CycleTest(_fixtures.FixtureTest):
 
     @testing.provide_metadata
     def test_optimized_get(self):
-
         Base = declarative_base(metadata=self.metadata)
 
         class Employee(Base):
diff --git a/test/base/test_concurrency_py3k.py b/test/base/test_concurrency_py3k.py
index 6a3098a6a..17a0fafb5 100644
--- a/test/base/test_concurrency_py3k.py
+++ b/test/base/test_concurrency_py3k.py
@@ -95,7 +95,12 @@ class TestAsyncioCompat(fixtures.TestBase):
         ):
             await_only(to_await)
 
-        # ensure no warning
+        # existing awaitable is done
+        with expect_raises(RuntimeError):
+            await greenlet_spawn(await_fallback, to_await)
+
+        # no warning for a new one...
+        to_await = run1()
         await greenlet_spawn(await_fallback, to_await)
 
     @async_test
@@ -118,7 +123,8 @@ class TestAsyncioCompat(fixtures.TestBase):
         ):
             await greenlet_spawn(go)
 
-        await to_await
+        with expect_raises(RuntimeError):
+            await to_await
 
     @async_test
     async def test_await_only_error(self):
@@ -141,7 +147,8 @@ class TestAsyncioCompat(fixtures.TestBase):
         ):
             await greenlet_spawn(go)
 
-        await to_await
+        with expect_raises(RuntimeError):
+            await to_await
 
     @async_test
     async def test_contextvars(self):
diff --git a/test/dialect/mysql/test_query.py b/test/dialect/mysql/test_query.py
index 0ce361182..921b5c52b 100644
--- a/test/dialect/mysql/test_query.py
+++ b/test/dialect/mysql/test_query.py
@@ -11,9 +11,9 @@ from sqlalchemy import or_
 from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
-from sqlalchemy import testing
 from sqlalchemy import true
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 
@@ -22,23 +22,26 @@ class IdiosyncrasyTest(fixtures.TestBase):
     __only_on__ = "mysql", "mariadb"
     __backend__ = True
 
-    @testing.emits_warning()
     def test_is_boolean_symbols_despite_no_native(self, connection):
+        with expect_warnings("Datatype BOOL does not support CAST"):
+            is_(
+                connection.scalar(select(cast(true().is_(true()), Boolean))),
+                True,
+            )
 
-        is_(
-            connection.scalar(select(cast(true().is_(true()), Boolean))),
-            True,
-        )
-
-        is_(
-            connection.scalar(select(cast(true().is_not(true()), Boolean))),
-            False,
-        )
+        with expect_warnings("Datatype BOOL does not support CAST"):
+            is_(
+                connection.scalar(
+                    select(cast(true().is_not(true()), Boolean))
+                ),
+                False,
+            )
 
-        is_(
-            connection.scalar(select(cast(false().is_(false()), Boolean))),
-            True,
-        )
+        with expect_warnings("Datatype BOOL does not support CAST"):
+            is_(
+                connection.scalar(select(cast(false().is_(false()), Boolean))),
+                True,
+            )
 
 
 class MatchTest(fixtures.TablesTest):
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index e99448a26..2c5f091b4 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -2044,6 +2044,7 @@ class EngineEventsTest(fixtures.TestBase):
                 select(1).compile(dialect=e1.dialect), (), {}
             )
 
+    @testing.emits_warning("The garbage collector is trying to clean up")
     def test_execute_events(self):
 
         stmts = []
@@ -2283,6 +2284,7 @@ class EngineEventsTest(fixtures.TestBase):
 
         eng_copy = copy.copy(eng)
         eng_copy.dispose(close=close)
+
         copy_conn = eng_copy.connect()
         dbapi_conn_two = copy_conn.connection.dbapi_connection
 
@@ -2294,6 +2296,9 @@ class EngineEventsTest(fixtures.TestBase):
         else:
             is_(dbapi_conn_one, conn.connection.dbapi_connection)
 
+        conn.close()
+        copy_conn.close()
+
     def test_retval_flag(self):
         canary = []
 
@@ -3702,8 +3707,8 @@ class DialectEventTest(fixtures.TestBase):
 
         conn.connection.invalidate(soft=True)
         conn.close()
-        conn = e.connect()
-        eq_(conn.info["boom"], "one")
+        with e.connect() as conn:
+            eq_(conn.info["boom"], "one")
 
     def test_connect_do_connect_info_there_after_invalidate(self):
         # test that info is maintained after the do_connect()
@@ -3720,8 +3725,9 @@ class DialectEventTest(fixtures.TestBase):
         eq_(conn.info["boom"], "one")
 
         conn.connection.invalidate()
-        conn = e.connect()
-        eq_(conn.info["boom"], "one")
+
+        with e.connect() as conn:
+            eq_(conn.info["boom"], "one")
 
 
 class SetInputSizesTest(fixtures.TablesTest):
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 6730d7012..cca6e2589 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -1705,6 +1705,7 @@ class QueuePoolTest(PoolTestBase):
         )
 
     @testing.combinations((True,), (False,))
+    @testing.emits_warning("The garbage collector")
     def test_userspace_disconnectionerror_weakref_finalizer(self, detach_gced):
         dbapi, pool = self._queuepool_dbapi_fixture(
             pool_size=1, max_overflow=2, _is_asyncio=detach_gced
@@ -1737,6 +1738,7 @@ class QueuePoolTest(PoolTestBase):
 
         not_closed_dbapi_conn = conn.dbapi_connection
         del conn
+
         gc_collect()
 
         if detach_gced:
@@ -1744,6 +1746,9 @@ class QueuePoolTest(PoolTestBase):
             eq_(not_closed_dbapi_conn.mock_calls, [])
         else:
             # new connection reset and returned to pool
+            # this creates a gc-level warning that is not easy to pin down,
+            # hence we use the testing.emits_warning() decorator just to squash
+            # it
             eq_(not_closed_dbapi_conn.mock_calls, [call.rollback()])
 
     @testing.requires.timing_intensive
diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py
index 36135a43d..fed59995f 100644
--- a/test/ext/asyncio/test_session_py3k.py
+++ b/test/ext/asyncio/test_session_py3k.py
@@ -741,6 +741,7 @@ class AsyncORMBehaviorsTest(AsyncFixture):
                 (exc.StatementError, exc.MissingGreenlet)
             ):
                 a1.b = b2
+
         else:
             a1.b = b2
 
diff --git a/test/ext/test_deprecations.py b/test/ext/test_deprecations.py
index 97c4172ba..653a02157 100644
--- a/test/ext/test_deprecations.py
+++ b/test/ext/test_deprecations.py
@@ -70,11 +70,11 @@ class HorizontalShardTest(fixtures.TestBase):
         m1 = mock.Mock()
 
         with testing.expect_deprecated(
-            "The ``query_chooser`` parameter is deprecated; please use"
+            "The ``query_chooser`` parameter is deprecated; please use",
         ):
             s = ShardedSession(
                 shard_chooser=m1.shard_chooser,
-                id_chooser=m1.id_chooser,
+                identity_chooser=m1.identity_chooser,
                 query_chooser=m1.query_chooser,
             )
 
diff --git a/test/orm/declarative/test_basic.py b/test/orm/declarative/test_basic.py
index d0e56819c..6dcecc0c1 100644
--- a/test/orm/declarative/test_basic.py
+++ b/test/orm/declarative/test_basic.py
@@ -1394,7 +1394,6 @@ class DeclarativeMultiBaseTest(
             r"non-schema SQLAlchemy expression object; ",
             r"Attribute 'y' on class <class .*Foo5.* appears to be a "
             r"non-schema SQLAlchemy expression object; ",
-            raise_on_any_unexpected=True,
         ):
 
             class Foo5(Base):
diff --git a/test/orm/declarative/test_inheritance.py b/test/orm/declarative/test_inheritance.py
index 333d24230..7d66c0d0b 100644
--- a/test/orm/declarative/test_inheritance.py
+++ b/test/orm/declarative/test_inheritance.py
@@ -1026,7 +1026,6 @@ class DeclarativeInheritanceTest(
         if not combine_on_b and not omit_from_statements:
             ctx = expect_warnings(
                 "Implicitly combining column a.extra with column b.extra",
-                raise_on_any_unexpected=True,
             )
         else:
             ctx = contextlib.nullcontext()
diff --git a/test/orm/inheritance/test_relationship.py b/test/orm/inheritance/test_relationship.py
index 6d168d2ce..ab9c29918 100644
--- a/test/orm/inheritance/test_relationship.py
+++ b/test/orm/inheritance/test_relationship.py
@@ -61,7 +61,6 @@ def _aliased_join_warning(arg):
     return testing.expect_warnings(
         r"An alias is being generated automatically against joined entity "
         r"Mapper\[%s\] due to overlapping tables" % (arg,),
-        raise_on_any_unexpected=True,
     )
 
 
diff --git a/test/orm/test_collection.py b/test/orm/test_collection.py
index 650b30aaa..85efbe0a3 100644
--- a/test/orm/test_collection.py
+++ b/test/orm/test_collection.py
@@ -2928,7 +2928,6 @@ class AttrKeyedDictKeysTest(fixtures.TestBase):
                 "'bs' was None;",
                 "Attribute keyed dictionary value for attribute "
                 "'B.a' was None;",
-                raise_on_any_unexpected=True,
             ):
                 a1 = A(bs={"k1": b1, "k2": b2})
         sess.add(a1)
@@ -2943,7 +2942,6 @@ class AttrKeyedDictKeysTest(fixtures.TestBase):
             with expect_warnings(
                 "Attribute keyed dictionary value for attribute "
                 "'unknown relationship' was None;",
-                raise_on_any_unexpected=True,
             ):
                 eq_(a1.bs, {None: b2})
 
@@ -3074,7 +3072,6 @@ class UnpopulatedAttrTest(fixtures.TestBase):
             with expect_warnings(
                 "Attribute keyed dictionary value for attribute "
                 "'B.a' was None;",
-                raise_on_any_unexpected=True,
             ):
                 b1.a = a1
         else:
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index a3e2f4ef7..db64f91f1 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -414,7 +414,6 @@ class MiscDeprecationsTest(fixtures.TestBase):
         with expect_deprecated(
             rf"The column_property.{paramname} parameter is deprecated "
             r"for column_property\(\)",
-            raise_on_any_unexpected=True,
         ):
             column_property(column("q"), **{paramname: value})
 
@@ -429,7 +428,6 @@ class MiscDeprecationsTest(fixtures.TestBase):
             r"for column_property\(\)",
             r"The column_property.kw_only parameter is deprecated "
             r"for column_property\(\)",
-            raise_on_any_unexpected=True,
         ):
 
             class MyClass(dc_decl_base):
@@ -488,7 +486,8 @@ class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
         s = addresses.select()
         sess = fixture_session()
         with testing.expect_deprecated(
-            "Implicit coercion of SELECT and " "textual SELECT constructs"
+            "Implicit coercion of SELECT and textual SELECT constructs",
+            "An alias is being generated automatically against joined entity",
         ):
             self.assert_compile(
                 sess.query(User).join(s, User.addresses),
@@ -1592,7 +1591,7 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
         def go():
             with testing.expect_deprecated(
                 "The AliasOption object is not necessary for entities to be "
-                "matched up to a query"
+                "matched up to a query",
             ):
                 result = (
                     q.options(
@@ -1624,7 +1623,8 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
 
         def go():
             with testing.expect_deprecated(
-                r"Using the Query.instances\(\) method without a context"
+                r"The Query.instances\(\) method is deprecated",
+                r"Using the Query.instances\(\) method without a context",
             ):
                 result = list(
                     q.options(contains_eager(User.addresses)).instances(
@@ -1639,7 +1639,8 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
 
         def go():
             with testing.expect_deprecated(
-                r"Using the Query.instances\(\) method without a context"
+                r"The Query.instances\(\) method is deprecated",
+                r"Using the Query.instances\(\) method without a context",
             ):
                 result = list(
                     q.options(contains_eager(User.addresses)).instances(
@@ -1674,7 +1675,6 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
                 r"Using the Query.instances\(\) method without a context",
                 r"The Query.instances\(\) method is deprecated and will be "
                 r"removed in a future release.",
-                raise_on_any_unexpected=True,
             ):
                 result = list(
                     q.options(
@@ -1721,7 +1721,6 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
                 r"Using the Query.instances\(\) method without a context",
                 r"The Query.instances\(\) method is deprecated and will be "
                 r"removed in a future release.",
-                raise_on_any_unexpected=True,
             ):
                 result = list(
                     q.options(
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index ef6d4b684..fa44dbf10 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -1656,7 +1656,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
         def go():
             eq_(u1.addresses[0].user, u1)
 
-        with testing.expect_warnings(raise_on_any_unexpected=True):
+        with testing.expect_warnings():
             self.assert_sql_execution(
                 testing.db,
                 go,
@@ -1707,7 +1707,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
         def go():
             eq_(u1.addresses[0].user, u1)
 
-        with testing.expect_warnings(raise_on_any_unexpected=True):
+        with testing.expect_warnings():
             self.assert_sql_execution(
                 testing.db,
                 go,
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 3120a160d..ff684d846 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -170,7 +170,6 @@ class InheritedJoinTest(InheritedTest, AssertsCompiledSQL):
             r"Mapper\[Manager\(managers\)\] due to overlapping",
             "An alias is being generated automatically against joined entity "
             r"Mapper\[Boss\(boss\)\] due to overlapping",
-            raise_on_any_unexpected=True,
         ):
             self.assert_compile(
                 q,
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 0cbcc01f3..12651fe36 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -33,7 +33,6 @@ from sqlalchemy.orm.interfaces import MANYTOONE
 from sqlalchemy.orm.interfaces import ONETOMANY
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import assert_warns_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises_message
@@ -656,18 +655,15 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
         add_bsub2_a_viewonly=False,
         add_b_a_overlaps=None,
     ):
-
         Base = self.mapper_registry.generate_base()
 
         class A(Base):
-
             __tablename__ = "a"
 
             id = Column(Integer, primary_key=True)
             a_members = relationship("AMember", backref="a")
 
         class AMember(Base):
-
             __tablename__ = "a_member"
 
             a_id = Column(Integer, ForeignKey("a.id"), primary_key=True)
@@ -707,14 +703,12 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
         # however, *no* warning should be emitted otherwise.
 
         class BSub1(B):
-
             if add_bsub1_a:
                 a = relationship("A")
 
             __mapper_args__ = {"polymorphic_identity": "bsub1"}
 
         class BSub2(B):
-
             if add_bsub2_a_viewonly:
                 a = relationship("A", viewonly=True)
 
@@ -729,7 +723,6 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
         return A, AMember, B, BSub1, BSub2
 
     def _fixture_two(self, setup_backrefs=False, setup_overlaps=False):
-
         Base = self.mapper_registry.generate_base()
 
         # purposely using the comma to make sure parsing the comma works
@@ -874,15 +867,13 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
 
     @testing.provide_metadata
     def test_simple_warn(self):
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship '(?:Child.parent|Parent.children)' will copy "
             r"column parent.id to column child.parent_id, which conflicts "
             r"with relationship\(s\): '(?:Parent.children|Child.parent)' "
-            r"\(copies parent.id to child.parent_id\).",
-            self._fixture_two,
-            setup_backrefs=False,
-        )
+            r"\(copies parent.id to child.parent_id\)."
+        ):
+            self._fixture_two(setup_backrefs=False)
 
     @testing.combinations((True,), (False,), argnames="set_overlaps")
     def test_fixture_five(self, metadata, set_overlaps):
@@ -965,15 +956,12 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
 
     @testing.provide_metadata
     def test_double_rel_same_mapper_warns(self):
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship 'Parent.child[12]' will copy column parent.id to "
             r"column child.parent_id, which conflicts with relationship\(s\): "
-            r"'Parent.child[12]' \(copies parent.id to child.parent_id\)",
-            self._fixture_three,
-            use_same_mappers=True,
-            setup_overlaps=False,
-        )
+            r"'Parent.child[12]' \(copies parent.id to child.parent_id\)"
+        ):
+            self._fixture_three(use_same_mappers=True, setup_overlaps=False)
 
     @testing.provide_metadata
     def test_double_rel_same_mapper_overlaps_works(self):
@@ -985,48 +973,37 @@ class OverlappingFksSiblingTest(fixtures.MappedTest):
 
     @testing.provide_metadata
     def test_warn_one(self):
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship '(?:BSub1.a|BSub2.a_member|B.a)' will copy column "
-            r"(?:a.id|a_member.a_id) to column b.a_id",
-            self._fixture_one,
-            add_b_a=True,
-            add_bsub1_a=True,
-        )
+            r"(?:a.id|a_member.a_id) to column b.a_id"
+        ):
+            self._fixture_one(add_b_a=True, add_bsub1_a=True)
 
     @testing.provide_metadata
     def test_warn_two(self):
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship '(?:BSub1.a|B.a_member)' will copy column "
-            r"(?:a.id|a_member.a_id) to column b.a_id",
-            self._fixture_one,
-            add_b_amember=True,
-            add_bsub1_a=True,
-        )
+            r"(?:a.id|a_member.a_id) to column b.a_id"
+        ):
+            self._fixture_one(add_b_amember=True, add_bsub1_a=True)
 
     @testing.provide_metadata
     def test_warn_three(self):
-        assert_warns_message(
-            exc.SAWarning,
-            r"relationship '(?:BSub1.a|B.a_member|B.a)' will copy column "
-            r"(?:a.id|a_member.a_id) to column b.a_id",
-            self._fixture_one,
-            add_b_amember=True,
-            add_bsub1_a=True,
-            add_b_a=True,
-        )
+        with expect_warnings(
+            r"relationship '(?:BSub1.a|B.a_member|BSub2.a_member|B.a)' "
+            r"will copy column (?:a.id|a_member.a_id) to column b.a_id",
+        ):
+            self._fixture_one(
+                add_b_amember=True, add_bsub1_a=True, add_b_a=True
+            )
 
     @testing.provide_metadata
     def test_warn_four(self):
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship '(?:B.a|BSub2.a_member|B.a)' will copy column "
-            r"(?:a.id|a_member.a_id) to column b.a_id",
-            self._fixture_one,
-            add_bsub2_a_viewonly=True,
-            add_b_a=True,
-        )
+            r"(?:a.id|a_member.a_id) to column b.a_id"
+        ):
+            self._fixture_one(add_bsub2_a_viewonly=True, add_b_a=True)
 
     @testing.provide_metadata
     def test_works_one(self):
@@ -1303,12 +1280,11 @@ class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
             },
         )
 
-        assert_warns_message(
-            exc.SAWarning,
+        with expect_warnings(
             r"relationship .* will copy column .* to column "
-            r"employee_t.company_id, which conflicts with relationship\(s\)",
-            configure_mappers,
-        )
+            r"employee_t.company_id, which conflicts with relationship\(s\)"
+        ):
+            configure_mappers()
 
     def test_annotated_no_overwriting(self):
         Employee, Company, employee_t, company_t = (
@@ -2537,7 +2513,6 @@ class JoinConditionErrorTest(fixtures.TestBase):
         argnames="argname, arg",
     )
     def test_invalid_string_args(self, registry, argname, arg):
-
         kw = {argname: arg}
         Base = registry.generate_base()
 
@@ -4789,7 +4764,6 @@ class SecondaryNestedJoinTest(
         )
 
     def test_render_lazyload(self):
-
         A = self.classes.A
         sess = fixture_session()
         a1 = sess.query(A).filter(A.name == "a1").first()
@@ -5968,7 +5942,6 @@ class InactiveHistoryNoRaiseTest(_fixtures.FixtureTest):
         delete,
         legacy_inactive_history_style,
     ):
-
         if delete:
             assert not backref, "delete and backref are mutually exclusive"
 
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 5a0431788..395603a77 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -36,6 +36,7 @@ from sqlalchemy.testing import config
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_raises_message
+from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_false
@@ -108,7 +109,6 @@ class ExecutionTest(_fixtures.FixtureTest):
         )
 
     def test_no_string_execute(self, connection):
-
         with Session(bind=connection) as sess:
             with expect_raises_message(
                 sa.exc.ArgumentError,
@@ -242,7 +242,6 @@ class TransScopingTest(_fixtures.FixtureTest):
         self.mapper_registry.map_imperatively(Address, addresses)
 
         with Session(testing.db) as sess:
-
             sess.add(User(name="u1"))
             sess.commit()
 
@@ -2293,13 +2292,13 @@ class FlushWarningsTest(fixtures.MappedTest):
         def evt(mapper, conn, instance):
             instance.addresses.append(Address(email="x1"))
 
-        self._test(evt, "collection append")
+        self._test(evt, "collection append", "related attribute set")
 
     def test_o2m_cascade_remove(self):
         def evt(mapper, conn, instance):
             del instance.addresses[0]
 
-        self._test(evt, "collection remove")
+        self._test(evt, "collection remove", "related attribute set")
 
     def test_m2o_cascade_add(self):
         User = self.classes.User
@@ -2308,14 +2307,19 @@ class FlushWarningsTest(fixtures.MappedTest):
             instance.addresses[0].user = User(name="u2")
 
         with expect_raises_message(orm_exc.FlushError, ".*Over 100"):
-            self._test(evt, "related attribute set")
+            self._test(
+                evt,
+                "related attribute set",
+                "collection remove",
+                "collection append",
+            )
 
     def test_m2o_cascade_remove(self):
         def evt(mapper, conn, instance):
             a1 = instance.addresses[0]
             del a1.user
 
-        self._test(evt, "related attribute delete")
+        self._test(evt, "related attribute delete", "collection remove")
 
     def test_plain_add(self):
         Address = self.classes.Address
@@ -2344,7 +2348,7 @@ class FlushWarningsTest(fixtures.MappedTest):
         ):
             self._test(evt, r"Session.delete\(\)")
 
-    def _test(self, fn, method):
+    def _test(self, fn, *methods):
         User = self.classes.User
         Address = self.classes.Address
 
@@ -2353,6 +2357,8 @@ class FlushWarningsTest(fixtures.MappedTest):
 
         u1 = User(name="u1", addresses=[Address(name="a1")])
         s.add(u1)
-        assert_warns_message(
-            sa.exc.SAWarning, "Usage of the '%s'" % method, s.commit
-        )
+
+        with expect_warnings(
+            *[f"Usage of the '{method}'" for method in methods]
+        ):
+            s.commit()
diff --git a/test/orm/test_utils.py b/test/orm/test_utils.py
index 3bd0230de..4d6c14863 100644
--- a/test/orm/test_utils.py
+++ b/test/orm/test_utils.py
@@ -78,7 +78,6 @@ class ContextualWarningsTest(fixtures.TestBase):
                 "bar.foo_id, which conflicts with relationship(s): 'Foo.bars' "
                 "(copies foo.id to bar.foo_id). "
             ),
-            raise_on_any_unexpected=True,
         ):
             decl_base.registry.configure()
 
@@ -96,7 +95,6 @@ class ContextualWarningsTest(fixtures.TestBase):
                 "invoked automatically in response to a user-initiated "
                 "operation.)"
             ),
-            raise_on_any_unexpected=True,
         ):
             FooAlias = aliased(Foo)
             assert hasattr(FooAlias, "bars")
@@ -115,7 +113,6 @@ class ContextualWarningsTest(fixtures.TestBase):
                 "invoked automatically in response to a user-initiated "
                 "operation.)"
             ),
-            raise_on_any_unexpected=True,
         ):
             foo = Foo()
             assert hasattr(foo, "bars")
@@ -145,7 +142,6 @@ class ContextualWarningsTest(fixtures.TestBase):
                 "process, which was invoked automatically in response to a "
                 "user-initiated operation.)"
             ),
-            raise_on_any_unexpected=True,
         ):
             sess.execute(select(Foo))
 
diff --git a/test/sql/test_delete.py b/test/sql/test_delete.py
index 5b7e5ebbe..9f9e104b6 100644
--- a/test/sql/test_delete.py
+++ b/test/sql/test_delete.py
@@ -76,11 +76,15 @@ class DeleteTest(_DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL):
     def test_where_empty(self):
         table1 = self.tables.mytable
 
-        with expect_deprecated():
+        with expect_deprecated(
+            r"Invoking and_\(\) without arguments is deprecated"
+        ):
             self.assert_compile(
                 table1.delete().where(and_()), "DELETE FROM mytable"
             )
-        with expect_deprecated():
+        with expect_deprecated(
+            r"Invoking or_\(\) without arguments is deprecated"
+        ):
             self.assert_compile(
                 table1.delete().where(or_()), "DELETE FROM mytable"
             )
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index f545671e7..0fee9bd35 100644
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -1239,7 +1239,6 @@ class IMVSentinelTest(fixtures.TestBase):
             ):
                 return expect_warnings(
                     "Batches were downgraded",
-                    raise_on_any_unexpected=True,
                 )
 
         return contextlib.nullcontext()
@@ -1982,7 +1981,6 @@ class IMVSentinelTest(fixtures.TestBase):
         ):
             with expect_warnings(
                 "Batches were downgraded for sorted INSERT",
-                raise_on_any_unexpected=True,
             ):
                 result = connection.execute(stmt, data)
         else:
@@ -2407,7 +2405,6 @@ class IMVSentinelTest(fixtures.TestBase):
 
         with expect_warnings(
             r".*but has no Python-side or server-side default ",
-            raise_on_any_unexpected=True,
         ):
             with expect_raises(exc.IntegrityError):
                 connection.execute(