Add config flow to history_stats helper (#121381)

This commit is contained in:
G Johansson 2024-07-07 21:06:37 +02:00 committed by GitHub
parent 72458d143d
commit 833ac4db49
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 605 additions and 25 deletions

View file

@ -1,6 +1,60 @@
"""The history_stats component."""
from homeassistant.const import Platform
from __future__ import annotations
DOMAIN = "history_stats"
PLATFORMS = [Platform.SENSOR]
from datetime import timedelta
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ENTITY_ID, CONF_STATE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.template import Template
from .const import CONF_DURATION, CONF_END, CONF_START, PLATFORMS
from .coordinator import HistoryStatsUpdateCoordinator
from .data import HistoryStats
type HistoryStatsConfigEntry = ConfigEntry[HistoryStatsUpdateCoordinator]
async def async_setup_entry(
hass: HomeAssistant, entry: HistoryStatsConfigEntry
) -> bool:
"""Set up History stats from a config entry."""
entity_id: str = entry.options[CONF_ENTITY_ID]
entity_states: list[str] = entry.options[CONF_STATE]
start: str | None = entry.options.get(CONF_START)
end: str | None = entry.options.get(CONF_END)
duration: timedelta | None = None
if duration_dict := entry.options.get(CONF_DURATION):
duration = timedelta(**duration_dict)
history_stats = HistoryStats(
hass,
entity_id,
entity_states,
Template(start, hass) if start else None,
Template(end, hass) if end else None,
duration,
)
coordinator = HistoryStatsUpdateCoordinator(hass, history_stats, entry.title)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True
async def async_unload_entry(
hass: HomeAssistant, entry: HistoryStatsConfigEntry
) -> bool:
"""Unload History stats config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

View file

@ -0,0 +1,103 @@
"""The history_stats component config flow."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, cast
import voluptuous as vol
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME, CONF_STATE, CONF_TYPE
from homeassistant.helpers.schema_config_entry_flow import (
SchemaCommonFlowHandler,
SchemaConfigFlowHandler,
SchemaFlowError,
SchemaFlowFormStep,
)
from homeassistant.helpers.selector import (
DurationSelector,
DurationSelectorConfig,
EntitySelector,
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TemplateSelector,
TextSelector,
TextSelectorConfig,
)
from .const import (
CONF_DURATION,
CONF_END,
CONF_PERIOD_KEYS,
CONF_START,
CONF_TYPE_KEYS,
CONF_TYPE_TIME,
DEFAULT_NAME,
DOMAIN,
)
async def validate_options(
handler: SchemaCommonFlowHandler, user_input: dict[str, Any]
) -> dict[str, Any]:
"""Validate options selected."""
if sum(param in user_input for param in CONF_PERIOD_KEYS) != 2:
raise SchemaFlowError("only_two_keys_allowed")
handler.parent_handler._async_abort_entries_match({**handler.options, **user_input}) # noqa: SLF001
return user_input
DATA_SCHEMA_SETUP = vol.Schema(
{
vol.Required(CONF_NAME, default=DEFAULT_NAME): TextSelector(),
vol.Required(CONF_ENTITY_ID): EntitySelector(),
vol.Required(CONF_STATE): TextSelector(TextSelectorConfig(multiple=True)),
vol.Required(CONF_TYPE, default=CONF_TYPE_TIME): SelectSelector(
SelectSelectorConfig(
options=CONF_TYPE_KEYS,
mode=SelectSelectorMode.DROPDOWN,
translation_key=CONF_TYPE,
)
),
}
)
DATA_SCHEMA_OPTIONS = vol.Schema(
{
vol.Optional(CONF_START): TemplateSelector(),
vol.Optional(CONF_END): TemplateSelector(),
vol.Optional(CONF_DURATION): DurationSelector(
DurationSelectorConfig(enable_day=True, allow_negative=False)
),
}
)
CONFIG_FLOW = {
"user": SchemaFlowFormStep(
schema=DATA_SCHEMA_SETUP,
next_step="options",
),
"options": SchemaFlowFormStep(
schema=DATA_SCHEMA_OPTIONS,
validate_user_input=validate_options,
),
}
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(
DATA_SCHEMA_OPTIONS,
validate_user_input=validate_options,
),
}
class StatisticsConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
"""Handle a config flow for History stats."""
config_flow = CONFIG_FLOW
options_flow = OPTIONS_FLOW
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title."""
return cast(str, options[CONF_NAME])

View file

@ -0,0 +1,18 @@
"""The history_stats component constants."""
from homeassistant.const import Platform
DOMAIN = "history_stats"
PLATFORMS = [Platform.SENSOR]
CONF_START = "start"
CONF_END = "end"
CONF_DURATION = "duration"
CONF_PERIOD_KEYS = [CONF_START, CONF_END, CONF_DURATION]
CONF_TYPE_TIME = "time"
CONF_TYPE_RATIO = "ratio"
CONF_TYPE_COUNT = "count"
CONF_TYPE_KEYS = [CONF_TYPE_TIME, CONF_TYPE_RATIO, CONF_TYPE_COUNT]
DEFAULT_NAME = "unnamed statistics"

View file

@ -27,7 +27,7 @@ UPDATE_INTERVAL = timedelta(minutes=1)
class HistoryStatsUpdateCoordinator(DataUpdateCoordinator[HistoryStatsState]):
"""DataUpdateCoordinator to gather data for a specific TPLink device."""
"""DataUpdateCoordinator for history stats."""
def __init__(
self,

View file

@ -2,8 +2,10 @@
"domain": "history_stats",
"name": "History Stats",
"codeowners": [],
"config_flow": true,
"dependencies": ["recorder"],
"documentation": "https://www.home-assistant.io/integrations/history_stats",
"integration_type": "helper",
"iot_class": "local_polling",
"quality_scale": "internal"
}

View file

@ -32,22 +32,24 @@ from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DOMAIN, PLATFORMS
from . import HistoryStatsConfigEntry
from .const import (
CONF_DURATION,
CONF_END,
CONF_PERIOD_KEYS,
CONF_START,
CONF_TYPE_COUNT,
CONF_TYPE_KEYS,
CONF_TYPE_RATIO,
CONF_TYPE_TIME,
DEFAULT_NAME,
DOMAIN,
PLATFORMS,
)
from .coordinator import HistoryStatsUpdateCoordinator
from .data import HistoryStats
from .helpers import pretty_ratio
CONF_START = "start"
CONF_END = "end"
CONF_DURATION = "duration"
CONF_PERIOD_KEYS = [CONF_START, CONF_END, CONF_DURATION]
CONF_TYPE_TIME = "time"
CONF_TYPE_RATIO = "ratio"
CONF_TYPE_COUNT = "count"
CONF_TYPE_KEYS = [CONF_TYPE_TIME, CONF_TYPE_RATIO, CONF_TYPE_COUNT]
DEFAULT_NAME = "unnamed statistics"
UNITS: dict[str, str] = {
CONF_TYPE_TIME: UnitOfTime.HOURS,
CONF_TYPE_RATIO: PERCENTAGE,
@ -82,7 +84,6 @@ PLATFORM_SCHEMA = vol.All(
)
# noinspection PyUnusedLocal
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
@ -113,6 +114,20 @@ async def async_setup_platform(
async_add_entities([HistoryStatsSensor(coordinator, sensor_type, name, unique_id)])
async def async_setup_entry(
hass: HomeAssistant,
entry: HistoryStatsConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the History stats sensor entry."""
sensor_type: str = entry.options[CONF_TYPE]
coordinator = entry.runtime_data
async_add_entities(
[HistoryStatsSensor(coordinator, sensor_type, entry.title, entry.entry_id)]
)
class HistoryStatsSensorBase(
CoordinatorEntity[HistoryStatsUpdateCoordinator], SensorEntity
):

