evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add integration for iotty Smart Home (#103073)

Browse source 
* Initial import 0.0.2

* Fixes to URL, and removed commits

* Initial import 0.0.2

* Fixes to URL, and removed commits

* Added first test for iotty

* First release

* Reviewers request #1
- Removed clutter
- Added support for new naming convention for IottySmartSwitch entity

* Removed commmented code

* Some modifications

* Modified REST EP for iotty CloudApi

* Initial import 0.0.2

* Fixes to URL, and removed commits

* Added first test for iotty

* First release

* Rebased and resolved conflicts

* Reviewers request #1
- Removed clutter
- Added support for new naming convention for IottySmartSwitch entity

* Removed commmented code

* Some modifications

* Modified REST EP for iotty CloudApi

* Removed empty entries in manifest.json

* Added test_config_flow

* Fix as requested by @edenhaus

* Added test_init

* Removed comments, added one assert

* Added TEST_CONFIG_FLOW

* Added test for STORE_ENTITY

* Increased code coverage

* Full coverage for api.py

* Added tests for switch component

* Converted INFO logs onto DEBUG logs

* Removed .gitignore from commits

* Modifications to SWITCH.PY

* Initial import 0.0.2

* Fixes to URL, and removed commits

* Added first test for iotty

* First release

* Rebased and resolved conflicts

* Fixed conflicts

* Reviewers request #1
- Removed clutter
- Added support for new naming convention for IottySmartSwitch entity

* Removed commmented code

* Some modifications

* Modified REST EP for iotty CloudApi

* Removed empty entries in manifest.json

* Added test_config_flow

* Some modifications

* Fix as requested by @edenhaus

* Added test_init

* Removed comments, added one assert

* Added TEST_CONFIG_FLOW

* Added test for STORE_ENTITY

* Increased code coverage

* Full coverage for api.py

* Added tests for switch component

* Converted INFO logs onto DEBUG logs

* Removed .gitignore from commits

* Modifications to SWITCH.PY

* Fixed tests for SWITCH

* First working implementation of Coordinator

* Increased code coverage

* Full code coverage

* Missing a line in testing

* Update homeassistant/components/iotty/__init__.py

Co-authored-by: Robert Resch <robert@resch.dev>

* Update homeassistant/components/iotty/__init__.py

Co-authored-by: Robert Resch <robert@resch.dev>

* Modified coordinator as per request by edenhaus

* use coordinator entities for switches

* move platforms to constants

* fix whitespace with ruff-format

* correct iotty entry in application_credentials list

* minor style improvements

* refactor function name

* handle new and deleted devices

* improve code for adding devices after first initialization

* use typed config entry instead of adding known devices to hass.data

* improve iotty entity removal

* test listeners update cycle

* handle iotty as devices and not only as entities

* fix test typing for mock config entry

* test with fewer mocks for an integration test style opposed to the previous unit test style

* remove useless tests and add more integration style tests

* check if device_to_remove is None

* integration style tests for turning switches on and off

* remove redundant coordinator tests

* check device status after issuing command in tests

* remove unused fixtures

* add strict typing for iotty

* additional asserts and named snapshots in tests

* fix mypy issues after enabling strict typing

* upgrade iottycloud version to 0.1.3

* move coordinator to runtime_data

* remove entity name

* fix typing issues

* coding style fixes

* improve tests coding style and assertion targets

* test edge cases when apis are not working

* improve tests comments and assertions

---------

Co-authored-by: Robert Resch <robert@resch.dev>
Co-authored-by: Shapour Nemati <shapour.nemati@iotty.com>
Co-authored-by: Erik Montnemery <erik@montnemery.com>
Co-authored-by: shapournemati-iotty <130070037+shapournemati-iotty@users.noreply.github.com>
This commit is contained in:
Paolo Burgio 2024-07-19 12:10:39 +02:00 • committed by GitHub
parent 4620a54582
commit c1c5cff993
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 1336 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -255,6 +255,7 @@ homeassistant.components.integration.*
homeassistant.components.intent.*
homeassistant.components.intent_script.*
homeassistant.components.ios.*
homeassistant.components.iotty.*
homeassistant.components.ipp.*
homeassistant.components.iqvia.*
homeassistant.components.islamic_prayer_times.*

2
CODEOWNERS
View file

@ -695,6 +695,8 @@ build.json @home-assistant/supervisor
/tests/components/ios/ @robbiet480
/homeassistant/components/iotawatt/ @gtdiehl @jyavenard
/tests/components/iotawatt/ @gtdiehl @jyavenard
/homeassistant/components/iotty/ @pburgio
/tests/components/iotty/ @pburgio
/homeassistant/components/iperf3/ @rohankapoorcom
/homeassistant/components/ipma/ @dgomes
/tests/components/ipma/ @dgomes

56
homeassistant/components/iotty/__init__.py Normal file
View file

@ -0,0 +1,56 @@
"""The iotty integration."""
from __future__ import annotations
from dataclasses import dataclass
import logging
from iottycloud.device import Device
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from . import coordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS: list[Platform] = [Platform.SWITCH]
type IottyConfigEntry = ConfigEntry[IottyConfigEntryData]
@dataclass
class IottyConfigEntryData:
"""Contains config entry data for iotty."""
known_devices: set[Device]
coordinator: coordinator.IottyDataUpdateCoordinator
async def async_setup_entry(hass: HomeAssistant, entry: IottyConfigEntry) -> bool:
"""Set up iotty from a config entry."""
_LOGGER.debug("async_setup_entry entry_id=%s", entry.entry_id)
implementation = await async_get_config_entry_implementation(hass, entry)
session = OAuth2Session(hass, entry, implementation)
data_update_coordinator = coordinator.IottyDataUpdateCoordinator(
hass, entry, session
)
entry.runtime_data = IottyConfigEntryData(set(), data_update_coordinator)
await data_update_coordinator.async_config_entry_first_refresh()
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

40
homeassistant/components/iotty/api.py Normal file
View file

@ -0,0 +1,40 @@
"""API for iotty bound to Home Assistant OAuth."""
from __future__ import annotations
from typing import Any
from aiohttp import ClientSession
from iottycloud.cloudapi import CloudApi
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
OAUTH2_CLIENT_ID = "hass-iotty"
IOTTYAPI_BASE = "https://homeassistant.iotty.com/"
class IottyProxy(CloudApi):
"""Provide iotty authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: HomeAssistant,
websession: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None:
"""Initialize iotty auth."""
super().__init__(websession, IOTTYAPI_BASE, OAUTH2_CLIENT_ID)
if oauth_session is None:
raise ValueError("oauth_session")
self._oauth_session = oauth_session
self._hass = hass
async def async_get_access_token(self) -> Any:
"""Return a valid access token."""
if not self._oauth_session.valid_token:
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]

17
homeassistant/components/iotty/application_credentials.py Normal file
View file

@ -0,0 +1,17 @@
"""Application credentials platform for iotty."""
from __future__ import annotations
from homeassistant.components.application_credentials import AuthorizationServer
from homeassistant.core import HomeAssistant
OAUTH2_AUTHORIZE = "https://auth.iotty.com/.auth/oauth2/login"
OAUTH2_TOKEN = "https://auth.iotty.com/.auth/oauth2/token"
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
"""Return authorization server."""
return AuthorizationServer(
authorize_url=OAUTH2_AUTHORIZE,
token_url=OAUTH2_TOKEN,
)

22
homeassistant/components/iotty/config_flow.py Normal file
View file

@ -0,0 +1,22 @@
"""Config flow for iotty."""
from __future__ import annotations
import logging
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle iotty OAuth2 authentication."""
DOMAIN = DOMAIN
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)

