Keep task references while running (#87970)

* Keep task references while running

* Update pilight tests pointing at correct logger call

* Fix graphite tests

* Fix profiler tests

* More graphite test fixes

* Remove extra sleep

* Fix tests

* Shutdown background tasks as part of stage 1

* Remove unnecessary sleep in test

* Remove unused method on mock hass

* Skip on cancelled too

* Remove background tasks

* Test trigger variables without actually sleeping

* Fix graphite

* One more graphite grrrrrrr

tests/common.py

@@ -3,14 +3,7 @@ from __future__ import annotations
 
 import asyncio
 from collections import OrderedDict
-from collections.abc import (
-    Awaitable,
-    Callable,
-    Collection,
-    Generator,
-    Mapping,
-    Sequence,
-)
+from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 import functools as ft
@@ -22,8 +15,6 @@ import os
 import pathlib
 import threading
 import time
-from time import monotonic
-import types
 from typing import Any, NoReturn
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -51,7 +42,6 @@ from homeassistant.const import (
     STATE_ON,
 )
 from homeassistant.core import (
-    BLOCK_LOG_TIMEOUT,
     CoreState,
     Event,
     HomeAssistant,
@@ -221,76 +211,9 @@ async def async_test_home_assistant(event_loop, load_registries=True):
 
         return orig_async_create_task(coroutine)
 
-    async def async_wait_for_task_count(self, max_remaining_tasks: int = 0) -> None:
-        """Block until at most max_remaining_tasks remain.
-
-        Based on HomeAssistant.async_block_till_done
-        """
-        # To flush out any call_soon_threadsafe
-        await asyncio.sleep(0)
-        start_time: float | None = None
-
-        while len(self._pending_tasks) > max_remaining_tasks:
-            pending: Collection[Awaitable[Any]] = [
-                task for task in self._pending_tasks if not task.done()
-            ]
-            self._pending_tasks.clear()
-            if len(pending) > max_remaining_tasks:
-                remaining_pending = await self._await_count_and_log_pending(
-                    pending, max_remaining_tasks=max_remaining_tasks
-                )
-                self._pending_tasks.extend(remaining_pending)
-
-                if start_time is None:
-                    # Avoid calling monotonic() until we know
-                    # we may need to start logging blocked tasks.
-                    start_time = 0
-                elif start_time == 0:
-                    # If we have waited twice then we set the start
-                    # time
-                    start_time = monotonic()
-                elif monotonic() - start_time > BLOCK_LOG_TIMEOUT:
-                    # We have waited at least three loops and new tasks
-                    # continue to block. At this point we start
-                    # logging all waiting tasks.
-                    for task in pending:
-                        _LOGGER.debug("Waiting for task: %s", task)
-            else:
-                self._pending_tasks.extend(pending)
-                await asyncio.sleep(0)
-
-    async def _await_count_and_log_pending(
-        self, pending: Collection[Awaitable[Any]], max_remaining_tasks: int = 0
-    ) -> Collection[Awaitable[Any]]:
-        """Block at most max_remaining_tasks remain and log tasks that take a long time.
-
-        Based on HomeAssistant._await_and_log_pending
-        """
-        wait_time = 0
-
-        return_when = asyncio.ALL_COMPLETED
-        if max_remaining_tasks:
-            return_when = asyncio.FIRST_COMPLETED
-
-        while len(pending) > max_remaining_tasks:
-            _, pending = await asyncio.wait(
-                pending, timeout=BLOCK_LOG_TIMEOUT, return_when=return_when
-            )
-            if not pending or max_remaining_tasks:
-                return pending
-            wait_time += BLOCK_LOG_TIMEOUT
-            for task in pending:
-                _LOGGER.debug("Waited %s seconds for task: %s", wait_time, task)
-
-        return []
-
     hass.async_add_job = async_add_job
     hass.async_add_executor_job = async_add_executor_job
     hass.async_create_task = async_create_task
-    hass.async_wait_for_task_count = types.MethodType(async_wait_for_task_count, hass)
-    hass._await_count_and_log_pending = types.MethodType(
-        _await_count_and_log_pending, hass
-    )
 
     hass.data[loader.DATA_CUSTOM_COMPONENTS] = {}
 
@@ -328,17 +251,6 @@ async def async_test_home_assistant(event_loop, load_registries=True):
 
     hass.state = CoreState.running
 
-    # Mock async_start
-    orig_start = hass.async_start
-
-    async def mock_async_start():
-        """Start the mocking."""
-        # We only mock time during tests and we want to track tasks
-        with patch.object(hass, "async_stop_track_tasks"):
-            await orig_start()
-
-    hass.async_start = mock_async_start
-
     @callback
     def clear_instance(event):
         """Clear global instance."""