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

1 files changed, 27 insertions, 0 deletions

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. "