5
homeassistant/components/iotty/const.py Normal file
View file

@ -0,0 +1,5 @@
"""Constants for the iotty integration."""
from __future__ import annotations
DOMAIN = "iotty"

108
homeassistant/components/iotty/coordinator.py Normal file
View file

@ -0,0 +1,108 @@
"""DataUpdateCoordinator for iotty."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import logging
from iottycloud.device import Device
from iottycloud.verbs import RESULT, STATUS
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client, device_registry as dr
from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import api
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
UPDATE_INTERVAL = timedelta(seconds=30)
@dataclass
class IottyData:
"""iotty data stored in the DataUpdateCoordinator."""
devices: list[Device]
class IottyDataUpdateCoordinator(DataUpdateCoordinator[IottyData]):
"""Class to manage fetching Iotty data."""
config_entry: ConfigEntry
_entities: dict[str, Entity]
_devices: list[Device]
_device_registry: dr.DeviceRegistry
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, session: OAuth2Session
) -> None:
"""Initialize the coordinator."""
_LOGGER.debug("Initializing iotty data update coordinator")
super().__init__(
hass,
_LOGGER,
name=f"{DOMAIN}_coordinator",
update_interval=UPDATE_INTERVAL,
)
self.config_entry = entry
self._entities = {}
self._devices = []
self.iotty = api.IottyProxy(
hass, aiohttp_client.async_get_clientsession(hass), session
)
self._device_registry = dr.async_get(hass)
async def async_config_entry_first_refresh(self) -> None:
"""Override the first refresh to also fetch iotty devices list."""
_LOGGER.debug("Fetching devices list from iottyCloud")
self._devices = await self.iotty.get_devices()
_LOGGER.debug("There are %d devices", len(self._devices))
await super().async_config_entry_first_refresh()
async def _async_update_data(self) -> IottyData:
"""Fetch data from iottyCloud device."""
_LOGGER.debug("Fetching devices status from iottyCloud")
current_devices = await self.iotty.get_devices()
removed_devices = [
d
for d in self._devices
if not any(x.device_id == d.device_id for x in current_devices)
]
for removed_device in removed_devices:
device_to_remove = self._device_registry.async_get_device(
{(DOMAIN, removed_device.device_id)}
)
if device_to_remove is not None:
self._device_registry.async_remove_device(device_to_remove.id)
self._devices = current_devices
for device in self._devices:
res = await self.iotty.get_status(device.device_id)
json = res.get(RESULT, {})
if (
not isinstance(res, dict)
or RESULT not in res
or not isinstance(json := res[RESULT], dict)
or not (status := json.get(STATUS))
):
_LOGGER.warning("Unable to read status for device %s", device.device_id)
else:
_LOGGER.debug(
"Retrieved status: '%s' for device %s", status, device.device_id
)
device.update_status(status)
return IottyData(self._devices)

11
homeassistant/components/iotty/manifest.json Normal file
View file

@ -0,0 +1,11 @@
{
"domain": "iotty",
"name": "iotty",
"codeowners": ["@pburgio"],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/iotty",
"integration_type": "device",
"iot_class": "cloud_polling",
"requirements": ["iottycloud==0.1.3"]
}

21
homeassistant/components/iotty/strings.json Normal file
View file

@ -0,0 +1,21 @@
{
"config": {
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
}
}

165
homeassistant/components/iotty/switch.py Normal file
View file

@ -0,0 +1,165 @@
"""Implement a iotty Light Switch Device."""
from __future__ import annotations
import logging
from typing import Any, cast
from iottycloud.device import Device
from iottycloud.lightswitch import LightSwitch
from iottycloud.verbs import LS_DEVICE_TYPE_UID
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import IottyConfigEntry
from .api import IottyProxy
from .const import DOMAIN
from .coordinator import IottyDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: IottyConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Activate the iotty LightSwitch component."""
_LOGGER.debug("Setup SWITCH entry id is %s", config_entry.entry_id)
coordinator = config_entry.runtime_data.coordinator
entities = [
IottyLightSwitch(
coordinator=coordinator, iotty_cloud=coordinator.iotty, iotty_device=d
)
for d in coordinator.data.devices
if d.device_type == LS_DEVICE_TYPE_UID
if (isinstance(d, LightSwitch))
]
_LOGGER.debug("Found %d LightSwitches", len(entities))
async_add_entities(entities)
known_devices: set = config_entry.runtime_data.known_devices
for known_device in coordinator.data.devices:
if known_device.device_type == LS_DEVICE_TYPE_UID:
known_devices.add(known_device)
@callback
def async_update_data() -> None:
"""Handle updated data from the API endpoint."""
if not coordinator.last_update_success:
return None
devices = coordinator.data.devices
entities = []
known_devices: set = config_entry.runtime_data.known_devices
# Add entities for devices which we've not yet seen
for device in devices:
if (
any(d.device_id == device.device_id for d in known_devices)
or device.device_type != LS_DEVICE_TYPE_UID
):
continue
iotty_entity = IottyLightSwitch(
coordinator=coordinator,
iotty_cloud=coordinator.iotty,
iotty_device=LightSwitch(
device.device_id,
device.serial_number,
device.device_type,
device.device_name,
),
)
entities.extend([iotty_entity])
known_devices.add(device)
async_add_entities(entities)
# Add a subscriber to the coordinator to discover new devices
coordinator.async_add_listener(async_update_data)
class IottyLightSwitch(SwitchEntity, CoordinatorEntity[IottyDataUpdateCoordinator]):
"""Haas entity class for iotty LightSwitch."""
_attr_has_entity_name = True
_attr_name = None
_attr_entity_category = None
_attr_device_class = SwitchDeviceClass.SWITCH
_iotty_cloud: IottyProxy
_iotty_device: LightSwitch
def __init__(
self,
coordinator: IottyDataUpdateCoordinator,
iotty_cloud: IottyProxy,
iotty_device: LightSwitch,
) -> None:
"""Initialize the LightSwitch device."""
super().__init__(coordinator=coordinator)
_LOGGER.debug(
"Creating new SWITCH (%s) %s",
iotty_device.device_type,
iotty_device.device_id,
)
self._iotty_cloud = iotty_cloud
self._iotty_device = iotty_device
self._attr_unique_id = iotty_device.device_id
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
identifiers={(DOMAIN, cast(str, self._attr_unique_id))},
name=self._iotty_device.name,
manufacturer="iotty",
)
@property
def is_on(self) -> bool:
"""Return true if the LightSwitch is on."""
_LOGGER.debug(
"Retrieve device status for %s ? %s",
self._iotty_device.device_id,
self._iotty_device.is_on,
)
return self._iotty_device.is_on
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the LightSwitch on."""
_LOGGER.debug("[%s] Turning on", self._iotty_device.device_id)
await self._iotty_cloud.command(
self._iotty_device.device_id, self._iotty_device.cmd_turn_on()
)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the LightSwitch off."""
_LOGGER.debug("[%s] Turning off", self._iotty_device.device_id)
await self._iotty_cloud.command(
self._iotty_device.device_id, self._iotty_device.cmd_turn_off()
)
await self.coordinator.async_request_refresh()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
device: Device = next(
device
for device in self.coordinator.data.devices
if device.device_id == self._iotty_device.device_id
)
if isinstance(device, LightSwitch):
self._iotty_device.is_on = device.is_on
self.async_write_ha_state()

