github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add Aprilaire integration (#95093) 

* Add Aprilaire integration

* Fix test errors

* Update constants

* Code review cleanup

* Reuse coordinator from config flow

* Code review fixes

* Remove unneeded tests

* Improve translation

* Code review fixes

* Remove unneeded fixture

* Code review fixes

* Code review updates

* Use base data coordinator

* Deduplicate based on MAC

* Fix tests

* Check mac address on init

* Fix mypy error

* Use config entry ID for entity unique ID

* Fix tests

* Code review updates

* Fix mypy errors

* Code review updates

* Add data_description

* Update homeassistant/components/aprilaire/coordinator.py

Co-authored-by: Jon Oberheide <506986+jonoberheide@users.noreply.github.com>

* Update .coveragerc

* Update homeassistant/components/aprilaire/coordinator.py

---------

Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: Jon Oberheide <506986+jonoberheide@users.noreply.github.com>
This commit is contained in:
Matthew FitzGerald-Chamberlain 2024-02-16 01:30:51 -06:00 committed by GitHub
parent f7b9b0da0e
commit ce8cf314f9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 880 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.coveragerc
View File

@ -73,6 +73,10 @@ omit =
homeassistant/components/apple_tv/browse_media.py
homeassistant/components/apple_tv/media_player.py
homeassistant/components/apple_tv/remote.py
homeassistant/components/aprilaire/__init__.py
homeassistant/components/aprilaire/climate.py
homeassistant/components/aprilaire/coordinator.py
homeassistant/components/aprilaire/entity.py
homeassistant/components/aqualogic/*
homeassistant/components/aquostv/media_player.py
homeassistant/components/arcam_fmj/__init__.py

2
CODEOWNERS
View File

@ -104,6 +104,8 @@ build.json @home-assistant/supervisor
/tests/components/application_credentials/ @home-assistant/core
/homeassistant/components/apprise/ @caronc
/tests/components/apprise/ @caronc
/homeassistant/components/aprilaire/ @chamberlain2007
/tests/components/aprilaire/ @chamberlain2007
/homeassistant/components/aprs/ @PhilRW
/tests/components/aprs/ @PhilRW
/homeassistant/components/aranet/ @aschmitz @thecode

69
homeassistant/components/aprilaire/__init__.py Normal file
View File

@ -0,0 +1,69 @@
"""The Aprilaire integration."""
from __future__ import annotations
import logging
from pyaprilaire.const import Attribute
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.device_registry import format_mac
from .const import DOMAIN
from .coordinator import AprilaireCoordinator
PLATFORMS: list[Platform] = [Platform.CLIMATE]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry for Aprilaire."""
host = entry.data[CONF_HOST]
port = entry.data[CONF_PORT]
coordinator = AprilaireCoordinator(hass, entry.unique_id, host, port)
await coordinator.start_listen()
hass.data.setdefault(DOMAIN, {})[entry.unique_id] = coordinator
async def ready_callback(ready: bool):
if ready:
mac_address = format_mac(coordinator.data[Attribute.MAC_ADDRESS])
if mac_address != entry.unique_id:
raise ConfigEntryAuthFailed("Invalid MAC address")
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
async def _async_close(_: Event) -> None:
coordinator.stop_listen()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_close)
)
else:
_LOGGER.error("Failed to wait for ready")
coordinator.stop_listen()
raise ConfigEntryNotReady()
await coordinator.wait_for_ready(ready_callback)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
coordinator: AprilaireCoordinator = hass.data[DOMAIN].pop(entry.unique_id)
coordinator.stop_listen()
return unload_ok

302
homeassistant/components/aprilaire/climate.py Normal file
View File

@ -0,0 +1,302 @@
"""The Aprilaire climate component."""
from __future__ import annotations
from typing import Any
from pyaprilaire.const import Attribute
from homeassistant.components.climate import (
FAN_AUTO,
FAN_ON,
PRESET_AWAY,
PRESET_NONE,
ClimateEntity,
ClimateEntityFeature,
HVACAction,
HVACMode,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PRECISION_HALVES, PRECISION_WHOLE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
DOMAIN,
FAN_CIRCULATE,
PRESET_PERMANENT_HOLD,
PRESET_TEMPORARY_HOLD,
PRESET_VACATION,
)
from .coordinator import AprilaireCoordinator
from .entity import BaseAprilaireEntity
HVAC_MODE_MAP = {
1: HVACMode.OFF,
2: HVACMode.HEAT,
3: HVACMode.COOL,
4: HVACMode.HEAT,
5: HVACMode.AUTO,
}
HVAC_MODES_MAP = {
1: [HVACMode.OFF, HVACMode.HEAT],
2: [HVACMode.OFF, HVACMode.COOL],
3: [HVACMode.OFF, HVACMode.HEAT, HVACMode.COOL],
4: [HVACMode.OFF, HVACMode.HEAT, HVACMode.COOL],
5: [HVACMode.OFF, HVACMode.HEAT, HVACMode.COOL, HVACMode.AUTO],
6: [HVACMode.OFF, HVACMode.HEAT, HVACMode.COOL, HVACMode.AUTO],
}
PRESET_MODE_MAP = {
1: PRESET_TEMPORARY_HOLD,
2: PRESET_PERMANENT_HOLD,
3: PRESET_AWAY,
4: PRESET_VACATION,
}
FAN_MODE_MAP = {
1: FAN_ON,
2: FAN_AUTO,
3: FAN_CIRCULATE,
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add climates for passed config_entry in HA."""
coordinator: AprilaireCoordinator = hass.data[DOMAIN][config_entry.unique_id]
async_add_entities([AprilaireClimate(coordinator, config_entry.unique_id)])
class AprilaireClimate(BaseAprilaireEntity, ClimateEntity):
"""Climate entity for Aprilaire."""
_attr_fan_modes = [FAN_AUTO, FAN_ON, FAN_CIRCULATE]
_attr_min_humidity = 10
_attr_max_humidity = 50
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_translation_key = "thermostat"
@property
def precision(self) -> float:
"""Get the precision based on the unit."""
return (
PRECISION_HALVES
if self.hass.config.units.temperature_unit == UnitOfTemperature.CELSIUS
else PRECISION_WHOLE
)
@property
def supported_features(self) -> ClimateEntityFeature:
"""Get supported features."""
features = 0
if self.coordinator.data.get(Attribute.MODE) == 5:
features = features | ClimateEntityFeature.TARGET_TEMPERATURE_RANGE
else:
features = features | ClimateEntityFeature.TARGET_TEMPERATURE
if self.coordinator.data.get(Attribute.HUMIDIFICATION_AVAILABLE) == 2:
features = features | ClimateEntityFeature.TARGET_HUMIDITY
features = features | ClimateEntityFeature.PRESET_MODE
features = features | ClimateEntityFeature.FAN_MODE
return features
@property
def current_humidity(self) -> int | None:
"""Get current humidity."""
return self.coordinator.data.get(
Attribute.INDOOR_HUMIDITY_CONTROLLING_SENSOR_VALUE
)
@property
def target_humidity(self) -> int | None:
"""Get current target humidity."""
return self.coordinator.data.get(Attribute.HUMIDIFICATION_SETPOINT)
@property
def hvac_mode(self) -> HVACMode | None:
"""Get HVAC mode."""
if mode := self.coordinator.data.get(Attribute.MODE):
if hvac_mode := HVAC_MODE_MAP.get(mode):
return hvac_mode
return None
@property
def hvac_modes(self) -> list[HVACMode]:
"""Get supported HVAC modes."""
if modes := self.coordinator.data.get(Attribute.THERMOSTAT_MODES):
if thermostat_modes := HVAC_MODES_MAP.get(modes):
return thermostat_modes
return []
@property
def hvac_action(self) -> HVACAction | None:
"""Get the current HVAC action."""
if self.coordinator.data.get(Attribute.HEATING_EQUIPMENT_STATUS, 0):
return HVACAction.HEATING
if self.coordinator.data.get(Attribute.COOLING_EQUIPMENT_STATUS, 0):
return HVACAction.COOLING
return HVACAction.IDLE
@property
def current_temperature(self) -> float | None:
"""Get current temperature."""
return self.coordinator.data.get(
Attribute.INDOOR_TEMPERATURE_CONTROLLING_SENSOR_VALUE
)
@property
def target_temperature(self) -> float | None:
"""Get the target temperature."""
hvac_mode = self.hvac_mode
if hvac_mode == HVACMode.COOL:
return self.target_temperature_high
if hvac_mode == HVACMode.HEAT:
return self.target_temperature_low
return None
@property
def target_temperature_step(self) -> float | None:
"""Get the step for the target temperature based on the unit."""
return (
0.5
if self.hass.config.units.temperature_unit == UnitOfTemperature.CELSIUS
else 1
)
@property
def target_temperature_high(self) -> float | None:
"""Get cool setpoint."""
return self.coordinator.data.get(Attribute.COOL_SETPOINT)
@property
def target_temperature_low(self) -> float | None:
"""Get heat setpoint."""
return self.coordinator.data.get(Attribute.HEAT_SETPOINT)
@property
def preset_mode(self) -> str | None:
"""Get the current preset mode."""
if hold := self.coordinator.data.get(Attribute.HOLD):
if preset_mode := PRESET_MODE_MAP.get(hold):
return preset_mode
return PRESET_NONE
@property
def preset_modes(self) -> list[str] | None:
"""Get the supported preset modes."""
presets = [PRESET_NONE, PRESET_VACATION]
if self.coordinator.data.get(Attribute.AWAY_AVAILABLE) == 1:
presets.append(PRESET_AWAY)
hold = self.coordinator.data.get(Attribute.HOLD, 0)
if hold == 1:
presets.append(PRESET_TEMPORARY_HOLD)
elif hold == 2:
presets.append(PRESET_PERMANENT_HOLD)
return presets
@property
def fan_mode(self) -> str | None:
"""Get fan mode."""
if mode := self.coordinator.data.get(Attribute.FAN_MODE):
if fan_mode := FAN_MODE_MAP.get(mode):
return fan_mode
return None
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
cool_setpoint = 0
heat_setpoint = 0
if temperature := kwargs.get("temperature"):
if self.coordinator.data.get(Attribute.MODE) == 3:
cool_setpoint = temperature
else:
heat_setpoint = temperature
else:
if target_temp_low := kwargs.get("target_temp_low"):
heat_setpoint = target_temp_low
if target_temp_high := kwargs.get("target_temp_high"):
cool_setpoint = target_temp_high
if cool_setpoint == 0 and heat_setpoint == 0:
return
await self.coordinator.client.update_setpoint(cool_setpoint, heat_setpoint)
await self.coordinator.client.read_control()
async def async_set_humidity(self, humidity: int) -> None:
"""Set the target humidification setpoint."""
await self.coordinator.client.set_humidification_setpoint(humidity)
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set the fan mode."""
try:
fan_mode_value_index = list(FAN_MODE_MAP.values()).index(fan_mode)
except ValueError as exc:
raise ValueError(f"Unsupported fan mode {fan_mode}") from exc
fan_mode_value = list(FAN_MODE_MAP.keys())[fan_mode_value_index]
await self.coordinator.client.update_fan_mode(fan_mode_value)
await self.coordinator.client.read_control()
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the HVAC mode."""
try:
mode_value_index = list(HVAC_MODE_MAP.values()).index(hvac_mode)
except ValueError as exc:
raise ValueError(f"Unsupported HVAC mode {hvac_mode}") from exc
mode_value = list(HVAC_MODE_MAP.keys())[mode_value_index]
await self.coordinator.client.update_mode(mode_value)
await self.coordinator.client.read_control()
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set the preset mode."""
if preset_mode == PRESET_AWAY:
await self.coordinator.client.set_hold(3)
elif preset_mode == PRESET_VACATION:
await self.coordinator.client.set_hold(4)
elif preset_mode == PRESET_NONE:
await self.coordinator.client.set_hold(0)
else:
raise ValueError(f"Unsupported preset mode {preset_mode}")
await self.coordinator.client.read_scheduling()

72
homeassistant/components/aprilaire/config_flow.py Normal file
View File

@ -0,0 +1,72 @@
"""Config flow for the Aprilaire integration."""
from __future__ import annotations
import logging
from typing import Any
from pyaprilaire.const import Attribute
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_PORT
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import format_mac
from .const import DOMAIN
from .coordinator import AprilaireCoordinator
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=7000): cv.port,
}
)
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Aprilaire."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
coordinator = AprilaireCoordinator(
self.hass, None, user_input[CONF_HOST], user_input[CONF_PORT]
)
await coordinator.start_listen()
async def ready_callback(ready: bool):
if not ready:
_LOGGER.error("Failed to wait for ready")
try:
ready = await coordinator.wait_for_ready(ready_callback)
finally:
coordinator.stop_listen()
mac_address = coordinator.data.get(Attribute.MAC_ADDRESS)
if ready and mac_address is not None:
await self.async_set_unique_id(format_mac(mac_address))
self._abort_if_unique_id_configured()
return self.async_create_entry(title="Aprilaire", data=user_input)
return self.async_show_form(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors={"base": "connection_failed"},
)

11
homeassistant/components/aprilaire/const.py Normal file
View File

@ -0,0 +1,11 @@
"""Constants for the Aprilaire integration."""
from __future__ import annotations
DOMAIN = "aprilaire"
FAN_CIRCULATE = "Circulate"
PRESET_TEMPORARY_HOLD = "Temporary"
PRESET_PERMANENT_HOLD = "Permanent"
PRESET_VACATION = "Vacation"

209
homeassistant/components/aprilaire/coordinator.py Normal file
View File

@ -0,0 +1,209 @@
"""The Aprilaire coordinator."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
import logging
from typing import Any, Optional
import pyaprilaire.client
from pyaprilaire.const import MODELS, Attribute, FunctionalDomain
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import BaseDataUpdateCoordinatorProtocol
from .const import DOMAIN
RECONNECT_INTERVAL = 60 * 60
RETRY_CONNECTION_INTERVAL = 10
WAIT_TIMEOUT = 30
_LOGGER = logging.getLogger(__name__)
class AprilaireCoordinator(BaseDataUpdateCoordinatorProtocol):
"""Coordinator for interacting with the thermostat."""
def __init__(
self,
hass: HomeAssistant,
unique_id: str | None,
host: str,
port: int,
) -> None:
"""Initialize the coordinator."""
self.hass = hass
self.unique_id = unique_id
self.data: dict[str, Any] = {}
self._listeners: dict[CALLBACK_TYPE, tuple[CALLBACK_TYPE, object | None]] = {}
self.client = pyaprilaire.client.AprilaireClient(
host,
port,
self.async_set_updated_data,
_LOGGER,
RECONNECT_INTERVAL,
RETRY_CONNECTION_INTERVAL,
)
if hasattr(self.client, "data") and self.client.data:
self.data = self.client.data
@callback
def async_add_listener(
self, update_callback: CALLBACK_TYPE, context: Any = None
) -> Callable[[], None]:
"""Listen for data updates."""
@callback
def remove_listener() -> None:
"""Remove update listener."""
self._listeners.pop(remove_listener)
self._listeners[remove_listener] = (update_callback, context)
return remove_listener
@callback
def async_update_listeners(self) -> None:
"""Update all registered listeners."""
for update_callback, _ in list(self._listeners.values()):
update_callback()
def async_set_updated_data(self, data: Any) -> None:
"""Manually update data, notify listeners and reset refresh interval."""
old_device_info = self.create_device_info(self.data)
self.data = self.data | data
self.async_update_listeners()
new_device_info = self.create_device_info(data)
if (
old_device_info is not None
and new_device_info is not None
and old_device_info != new_device_info
):
device_registry = dr.async_get(self.hass)
device = device_registry.async_get_device(old_device_info["identifiers"])
if device is not None:
new_device_info.pop("identifiers", None)
new_device_info.pop("connections", None)
device_registry.async_update_device(
device_id=device.id,
**new_device_info, # type: ignore[misc]
)
async def start_listen(self):
"""Start listening for data."""
await self.client.start_listen()
def stop_listen(self):
"""Stop listening for data."""
self.client.stop_listen()
async def wait_for_ready(
self, ready_callback: Callable[[bool], Awaitable[bool]]
) -> bool:
"""Wait for the client to be ready."""
if not self.data or Attribute.MAC_ADDRESS not in self.data:
data = await self.client.wait_for_response(
FunctionalDomain.IDENTIFICATION, 2, WAIT_TIMEOUT
)
if not data or Attribute.MAC_ADDRESS not in data:
_LOGGER.error("Missing MAC address")
await ready_callback(False)
return False
if not self.data or Attribute.NAME not in self.data:
await self.client.wait_for_response(
FunctionalDomain.IDENTIFICATION, 4, WAIT_TIMEOUT
)
if not self.data or Attribute.THERMOSTAT_MODES not in self.data:
await self.client.wait_for_response(
FunctionalDomain.CONTROL, 7, WAIT_TIMEOUT
)
if (
not self.data
or Attribute.INDOOR_TEMPERATURE_CONTROLLING_SENSOR_STATUS not in self.data
):
await self.client.wait_for_response(
FunctionalDomain.SENSORS, 2, WAIT_TIMEOUT
)
await ready_callback(True)
return True
@property
def device_name(self) -> str:
"""Get the name of the thermostat."""
return self.create_device_name(self.data)
def create_device_name(self, data: Optional[dict[str, Any]]) -> str:
"""Create the name of the thermostat."""
name = data.get(Attribute.NAME) if data else None
return name if name else "Aprilaire"
def get_hw_version(self, data: dict[str, Any]) -> str:
"""Get the hardware version."""
if hardware_revision := data.get(Attribute.HARDWARE_REVISION):
return (
f"Rev. {chr(hardware_revision)}"
if hardware_revision > ord("A")
else str(hardware_revision)
)
return "Unknown"
@property
def device_info(self) -> DeviceInfo | None:
"""Get the device info for the thermostat."""
return self.create_device_info(self.data)
def create_device_info(self, data: dict[str, Any]) -> DeviceInfo | None:
"""Create the device info for the thermostat."""
if data is None or Attribute.MAC_ADDRESS not in data or self.unique_id is None:
return None
device_info = DeviceInfo(
identifiers={(DOMAIN, self.unique_id)},
name=self.create_device_name(data),
manufacturer="Aprilaire",
)
model_number = data.get(Attribute.MODEL_NUMBER)
if model_number is not None:
device_info["model"] = MODELS.get(model_number, f"Unknown ({model_number})")
device_info["hw_version"] = self.get_hw_version(data)
firmware_major_revision = data.get(Attribute.FIRMWARE_MAJOR_REVISION)
firmware_minor_revision = data.get(Attribute.FIRMWARE_MINOR_REVISION)
if firmware_major_revision is not None:
device_info["sw_version"] = (
str(firmware_major_revision)
if firmware_minor_revision is None
else f"{firmware_major_revision}.{firmware_minor_revision:02}"
)
return device_info

46
homeassistant/components/aprilaire/entity.py Normal file
View File

@ -0,0 +1,46 @@
"""Base functionality for Aprilaire entities."""
from __future__ import annotations
import logging
from pyaprilaire.const import Attribute
from homeassistant.helpers.update_coordinator import BaseCoordinatorEntity
from .coordinator import AprilaireCoordinator
_LOGGER = logging.getLogger(__name__)
class BaseAprilaireEntity(BaseCoordinatorEntity[AprilaireCoordinator]):
"""Base for Aprilaire entities."""
_attr_available = False
_attr_has_entity_name = True
def __init__(
self, coordinator: AprilaireCoordinator, unique_id: str | None
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._attr_device_info = coordinator.device_info
self._attr_unique_id = f"{unique_id}_{self.translation_key}"
self._update_available()
def _update_available(self):
"""Update the entity availability."""
connected: bool = self.coordinator.data.get(
Attribute.CONNECTED, None
) or self.coordinator.data.get(Attribute.RECONNECTING, None)
stopped: bool = self.coordinator.data.get(Attribute.STOPPED, None)
self._attr_available = connected and not stopped
async def async_update(self) -> None:
"""Implement abstract base method."""

11
homeassistant/components/aprilaire/manifest.json Normal file
View File

@ -0,0 +1,11 @@
{
"domain": "aprilaire",
"name": "Aprilaire",
"codeowners": ["@chamberlain2007"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/aprilaire",
"integration_type": "device",
"iot_class": "local_push",
"loggers": ["pyaprilaire"],
"requirements": ["pyaprilaire==0.7.0"]
}

28
homeassistant/components/aprilaire/strings.json Normal file
View File

@ -0,0 +1,28 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]"
},
"data_description": {
"port": "Usually 7000 or 8000"
}
}
},
"error": {
"connection_failed": "Connection failed. Please check that the host and port is correct."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"entity": {
"climate": {
"thermostat": {
"name": "Thermostat"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -52,6 +52,7 @@ FLOWS = {
"aosmith",
"apcupsd",
"apple_tv",
"aprilaire",
"aranet",
"arcam_fmj",
"aseko_pool_live",

6
homeassistant/generated/integrations.json
View File

@ -383,6 +383,12 @@
"config_flow": false,
"iot_class": "cloud_push"
},
"aprilaire": {
"name": "Aprilaire",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_push"
},
"aprs": {
"name": "APRS",
"integration_type": "hub",

3
requirements_all.txt
View File

@ -1684,6 +1684,9 @@ pyairnow==1.2.1
# homeassistant.components.airvisual_pro
pyairvisual==2023.08.1
# homeassistant.components.aprilaire
pyaprilaire==0.7.0
# homeassistant.components.asuswrt
pyasuswrt==0.1.21

3
requirements_test_all.txt
View File

@ -1313,6 +1313,9 @@ pyairnow==1.2.1
# homeassistant.components.airvisual_pro
pyairvisual==2023.08.1
# homeassistant.components.aprilaire
pyaprilaire==0.7.0
# homeassistant.components.asuswrt
pyasuswrt==0.1.21

1
tests/components/aprilaire/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for Aprilaire."""

112
tests/components/aprilaire/test_config_flow.py Normal file
View File

@ -0,0 +1,112 @@
"""Tests for the Aprilaire config flow."""
from unittest.mock import AsyncMock, Mock, patch
from pyaprilaire.client import AprilaireClient
from pyaprilaire.const import FunctionalDomain
import pytest
from homeassistant.components.aprilaire.config_flow import (
STEP_USER_DATA_SCHEMA,
ConfigFlow,
)
from homeassistant.core import HomeAssistant
@pytest.fixture
def client() -> AprilaireClient:
"""Return a mock client."""
return AsyncMock(AprilaireClient)
async def test_user_input_step() -> None:
"""Test the user input step."""
show_form_mock = Mock()
config_flow = ConfigFlow()
config_flow.async_show_form = show_form_mock
await config_flow.async_step_user(None)
show_form_mock.assert_called_once_with(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
async def test_config_flow_invalid_data(client: AprilaireClient) -> None:
"""Test that the flow is aborted with invalid data."""
show_form_mock = Mock()
set_unique_id_mock = AsyncMock()
async_abort_entries_match_mock = Mock()
config_flow = ConfigFlow()
config_flow.async_show_form = show_form_mock
config_flow.async_set_unique_id = set_unique_id_mock
config_flow._async_abort_entries_match = async_abort_entries_match_mock
with patch("pyaprilaire.client.AprilaireClient", return_value=client):
await config_flow.async_step_user(
{
"host": "localhost",
"port": 7000,
}
)
client.start_listen.assert_called_once()
client.wait_for_response.assert_called_once_with(
FunctionalDomain.IDENTIFICATION, 2, 30
)
client.stop_listen.assert_called_once()
show_form_mock.assert_called_once_with(
step_id="user",
data_schema=STEP_USER_DATA_SCHEMA,
errors={"base": "connection_failed"},
)
async def test_config_flow_data(client: AprilaireClient, hass: HomeAssistant) -> None:
"""Test the config flow with valid data."""
client.data = {"mac_address": "1:2:3:4:5:6"}
show_form_mock = Mock()
set_unique_id_mock = AsyncMock()
abort_if_unique_id_configured_mock = Mock()
create_entry_mock = Mock()
config_flow = ConfigFlow()
config_flow.hass = hass
config_flow.async_show_form = show_form_mock
config_flow.async_set_unique_id = set_unique_id_mock
config_flow._abort_if_unique_id_configured = abort_if_unique_id_configured_mock
config_flow.async_create_entry = create_entry_mock
client.wait_for_response = AsyncMock(return_value={"mac_address": "1:2:3:4:5:6"})
with patch("pyaprilaire.client.AprilaireClient", return_value=client):
await config_flow.async_step_user(
{
"host": "localhost",
"port": 7000,
}
)
client.start_listen.assert_called_once()
client.wait_for_response.assert_any_call(FunctionalDomain.IDENTIFICATION, 4, 30)
client.wait_for_response.assert_any_call(FunctionalDomain.CONTROL, 7, 30)
client.wait_for_response.assert_any_call(FunctionalDomain.SENSORS, 2, 30)
client.stop_listen.assert_called_once()
set_unique_id_mock.assert_called_once_with("1:2:3:4:5:6")
abort_if_unique_id_configured_mock.assert_called_once()
create_entry_mock.assert_called_once_with(
title="Aprilaire",
data={
"host": "localhost",
"port": 7000,
},
)