View file

@ -1,4 +1,74 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"only_two_keys_allowed": "The sensor configuration must provide two out of 'start', 'end', 'duration'"
},
"step": {
"user": {
"description": "Add a history stats sensor",
"data": {
"name": "[%key:common::config_flow::data::name%]",
"entity_id": "Entity",
"state": "State",
"type": "Type"
},
"data_description": {
"name": "Name for the created entity.",
"entity_id": "Entity to get statistics from.",
"state": "The states you want to track from the entity.",
"type": "The type of sensor, one of 'time', 'ratio' or 'count'"
}
},
"options": {
"description": "Read the documention for further details on how to configure the history stats sensor using these options.",
"data": {
"start": "Start",
"end": "End",
"duration": "Duration"
},
"data_description": {
"start": "When to start the measure (timestamp or datetime). Can be a template.",
"end": "When to stop the measure (timestamp or datetime). Can be a template",
"duration": "Duration of the measure."
}
}
}
},
"options": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"only_two_keys_allowed": "[%key:component::history_stats::config::error::only_two_keys_allowed%]"
},
"step": {
"init": {
"description": "[%key:component::history_stats::config::step::options::description%]",
"data": {
"start": "[%key:component::history_stats::config::step::options::data::start%]",
"end": "[%key:component::history_stats::config::step::options::data::end%]",
"duration": "[%key:component::history_stats::config::step::options::data::duration%]"
},
"data_description": {
"start": "[%key:component::history_stats::config::step::options::data_description::start%]",
"end": "[%key:component::history_stats::config::step::options::data_description::end%]",
"duration": "[%key:component::history_stats::config::step::options::data_description::duration%]"
}
}
}
},
"selector": {
"type": {
"options": {
"time": "Time",
"ratio": "Ratio",
"count": "Count"
}
}
},
"services": {
"reload": {
"name": "[%key:common::action::reload%]",

View file

@ -9,6 +9,7 @@ FLOWS = {
"generic_hygrostat",
"generic_thermostat",
"group",
"history_stats",
"integration",
"min_max",
"random",

View file

@ -2524,12 +2524,6 @@
"config_flow": true,
"iot_class": "local_polling"
},
"history_stats": {
"name": "History Stats",
"integration_type": "hub",
"config_flow": false,
"iot_class": "local_polling"
},
"hitron_coda": {
"name": "Rogers Hitron CODA",
"integration_type": "hub",
@ -7174,6 +7168,12 @@
"config_flow": true,
"iot_class": "calculated"
},
"history_stats": {
"name": "History Stats",
"integration_type": "helper",
"config_flow": true,
"iot_class": "local_polling"
},
"input_boolean": {
"integration_type": "helper",
"config_flow": false

View file

@ -0,0 +1,93 @@
"""Fixtures for the History stats integration."""
from __future__ import annotations
from collections.abc import Generator
from datetime import timedelta
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.history_stats.const import (
CONF_END,
CONF_START,
DEFAULT_NAME,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME, CONF_STATE, CONF_TYPE
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Automatically patch history stats setup."""
with patch(
"homeassistant.components.history_stats.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture(name="get_config")
async def get_config_to_integration_load() -> dict[str, Any]:
"""Return configuration.
To override the config, tests can be marked with:
@pytest.mark.parametrize("get_config", [{...}])
"""
return {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
}
@pytest.fixture(name="loaded_entry")
async def load_integration(
hass: HomeAssistant, get_config: dict[str, Any]
) -> MockConfigEntry:
"""Set up the History stats integration in Home Assistant."""
start_time = dt_util.utcnow() - timedelta(minutes=60)
t0 = start_time + timedelta(minutes=20)
t1 = t0 + timedelta(minutes=10)
t2 = t1 + timedelta(minutes=10)
def _fake_states(*args, **kwargs):
return {
"binary_sensor.test_monitored": [
State("binary_sensor.test_monitored", "off", last_changed=start_time),
State("binary_sensor.test_monitored", "on", last_changed=t0),
State("binary_sensor.test_monitored", "off", last_changed=t1),
State("binary_sensor.test_monitored", "on", last_changed=t2),
]
}
config_entry = MockConfigEntry(
domain=DOMAIN,
title=DEFAULT_NAME,
source=SOURCE_USER,
options=get_config,
entry_id="1",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.recorder.history.state_changes_during_period",
_fake_states,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
await async_update_entity(hass, "sensor.test")
await hass.async_block_till_done()
return config_entry

View file

@ -0,0 +1,195 @@
"""Test the History stats config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock
from homeassistant import config_entries
from homeassistant.components.history_stats.const import (
CONF_DURATION,
CONF_END,
CONF_START,
DEFAULT_NAME,
DOMAIN,
)
from homeassistant.components.recorder import Recorder
from homeassistant.const import CONF_ENTITY_ID, CONF_NAME, CONF_STATE, CONF_TYPE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_form(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_options_flow(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test options flow."""
result = await hass.config_entries.options.async_init(loaded_entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_END: "{{ utcnow() }}",
CONF_DURATION: {"hours": 8, "minutes": 0, "seconds": 0, "days": 20},
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"] == {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
CONF_END: "{{ utcnow() }}",
CONF_DURATION: {"hours": 8, "minutes": 0, "seconds": 0, "days": 20},
}
await hass.async_block_till_done()
# Check the entity was updated, no new entity was created
assert len(hass.states.async_all()) == 1
state = hass.states.get("sensor.unnamed_statistics")
assert state is not None
async def test_validation_options(
recorder_mock: Recorder, hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test validation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
CONF_DURATION: {"hours": 8, "minutes": 0, "seconds": 0, "days": 20},
},
)
await hass.async_block_till_done()
assert result["step_id"] == "options"
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "only_two_keys_allowed"}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["version"] == 1
assert result["options"] == {
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_entry_already_exist(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test abort when entry already exist."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_NAME: DEFAULT_NAME,
CONF_ENTITY_ID: "binary_sensor.test_monitored",
CONF_STATE: ["on"],
CONF_TYPE: "count",
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_START: "{{ as_timestamp(utcnow()) - 3600 }}",
CONF_END: "{{ utcnow() }}",
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View file

@ -0,0 +1,20 @@
"""Test History stats component setup process."""
from __future__ import annotations
from homeassistant.components.recorder import Recorder
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_unload_entry(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test unload an entry."""
assert loaded_entry.state is ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(loaded_entry.entry_id)
await hass.async_block_till_done()
assert loaded_entry.state is ConfigEntryState.NOT_LOADED

View file

@ -8,7 +8,7 @@ import pytest
import voluptuous as vol
from homeassistant import config as hass_config
from homeassistant.components.history_stats import DOMAIN
from homeassistant.components.history_stats.const import DOMAIN
from homeassistant.components.history_stats.sensor import (
PLATFORM_SCHEMA as SENSOR_SCHEMA,
)
@ -21,7 +21,7 @@ from homeassistant.helpers.entity_component import async_update_entity
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from tests.common import async_fire_time_changed, get_fixture_path
from tests.common import MockConfigEntry, async_fire_time_changed, get_fixture_path
from tests.components.recorder.common import async_wait_recording_done
from tests.typing import RecorderInstanceGenerator
@ -48,6 +48,15 @@ async def test_setup(recorder_mock: Recorder, hass: HomeAssistant) -> None:
assert state.state == "0.0"
async def test_setup_config_entry(
recorder_mock: Recorder, hass: HomeAssistant, loaded_entry: MockConfigEntry
) -> None:
"""Test the history statistics sensor setup from a config entry."""
state = hass.states.get("sensor.unnamed_statistics")
assert state.state == "2"
async def test_setup_multiple_states(
recorder_mock: Recorder, hass: HomeAssistant
) -> None: