Add config flow to random (#100858)

Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Joost Lekkerkerker 2023-10-25 13:01:27 +02:00 committed by GitHub
parent 6fae50cb75
commit 0658c7b307
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 524 additions and 28 deletions

View file

@ -1 +1,24 @@
"""The random component."""
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry."""
await hass.config_entries.async_forward_entry_setups(
entry, (entry.options["entity_type"],)
)
entry.async_on_unload(entry.add_update_listener(config_entry_update_listener))
return True
async def config_entry_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Update listener, called when the config entry options are changed."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(
entry, (entry.options["entity_type"],)
)

View file

@ -1,7 +1,9 @@
"""Support for showing random states."""
from __future__ import annotations
from collections.abc import Mapping
from random import getrandbits
from typing import Any
import voluptuous as vol
@ -10,6 +12,7 @@ from homeassistant.components.binary_sensor import (
PLATFORM_SCHEMA,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_CLASS, CONF_NAME
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
@ -33,20 +36,32 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Random binary sensor."""
name = config.get(CONF_NAME)
device_class = config.get(CONF_DEVICE_CLASS)
async_add_entities([RandomSensor(name, device_class)], True)
async_add_entities([RandomBinarySensor(config)], True)
class RandomSensor(BinarySensorEntity):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Initialize config entry."""
async_add_entities(
[RandomBinarySensor(config_entry.options, config_entry.entry_id)], True
)
class RandomBinarySensor(BinarySensorEntity):
"""Representation of a Random binary sensor."""
def __init__(self, name, device_class):
_state: bool | None = None
def __init__(self, config: Mapping[str, Any], entry_id: str | None = None) -> None:
"""Initialize the Random binary sensor."""
self._name = name
self._device_class = device_class
self._state = None
self._name = config.get(CONF_NAME)
self._device_class = config.get(CONF_DEVICE_CLASS)
if entry_id:
self._attr_unique_id = entry_id
@property
def name(self):

View file

@ -0,0 +1,186 @@
"""Config flow for Random helper."""
from collections.abc import Callable, Coroutine, Mapping
from enum import StrEnum
from typing import Any, cast
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.components.sensor import DEVICE_CLASS_UNITS, SensorDeviceClass
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_MAXIMUM,
CONF_MINIMUM,
CONF_NAME,
CONF_UNIT_OF_MEASUREMENT,
Platform,
)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.schema_config_entry_flow import (
SchemaCommonFlowHandler,
SchemaConfigFlowHandler,
SchemaFlowFormStep,
SchemaFlowMenuStep,
)
from homeassistant.helpers.selector import (
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
)
from .const import DOMAIN
from .sensor import DEFAULT_MAX, DEFAULT_MIN
class _FlowType(StrEnum):
CONFIG = "config"
OPTION = "option"
def _generate_schema(domain: str, flow_type: _FlowType) -> vol.Schema:
"""Generate schema."""
schema: dict[vol.Marker, Any] = {}
if flow_type == _FlowType.CONFIG:
schema[vol.Required(CONF_NAME)] = TextSelector()
if domain == Platform.BINARY_SENSOR:
schema[vol.Optional(CONF_DEVICE_CLASS)] = SelectSelector(
SelectSelectorConfig(
options=[cls.value for cls in BinarySensorDeviceClass],
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key="binary_sensor_device_class",
),
)
if domain == Platform.SENSOR:
schema.update(
{
vol.Optional(CONF_MINIMUM, default=DEFAULT_MIN): cv.positive_int,
vol.Optional(CONF_MAXIMUM, default=DEFAULT_MAX): cv.positive_int,
vol.Optional(CONF_DEVICE_CLASS): SelectSelector(
SelectSelectorConfig(
options=[
cls.value
for cls in SensorDeviceClass
if cls != SensorDeviceClass.ENUM
],
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key="sensor_device_class",
),
),
vol.Optional(CONF_UNIT_OF_MEASUREMENT): SelectSelector(
SelectSelectorConfig(
options=[
str(unit)
for units in DEVICE_CLASS_UNITS.values()
for unit in units
if unit is not None
],
sort=True,
mode=SelectSelectorMode.DROPDOWN,
translation_key="sensor_unit_of_measurement",
custom_value=True,
),
),
}
)
return vol.Schema(schema)
async def choose_options_step(options: dict[str, Any]) -> str:
"""Return next step_id for options flow according to template_type."""
return cast(str, options["entity_type"])
def _validate_unit(options: dict[str, Any]) -> None:
"""Validate unit of measurement."""
if (
(device_class := options.get(CONF_DEVICE_CLASS))
and (units := DEVICE_CLASS_UNITS.get(device_class))
and (unit := options.get(CONF_UNIT_OF_MEASUREMENT)) not in units
):
sorted_units = sorted(
[f"'{str(unit)}'" if unit else "no unit of measurement" for unit in units],
key=str.casefold,
)
if len(sorted_units) == 1:
units_string = sorted_units[0]
else:
units_string = f"one of {', '.join(sorted_units)}"
raise vol.Invalid(
f"'{unit}' is not a valid unit for device class '{device_class}'; "
f"expected {units_string}"
)
def validate_user_input(
template_type: str,
) -> Callable[
[SchemaCommonFlowHandler, dict[str, Any]],
Coroutine[Any, Any, dict[str, Any]],
]:
"""Do post validation of user input.
For sensors: Validate unit of measurement.
"""
async def _validate_user_input(
_: SchemaCommonFlowHandler,
user_input: dict[str, Any],
) -> dict[str, Any]:
"""Add template type to user input."""
if template_type == Platform.SENSOR:
_validate_unit(user_input)
return {"entity_type": template_type} | user_input
return _validate_user_input
RANDOM_TYPES = [
Platform.BINARY_SENSOR.value,
Platform.SENSOR.value,
]
CONFIG_FLOW = {
"user": SchemaFlowMenuStep(RANDOM_TYPES),
Platform.BINARY_SENSOR: SchemaFlowFormStep(
_generate_schema(Platform.BINARY_SENSOR, _FlowType.CONFIG),
validate_user_input=validate_user_input(Platform.BINARY_SENSOR),
),
Platform.SENSOR: SchemaFlowFormStep(
_generate_schema(Platform.SENSOR, _FlowType.CONFIG),
validate_user_input=validate_user_input(Platform.SENSOR),
),
}
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(next_step=choose_options_step),
Platform.BINARY_SENSOR: SchemaFlowFormStep(
_generate_schema(Platform.BINARY_SENSOR, _FlowType.OPTION),
validate_user_input=validate_user_input(Platform.BINARY_SENSOR),
),
Platform.SENSOR: SchemaFlowFormStep(
_generate_schema(Platform.SENSOR, _FlowType.OPTION),
validate_user_input=validate_user_input(Platform.SENSOR),
),
}
class RandomConfigFlowHandler(SchemaConfigFlowHandler, domain=DOMAIN):
"""Handle config flow for random helper."""
config_flow = CONFIG_FLOW
options_flow = OPTIONS_FLOW
@callback
def async_config_entry_title(self, options: Mapping[str, Any]) -> str:
"""Return config entry title."""
return cast(str, options["name"])

View file

@ -0,0 +1,5 @@
"""Constants for random helper."""
DOMAIN = "random"
DEFAULT_MIN = 0
DEFAULT_MAX = 20

View file

@ -2,7 +2,9 @@
"domain": "random",
"name": "Random",
"codeowners": ["@fabaff"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/random",
"iot_class": "local_polling",
"integration_type": "helper",
"iot_class": "calculated",
"quality_scale": "internal"
}

View file

@ -1,12 +1,16 @@
"""Support for showing random numbers."""
from __future__ import annotations
from collections.abc import Mapping
from random import randrange
from typing import Any
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_MAXIMUM,
CONF_MINIMUM,
CONF_NAME,
@ -17,12 +21,12 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import DEFAULT_MAX, DEFAULT_MIN
ATTR_MAXIMUM = "maximum"
ATTR_MINIMUM = "minimum"
DEFAULT_NAME = "Random Sensor"
DEFAULT_MIN = 0
DEFAULT_MAX = 20
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@ -42,26 +46,37 @@ async def async_setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Random number sensor."""
name = config.get(CONF_NAME)
minimum = config.get(CONF_MINIMUM)
maximum = config.get(CONF_MAXIMUM)
unit = config.get(CONF_UNIT_OF_MEASUREMENT)
async_add_entities([RandomSensor(name, minimum, maximum, unit)], True)
async_add_entities([RandomSensor(config)], True)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Initialize config entry."""
async_add_entities(
[RandomSensor(config_entry.options, config_entry.entry_id)], True
)
class RandomSensor(SensorEntity):
"""Representation of a Random number sensor."""
_attr_icon = "mdi:hanger"
_state: int | None = None
def __init__(self, name, minimum, maximum, unit_of_measurement):
def __init__(self, config: Mapping[str, Any], entry_id: str | None = None) -> None:
"""Initialize the Random sensor."""
self._name = name
self._minimum = minimum
self._maximum = maximum
self._unit_of_measurement = unit_of_measurement
self._state = None
self._name = config.get(CONF_NAME)
self._minimum = config.get(CONF_MINIMUM, DEFAULT_MIN)
self._maximum = config.get(CONF_MAXIMUM, DEFAULT_MAX)
self._unit_of_measurement = config.get(CONF_UNIT_OF_MEASUREMENT)
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
if entry_id:
self._attr_unique_id = entry_id
@property
def name(self):

View file

@ -0,0 +1,48 @@
{
"config": {
"step": {
"binary_sensor": {
"data": {
"device_class": "[%key:component::random::config::step::sensor::data::device_class%]",
"name": "[%key:common::config_flow::data::name%]"
},
"title": "Random binary sensor"
},
"sensor": {
"data": {
"device_class": "Device class",
"name": "[%key:common::config_flow::data::name%]",
"minimum": "Minimum",
"maximum": "Maximum",
"unit_of_measurement": "Unit of measurement"
},
"title": "Random sensor"
},
"user": {
"description": "This helper allow you to create a helper that emits a random value.",
"menu_options": {
"binary_sensor": "Random binary sensor",
"sensor": "Random sensor"
},
"title": "Random helper"
}
}
},
"options": {
"step": {
"binary_sensor": {
"title": "[%key:component::random::config::step::binary_sensor::title%]",
"description": "This helper does not have any options."
},
"sensor": {
"data": {
"device_class": "[%key:component::random::config::step::sensor::data::device_class%]",
"minimum": "[%key:component::random::config::step::sensor::data::minimum%]",
"maximum": "[%key:component::random::config::step::sensor::data::maximum%]",
"unit_of_measurement": "[%key:component::random::config::step::sensor::data::unit_of_measurement%]"
},
"title": "[%key:component::random::config::step::sensor::title%]"
}
}
}
}

View file

@ -9,6 +9,7 @@ FLOWS = {
"group",
"integration",
"min_max",
"random",
"switch_as_x",
"template",
"threshold",

View file

@ -4596,12 +4596,6 @@
"config_flow": true,
"iot_class": "local_polling"
},
"random": {
"name": "Random",
"integration_type": "hub",
"config_flow": false,
"iot_class": "local_polling"
},
"rapt_ble": {
"name": "RAPT Bluetooth",
"integration_type": "hub",
@ -6769,6 +6763,12 @@
"config_flow": true,
"iot_class": "calculated"
},
"random": {
"name": "Random",
"integration_type": "helper",
"config_flow": true,
"iot_class": "calculated"
},
"schedule": {
"integration_type": "helper",
"config_flow": false

View file

@ -0,0 +1,201 @@
"""Test the Random config flow."""
from typing import Any
from unittest.mock import patch
import pytest
from voluptuous import Invalid
from homeassistant import config_entries
from homeassistant.components.random import async_setup_entry
from homeassistant.components.random.const import DOMAIN
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import UnitOfEnergy, UnitOfPower
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
(
"entity_type",
"extra_input",
"extra_options",
),
(
(
"binary_sensor",
{},
{},
),
(
"sensor",
{
"device_class": SensorDeviceClass.POWER,
"unit_of_measurement": UnitOfPower.WATT,
},
{
"device_class": SensorDeviceClass.POWER,
"unit_of_measurement": UnitOfPower.WATT,
"minimum": 0,
"maximum": 20,
},
),
(
"sensor",
{},
{"minimum": 0, "maximum": 20},
),
),
)
async def test_config_flow(
hass: HomeAssistant,
entity_type: str,
extra_input: dict[str, Any],
extra_options: dict[str, Any],
) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.MENU
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": entity_type},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == entity_type
with patch(
"homeassistant.components.random.async_setup_entry", wraps=async_setup_entry
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"name": "My random entity",
**extra_input,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "My random entity"
assert result["data"] == {}
assert result["options"] == {
"name": "My random entity",
"entity_type": entity_type,
**extra_options,
}
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("device_class", "unit_of_measurement"),
[
(SensorDeviceClass.POWER, UnitOfEnergy.WATT_HOUR),
(SensorDeviceClass.ILLUMINANCE, UnitOfEnergy.WATT_HOUR),
],
)
async def test_wrong_uom(
hass: HomeAssistant, device_class: SensorDeviceClass, unit_of_measurement: str
) -> None:
"""Test entering a wrong unit of measurement."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.MENU
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"next_step_id": "sensor"},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "sensor"
with pytest.raises(Invalid, match="is not a valid unit for device class"):
await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"name": "My random entity",
"device_class": device_class,
"unit_of_measurement": unit_of_measurement,
},
)
@pytest.mark.parametrize(
(
"entity_type",
"extra_options",
"options_options",
),
(
(
"sensor",
{
"device_class": SensorDeviceClass.ENERGY,
"unit_of_measurement": UnitOfEnergy.WATT_HOUR,
"minimum": 0,
"maximum": 20,
},
{
"minimum": 10,
"maximum": 20,
"device_class": SensorDeviceClass.POWER,
"unit_of_measurement": UnitOfPower.WATT,
},
),
),
)
async def test_options(
hass: HomeAssistant,
entity_type: str,
extra_options,
options_options,
) -> None:
"""Test reconfiguring."""
random_config_entry = MockConfigEntry(
data={},
domain=DOMAIN,
options={
"name": "My random",
"entity_type": entity_type,
**extra_options,
},
title="My random",
)
random_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(random_config_entry.entry_id)
await hass.async_block_till_done()
config_entry = hass.config_entries.async_entries(DOMAIN)[0]
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == entity_type
assert "name" not in result["data_schema"].schema
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=options_options,
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"] == {
"name": "My random",
"entity_type": entity_type,
**options_options,
}
assert config_entry.data == {}
assert config_entry.options == {
"name": "My random",
"entity_type": entity_type,
**options_options,
}
assert config_entry.title == "My random"