Add missing hass type hint in history and recorder tests (#124204)
parent
5470d14a11
commit
c76d68503a
|
@ -1,6 +1,6 @@
|
||||||
"""The tests the History component."""
|
"""The tests the History component."""
|
||||||
|
|
||||||
from datetime import timedelta
|
from datetime import datetime, timedelta
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
import json
|
import json
|
||||||
from unittest.mock import sentinel
|
from unittest.mock import sentinel
|
||||||
|
@ -13,7 +13,7 @@ from homeassistant.components.recorder import Recorder
|
||||||
from homeassistant.components.recorder.history import get_significant_states
|
from homeassistant.components.recorder.history import get_significant_states
|
||||||
from homeassistant.components.recorder.models import process_timestamp
|
from homeassistant.components.recorder.models import process_timestamp
|
||||||
from homeassistant.const import EVENT_HOMEASSISTANT_FINAL_WRITE
|
from homeassistant.const import EVENT_HOMEASSISTANT_FINAL_WRITE
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant, State
|
||||||
from homeassistant.helpers.json import JSONEncoder
|
from homeassistant.helpers.json import JSONEncoder
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
@ -291,13 +291,9 @@ async def test_get_significant_states_only(hass: HomeAssistant, hass_history) ->
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def check_significant_states(hass, zero, four, states, config):
|
async def async_record_states(
|
||||||
"""Check if significant states are retrieved."""
|
hass: HomeAssistant,
|
||||||
hist = get_significant_states(hass, zero, four)
|
) -> tuple[datetime, datetime, dict[str, list[State | None]]]:
|
||||||
assert_dict_of_states_equal_without_context_and_last_changed(states, hist)
|
|
||||||
|
|
||||||
|
|
||||||
async def async_record_states(hass):
|
|
||||||
"""Record some test states.
|
"""Record some test states.
|
||||||
|
|
||||||
We inject a bunch of state updates from media player, zone and
|
We inject a bunch of state updates from media player, zone and
|
||||||
|
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import timedelta
|
from datetime import datetime, timedelta
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
import json
|
import json
|
||||||
from unittest.mock import patch, sentinel
|
from unittest.mock import patch, sentinel
|
||||||
|
@ -14,7 +14,7 @@ from homeassistant.components import recorder
|
||||||
from homeassistant.components.recorder import Recorder
|
from homeassistant.components.recorder import Recorder
|
||||||
from homeassistant.components.recorder.history import get_significant_states
|
from homeassistant.components.recorder.history import get_significant_states
|
||||||
from homeassistant.components.recorder.models import process_timestamp
|
from homeassistant.components.recorder.models import process_timestamp
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant, State
|
||||||
from homeassistant.helpers.json import JSONEncoder
|
from homeassistant.helpers.json import JSONEncoder
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
@ -309,13 +309,9 @@ async def test_get_significant_states_only(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def check_significant_states(hass, zero, four, states, config):
|
async def async_record_states(
|
||||||
"""Check if significant states are retrieved."""
|
hass: HomeAssistant,
|
||||||
hist = get_significant_states(hass, zero, four)
|
) -> tuple[datetime, datetime, dict[str, list[State | None]]]:
|
||||||
assert_dict_of_states_equal_without_context_and_last_changed(states, hist)
|
|
||||||
|
|
||||||
|
|
||||||
async def async_record_states(hass):
|
|
||||||
"""Record some test states.
|
"""Record some test states.
|
||||||
|
|
||||||
We inject a bunch of state updates from media player, zone and
|
We inject a bunch of state updates from media player, zone and
|
||||||
|
|
|
@ -265,12 +265,16 @@ def assert_dict_of_states_equal_without_context_and_last_changed(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def async_record_states(hass: HomeAssistant):
|
async def async_record_states(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
) -> tuple[datetime, datetime, dict[str, list[State | None]]]:
|
||||||
"""Record some test states."""
|
"""Record some test states."""
|
||||||
return await hass.async_add_executor_job(record_states, hass)
|
return await hass.async_add_executor_job(record_states, hass)
|
||||||
|
|
||||||
|
|
||||||
def record_states(hass):
|
def record_states(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
) -> tuple[datetime, datetime, dict[str, list[State | None]]]:
|
||||||
"""Record some test states.
|
"""Record some test states.
|
||||||
|
|
||||||
We inject a bunch of state updates temperature sensors.
|
We inject a bunch of state updates temperature sensors.
|
||||||
|
|
|
@ -72,12 +72,13 @@ from homeassistant.const import (
|
||||||
STATE_LOCKED,
|
STATE_LOCKED,
|
||||||
STATE_UNLOCKED,
|
STATE_UNLOCKED,
|
||||||
)
|
)
|
||||||
from homeassistant.core import Context, CoreState, Event, HomeAssistant, callback
|
from homeassistant.core import Context, CoreState, Event, HomeAssistant, State, callback
|
||||||
from homeassistant.helpers import (
|
from homeassistant.helpers import (
|
||||||
entity_registry as er,
|
entity_registry as er,
|
||||||
issue_registry as ir,
|
issue_registry as ir,
|
||||||
recorder as recorder_helper,
|
recorder as recorder_helper,
|
||||||
)
|
)
|
||||||
|
from homeassistant.helpers.typing import ConfigType
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
from homeassistant.util import dt as dt_util
|
from homeassistant.util import dt as dt_util
|
||||||
from homeassistant.util.json import json_loads
|
from homeassistant.util.json import json_loads
|
||||||
|
@ -123,7 +124,7 @@ def small_cache_size() -> Generator[None]:
|
||||||
yield
|
yield
|
||||||
|
|
||||||
|
|
||||||
def _default_recorder(hass):
|
def _default_recorder(hass: HomeAssistant) -> Recorder:
|
||||||
"""Return a recorder with reasonable defaults."""
|
"""Return a recorder with reasonable defaults."""
|
||||||
return Recorder(
|
return Recorder(
|
||||||
hass,
|
hass,
|
||||||
|
@ -581,7 +582,7 @@ async def test_saving_state_with_commit_interval_zero(
|
||||||
assert db_states[0].event_id is None
|
assert db_states[0].event_id is None
|
||||||
|
|
||||||
|
|
||||||
async def _add_entities(hass, entity_ids):
|
async def _add_entities(hass: HomeAssistant, entity_ids: list[str]) -> list[State]:
|
||||||
"""Add entities."""
|
"""Add entities."""
|
||||||
attributes = {"test_attr": 5, "test_attr_10": "nice"}
|
attributes = {"test_attr": 5, "test_attr_10": "nice"}
|
||||||
for idx, entity_id in enumerate(entity_ids):
|
for idx, entity_id in enumerate(entity_ids):
|
||||||
|
@ -605,7 +606,7 @@ async def _add_entities(hass, entity_ids):
|
||||||
return states
|
return states
|
||||||
|
|
||||||
|
|
||||||
def _state_with_context(hass, entity_id):
|
def _state_with_context(hass: HomeAssistant, entity_id: str) -> State | None:
|
||||||
# We don't restore context unless we need it by joining the
|
# We don't restore context unless we need it by joining the
|
||||||
# events table on the event_id for state_changed events
|
# events table on the event_id for state_changed events
|
||||||
return hass.states.get(entity_id)
|
return hass.states.get(entity_id)
|
||||||
|
@ -1004,7 +1005,7 @@ async def test_defaults_set(hass: HomeAssistant) -> None:
|
||||||
"""Test the config defaults are set."""
|
"""Test the config defaults are set."""
|
||||||
recorder_config = None
|
recorder_config = None
|
||||||
|
|
||||||
async def mock_setup(hass, config):
|
async def mock_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||||
"""Mock setup."""
|
"""Mock setup."""
|
||||||
nonlocal recorder_config
|
nonlocal recorder_config
|
||||||
recorder_config = config["recorder"]
|
recorder_config = config["recorder"]
|
||||||
|
|
|
@ -29,7 +29,7 @@ from homeassistant.components.recorder.db_schema import (
|
||||||
States,
|
States,
|
||||||
)
|
)
|
||||||
from homeassistant.components.recorder.util import session_scope
|
from homeassistant.components.recorder.util import session_scope
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant, State
|
||||||
from homeassistant.helpers import recorder as recorder_helper
|
from homeassistant.helpers import recorder as recorder_helper
|
||||||
import homeassistant.util.dt as dt_util
|
import homeassistant.util.dt as dt_util
|
||||||
|
|
||||||
|
@ -47,7 +47,7 @@ async def mock_recorder_before_hass(
|
||||||
"""Set up recorder."""
|
"""Set up recorder."""
|
||||||
|
|
||||||
|
|
||||||
def _get_native_states(hass, entity_id):
|
def _get_native_states(hass: HomeAssistant, entity_id: str) -> list[State]:
|
||||||
with session_scope(hass=hass, read_only=True) as session:
|
with session_scope(hass=hass, read_only=True) as session:
|
||||||
instance = recorder.get_instance(hass)
|
instance = recorder.get_instance(hass)
|
||||||
metadata_id = instance.states_meta_manager.get(entity_id, session, True)
|
metadata_id = instance.states_meta_manager.get(entity_id, session, True)
|
||||||
|
|
|
@ -1330,7 +1330,9 @@ async def test_purge_filtered_events_state_changed(
|
||||||
async def test_purge_entities(hass: HomeAssistant, recorder_mock: Recorder) -> None:
|
async def test_purge_entities(hass: HomeAssistant, recorder_mock: Recorder) -> None:
|
||||||
"""Test purging of specific entities."""
|
"""Test purging of specific entities."""
|
||||||
|
|
||||||
async def _purge_entities(hass, entity_ids, domains, entity_globs):
|
async def _purge_entities(
|
||||||
|
hass: HomeAssistant, entity_ids: str, domains: str, entity_globs: str
|
||||||
|
) -> None:
|
||||||
service_data = {
|
service_data = {
|
||||||
"entity_id": entity_ids,
|
"entity_id": entity_ids,
|
||||||
"domains": domains,
|
"domains": domains,
|
||||||
|
|
Loading…
Reference in New Issue