Make sure all executors and task are closed once hass closes

tests/conftest.py

@@ -12,6 +12,7 @@
 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 +188,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 +206,13 @@ def verify_cleanup():
     for thread in threads:
         assert isinstance(thread, threading._DummyThread)
 
+    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))
+
 
 @pytest.fixture(autouse=True)
 def bcrypt_cost():
@@ -381,7 +391,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 +412,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