1
homeassistant/generated/application_credentials.py
View file

@ -14,6 +14,7 @@ APPLICATION_CREDENTIALS = [
"google_tasks",
"home_connect",
"husqvarna_automower",
"iotty",
"lametric",
"lyric",
"microbees",

1
homeassistant/generated/config_flows.py
View file

@ -269,6 +269,7 @@ FLOWS = {
"intellifire",
"ios",
"iotawatt",
"iotty",
"ipma",
"ipp",
"iqvia",

6
homeassistant/generated/integrations.json
View file

@ -2857,6 +2857,12 @@
"config_flow": true,
"iot_class": "local_polling"
},
"iotty": {
"name": "iotty",
"integration_type": "device",
"config_flow": true,
"iot_class": "cloud_polling"
},
"iperf3": {
"name": "Iperf3",
"integration_type": "hub",

10
mypy.ini
View file

@ -2306,6 +2306,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.iotty.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.ipp.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View file

@ -1166,6 +1166,9 @@ insteon-frontend-home-assistant==0.5.0
# homeassistant.components.intellifire
intellifire4py==2.2.2
# homeassistant.components.iotty
iottycloud==0.1.3
# homeassistant.components.iperf3
iperf3==0.1.11

3
requirements_test_all.txt
View file

@ -968,6 +968,9 @@ insteon-frontend-home-assistant==0.5.0
# homeassistant.components.intellifire
intellifire4py==2.2.2
# homeassistant.components.iotty
iottycloud==0.1.3
# homeassistant.components.isal
isal==1.6.1

1
tests/components/iotty/__init__.py Normal file
View file

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

180
tests/components/iotty/conftest.py Normal file
View file

@ -0,0 +1,180 @@
"""Fixtures for iotty integration tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from aiohttp import ClientSession
from iottycloud.device import Device
from iottycloud.lightswitch import LightSwitch
from iottycloud.verbs import LS_DEVICE_TYPE_UID, RESULT, STATUS, STATUS_OFF, STATUS_ON
import pytest
from homeassistant import setup
from homeassistant.components.iotty.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker, mock_aiohttp_client
CLIENT_ID = "client_id"
CLIENT_SECRET = "client_secret"
REDIRECT_URI = "https://example.com/auth/external/callback"
test_devices = [
Device("TestDevice0", "TEST_SERIAL_0", LS_DEVICE_TYPE_UID, "[TEST] Device Name 0"),
Device("TestDevice1", "TEST_SERIAL_1", LS_DEVICE_TYPE_UID, "[TEST] Device Name 1"),
]
ls_0 = LightSwitch(
"TestLS", "TEST_SERIAL_0", LS_DEVICE_TYPE_UID, "[TEST] Light switch 0"
)
ls_1 = LightSwitch(
"TestLS1", "TEST_SERIAL_1", LS_DEVICE_TYPE_UID, "[TEST] Light switch 1"
)
ls_2 = LightSwitch(
"TestLS2", "TEST_SERIAL_2", LS_DEVICE_TYPE_UID, "[TEST] Light switch 2"
)
test_ls = [ls_0, ls_1]
test_ls_one_removed = [ls_0]
test_ls_one_added = [
ls_0,
ls_1,
ls_2,
]
@pytest.fixture
async def local_oauth_impl(hass: HomeAssistant):
"""Local implementation."""
assert await setup.async_setup_component(hass, "auth", {})
return config_entry_oauth2_flow.LocalOAuth2Implementation(
hass, DOMAIN, "client_id", "client_secret", "authorize_url", "https://token.url"
)
@pytest.fixture
def aiohttp_client_session() -> None:
"""AIOHTTP client session."""
return ClientSession
@pytest.fixture
def mock_aioclient() -> Generator[AiohttpClientMocker, None, None]:
"""Fixture to mock aioclient calls."""
with mock_aiohttp_client() as mock_session:
yield mock_session
@pytest.fixture
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="IOTTY00001",
domain=DOMAIN,
data={
"auth_implementation": DOMAIN,
"token": {
"refresh_token": "REFRESH_TOKEN",
"access_token": "ACCESS_TOKEN_1",
"expires_in": 10,
"expires_at": 0,
"token_type": "bearer",
"random_other_data": "should_stay",
},
CONF_HOST: "127.0.0.1",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
CONF_PORT: 9123,
},
unique_id="IOTTY00001",
)
@pytest.fixture
def mock_config_entries_async_forward_entry_setup() -> Generator[AsyncMock, None, None]:
"""Mock async_forward_entry_setup."""
with patch(
"homeassistant.config_entries.ConfigEntries.async_forward_entry_setups"
) as mock_fn:
yield mock_fn
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Mock setting up a config entry."""
with patch(
"homeassistant.components.iotty.async_setup_entry", return_value=True
) as mock_setup:
yield mock_setup
@pytest.fixture
def mock_iotty() -> Generator[None, MagicMock, None]:
"""Mock IottyProxy."""
with patch(
"homeassistant.components.iotty.api.IottyProxy", autospec=True
) as iotty_mock:
yield iotty_mock
@pytest.fixture
def mock_coordinator() -> Generator[None, MagicMock, None]:
"""Mock IottyDataUpdateCoordinator."""
with patch(
"homeassistant.components.iotty.coordinator.IottyDataUpdateCoordinator",
autospec=True,
) as coordinator_mock:
yield coordinator_mock
@pytest.fixture
def mock_get_devices_nodevices() -> Generator[AsyncMock, None, None]:
"""Mock for get_devices, returning two objects."""
with patch("iottycloud.cloudapi.CloudApi.get_devices") as mock_fn:
yield mock_fn
@pytest.fixture
def mock_get_devices_twolightswitches() -> Generator[AsyncMock, None, None]:
"""Mock for get_devices, returning two objects."""
with patch(
"iottycloud.cloudapi.CloudApi.get_devices", return_value=test_ls
) as mock_fn:
yield mock_fn
@pytest.fixture
def mock_command_fn() -> Generator[AsyncMock, None, None]:
"""Mock for command."""
with patch("iottycloud.cloudapi.CloudApi.command", return_value=None) as mock_fn:
yield mock_fn
@pytest.fixture
def mock_get_status_filled_off() -> Generator[AsyncMock, None, None]:
"""Mock setting up a get_status."""
retval = {RESULT: {STATUS: STATUS_OFF}}
with patch(
"iottycloud.cloudapi.CloudApi.get_status", return_value=retval
) as mock_fn:
yield mock_fn
@pytest.fixture
def mock_get_status_filled() -> Generator[AsyncMock, None, None]:
"""Mock setting up a get_status."""
retval = {RESULT: {STATUS: STATUS_ON}}
with patch(
"iottycloud.cloudapi.CloudApi.get_status", return_value=retval
) as mock_fn:
yield mock_fn

126
tests/components/iotty/snapshots/test_switch.ambr Normal file
View file

@ -0,0 +1,126 @@
# serializer version: 1
# name: test_api_not_ok_entities_stay_the_same_as_before
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
])
# ---
# name: test_api_throws_response_entities_stay_the_same_as_before
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
])
# ---
# name: test_devices_creaction_ok[device]
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'iotty',
'TestLS',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'iotty',
'model': None,
'model_id': None,
'name': '[TEST] Light switch 0 (TEST_SERIAL_0)',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'suggested_area': None,
'sw_version': None,
'via_device_id': None,
})
# ---
# name: test_devices_creaction_ok[entity-ids]
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
])
# ---
# name: test_devices_creaction_ok[entity]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.test_light_switch_0_test_serial_0',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SwitchDeviceClass.SWITCH: 'switch'>,
'original_icon': None,
'original_name': None,
'platform': 'iotty',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'TestLS',
'unit_of_measurement': None,
})
# ---
# name: test_devices_creaction_ok[state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'switch',
'friendly_name': '[TEST] Light switch 0 (TEST_SERIAL_0)',
}),
'context': <ANY>,
'entity_id': 'switch.test_light_switch_0_test_serial_0',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_devices_deletion_ok
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
])
# ---
# name: test_devices_deletion_ok.1
list([
'switch.test_light_switch_0_test_serial_0',
])
# ---
# name: test_devices_insertion_ok
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
])
# ---
# name: test_devices_insertion_ok.1
list([
'switch.test_light_switch_0_test_serial_0',
'switch.test_light_switch_1_test_serial_1',
'switch.test_light_switch_2_test_serial_2',
])
# ---
# name: test_setup_entry_ok_nodevices
list([
])
# ---

