Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add basic Aranet integration (#80865)

pull/81287/head
aschmitz 2022-10-30 22:50:46 -05:00 committed by GitHub
parent ccefc510c3
commit a766b41b13
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 880 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -94,6 +94,8 @@ build.json @home-assistant/supervisor
/tests/components/apprise/ @caronc
/homeassistant/components/aprs/ @PhilRW
/tests/components/aprs/ @PhilRW
/homeassistant/components/aranet/ @aschmitz
/tests/components/aranet/ @aschmitz
/homeassistant/components/arcam_fmj/ @elupus
/tests/components/arcam_fmj/ @elupus
/homeassistant/components/arris_tg2492lg/ @vanbalken

56
homeassistant/components/aranet/__init__.py Normal file
View File

@ -0,0 +1,56 @@
"""The Aranet integration."""
from __future__ import annotations
import logging
from aranet4.client import Aranet4Advertisement
from homeassistant.components.bluetooth import BluetoothScanningMode
from homeassistant.components.bluetooth.models import BluetoothServiceInfoBleak
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .const import DOMAIN
PLATFORMS: list[Platform] = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
def _service_info_to_adv(
service_info: BluetoothServiceInfoBleak,
) -> Aranet4Advertisement:
return Aranet4Advertisement(service_info.device, service_info.advertisement)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Aranet from a config entry."""
address = entry.unique_id
assert address is not None
coordinator = hass.data.setdefault(DOMAIN, {})[
entry.entry_id
] = PassiveBluetoothProcessorCoordinator(
hass,
_LOGGER,
address=address,
mode=BluetoothScanningMode.PASSIVE,
update_method=_service_info_to_adv,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(
coordinator.async_start()
) # only start after all platforms have had a chance to subscribe
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

123
homeassistant/components/aranet/config_flow.py Normal file
View File

@ -0,0 +1,123 @@
"""Config flow for Aranet integration."""
from __future__ import annotations
import logging
from typing import Any
from aranet4.client import Aranet4Advertisement, Version as AranetVersion
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_discovered_service_info,
)
from homeassistant.const import CONF_ADDRESS
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
MIN_VERSION = AranetVersion(1, 2, 0)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Aranet."""
VERSION = 1
def __init__(self) -> None:
"""Set up a new config flow for Aranet."""
self._discovery_info: BluetoothServiceInfoBleak | None = None
self._discovered_device: Aranet4Advertisement | None = None
self._discovered_devices: dict[str, tuple[str, Aranet4Advertisement]] = {}
def _raise_for_advertisement_errors(self, adv: Aranet4Advertisement) -> None:
"""Raise any configuration errors that apply to an advertisement."""
# Old versions of firmware don't expose sensor data in advertisements.
if not adv.manufacturer_data or adv.manufacturer_data.version < MIN_VERSION:
raise AbortFlow("outdated_version")
# If integrations are disabled, we get no sensor data.
if not adv.manufacturer_data.integrations:
raise AbortFlow("integrations_disabled")
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
) -> FlowResult:
"""Handle the Bluetooth discovery step."""
await self.async_set_unique_id(discovery_info.address)
self._abort_if_unique_id_configured()
adv = Aranet4Advertisement(discovery_info.device, discovery_info.advertisement)
self._raise_for_advertisement_errors(adv)
self._discovery_info = discovery_info
self._discovered_device = adv
return await self.async_step_bluetooth_confirm()
async def async_step_bluetooth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm discovery."""
assert self._discovered_device is not None
adv = self._discovered_device
assert self._discovery_info is not None
discovery_info = self._discovery_info
title = adv.readings.name if adv.readings else discovery_info.name
if user_input is not None:
return self.async_create_entry(title=title, data={})
self._set_confirm_only()
placeholders = {"name": title}
self.context["title_placeholders"] = placeholders
return self.async_show_form(
step_id="bluetooth_confirm", description_placeholders=placeholders
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step to pick discovered device."""
if user_input is not None:
address = user_input[CONF_ADDRESS]
adv = self._discovered_devices[address][1]
self._raise_for_advertisement_errors(adv)
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self._discovered_devices[address][0], data={}
)
current_addresses = self._async_current_ids()
for discovery_info in async_discovered_service_info(self.hass, False):
address = discovery_info.address
if address in current_addresses or address in self._discovered_devices:
continue
adv = Aranet4Advertisement(
discovery_info.device, discovery_info.advertisement
)
if adv.manufacturer_data:
self._discovered_devices[address] = (
adv.readings.name if adv.readings else discovery_info.name,
adv,
)
if not self._discovered_devices:
return self.async_abort(reason="no_devices_found")
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_ADDRESS): vol.In(
{
addr: dev[0]
for (addr, dev) in self._discovered_devices.items()
}
)
}
),
)

