jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 08:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Awair Local API support (#75535)

Browse Source
This commit is contained in:
Zach Berger 2022-08-11 06:01:35 -07:00 committed by GitHub
parent 078a4974e1
commit ebbff7b60e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 603 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
homeassistant/components/awair/__init__.py
View File

@ -2,20 +2,27 @@
from __future__ import annotations
from asyncio import gather
from typing import Any
from async_timeout import timeout
from python_awair import Awair
from python_awair.exceptions import AuthError
from python_awair import Awair, AwairLocal
from python_awair.devices import AwairBaseDevice, AwairLocalDevice
from python_awair.exceptions import AuthError, AwairError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL, AwairResult
from .const import (
API_TIMEOUT,
DOMAIN,
LOGGER,
UPDATE_INTERVAL_CLOUD,
UPDATE_INTERVAL_LOCAL,
AwairResult,
)
PLATFORMS = [Platform.SENSOR]
@ -23,7 +30,13 @@ PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up Awair integration from a config entry."""
session = async_get_clientsession(hass)
coordinator = AwairDataUpdateCoordinator(hass, config_entry, session)
coordinator: AwairDataUpdateCoordinator
if CONF_HOST in config_entry.data:
coordinator = AwairLocalDataUpdateCoordinator(hass, config_entry, session)
else:
coordinator = AwairCloudDataUpdateCoordinator(hass, config_entry, session)
await coordinator.async_config_entry_first_refresh()
@ -50,15 +63,31 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
class AwairDataUpdateCoordinator(DataUpdateCoordinator):
"""Define a wrapper class to update Awair data."""
def __init__(self, hass, config_entry, session) -> None:
def __init__(self, hass, config_entry, update_interval) -> None:
"""Set up the AwairDataUpdateCoordinator class."""
access_token = config_entry.data[CONF_ACCESS_TOKEN]
self._awair = Awair(access_token=access_token, session=session)
self._config_entry = config_entry
super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
super().__init__(hass, LOGGER, name=DOMAIN, update_interval=update_interval)
async def _async_update_data(self) -> Any | None:
async def _fetch_air_data(self, device: AwairBaseDevice):
"""Fetch latest air quality data."""
LOGGER.debug("Fetching data for %s", device.uuid)
air_data = await device.air_data_latest()
LOGGER.debug(air_data)
return AwairResult(device=device, air_data=air_data)
class AwairCloudDataUpdateCoordinator(AwairDataUpdateCoordinator):
"""Define a wrapper class to update Awair data from Cloud API."""
def __init__(self, hass, config_entry, session) -> None:
"""Set up the AwairCloudDataUpdateCoordinator class."""
access_token = config_entry.data[CONF_ACCESS_TOKEN]
self._awair = Awair(access_token=access_token, session=session)
super().__init__(hass, config_entry, UPDATE_INTERVAL_CLOUD)
async def _async_update_data(self) -> dict[str, AwairResult] | None:
"""Update data via Awair client library."""
async with timeout(API_TIMEOUT):
try:
@ -74,9 +103,30 @@ class AwairDataUpdateCoordinator(DataUpdateCoordinator):
except Exception as err:
raise UpdateFailed(err) from err
async def _fetch_air_data(self, device):
"""Fetch latest air quality data."""
LOGGER.debug("Fetching data for %s", device.uuid)
air_data = await device.air_data_latest()
LOGGER.debug(air_data)
return AwairResult(device=device, air_data=air_data)
class AwairLocalDataUpdateCoordinator(AwairDataUpdateCoordinator):
"""Define a wrapper class to update Awair data from the local API."""
_device: AwairLocalDevice | None = None
def __init__(self, hass, config_entry, session) -> None:
"""Set up the AwairLocalDataUpdateCoordinator class."""
self._awair = AwairLocal(
session=session, device_addrs=[config_entry.data[CONF_HOST]]
)
super().__init__(hass, config_entry, UPDATE_INTERVAL_LOCAL)
async def _async_update_data(self) -> dict[str, AwairResult] | None:
"""Update data via Awair client library."""
async with timeout(API_TIMEOUT):
try:
if self._device is None:
LOGGER.debug("Fetching devices")
devices = await self._awair.devices()
self._device = devices[0]
result = await self._fetch_air_data(self._device)
return {result.device.uuid: result}
except AwairError as err:
LOGGER.error("Unexpected API error: %s", err)
raise UpdateFailed(err) from err

124
homeassistant/components/awair/config_flow.py
View File

@ -4,12 +4,14 @@ from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from python_awair import Awair
from aiohttp.client_exceptions import ClientConnectorError
from python_awair import Awair, AwairLocal, AwairLocalDevice
from python_awair.exceptions import AuthError, AwairError
import voluptuous as vol
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -21,20 +23,76 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
_device: AwairLocalDevice
async def async_step_zeroconf(
self, discovery_info: zeroconf.ZeroconfServiceInfo
) -> FlowResult:
"""Handle zeroconf discovery."""
host = discovery_info.host
LOGGER.debug("Discovered device: %s", host)
self._device, _ = await self._check_local_connection(host)
if self._device is not None:
await self.async_set_unique_id(self._device.mac_address)
self._abort_if_unique_id_configured(error="already_configured_device")
self.context.update(
{
"title_placeholders": {
"model": self._device.model,
"device_id": self._device.device_id,
}
}
)
else:
return self.async_abort(reason="unreachable")
return await self.async_step_discovery_confirm()
async def async_step_discovery_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm discovery."""
if user_input is not None:
title = f"{self._device.model} ({self._device.device_id})"
return self.async_create_entry(
title=title,
data={CONF_HOST: self._device.device_addr},
)
self._set_confirm_only()
placeholders = {
"model": self._device.model,
"device_id": self._device.device_id,
}
return self.async_show_form(
step_id="discovery_confirm",
description_placeholders=placeholders,
)
async def async_step_user(
self, user_input: dict[str, str] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
return self.async_show_menu(step_id="user", menu_options=["local", "cloud"])
async def async_step_cloud(self, user_input: Mapping[str, Any]) -> FlowResult:
"""Handle collecting and verifying Awair Cloud API credentials."""
errors = {}
if user_input is not None:
user, error = await self._check_connection(user_input[CONF_ACCESS_TOKEN])
user, error = await self._check_cloud_connection(
user_input[CONF_ACCESS_TOKEN]
)
if user is not None:
await self.async_set_unique_id(user.email)
self._abort_if_unique_id_configured()
self._abort_if_unique_id_configured(error="already_configured_account")
title = f"{user.email} ({user.user_id})"
title = user.email
return self.async_create_entry(title=title, data=user_input)
if error != "invalid_access_token":
@ -43,8 +101,39 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
errors = {CONF_ACCESS_TOKEN: "invalid_access_token"}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str}),
step_id="cloud",
data_schema=vol.Schema({vol.Optional(CONF_ACCESS_TOKEN): str}),
description_placeholders={
"url": "https://developer.getawair.com/onboard/login"
},
errors=errors,
)
async def async_step_local(self, user_input: Mapping[str, Any]) -> FlowResult:
"""Handle collecting and verifying Awair Local API hosts."""
errors = {}
if user_input is not None:
self._device, error = await self._check_local_connection(
user_input[CONF_HOST]
)
if self._device is not None:
await self.async_set_unique_id(self._device.mac_address)
self._abort_if_unique_id_configured(error="already_configured_device")
title = f"{self._device.model} ({self._device.device_id})"
return self.async_create_entry(title=title, data=user_input)
if error is not None:
errors = {CONF_HOST: error}
return self.async_show_form(
step_id="local",
data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
description_placeholders={
"url": "https://support.getawair.com/hc/en-us/articles/360049221014-Awair-Element-Local-API-Feature"
},
errors=errors,
)
@ -60,7 +149,7 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input is not None:
access_token = user_input[CONF_ACCESS_TOKEN]
_, error = await self._check_connection(access_token)
_, error = await self._check_cloud_connection(access_token)
if error is None:
entry = await self.async_set_unique_id(self.unique_id)
@ -79,7 +168,24 @@ class AwairFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def _check_connection(self, access_token: str):
async def _check_local_connection(self, device_address: str):
"""Check the access token is valid."""
session = async_get_clientsession(self.hass)
awair = AwairLocal(session=session, device_addrs=[device_address])
try:
devices = await awair.devices()
return (devices[0], None)
except ClientConnectorError as err:
LOGGER.error("Unable to connect error: %s", err)
return (None, "unreachable")
except AwairError as err:
LOGGER.error("Unexpected API error: %s", err)
return (None, "unknown")
async def _check_cloud_connection(self, access_token: str):
"""Check the access token is valid."""
session = async_get_clientsession(self.hass)
awair = Awair(access_token=access_token, session=session)