82
tests/components/iotty/test_api.py Normal file
View file

@ -0,0 +1,82 @@
"""Unit tests for iottycloud API."""
from unittest.mock import patch
from aiohttp import ClientSession
import pytest
from homeassistant.components.iotty import api
from homeassistant.components.iotty.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
async def test_api_create_fail(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test API creation with no session."""
with pytest.raises(ValueError, match="websession"):
api.IottyProxy(hass, None, None)
with pytest.raises(ValueError, match="oauth_session"):
api.IottyProxy(hass, aioclient_mock, None)
async def test_api_create_ok(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
aiohttp_client_session: None,
local_oauth_impl: ClientSession,
) -> None:
"""Test API creation. We're checking that we can create an IottyProxy without raising."""
mock_config_entry.add_to_hass(hass)
assert mock_config_entry.data["auth_implementation"] is not None
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
api.IottyProxy(hass, aiohttp_client_session, local_oauth_impl)
@patch(
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.valid_token", False
)
async def test_api_getaccesstoken_tokennotvalid_reloadtoken(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_aioclient: None,
aiohttp_client_session: ClientSession,
) -> None:
"""Test getting access token.
If a request with an invalid token is made, a request for a new token is done,
and the resulting token is used for future calls.
"""
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
new_token = "ACCESS_TOKEN_1"
mock_aioclient.post(
"https://token.url", json={"access_token": new_token, "expires_in": 100}
)
mock_aioclient.post("https://example.com", status=201)
mock_config_entry.add_to_hass(hass)
oauth2_session = config_entry_oauth2_flow.OAuth2Session(
hass, mock_config_entry, local_oauth_impl
)
iotty = api.IottyProxy(hass, aiohttp_client_session, oauth2_session)
tok = await iotty.async_get_access_token()
assert tok == new_token

102
tests/components/iotty/test_config_flow.py Normal file
View file

@ -0,0 +1,102 @@
"""Test the iotty config flow."""
from http import HTTPStatus
from unittest.mock import AsyncMock, MagicMock
import multidict
import pytest
from homeassistant import config_entries
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.iotty.application_credentials import OAUTH2_TOKEN
from homeassistant.components.iotty.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.setup import async_setup_component
from .conftest import CLIENT_ID, CLIENT_SECRET, REDIRECT_URI
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
@pytest.fixture
async def setup_credentials(hass: HomeAssistant) -> None:
"""Fixture to setup application credentials component."""
await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, CLIENT_SECRET),
)
@pytest.fixture
def current_request_with_host(current_request: MagicMock) -> None:
"""Mock current request with a host header."""
new_headers = multidict.CIMultiDict(current_request.get.return_value.headers)
new_headers[config_entry_oauth2_flow.HEADER_FRONTEND_BASE] = "https://example.com"
current_request.get.return_value = current_request.get.return_value.clone(
headers=new_headers
)
async def test_config_flow_no_credentials(hass: HomeAssistant) -> None:
"""Test config flow base case with no credentials registered."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "missing_credentials"
@pytest.mark.usefixtures("current_request_with_host", "setup_credentials")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_setup_entry: AsyncMock,
) -> None:
"""Check full flow."""
await async_import_client_credential(
hass, DOMAIN, ClientCredential(CLIENT_ID, CLIENT_SECRET)
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER, "entry_id": DOMAIN}
)
assert result.get("type") == FlowResultType.EXTERNAL_STEP
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URI,
},
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == HTTPStatus.OK
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup_entry.mock_calls) == 1

73
tests/components/iotty/test_init.py Normal file
View file

@ -0,0 +1,73 @@
"""Tests for the iotty integration."""
from unittest.mock import MagicMock
from homeassistant.components.iotty.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
async def test_load_unload_coordinator_called(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_coordinator: MagicMock,
local_oauth_impl,
) -> None:
"""Test the configuration entry loading/unloading."""
mock_config_entry.add_to_hass(hass)
assert mock_config_entry.data["auth_implementation"] is not None
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.async_block_till_done()
await hass.config_entries.async_setup(mock_config_entry.entry_id)
mock_coordinator.assert_called_once()
assert mock_config_entry.state is ConfigEntryState.LOADED
method_call = mock_coordinator.method_calls[0]
name, _, _ = method_call
assert name == "().async_config_entry_first_refresh"
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_load_unload_iottyproxy_called(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_iotty: MagicMock,
local_oauth_impl,
mock_config_entries_async_forward_entry_setup,
) -> None:
"""Test the configuration entry loading/unloading."""
mock_config_entry.add_to_hass(hass)
assert mock_config_entry.data["auth_implementation"] is not None
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_iotty.assert_called_once()
assert mock_config_entry.state is ConfigEntryState.LOADED
method_call = mock_iotty.method_calls[0]
name, _, _ = method_call
assert name == "().get_devices"
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED

300
tests/components/iotty/test_switch.py Normal file
View file

@ -0,0 +1,300 @@
"""Unit tests the Hass SWITCH component."""
from aiohttp import ClientSession
from freezegun.api import FrozenDateTimeFactory
from iottycloud.verbs import RESULT, STATUS, STATUS_OFF, STATUS_ON
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.iotty.const import DOMAIN
from homeassistant.components.iotty.coordinator import UPDATE_INTERVAL
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
config_entry_oauth2_flow,
device_registry as dr,
entity_registry as er,
)
from .conftest import test_ls_one_added, test_ls_one_removed
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_turn_on_ok(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled_off,
mock_command_fn,
) -> None:
"""Issue a turnon command."""
entity_id = "switch.test_light_switch_0_test_serial_0"
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert (state := hass.states.get(entity_id))
assert state.state == STATUS_OFF
mock_get_status_filled_off.return_value = {RESULT: {STATUS: STATUS_ON}}
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
mock_command_fn.assert_called_once()
assert (state := hass.states.get(entity_id))
assert state.state == STATUS_ON
async def test_turn_off_ok(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
mock_command_fn,
) -> None:
"""Issue a turnoff command."""
entity_id = "switch.test_light_switch_0_test_serial_0"
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert (state := hass.states.get(entity_id))
assert state.state == STATUS_ON
mock_get_status_filled.return_value = {RESULT: {STATUS: STATUS_OFF}}
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
mock_command_fn.assert_called_once()
assert (state := hass.states.get(entity_id))
assert state.state == STATUS_OFF
async def test_setup_entry_ok_nodevices(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_status_filled,
snapshot: SnapshotAssertion,
mock_get_devices_nodevices,
) -> None:
"""Correctly setup, with no iotty Devices to add to Hass."""
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert hass.states.async_entity_ids_count() == 0
assert hass.states.async_entity_ids() == snapshot
async def test_devices_creaction_ok(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
snapshot: SnapshotAssertion,
) -> None:
"""Test iotty switch creation."""
entity_id = "switch.test_light_switch_0_test_serial_0"
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
assert (state := hass.states.get(entity_id))
assert state == snapshot(name="state")
assert (entry := entity_registry.async_get(entity_id))
assert entry == snapshot(name="entity")
assert entry.device_id
assert (device_entry := device_registry.async_get(entry.device_id))
assert device_entry == snapshot(name="device")
assert hass.states.async_entity_ids_count() == 2
assert hass.states.async_entity_ids() == snapshot(name="entity-ids")
async def test_devices_deletion_ok(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
snapshot: SnapshotAssertion,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test iotty switch deletion."""
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Should have two devices
assert hass.states.async_entity_ids_count() == 2
assert hass.states.async_entity_ids() == snapshot
mock_get_devices_twolightswitches.return_value = test_ls_one_removed
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should have one device
assert hass.states.async_entity_ids_count() == 1
assert hass.states.async_entity_ids() == snapshot
async def test_devices_insertion_ok(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
snapshot: SnapshotAssertion,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test iotty switch insertion."""
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Should have two devices
assert hass.states.async_entity_ids_count() == 2
assert hass.states.async_entity_ids() == snapshot
mock_get_devices_twolightswitches.return_value = test_ls_one_added
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should have three devices
assert hass.states.async_entity_ids_count() == 3
assert hass.states.async_entity_ids() == snapshot
async def test_api_not_ok_entities_stay_the_same_as_before(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
snapshot: SnapshotAssertion,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test case of incorrect response from iotty API on getting device status."""
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Should have two devices
assert hass.states.async_entity_ids_count() == 2
entity_ids = hass.states.async_entity_ids()
assert entity_ids == snapshot
mock_get_status_filled.return_value = {RESULT: "Not a valid restul"}
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should still have have two devices
assert hass.states.async_entity_ids_count() == 2
assert hass.states.async_entity_ids() == entity_ids
async def test_api_throws_response_entities_stay_the_same_as_before(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
local_oauth_impl: ClientSession,
mock_get_devices_twolightswitches,
mock_get_status_filled,
snapshot: SnapshotAssertion,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test case of incorrect response from iotty API on getting device status."""
mock_config_entry.add_to_hass(hass)
config_entry_oauth2_flow.async_register_implementation(
hass, DOMAIN, local_oauth_impl
)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Should have two devices
assert hass.states.async_entity_ids_count() == 2
entity_ids = hass.states.async_entity_ids()
assert entity_ids == snapshot
mock_get_devices_twolightswitches.return_value = test_ls_one_added
mock_get_status_filled.side_effect = Exception("Something went wrong")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Should still have have two devices
assert hass.states.async_entity_ids_count() == 2
assert hass.states.async_entity_ids() == entity_ids