Remove test helper get_test_home_assistant (#124177)

This commit is contained in:
Erik Montnemery 2024-08-19 10:23:25 +02:00 committed by GitHub
parent 013b91394e
commit 511ec4ba8a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -13,12 +13,7 @@ from collections.abc import (
Mapping, Mapping,
Sequence, Sequence,
) )
from contextlib import ( from contextlib import asynccontextmanager, contextmanager, suppress
AbstractAsyncContextManager,
asynccontextmanager,
contextmanager,
suppress,
)
from datetime import UTC, datetime, timedelta from datetime import UTC, datetime, timedelta
from enum import Enum from enum import Enum
import functools as ft import functools as ft
@ -28,7 +23,6 @@ import json
import logging import logging
import os import os
import pathlib import pathlib
import threading
import time import time
from types import FrameType, ModuleType from types import FrameType, ModuleType
from typing import Any, Literal, NoReturn from typing import Any, Literal, NoReturn
@ -179,56 +173,6 @@ def get_test_config_dir(*add_path):
return os.path.join(os.path.dirname(__file__), "testing_config", *add_path) return os.path.join(os.path.dirname(__file__), "testing_config", *add_path)
@contextmanager
def get_test_home_assistant() -> Generator[HomeAssistant]:
"""Return a Home Assistant object pointing at test config directory."""
hass_created_event = threading.Event()
loop_stop_event = threading.Event()
context_manager: AbstractAsyncContextManager = None
hass: HomeAssistant = None
loop: asyncio.AbstractEventLoop = None
orig_stop: Callable = None
def run_loop() -> None:
"""Create and run event loop."""
nonlocal context_manager, hass, loop, orig_stop
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
context_manager = async_test_home_assistant(loop)
hass = loop.run_until_complete(context_manager.__aenter__())
orig_stop = hass.stop
hass._stopped = Mock(set=loop.stop)
hass.start = start_hass
hass.stop = stop_hass
hass_created_event.set()
loop.run_forever()
loop_stop_event.set()
def start_hass(*mocks: Any) -> None:
"""Start hass."""
asyncio.run_coroutine_threadsafe(hass.async_start(), loop).result()
def stop_hass() -> None:
"""Stop hass."""
orig_stop()
loop_stop_event.wait()
threading.Thread(name="LoopThread", target=run_loop, daemon=False).start()
hass_created_event.wait()
try:
yield hass
finally:
loop.run_until_complete(context_manager.__aexit__(None, None, None))
loop.close()
class StoreWithoutWriteLoad[_T: (Mapping[str, Any] | Sequence[Any])](storage.Store[_T]): class StoreWithoutWriteLoad[_T: (Mapping[str, Any] | Sequence[Any])](storage.Store[_T]):
"""Fake store that does not write or load. Used for testing.""" """Fake store that does not write or load. Used for testing."""