3
homeassistant/components/aranet/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Constants for the Aranet integration."""
DOMAIN = "aranet"

23
homeassistant/components/aranet/manifest.json Normal file
View File

@ -0,0 +1,23 @@
{
"domain": "aranet",
"name": "Aranet",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/aranet",
"requirements": ["aranet4==2.1.3"],
"dependencies": ["bluetooth"],
"codeowners": ["@aschmitz"],
"iot_class": "local_push",
"integration_type": "device",
"bluetooth": [
{
"manufacturer_id": 1794,
"service_uuid": "f0cd1400-95da-4f4b-9ac8-aa55d312af0c",
"connectable": false
},
{
"manufacturer_id": 1794,
"service_uuid": "0000fce0-0000-1000-8000-00805f9b34fb",
"connectable": false
}
]
}

169
homeassistant/components/aranet/sensor.py Normal file
View File

@ -0,0 +1,169 @@
"""Support for Aranet sensors."""
from __future__ import annotations
from typing import Optional, Union
from aranet4.client import Aranet4Advertisement
from bleak.backends.device import BLEDevice
from homeassistant import config_entries
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothDataProcessor,
PassiveBluetoothDataUpdate,
PassiveBluetoothEntityKey,
PassiveBluetoothProcessorCoordinator,
PassiveBluetoothProcessorEntity,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
ATTR_NAME,
ATTR_SW_VERSION,
CONCENTRATION_PARTS_PER_MILLION,
PERCENTAGE,
PRESSURE_HPA,
TEMP_CELSIUS,
TIME_SECONDS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
SENSOR_DESCRIPTIONS = {
"temperature": SensorEntityDescription(
key="temperature",
name="Temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=TEMP_CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
),
"humidity": SensorEntityDescription(
key="humidity",
name="Humidity",
device_class=SensorDeviceClass.HUMIDITY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
"pressure": SensorEntityDescription(
key="pressure",
name="Pressure",
device_class=SensorDeviceClass.PRESSURE,
native_unit_of_measurement=PRESSURE_HPA,
state_class=SensorStateClass.MEASUREMENT,
),
"co2": SensorEntityDescription(
key="co2",
name="Carbon Dioxide",
device_class=SensorDeviceClass.CO2,
native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
state_class=SensorStateClass.MEASUREMENT,
),
"battery": SensorEntityDescription(
key="battery",
name="Battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
),
"interval": SensorEntityDescription(
key="update_interval",
name="Update Interval",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=TIME_SECONDS,
state_class=SensorStateClass.MEASUREMENT,
),
}
def _device_key_to_bluetooth_entity_key(
device: BLEDevice,
key: str,
) -> PassiveBluetoothEntityKey:
"""Convert a device key to an entity key."""
return PassiveBluetoothEntityKey(key, device.address)
def _sensor_device_info_to_hass(
adv: Aranet4Advertisement,
) -> DeviceInfo:
"""Convert a sensor device info to hass device info."""
hass_device_info = DeviceInfo({})
if adv.readings and adv.readings.name:
hass_device_info[ATTR_NAME] = adv.readings.name
if adv.manufacturer_data:
hass_device_info[ATTR_SW_VERSION] = str(adv.manufacturer_data.version)
return hass_device_info
def sensor_update_to_bluetooth_data_update(
adv: Aranet4Advertisement,
) -> PassiveBluetoothDataUpdate:
"""Convert a sensor update to a Bluetooth data update."""
return PassiveBluetoothDataUpdate(
devices={adv.device.address: _sensor_device_info_to_hass(adv)},
entity_descriptions={
_device_key_to_bluetooth_entity_key(adv.device, key): desc
for key, desc in SENSOR_DESCRIPTIONS.items()
},
entity_data={
_device_key_to_bluetooth_entity_key(adv.device, key): getattr(
adv.readings, key, None
)
for key in SENSOR_DESCRIPTIONS
},
entity_names={
_device_key_to_bluetooth_entity_key(adv.device, key): desc.name
for key, desc in SENSOR_DESCRIPTIONS.items()
},
)
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Aranet sensors."""
coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
entry.entry_id
]
processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update)
entry.async_on_unload(
processor.async_add_entities_listener(
Aranet4BluetoothSensorEntity, async_add_entities
)
)
entry.async_on_unload(coordinator.async_register_processor(processor))
class Aranet4BluetoothSensorEntity(
PassiveBluetoothProcessorEntity[
PassiveBluetoothDataProcessor[Optional[Union[float, int]]]
],
SensorEntity,
):
"""Representation of an Aranet sensor."""
@property
def available(self) -> bool:
"""Return whether the entity was available in the last update."""
# Our superclass covers "did the device disappear entirely", but if the
# device has smart home integrations disabled, it will send BLE beacons
# without data, which we turn into Nones here. Because None is never a
# valid value for any of the Aranet sensors, that means the entity is
# actually unavailable.
return (
super().available
and self.processor.entity_data.get(self.entity_key) is not None
)
@property
def native_value(self) -> int | float | None:
"""Return the native value."""
return self.processor.entity_data.get(self.entity_key)

