Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Remove hass_recorder test fixture (#120295)

Browse Source
This commit is contained in:
Erik Montnemery
2024-06-24 11:14:08 +02:00
committed by GitHub
parent be6dfc7a70
commit e32a27a8ff
3 changed files with 0 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
tests/conftest.py
View File

@@ -106,8 +106,6 @@ from .common import ( # noqa: E402, isort:skip
MockUser,
async_fire_mqtt_message,
async_test_home_assistant,
get_test_home_assistant,
init_recorder_component,
mock_storage,
patch_yaml_files,
extract_stack_to_frame,
@@ -1350,120 +1348,6 @@ def recorder_db_url(
sqlalchemy_utils.drop_database(db_url)
@pytest.fixture
def hass_recorder(
recorder_db_url: str,
enable_nightly_purge: bool,
enable_statistics: bool,
enable_schema_validation: bool,
enable_migrate_context_ids: bool,
enable_migrate_event_type_ids: bool,
enable_migrate_entity_ids: bool,
hass_storage: dict[str, Any],
) -> Generator[Callable[..., HomeAssistant]]:
"""Home Assistant fixture with in-memory recorder."""
# pylint: disable-next=import-outside-toplevel
from homeassistant.components import recorder
# pylint: disable-next=import-outside-toplevel
from homeassistant.components.recorder import migration
with get_test_home_assistant() as hass:
nightly = (
recorder.Recorder.async_nightly_tasks if enable_nightly_purge else None
)
stats = (
recorder.Recorder.async_periodic_statistics if enable_statistics else None
)
compile_missing = (
recorder.Recorder._schedule_compile_missing_statistics
if enable_statistics
else None
)
schema_validate = (
migration._find_schema_errors
if enable_schema_validation
else itertools.repeat(set())
)
migrate_states_context_ids = (
recorder.Recorder._migrate_states_context_ids
if enable_migrate_context_ids
else None
)
migrate_events_context_ids = (
recorder.Recorder._migrate_events_context_ids
if enable_migrate_context_ids
else None
)
migrate_event_type_ids = (
recorder.Recorder._migrate_event_type_ids
if enable_migrate_event_type_ids
else None
)
migrate_entity_ids = (
recorder.Recorder._migrate_entity_ids if enable_migrate_entity_ids else None
)
with (
patch(
"homeassistant.components.recorder.Recorder.async_nightly_tasks",
side_effect=nightly,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder.async_periodic_statistics",
side_effect=stats,
autospec=True,
),
patch(
"homeassistant.components.recorder.migration._find_schema_errors",
side_effect=schema_validate,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder._migrate_events_context_ids",
side_effect=migrate_events_context_ids,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder._migrate_states_context_ids",
side_effect=migrate_states_context_ids,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder._migrate_event_type_ids",
side_effect=migrate_event_type_ids,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder._migrate_entity_ids",
side_effect=migrate_entity_ids,
autospec=True,
),
patch(
"homeassistant.components.recorder.Recorder._schedule_compile_missing_statistics",
side_effect=compile_missing,
autospec=True,
),
):
def setup_recorder(
*, config: dict[str, Any] | None = None, timezone: str | None = None
) -> HomeAssistant:
"""Set up with params."""
if timezone is not None:
asyncio.run_coroutine_threadsafe(
hass.config.async_set_time_zone(timezone), hass.loop
).result()
init_recorder_component(hass, config, recorder_db_url)
hass.start()
hass.block_till_done()
hass.data[recorder.DATA_INSTANCE].block_till_done()
return hass
yield setup_recorder
hass.stop()
async def _async_init_recorder_component(
hass: HomeAssistant,
add_config: dict[str, Any] | None = None,