Make sure executors, tasks and timers are closed

Some test will trigger warnings on garbage collect, these warnings
spills over into next test.

Some test trigger tasks that raise errors on shutdown, these spill
over into next test.

This is to mimic older pytest-aiohttp and it's behaviour on test
cleanup.

Discussions on similar changes for pytest-aiohttp are here:
pytest-dev/pytest-asyncio#309

Author: elupus

tests/conftest.py

@@ -5,13 +5,15 @@
 from collections.abc import AsyncGenerator, Callable, Generator
 from contextlib import asynccontextmanager
 import functools
+import gc
 from json import JSONDecoder, loads
 import logging
 import sqlite3
 import ssl
 import threading
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
+import warnings
 
 from aiohttp import client
 from aiohttp.pytest_plugin import AiohttpClient
@@ -187,12 +189,14 @@ async def guard_func(*args, **kwargs):
 
 
 @pytest.fixture(autouse=True)
-def verify_cleanup():
+def verify_cleanup(event_loop: asyncio.AbstractEventLoop):
     """Verify that the test has cleaned up resources correctly."""
     threads_before = frozenset(threading.enumerate())
-
+    tasks_before = asyncio.all_tasks(event_loop)
     yield
 
+    event_loop.run_until_complete(event_loop.shutdown_default_executor())
+
     if len(INSTANCES) >= 2:
         count = len(INSTANCES)
         for inst in INSTANCES:
@@ -203,6 +207,26 @@ def verify_cleanup():
     for thread in threads:
         assert isinstance(thread, threading._DummyThread)
 
+    # Warn and clean-up lingering tasks and timers
+    # before moving on to the next test.
+    tasks = asyncio.all_tasks(event_loop) - tasks_before
+    for task in tasks:
+        warnings.warn(f"Linger task after test {task}")
+        task.cancel()
+    if tasks:
+        event_loop.run_until_complete(asyncio.wait(tasks))
+
+    for handle in event_loop._scheduled:  # pylint: disable=protected-access
+        if not handle.cancelled():
+            warnings.warn(f"Lingering timer after test {handle}")
+            handle.cancel()
+
+    # Make sure garbage collect run in same test as allocation
+    # this is to mimic the behavior of pytest-aiohttp, and is
+    # required to avoid warnings from spilling over into next
+    # test case.
+    gc.collect()
+
 
 @pytest.fixture(autouse=True)
 def bcrypt_cost():
@@ -381,7 +405,7 @@ def exc_handle(loop, context):
 
 
 @pytest.fixture
-async def stop_hass():
+async def stop_hass(event_loop):
     """Make sure all hass are stopped."""
     orig_hass = ha.HomeAssistant
 
@@ -402,6 +426,7 @@ def mock_hass():
         with patch.object(hass_inst.loop, "stop"):
             await hass_inst.async_block_till_done()
             await hass_inst.async_stop(force=True)
+            await event_loop.shutdown_default_executor()
 
 
 @pytest.fixture