25
homeassistant/components/aranet/strings.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"step": {
"user": {
"description": "[%key:component::bluetooth::config::step::user::description%]",
"data": {
"address": "[%key:component::bluetooth::config::step::user::data::address%]"
}
},
"bluetooth_confirm": {
"description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
}
},
"flow_title": "[%key:component::bluetooth::config::flow_title%]",
"error": {
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"integrations_diabled": "This device doesn't have integrations enabled. Please enable smart home integrations using the app and try again.",
"no_devices_found": "No unconfigured Aranet devices found.",
"outdated_version": "This device is using outdated firmware. Please update it to at least v1.2.0 and try again."
}
}
}

25
homeassistant/components/aranet/translations/en.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"integrations_diabled": "This device doesn't have integrations enabled. Please enable smart home integrations using the app and try again.",
"no_devices_found": "No unconfigured Aranet4 devices found.",
"outdated_version": "This device is using outdated firmware. Please update it to at least v1.2.0 and try again."
},
"error": {
"unknown": "Unexpected error"
},
"flow_title": "{name}",
"step": {
"bluetooth_confirm": {
"description": "Do you want to setup {name}?"
},
"user": {
"data": {
"address": "Device"
},
"description": "Choose a device to setup"
}
}
}
}

12
homeassistant/generated/bluetooth.py
View File