7
homeassistant/components/awair/const.py
View File

@ -6,7 +6,7 @@ from datetime import timedelta
import logging
from python_awair.air_data import AirData
from python_awair.devices import AwairDevice
from python_awair.devices import AwairBaseDevice
from homeassistant.components.sensor import SensorDeviceClass, SensorEntityDescription
from homeassistant.const import (
@ -39,7 +39,8 @@ DUST_ALIASES = [API_PM25, API_PM10]
LOGGER = logging.getLogger(__package__)
UPDATE_INTERVAL = timedelta(minutes=5)
UPDATE_INTERVAL_CLOUD = timedelta(minutes=5)
UPDATE_INTERVAL_LOCAL = timedelta(seconds=30)
@dataclass
@ -129,5 +130,5 @@ SENSOR_TYPES_DUST: tuple[AwairSensorEntityDescription, ...] = (
class AwairResult:
"""Wrapper class to hold an awair device and set of air data."""
device: AwairDevice
device: AwairBaseDevice
air_data: AirData

10
homeassistant/components/awair/manifest.json
View File

@ -5,6 +5,12 @@
"requirements": ["python_awair==0.2.3"],
"codeowners": ["@ahayworth", "@danielsjf"],
"config_flow": true,
"iot_class": "cloud_polling",
"loggers": ["python_awair"]
"iot_class": "local_polling",
"loggers": ["python_awair"],
"zeroconf": [
{
"type": "_http._tcp.local.",
"name": "awair*"
}
]
}

4
homeassistant/components/awair/sensor.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
from python_awair.air_data import AirData
from python_awair.devices import AwairDevice
from python_awair.devices import AwairBaseDevice
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
@ -76,7 +76,7 @@ class AwairSensor(CoordinatorEntity[AwairDataUpdateCoordinator], SensorEntity):
def __init__(
self,
device: AwairDevice,
device: AwairBaseDevice,
coordinator: AwairDataUpdateCoordinator,
description: AwairSensorEntityDescription,
) -> None:

32
homeassistant/components/awair/strings.json
View File

@ -1,29 +1,49 @@
{
"config": {
"step": {
"user": {
"description": "You must register for an Awair developer access token at: https://developer.getawair.com/onboard/login",
"cloud": {
"description": "You must register for an Awair developer access token at: {url}",
"data": {
"access_token": "[%key:common::config_flow::data::access_token%]",
"email": "[%key:common::config_flow::data::email%]"
}
},
"local": {
"data": {
"host": "[%key:common::config_flow::data::ip%]"
},
"description": "Awair Local API must be enabled following these steps: {url}"
},
"reauth_confirm": {
"description": "Please re-enter your Awair developer access token.",
"data": {
"access_token": "[%key:common::config_flow::data::access_token%]",
"email": "[%key:common::config_flow::data::email%]"
}
},
"discovery_confirm": {
"description": "Do you want to setup {model} ({device_id})?"
},
"user": {
"menu_options": {
"cloud": "Connect via the cloud",
"local": "Connect locally (preferred)"
},
"description": "Pick local for the best experience. Only use cloud if the device is not connected to the same network as Home Assistant, or if you have a legacy device."
}
},
"error": {
"invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
"unknown": "[%key:common::config_flow::error::unknown%]",
"unreachable": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]",
"already_configured_device": "[%key:common::config_flow::abort::already_configured_device%]",
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"unreachable": "[%key:common::config_flow::error::cannot_connect%]"
},
"flow_title": "{model} ({device_id})"
}
}

4
homeassistant/config_entries.py
View File

@ -1297,6 +1297,8 @@ class ConfigFlow(data_entry_flow.FlowHandler):
self,
updates: dict[str, Any] | None = None,
reload_on_update: bool = True,
*,
error: str = "already_configured",
) -> None:
"""Abort if the unique ID is already configured."""
if self.unique_id is None:
@ -1332,7 +1334,7 @@ class ConfigFlow(data_entry_flow.FlowHandler):
self.hass.async_create_task(
self.hass.config_entries.async_reload(entry.entry_id)
)
raise data_entry_flow.AbortFlow("already_configured")
raise data_entry_flow.AbortFlow(error)
async def async_set_unique_id(
self, unique_id: str | None = None, *, raise_on_progress: bool = True

4
homeassistant/generated/zeroconf.py
View File

@ -183,6 +183,10 @@ ZEROCONF = {
}
],
"_http._tcp.local.": [
{
"domain": "awair",
"name": "awair*"
},
{
"domain": "bosch_shc",
"name": "bosch shc*"

73
tests/components/awair/conftest.py Normal file
View File

@ -0,0 +1,73 @@
"""Fixtures for testing Awair integration."""
import json
import pytest
from tests.common import load_fixture
@pytest.fixture(name="cloud_devices", scope="session")
def cloud_devices_fixture():
"""Fixture representing devices returned by Awair Cloud API."""
return json.loads(load_fixture("awair/cloud_devices.json"))
@pytest.fixture(name="local_devices", scope="session")
def local_devices_fixture():
"""Fixture representing devices returned by Awair local API."""
return json.loads(load_fixture("awair/local_devices.json"))
@pytest.fixture(name="gen1_data", scope="session")
def gen1_data_fixture():
"""Fixture representing data returned from Gen1 Awair device."""
return json.loads(load_fixture("awair/awair.json"))
@pytest.fixture(name="gen2_data", scope="session")
def gen2_data_fixture():
"""Fixture representing data returned from Gen2 Awair device."""
return json.loads(load_fixture("awair/awair-r2.json"))
@pytest.fixture(name="glow_data", scope="session")
def glow_data_fixture():
"""Fixture representing data returned from Awair glow device."""
return json.loads(load_fixture("awair/glow.json"))
@pytest.fixture(name="mint_data", scope="session")
def mint_data_fixture():
"""Fixture representing data returned from Awair mint device."""
return json.loads(load_fixture("awair/mint.json"))
@pytest.fixture(name="no_devices", scope="session")
def no_devicess_fixture():
"""Fixture representing when no devices are found in Awair's cloud API."""
return json.loads(load_fixture("awair/no_devices.json"))
@pytest.fixture(name="awair_offline", scope="session")
def awair_offline_fixture():
"""Fixture representing when Awair devices are offline."""
return json.loads(load_fixture("awair/awair-offline.json"))
@pytest.fixture(name="omni_data", scope="session")
def omni_data_fixture():
"""Fixture representing data returned from Awair omni device."""
return json.loads(load_fixture("awair/omni.json"))
@pytest.fixture(name="user", scope="session")
def user_fixture():
"""Fixture representing the User object returned from Awair's Cloud API."""
return json.loads(load_fixture("awair/user.json"))
@pytest.fixture(name="local_data", scope="session")
def local_data_fixture():
"""Fixture representing data returned from Awair local device."""
return json.loads(load_fixture("awair/awair-local.json"))

31
tests/components/awair/const.py
View File

@ -1,20 +1,19 @@
"""Constants used in Awair tests."""
import json
from homeassistant.const import CONF_ACCESS_TOKEN
from tests.common import load_fixture
from homeassistant.components import zeroconf
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST
AWAIR_UUID = "awair_24947"
CONFIG = {CONF_ACCESS_TOKEN: "12345"}
UNIQUE_ID = "foo@bar.com"
DEVICES_FIXTURE = json.loads(load_fixture("awair/devices.json"))
GEN1_DATA_FIXTURE = json.loads(load_fixture("awair/awair.json"))
GEN2_DATA_FIXTURE = json.loads(load_fixture("awair/awair-r2.json"))
GLOW_DATA_FIXTURE = json.loads(load_fixture("awair/glow.json"))
MINT_DATA_FIXTURE = json.loads(load_fixture("awair/mint.json"))
NO_DEVICES_FIXTURE = json.loads(load_fixture("awair/no_devices.json"))
OFFLINE_FIXTURE = json.loads(load_fixture("awair/awair-offline.json"))
OMNI_DATA_FIXTURE = json.loads(load_fixture("awair/omni.json"))
USER_FIXTURE = json.loads(load_fixture("awair/user.json"))
CLOUD_CONFIG = {CONF_ACCESS_TOKEN: "12345"}
LOCAL_CONFIG = {CONF_HOST: "192.0.2.5"}
CLOUD_UNIQUE_ID = "foo@bar.com"
LOCAL_UNIQUE_ID = "00:B0:D0:63:C2:26"
ZEROCONF_DISCOVERY = zeroconf.ZeroconfServiceInfo(
host="192.0.2.5",
addresses=["192.0.2.5"],
hostname="mock_hostname",
name="awair12345",
port=None,
type="_http._tcp.local.",
properties={},
)

17
tests/components/awair/fixtures/awair-local.json Normal file
View File

@ -0,0 +1,17 @@
{
"timestamp": "2022-08-11T05:04:12.108Z",
"score": 94,
"dew_point": 14.47,
"temp": 23.64,
"humid": 56.45,
"abs_humid": 12.0,
"co2": 426,
"co2_est": 489,
"co2_est_baseline": 37021,
"voc": 149,
"voc_baseline": 37783,
"voc_h2_raw": 26,
"voc_ethanol_raw": 37,
"pm25": 2,
"pm10_est": 3
}

0
tests/components/awair/fixtures/devices.json → tests/components/awair/fixtures/cloud_devices.json
View File

16
tests/components/awair/fixtures/local_devices.json Normal file
View File

@ -0,0 +1,16 @@
{
"device_uuid": "awair-element_24947",
"wifi_mac": "00:B0:D0:63:C2:26",
"ssid": "Internet of Things",
"ip": "192.0.2.5",
"netmask": "255.255.255.0",
"gateway": "none",
"fw_version": "1.2.8",
"timezone": "America/Los_Angeles",
"display": "score",
"led": {
"mode": "auto",
"brightness": 179
},
"voc_feature_set": 34
}

251
tests/components/awair/test_config_flow.py
View File

@ -1,99 +1,143 @@
"""Define tests for the Awair config flow."""
from unittest.mock import Mock, patch
from unittest.mock import patch
from aiohttp.client_exceptions import ClientConnectorError
from python_awair.exceptions import AuthError, AwairError
from homeassistant import data_entry_flow
from homeassistant.components.awair.const import DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_HOST
from homeassistant.core import HomeAssistant
from .const import CONFIG, DEVICES_FIXTURE, NO_DEVICES_FIXTURE, UNIQUE_ID, USER_FIXTURE
from .const import (
CLOUD_CONFIG,
CLOUD_UNIQUE_ID,
LOCAL_CONFIG,
LOCAL_UNIQUE_ID,
ZEROCONF_DISCOVERY,
)
from tests.common import MockConfigEntry
async def test_show_form(hass):
async def test_show_form(hass: HomeAssistant):
"""Test that the form is served with no input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["type"] == data_entry_flow.FlowResultType.MENU
assert result["step_id"] == SOURCE_USER
async def test_invalid_access_token(hass):
async def test_invalid_access_token(hass: HomeAssistant):
"""Test that errors are shown when the access token is invalid."""
with patch("python_awair.AwairClient.query", side_effect=AuthError()):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CLOUD_CONFIG
)
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "cloud"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
CLOUD_CONFIG,
)
assert result["errors"] == {CONF_ACCESS_TOKEN: "invalid_access_token"}
async def test_unexpected_api_error(hass):
async def test_unexpected_api_error(hass: HomeAssistant):
"""Test that we abort on generic errors."""
with patch("python_awair.AwairClient.query", side_effect=AwairError()):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CLOUD_CONFIG
)
assert result["type"] == "abort"
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "cloud"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "unknown"
async def test_duplicate_error(hass):
async def test_duplicate_error(hass: HomeAssistant, user, cloud_devices):
"""Test that errors are shown when adding a duplicate config."""
with patch(
"python_awair.AwairClient.query", side_effect=[USER_FIXTURE, DEVICES_FIXTURE]
), patch(
"homeassistant.components.awair.sensor.async_setup_entry",
return_value=True,
"python_awair.AwairClient.query",
side_effect=[user, cloud_devices],
):
MockConfigEntry(domain=DOMAIN, unique_id=UNIQUE_ID, data=CONFIG).add_to_hass(
hass
MockConfigEntry(
domain=DOMAIN, unique_id=CLOUD_UNIQUE_ID, data=CLOUD_CONFIG
).add_to_hass(hass)
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CLOUD_CONFIG
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "cloud"},
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured_account"
async def test_no_devices_error(hass):
async def test_no_devices_error(hass: HomeAssistant, user, no_devices):
"""Test that errors are shown when the API returns no devices."""
with patch(
"python_awair.AwairClient.query", side_effect=[USER_FIXTURE, NO_DEVICES_FIXTURE]
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
with patch("python_awair.AwairClient.query", side_effect=[user, no_devices]):
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CLOUD_CONFIG
)
assert result["type"] == "abort"
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "cloud"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_devices_found"
async def test_reauth(hass: HomeAssistant) -> None:
async def test_reauth(hass: HomeAssistant, user, cloud_devices) -> None:
"""Test reauth flow."""
mock_config = MockConfigEntry(
domain=DOMAIN, unique_id=UNIQUE_ID, data={**CONFIG, CONF_ACCESS_TOKEN: "blah"}
domain=DOMAIN,
unique_id=CLOUD_UNIQUE_ID,
data={**CLOUD_CONFIG, CONF_ACCESS_TOKEN: "blah"},
)
mock_config.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH, "unique_id": UNIQUE_ID},
data={**CONFIG, CONF_ACCESS_TOKEN: "blah"},
context={"source": SOURCE_REAUTH, "unique_id": CLOUD_UNIQUE_ID},
data={**CLOUD_CONFIG, CONF_ACCESS_TOKEN: "blah"},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
@ -102,7 +146,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
with patch("python_awair.AwairClient.query", side_effect=AuthError()):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG,
user_input=CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
@ -110,11 +154,12 @@ async def test_reauth(hass: HomeAssistant) -> None:
assert result["errors"] == {CONF_ACCESS_TOKEN: "invalid_access_token"}
with patch(
"python_awair.AwairClient.query", side_effect=[USER_FIXTURE, DEVICES_FIXTURE]
"python_awair.AwairClient.query",
side_effect=[user, cloud_devices],
), patch("homeassistant.components.awair.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG,
user_input=CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
@ -124,14 +169,16 @@ async def test_reauth(hass: HomeAssistant) -> None:
async def test_reauth_error(hass: HomeAssistant) -> None:
"""Test reauth flow."""
mock_config = MockConfigEntry(
domain=DOMAIN, unique_id=UNIQUE_ID, data={**CONFIG, CONF_ACCESS_TOKEN: "blah"}
domain=DOMAIN,
unique_id=CLOUD_UNIQUE_ID,
data={**CLOUD_CONFIG, CONF_ACCESS_TOKEN: "blah"},
)
mock_config.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH, "unique_id": UNIQUE_ID},
data={**CONFIG, CONF_ACCESS_TOKEN: "blah"},
context={"source": SOURCE_REAUTH, "unique_id": CLOUD_UNIQUE_ID},
data={**CLOUD_CONFIG, CONF_ACCESS_TOKEN: "blah"},
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
@ -140,27 +187,127 @@ async def test_reauth_error(hass: HomeAssistant) -> None:
with patch("python_awair.AwairClient.query", side_effect=AwairError()):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONFIG,
user_input=CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "unknown"
async def test_create_entry(hass):
"""Test overall flow."""
async def test_create_cloud_entry(hass: HomeAssistant, user, cloud_devices):
"""Test overall flow when using cloud api."""
with patch(
"python_awair.AwairClient.query", side_effect=[USER_FIXTURE, DEVICES_FIXTURE]
"python_awair.AwairClient.query",
side_effect=[user, cloud_devices],
), patch(
"homeassistant.components.awair.sensor.async_setup_entry",
"homeassistant.components.awair.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CONFIG
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=CLOUD_CONFIG
)
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "cloud"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
CLOUD_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "foo@bar.com (32406)"
assert result["data"][CONF_ACCESS_TOKEN] == CONFIG[CONF_ACCESS_TOKEN]
assert result["result"].unique_id == UNIQUE_ID
assert result["title"] == "foo@bar.com"
assert result["data"][CONF_ACCESS_TOKEN] == CLOUD_CONFIG[CONF_ACCESS_TOKEN]
assert result["result"].unique_id == CLOUD_UNIQUE_ID
async def test_create_local_entry(hass: HomeAssistant, local_devices):
"""Test overall flow when using local API."""
with patch("python_awair.AwairClient.query", side_effect=[local_devices]), patch(
"homeassistant.components.awair.async_setup_entry",
return_value=True,
):
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=LOCAL_CONFIG
)
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "local"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
LOCAL_CONFIG,
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Awair Element (24947)"
assert result["data"][CONF_HOST] == LOCAL_CONFIG[CONF_HOST]
assert result["result"].unique_id == LOCAL_UNIQUE_ID
async def test_create_local_entry_awair_error(hass: HomeAssistant):
"""Test overall flow when using local API and device is returns error."""
with patch(
"python_awair.AwairClient.query",
side_effect=AwairError(),
):
menu_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=LOCAL_CONFIG
)
form_step = await hass.config_entries.flow.async_configure(
menu_step["flow_id"],
{"next_step_id": "local"},
)
result = await hass.config_entries.flow.async_configure(
form_step["flow_id"],
LOCAL_CONFIG,
)
# User is returned to form to try again
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "local"
async def test_create_zeroconf_entry(hass: HomeAssistant, local_devices):
"""Test overall flow when using discovery."""
with patch("python_awair.AwairClient.query", side_effect=[local_devices]), patch(
"homeassistant.components.awair.async_setup_entry",
return_value=True,
):
confirm_step = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_DISCOVERY
)
result = await hass.config_entries.flow.async_configure(
confirm_step["flow_id"],
{},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == "Awair Element (24947)"
assert result["data"][CONF_HOST] == ZEROCONF_DISCOVERY.host
assert result["result"].unique_id == LOCAL_UNIQUE_ID
async def test_unsuccessful_create_zeroconf_entry(hass: HomeAssistant):
"""Test overall flow when using discovery and device is unreachable."""
with patch(
"python_awair.AwairClient.query",
side_effect=ClientConnectorError(Mock(), OSError()),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_ZEROCONF}, data=ZEROCONF_DISCOVERY
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT

89
tests/components/awair/test_sensor.py
View File

@ -26,21 +26,16 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_component import async_update_entity
from .const import (
AWAIR_UUID,
CONFIG,
DEVICES_FIXTURE,
GEN1_DATA_FIXTURE,
GEN2_DATA_FIXTURE,
GLOW_DATA_FIXTURE,
MINT_DATA_FIXTURE,
OFFLINE_FIXTURE,
OMNI_DATA_FIXTURE,
UNIQUE_ID,
USER_FIXTURE,
CLOUD_CONFIG,
CLOUD_UNIQUE_ID,
LOCAL_CONFIG,
LOCAL_UNIQUE_ID,
)
from tests.common import MockConfigEntry
@ -50,10 +45,10 @@ SENSOR_TYPES_MAP = {
}
async def setup_awair(hass, fixtures):
async def setup_awair(hass: HomeAssistant, fixtures, unique_id, data):
"""Add Awair devices to hass, using specified fixtures for data."""
entry = MockConfigEntry(domain=DOMAIN, unique_id=UNIQUE_ID, data=CONFIG)
entry = MockConfigEntry(domain=DOMAIN, unique_id=unique_id, data=data)
with patch("python_awair.AwairClient.query", side_effect=fixtures):
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
@ -61,7 +56,12 @@ async def setup_awair(hass, fixtures):
def assert_expected_properties(
hass, registry, name, unique_id, state_value, attributes
hass: HomeAssistant,
registry: er.RegistryEntry,
name,
unique_id,
state_value,
attributes: dict,
):
"""Assert expected properties from a dict."""
@ -74,11 +74,11 @@ def assert_expected_properties(
assert state.attributes.get(attr) == value
async def test_awair_gen1_sensors(hass):
async def test_awair_gen1_sensors(hass: HomeAssistant, user, cloud_devices, gen1_data):
"""Test expected sensors on a 1st gen Awair."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, GEN1_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, gen1_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -166,11 +166,11 @@ async def test_awair_gen1_sensors(hass):
assert hass.states.get("sensor.living_room_illuminance") is None
async def test_awair_gen2_sensors(hass):
async def test_awair_gen2_sensors(hass: HomeAssistant, user, cloud_devices, gen2_data):
"""Test expected sensors on a 2nd gen Awair."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, GEN2_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, gen2_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -199,11 +199,28 @@ async def test_awair_gen2_sensors(hass):
assert hass.states.get("sensor.living_room_pm10") is None
async def test_awair_mint_sensors(hass):
async def test_local_awair_sensors(hass: HomeAssistant, local_devices, local_data):
"""Test expected sensors on a local Awair."""
fixtures = [local_devices, local_data]
await setup_awair(hass, fixtures, LOCAL_UNIQUE_ID, LOCAL_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
hass,
registry,
"sensor.awair_score",
f"{local_devices['device_uuid']}_{SENSOR_TYPES_MAP[API_SCORE].unique_id_tag}",
"94",
{},
)
async def test_awair_mint_sensors(hass: HomeAssistant, user, cloud_devices, mint_data):
"""Test expected sensors on an Awair mint."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, MINT_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, mint_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -240,11 +257,11 @@ async def test_awair_mint_sensors(hass):
assert hass.states.get("sensor.living_room_carbon_dioxide") is None
async def test_awair_glow_sensors(hass):
async def test_awair_glow_sensors(hass: HomeAssistant, user, cloud_devices, glow_data):
"""Test expected sensors on an Awair glow."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, GLOW_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, glow_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -260,11 +277,11 @@ async def test_awair_glow_sensors(hass):
assert hass.states.get("sensor.living_room_pm2_5") is None
async def test_awair_omni_sensors(hass):
async def test_awair_omni_sensors(hass: HomeAssistant, user, cloud_devices, omni_data):
"""Test expected sensors on an Awair omni."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, OMNI_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, omni_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -295,11 +312,11 @@ async def test_awair_omni_sensors(hass):
)
async def test_awair_offline(hass):
async def test_awair_offline(hass: HomeAssistant, user, cloud_devices, awair_offline):
"""Test expected behavior when an Awair is offline."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, OFFLINE_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, awair_offline]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
# The expected behavior is that we won't have any sensors
# if the device is not online when we set it up. python_awair
@ -313,11 +330,13 @@ async def test_awair_offline(hass):
assert hass.states.get("sensor.living_room_awair_score") is None
async def test_awair_unavailable(hass):
async def test_awair_unavailable(
hass: HomeAssistant, user, cloud_devices, gen1_data, awair_offline
):
"""Test expected behavior when an Awair becomes offline later."""
fixtures = [USER_FIXTURE, DEVICES_FIXTURE, GEN1_DATA_FIXTURE]
await setup_awair(hass, fixtures)
fixtures = [user, cloud_devices, gen1_data]
await setup_awair(hass, fixtures, CLOUD_UNIQUE_ID, CLOUD_CONFIG)
registry = er.async_get(hass)
assert_expected_properties(
@ -329,7 +348,7 @@ async def test_awair_unavailable(hass):
{},
)
with patch("python_awair.AwairClient.query", side_effect=OFFLINE_FIXTURE):
with patch("python_awair.AwairClient.query", side_effect=awair_offline):
await async_update_entity(hass, "sensor.living_room_awair_score")
assert_expected_properties(
hass,