@ -9,6 +9,18 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
"domain": "airthings_ble",
"manufacturer_id": 820,
},
{
"domain": "aranet",
"manufacturer_id": 1794,
"service_uuid": "f0cd1400-95da-4f4b-9ac8-aa55d312af0c",
"connectable": False,
},
{
"domain": "aranet",
"manufacturer_id": 1794,
"service_uuid": "0000fce0-0000-1000-8000-00805f9b34fb",
"connectable": False,
},
{
"domain": "bluemaestro",
"manufacturer_id": 307,

1
homeassistant/generated/config_flows.py
View File

@ -31,6 +31,7 @@ FLOWS = {
"anthemav",
"apcupsd",
"apple_tv",
"aranet",
"arcam_fmj",
"aseko_pool_live",
"asuswrt",

6
homeassistant/generated/integrations.json
View File

@ -313,6 +313,12 @@
"config_flow": false,
"iot_class": "local_polling"
},
"aranet": {
"name": "Aranet",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_push"
},
"arcam_fmj": {
"name": "Arcam FMJ Receivers",
"integration_type": "hub",

3
requirements_all.txt
View File

@ -335,6 +335,9 @@ aprslib==0.7.0
# homeassistant.components.aqualogic
aqualogic==2.6
# homeassistant.components.aranet
aranet4==2.1.3
# homeassistant.components.arcam_fmj
arcam-fmj==0.12.0

3
requirements_test_all.txt
View File

@ -298,6 +298,9 @@ apprise==1.1.0
# homeassistant.components.aprs
aprslib==0.7.0
# homeassistant.components.aranet
aranet4==2.1.3
# homeassistant.components.arcam_fmj
arcam-fmj==0.12.0

58
tests/components/aranet/__init__.py Normal file
View File

@ -0,0 +1,58 @@
"""Tests for the Aranet integration."""
from time import time
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
def fake_service_info(name, service_uuid, manufacturer_data):
"""Return a BluetoothServiceInfoBleak for use in testing."""
return BluetoothServiceInfoBleak(
name=name,
address="aa:bb:cc:dd:ee:ff",
rssi=-60,
manufacturer_data=manufacturer_data,
service_data={},
service_uuids=[service_uuid],
source="local",
connectable=False,
time=time(),
device=BLEDevice("aa:bb:cc:dd:ee:ff", name=name),
advertisement=AdvertisementData(
local_name=name,
manufacturer_data=manufacturer_data,
service_data={},
service_uuids=[service_uuid],
rssi=-60,
tx_power=-127,
platform_data=(),
),
)
NOT_ARANET4_SERVICE_INFO = fake_service_info(
"Not it", "61DE521B-F0BF-9F44-64D4-75BBE1738105", {3234: b"\x00\x01"}
)
OLD_FIRMWARE_SERVICE_INFO = fake_service_info(
"Aranet4 12345",
"f0cd1400-95da-4f4b-9ac8-aa55d312af0c",
{1794: b"\x21\x0a\x04\x00\x00\x00\x00\x00"},
)
DISABLED_INTEGRATIONS_SERVICE_INFO = fake_service_info(
"Aranet4 12345",
"0000fce0-0000-1000-8000-00805f9b34fb",
{1794: b"\x01\x00\x02\x01\x00\x00\x00\x00"},
)
VALID_DATA_SERVICE_INFO = fake_service_info(
"Aranet4 12345",
"0000fce0-0000-1000-8000-00805f9b34fb",
{
1794: b'\x21\x00\x02\x01\x00\x00\x00\x01\x8a\x02\xa5\x01\xb1&"Y\x01,\x01\xe8\x00\x88'
},
)

8
tests/components/aranet/conftest.py Normal file
View File

@ -0,0 +1,8 @@
"""Aranet session fixtures."""
import pytest
@pytest.fixture(autouse=True)
def mock_bluetooth(enable_bluetooth):
"""Auto mock bluetooth."""

252
tests/components/aranet/test_config_flow.py Normal file
View File

@ -0,0 +1,252 @@
"""Test the Aranet config flow."""
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.aranet.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import (
DISABLED_INTEGRATIONS_SERVICE_INFO,
NOT_ARANET4_SERVICE_INFO,
OLD_FIRMWARE_SERVICE_INFO,
VALID_DATA_SERVICE_INFO,
)
from tests.common import MockConfigEntry
async def test_async_step_bluetooth_valid_device(hass):
"""Test discovery via bluetooth with a valid device."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VALID_DATA_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Aranet4 12345"
assert result2["data"] == {}
assert result2["result"].unique_id == "aa:bb:cc:dd:ee:ff"
async def test_async_step_bluetooth_not_aranet4(hass):
"""Test that we reject discovery via Bluetooth for an unrelated device."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=NOT_ARANET4_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
async def test_async_step_bluetooth_devices_already_setup(hass):
"""Test we can't start a flow if there is already a config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="aa:bb:cc:dd:ee:ff",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VALID_DATA_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_async_step_bluetooth_already_in_progress(hass):
"""Test we can't start a flow for the same device twice."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VALID_DATA_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VALID_DATA_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "already_in_progress"
async def test_async_step_user_takes_precedence_over_discovery(hass):
"""Test manual setup takes precedence over discovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_BLUETOOTH},
data=VALID_DATA_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[VALID_DATA_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "aa:bb:cc:dd:ee:ff"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Aranet4 12345"
assert result2["data"] == {}
assert result2["result"].unique_id == "aa:bb:cc:dd:ee:ff"
# Verify the original one was aborted
assert not hass.config_entries.flow.async_progress(DOMAIN)
async def test_async_step_user_no_devices_found(hass: HomeAssistant):
"""Test setup from service info cache with no devices found."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_async_step_user_only_other_devices_found(hass: HomeAssistant):
"""Test setup from service info cache with only other devices found."""
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[NOT_ARANET4_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_async_step_user_with_found_devices(hass: HomeAssistant):
"""Test setup from service info cache with devices found."""
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[VALID_DATA_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "aa:bb:cc:dd:ee:ff"},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "Aranet4 12345"
assert result2["data"] == {}
assert result2["result"].unique_id == "aa:bb:cc:dd:ee:ff"
async def test_async_step_user_device_added_between_steps(hass: HomeAssistant):
"""Test the device gets added via another flow between steps."""
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[VALID_DATA_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="aa:bb:cc:dd:ee:ff",
)
entry.add_to_hass(hass)
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "aa:bb:cc:dd:ee:ff"},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
async def test_async_step_user_with_found_devices_already_setup(hass: HomeAssistant):
"""Test setup from service info cache with devices found."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="aa:bb:cc:dd:ee:ff",
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[VALID_DATA_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_async_step_user_old_firmware(hass: HomeAssistant):
"""Test we can't set up a device with firmware too old to report measurements."""
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[OLD_FIRMWARE_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "aa:bb:cc:dd:ee:ff"},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "outdated_version"
async def test_async_step_user_integrations_disabled(hass: HomeAssistant):
"""Test we can't set up a device the device's integration setting disabled."""
with patch(
"homeassistant.components.aranet.config_flow.async_discovered_service_info",
return_value=[DISABLED_INTEGRATIONS_SERVICE_INFO],
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
with patch("homeassistant.components.aranet.async_setup_entry", return_value=True):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"address": "aa:bb:cc:dd:ee:ff"},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "integrations_disabled"

111
tests/components/aranet/test_sensor.py Normal file
View File

@ -0,0 +1,111 @@
"""Test the Aranet sensors."""
from homeassistant.components.aranet.const import DOMAIN
from homeassistant.components.sensor import ATTR_STATE_CLASS
from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT
from . import DISABLED_INTEGRATIONS_SERVICE_INFO, VALID_DATA_SERVICE_INFO
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
async def test_sensors(hass):
"""Test setting up creates the sensors."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="aa:bb:cc:dd:ee:ff",
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all("sensor")) == 0
inject_bluetooth_service_info(hass, VALID_DATA_SERVICE_INFO)
await hass.async_block_till_done()
assert len(hass.states.async_all("sensor")) == 6
batt_sensor = hass.states.get("sensor.aranet4_12345_battery")
batt_sensor_attrs = batt_sensor.attributes
assert batt_sensor.state == "89"
assert batt_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Battery"
assert batt_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "%"
assert batt_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
co2_sensor = hass.states.get("sensor.aranet4_12345_carbon_dioxide")
co2_sensor_attrs = co2_sensor.attributes
assert co2_sensor.state == "650"
assert co2_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Carbon Dioxide"
assert co2_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "ppm"
assert co2_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
humid_sensor = hass.states.get("sensor.aranet4_12345_humidity")
humid_sensor_attrs = humid_sensor.attributes
assert humid_sensor.state == "34"
assert humid_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Humidity"
assert humid_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "%"
assert humid_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
temp_sensor = hass.states.get("sensor.aranet4_12345_temperature")
temp_sensor_attrs = temp_sensor.attributes
assert temp_sensor.state == "21.1"
assert temp_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Temperature"
assert temp_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "°C"
assert temp_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
press_sensor = hass.states.get("sensor.aranet4_12345_pressure")
press_sensor_attrs = press_sensor.attributes
assert press_sensor.state == "990.5"
assert press_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Pressure"
assert press_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "hPa"
assert press_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
interval_sensor = hass.states.get("sensor.aranet4_12345_update_interval")
interval_sensor_attrs = interval_sensor.attributes
assert interval_sensor.state == "300"
assert interval_sensor_attrs[ATTR_FRIENDLY_NAME] == "Aranet4 12345 Update Interval"
assert interval_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "s"
assert interval_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_smart_home_integration_disabled(hass):
"""Test disabling smart home integration marks entities as unavailable."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="aa:bb:cc:dd:ee:ff",
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all("sensor")) == 0
inject_bluetooth_service_info(hass, DISABLED_INTEGRATIONS_SERVICE_INFO)
await hass.async_block_till_done()
assert len(hass.states.async_all("sensor")) == 6
batt_sensor = hass.states.get("sensor.aranet4_12345_battery")
assert batt_sensor.state == "unavailable"
co2_sensor = hass.states.get("sensor.aranet4_12345_carbon_dioxide")
assert co2_sensor.state == "unavailable"
humid_sensor = hass.states.get("sensor.aranet4_12345_humidity")
assert humid_sensor.state == "unavailable"
temp_sensor = hass.states.get("sensor.aranet4_12345_temperature")
assert temp_sensor.state == "unavailable"
press_sensor = hass.states.get("sensor.aranet4_12345_pressure")
assert press_sensor.state == "unavailable"
interval_sensor = hass.states.get("sensor.aranet4_12345_update_interval")
assert interval_sensor.state == "unavailable"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()