Add sharkiq integration for Shark IQ robot vacuums (#38272)

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
pull/39421/head
Andrew Marks 2020-08-30 09:26:11 -04:00 committed by GitHub
parent ab7b42c022
commit 3d1ff5b8d0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1183 additions and 0 deletions

View File

@ -751,6 +751,7 @@ omit =
homeassistant/components/sesame/lock.py
homeassistant/components/seven_segments/image_processing.py
homeassistant/components/seventeentrack/sensor.py
homeassistant/components/sharkiq/vacuum.py
homeassistant/components/shiftr/*
homeassistant/components/shodan/sensor.py
homeassistant/components/shelly/__init__.py

View File

@ -366,6 +366,7 @@ homeassistant/components/sentry/* @dcramer @frenck
homeassistant/components/serial/* @fabaff
homeassistant/components/seven_segments/* @fabaff
homeassistant/components/seventeentrack/* @bachya
homeassistant/components/sharkiq/* @ajmarks
homeassistant/components/shell_command/* @home-assistant/core
homeassistant/components/shelly/* @balloob
homeassistant/components/shiftr/* @fabaff

View File

@ -0,0 +1,119 @@
"""Shark IQ Integration."""
import asyncio
import async_timeout
from sharkiqpy import (
AylaApi,
SharkIqAuthError,
SharkIqAuthExpiringError,
SharkIqNotAuthedError,
get_ayla_api,
)
import voluptuous as vol
from homeassistant import exceptions
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from .const import API_TIMEOUT, COMPONENTS, DOMAIN, LOGGER
from .update_coordinator import SharkIqUpdateCoordinator
CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.Schema({})}, extra=vol.ALLOW_EXTRA)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
async def async_setup(hass, config):
"""Set up the sharkiq environment."""
hass.data.setdefault(DOMAIN, {})
if DOMAIN not in config:
return True
async def async_connect_or_timeout(ayla_api: AylaApi) -> bool:
"""Connect to vacuum."""
try:
with async_timeout.timeout(API_TIMEOUT):
LOGGER.debug("Initialize connection to Ayla networks API")
await ayla_api.async_sign_in()
except SharkIqAuthError as exc:
LOGGER.error("Authentication error connecting to Shark IQ api", exc_info=exc)
return False
except asyncio.TimeoutError as exc:
LOGGER.error("Timeout expired", exc_info=exc)
raise CannotConnect from exc
return True
async def async_setup_entry(hass, config_entry):
"""Initialize the sharkiq platform via config entry."""
ayla_api = get_ayla_api(
username=config_entry.data[CONF_USERNAME],
password=config_entry.data[CONF_PASSWORD],
websession=hass.helpers.aiohttp_client.async_get_clientsession(),
)
try:
if not await async_connect_or_timeout(ayla_api):
return False
except CannotConnect as exc:
raise exceptions.ConfigEntryNotReady from exc
shark_vacs = await ayla_api.async_get_devices(False)
device_names = ", ".join([d.name for d in shark_vacs])
LOGGER.debug("Found %d Shark IQ device(s): %s", len(device_names), device_names)
coordinator = SharkIqUpdateCoordinator(hass, config_entry, ayla_api, shark_vacs)
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise exceptions.ConfigEntryNotReady
hass.data[DOMAIN][config_entry.entry_id] = coordinator
for component in COMPONENTS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, component)
)
return True
async def async_disconnect_or_timeout(coordinator: SharkIqUpdateCoordinator):
"""Disconnect to vacuum."""
LOGGER.debug("Disconnecting from Ayla Api")
with async_timeout.timeout(5):
try:
await coordinator.ayla_api.async_sign_out()
except (SharkIqAuthError, SharkIqAuthExpiringError, SharkIqNotAuthedError):
pass
return True
async def async_update_options(hass, config_entry):
"""Update options."""
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_unload_entry(hass, config_entry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(config_entry, component)
for component in COMPONENTS
]
)
)
if unload_ok:
domain_data = hass.data[DOMAIN][config_entry.entry_id]
try:
await async_disconnect_or_timeout(coordinator=domain_data)
except SharkIqAuthError:
pass
hass.data[DOMAIN].pop(config_entry.entry_id)
return unload_ok

View File

@ -0,0 +1,107 @@
"""Config flow for Shark IQ integration."""
import asyncio
from typing import Dict, Optional
import aiohttp
import async_timeout
from sharkiqpy import SharkIqAuthError, get_ayla_api
import voluptuous as vol
from homeassistant import config_entries, core, exceptions
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from .const import DOMAIN, LOGGER # pylint:disable=unused-import
SHARKIQ_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect."""
ayla_api = get_ayla_api(
username=data[CONF_USERNAME],
password=data[CONF_PASSWORD],
websession=hass.helpers.aiohttp_client.async_get_clientsession(hass),
)
try:
with async_timeout.timeout(10):
LOGGER.debug("Initialize connection to Ayla networks API")
await ayla_api.async_sign_in()
except (asyncio.TimeoutError, aiohttp.ClientError):
raise CannotConnect
except SharkIqAuthError:
raise InvalidAuth
# Return info that you want to store in the config entry.
return {"title": data[CONF_USERNAME]}
class SharkIqConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Shark IQ."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
async def _async_validate_input(self, user_input):
"""Validate form input."""
errors = {}
info = None
if user_input is not None:
# noinspection PyBroadException
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
return info, errors
async def async_step_user(self, user_input: Optional[Dict] = None):
"""Handle the initial step."""
errors = {}
if user_input is not None:
info, errors = await self._async_validate_input(user_input)
if info:
return self.async_create_entry(title=info["title"], data=user_input)
return self.async_show_form(
step_id="user", data_schema=SHARKIQ_SCHEMA, errors=errors
)
async def async_step_reauth(self, user_input: Optional[dict] = None):
"""Handle re-auth if login is invalid."""
errors = {}
if user_input is not None:
_, errors = await self._async_validate_input(user_input)
if not errors:
for entry in self._async_current_entries():
if entry.unique_id == self.unique_id:
self.hass.config_entries.async_update_entry(
entry, data=user_input
)
return self.async_abort(reason="reauth_successful")
if errors["base"] != "invalid_auth":
return self.async_abort(reason=errors["base"])
return self.async_show_form(
step_id="reauth", data_schema=SHARKIQ_SCHEMA, errors=errors,
)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
class InvalidAuth(exceptions.HomeAssistantError):
"""Error to indicate there is invalid auth."""

View File

@ -0,0 +1,11 @@
"""Shark IQ Constants."""
from datetime import timedelta
import logging
API_TIMEOUT = 20
COMPONENTS = ["vacuum"]
DOMAIN = "sharkiq"
LOGGER = logging.getLogger(__package__)
SHARK = "Shark"
UPDATE_INTERVAL = timedelta(seconds=30)

View File

@ -0,0 +1,9 @@
{
"domain": "sharkiq",
"name": "Shark IQ",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/sharkiq",
"requirements": ["sharkiqpy==0.1.8"],
"dependencies": [],
"codeowners": ["@ajmarks"]
}

View File

@ -0,0 +1,20 @@
{
"config": {
"step": {
"user": {
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]"
}
}
}

View File

@ -0,0 +1,27 @@
{
"title": "Shark IQ",
"config": {
"step": {
"init": {
"data": {
"username": "Username",
"password": "Password"
}
},
"user": {
"data": {
"username": "Username",
"password": "Password"
}
}
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error"
},
"abort": {
"already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]"
}
}
}

View File

@ -0,0 +1,100 @@
"""Data update coordinator for shark iq vacuums."""
from typing import Dict, List, Set
from async_timeout import timeout
from sharkiqpy import (
AylaApi,
SharkIqAuthError,
SharkIqAuthExpiringError,
SharkIqNotAuthedError,
SharkIqVacuum,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL
class SharkIqUpdateCoordinator(DataUpdateCoordinator):
"""Define a wrapper class to update Shark IQ data."""
def __init__(
self,
hass: HomeAssistant,
config_entry: ConfigEntry,
ayla_api: AylaApi,
shark_vacs: List[SharkIqVacuum],
) -> None:
"""Set up the SharkIqUpdateCoordinator class."""
self.ayla_api = ayla_api
self.shark_vacs: Dict[SharkIqVacuum] = {
sharkiq.serial_number: sharkiq for sharkiq in shark_vacs
}
self._config_entry = config_entry
self._online_dsns = set()
super().__init__(hass, LOGGER, name=DOMAIN, update_interval=UPDATE_INTERVAL)
@property
def online_dsns(self) -> Set[str]:
"""Get the set of all online DSNs."""
return self._online_dsns
def device_is_online(self, dsn: str) -> bool:
"""Return the online state of a given vacuum dsn."""
return dsn in self._online_dsns
@staticmethod
async def _async_update_vacuum(sharkiq: SharkIqVacuum) -> None:
"""Asynchronously update the data for a single vacuum."""
dsn = sharkiq.serial_number
LOGGER.info("Updating sharkiq data for device DSN %s", dsn)
with timeout(API_TIMEOUT):
await sharkiq.async_update()
async def _async_update_data(self) -> bool:
"""Update data device by device."""
try:
all_vacuums = await self.ayla_api.async_list_devices()
self._online_dsns = {
v["dsn"]
for v in all_vacuums
if v["connection_status"] == "Online" and v["dsn"] in self.shark_vacs
}
LOGGER.info("Updating sharkiq data")
for dsn in self._online_dsns:
await self._async_update_vacuum(self.shark_vacs[dsn])
except (
SharkIqAuthError,
SharkIqNotAuthedError,
SharkIqAuthExpiringError,
) as err:
LOGGER.exception("Bad auth state", exc_info=err)
flow_context = {
"source": "reauth",
"unique_id": self._config_entry.unique_id,
}
matching_flows = [
flow
for flow in self.hass.config_entries.flow.async_progress()
if flow["context"] == flow_context
]
if not matching_flows:
self.hass.async_create_task(
self.hass.config_entries.flow.async_init(
DOMAIN, context=flow_context, data=self._config_entry.data,
)
)
raise UpdateFailed(err)
except Exception as err: # pylint: disable=broad-except
LOGGER.exception("Unexpected error updating SharkIQ", exc_info=err)
raise UpdateFailed(err)
return True

View File

@ -0,0 +1,286 @@
"""Shark IQ Wrapper."""
import logging
from typing import Dict, Iterable, Optional
from sharkiqpy import OperatingModes, PowerModes, Properties, SharkIqVacuum
from homeassistant.components.vacuum import (
STATE_CLEANING,
STATE_DOCKED,
STATE_IDLE,
STATE_PAUSED,
STATE_RETURNING,
SUPPORT_BATTERY,
SUPPORT_FAN_SPEED,
SUPPORT_LOCATE,
SUPPORT_PAUSE,
SUPPORT_RETURN_HOME,
SUPPORT_START,
SUPPORT_STATE,
SUPPORT_STATUS,
SUPPORT_STOP,
StateVacuumEntity,
)
from .const import DOMAIN, SHARK
from .update_coordinator import SharkIqUpdateCoordinator
LOGGER = logging.getLogger(__name__)
# Supported features
SUPPORT_SHARKIQ = (
SUPPORT_BATTERY
| SUPPORT_FAN_SPEED
| SUPPORT_PAUSE
| SUPPORT_RETURN_HOME
| SUPPORT_START
| SUPPORT_STATE
| SUPPORT_STATUS
| SUPPORT_STOP
| SUPPORT_LOCATE
)
OPERATING_STATE_MAP = {
OperatingModes.PAUSE: STATE_PAUSED,
OperatingModes.START: STATE_CLEANING,
OperatingModes.STOP: STATE_IDLE,
OperatingModes.RETURN: STATE_RETURNING,
}
FAN_SPEEDS_MAP = {
"Eco": PowerModes.ECO,
"Normal": PowerModes.NORMAL,
"Max": PowerModes.MAX,
}
STATE_RECHARGING_TO_RESUME = "recharging_to_resume"
# Attributes to expose
ATTR_ERROR_CODE = "last_error_code"
ATTR_ERROR_MSG = "last_error_message"
ATTR_LOW_LIGHT = "low_light"
ATTR_RECHARGE_RESUME = "recharge_and_resume"
ATTR_RSSI = "rssi"
class SharkVacuumEntity(StateVacuumEntity):
"""Shark IQ vacuum entity."""
def __init__(self, sharkiq: SharkIqVacuum, coordinator: SharkIqUpdateCoordinator):
"""Create a new SharkVacuumEntity."""
if sharkiq.serial_number not in coordinator.shark_vacs:
raise RuntimeError(
f"Shark IQ robot {sharkiq.serial_number} is not known to the coordinator"
)
self.coordinator = coordinator
self.sharkiq = sharkiq
@property
def should_poll(self):
"""Don't poll this entity. Polling is done via the coordinator."""
return False
def clean_spot(self, **kwargs):
"""Clean a spot. Not yet implemented."""
raise NotImplementedError()
def send_command(self, command, params=None, **kwargs):
"""Send a command to the vacuum. Not yet implemented."""
raise NotImplementedError()
@property
def is_online(self) -> bool:
"""Tell us if the device is online."""
return self.coordinator.device_is_online(self.sharkiq.serial_number)
@property
def name(self) -> str:
"""Device name."""
return self.sharkiq.name
@property
def serial_number(self) -> str:
"""Vacuum API serial number (DSN)."""
return self.sharkiq.serial_number
@property
def model(self) -> str:
"""Vacuum model number."""
if self.sharkiq.vac_model_number:
return self.sharkiq.vac_model_number
return self.sharkiq.oem_model_number
@property
def device_info(self) -> Dict:
"""Device info dictionary."""
return {
"identifiers": {(DOMAIN, self.serial_number)},
"name": self.name,
"manufacturer": SHARK,
"model": self.model,
"sw_version": self.sharkiq.get_property_value(
Properties.ROBOT_FIRMWARE_VERSION
),
}
@property
def supported_features(self) -> int:
"""Flag vacuum cleaner robot features that are supported."""
return SUPPORT_SHARKIQ
@property
def is_docked(self) -> Optional[bool]:
"""Is vacuum docked."""
return self.sharkiq.get_property_value(Properties.DOCKED_STATUS)
@property
def error_code(self) -> Optional[int]:
"""Return the last observed error code (or None)."""
return self.sharkiq.error_code
@property
def error_message(self) -> Optional[str]:
"""Return the last observed error message (or None)."""
if not self.error_code:
return None
return self.sharkiq.error_text
@property
def operating_mode(self) -> Optional[str]:
"""Operating mode.."""
op_mode = self.sharkiq.get_property_value(Properties.OPERATING_MODE)
return OPERATING_STATE_MAP.get(op_mode)
@property
def recharging_to_resume(self) -> Optional[int]:
"""Return True if vacuum set to recharge and resume cleaning."""
return self.sharkiq.get_property_value(Properties.RECHARGING_TO_RESUME)
@property
def state(self):
"""
Get the current vacuum state.
NB: Currently, we do not return an error state because they can be very, very stale.
In the app, these are (usually) handled by showing the robot as stopped and sending the
user a notification.
"""
if self.recharging_to_resume:
return STATE_RECHARGING_TO_RESUME
if self.is_docked:
return STATE_DOCKED
return self.operating_mode
@property
def unique_id(self) -> str:
"""Return the unique id of the vacuum cleaner."""
return self.serial_number
@property
def available(self) -> bool:
"""Determine if the sensor is available based on API results."""
# If the last update was successful...
return self.coordinator.last_update_success and self.is_online
@property
def battery_level(self):
"""Get the current battery level."""
return self.sharkiq.get_property_value(Properties.BATTERY_CAPACITY)
async def async_update(self):
"""Update the known properties asynchronously."""
await self.coordinator.async_request_refresh()
async def async_added_to_hass(self) -> None:
"""Connect to dispatcher listening for entity data notifications."""
self.async_on_remove(
self.coordinator.async_add_listener(self.async_write_ha_state)
)
async def async_return_to_base(self, **kwargs):
"""Have the device return to base."""
await self.sharkiq.async_set_operating_mode(OperatingModes.RETURN)
await self.coordinator.async_refresh()
async def async_pause(self):
"""Pause the cleaning task."""
await self.sharkiq.async_set_operating_mode(OperatingModes.PAUSE)
await self.coordinator.async_refresh()
async def async_start(self):
"""Start the device."""
await self.sharkiq.async_set_operating_mode(OperatingModes.START)
await self.coordinator.async_refresh()
async def async_stop(self, **kwargs):
"""Stop the device."""
await self.sharkiq.async_set_operating_mode(OperatingModes.STOP)
await self.coordinator.async_refresh()
async def async_locate(self, **kwargs):
"""Cause the device to generate a loud chirp."""
await self.sharkiq.async_find_device()
@property
def fan_speed(self) -> str:
"""Return the current fan speed."""
fan_speed = None
speed_level = self.sharkiq.get_property_value(Properties.POWER_MODE)
for k, val in FAN_SPEEDS_MAP.items():
if val == speed_level:
fan_speed = k
return fan_speed
async def async_set_fan_speed(self, fan_speed: str, **kwargs):
"""Set the fan speed."""
await self.sharkiq.async_set_property_value(
Properties.POWER_MODE, FAN_SPEEDS_MAP.get(fan_speed.capitalize())
)
await self.coordinator.async_refresh()
@property
def fan_speed_list(self):
"""Get the list of available fan speed steps of the vacuum cleaner."""
return list(FAN_SPEEDS_MAP.keys())
# Various attributes we want to expose
@property
def recharge_resume(self) -> Optional[bool]:
"""Recharge and resume mode active."""
return self.sharkiq.get_property_value(Properties.RECHARGE_RESUME)
@property
def rssi(self) -> Optional[int]:
"""Get the WiFi RSSI."""
return self.sharkiq.get_property_value(Properties.RSSI)
@property
def low_light(self):
"""Let us know if the robot is operating in low-light mode."""
return self.sharkiq.get_property_value(Properties.LOW_LIGHT_MISSION)
@property
def device_state_attributes(self) -> Dict:
"""Return a dictionary of device state attributes specific to sharkiq."""
data = {
ATTR_ERROR_CODE: self.error_code,
ATTR_ERROR_MSG: self.sharkiq.error_text,
ATTR_LOW_LIGHT: self.low_light,
ATTR_RECHARGE_RESUME: self.recharge_resume,
}
return data
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the Shark IQ vacuum cleaner."""
coordinator: SharkIqUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
devices: Iterable["SharkIqVacuum"] = coordinator.shark_vacs.values()
device_names = [d.name for d in devices]
LOGGER.debug(
"Found %d Shark IQ device(s): %s",
len(device_names),
", ".join([d.name for d in devices]),
)
async_add_entities([SharkVacuumEntity(d, coordinator) for d in devices])

View File

@ -152,6 +152,7 @@ FLOWS = [
"samsungtv",
"sense",
"sentry",
"sharkiq",
"shelly",
"shopping_list",
"simplisafe",

View File

@ -1958,6 +1958,9 @@ sense_energy==0.7.2
# homeassistant.components.sentry
sentry-sdk==0.16.5
# homeassistant.components.sharkiq
sharkiqpy==0.1.8
# homeassistant.components.aquostv
sharp_aquos_rc==0.3.2

View File

@ -907,6 +907,9 @@ sense_energy==0.7.2
# homeassistant.components.sentry
sentry-sdk==0.16.5
# homeassistant.components.sharkiq
sharkiqpy==0.1.8
# homeassistant.components.sighthound
simplehound==0.3

View File

@ -0,0 +1 @@
"""Tests for the Shark IQ integration."""

View File

@ -0,0 +1,73 @@
"""Constants used in shark iq tests."""
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
# Dummy device dict of the form returned by AylaApi.list_devices()
SHARK_DEVICE_DICT = {
"product_name": "Sharknado",
"model": "AY001MRT1",
"dsn": "AC000Wxxxxxxxxx",
"oem_model": "RV1000A",
"sw_version": "devd 1.7 2020-05-13 11:50:36",
"template_id": 99999,
"mac": "ffffffffffff",
"unique_hardware_id": None,
"lan_ip": "192.168.0.123",
"connected_at": "2020-07-31T08:03:05Z",
"key": 26517570,
"lan_enabled": False,
"has_properties": True,
"product_class": None,
"connection_status": "Online",
"lat": "99.9999",
"lng": "-99.9999",
"locality": "99999",
"device_type": "Wifi",
}
# Dummy response for get_metadata
SHARK_METADATA_DICT = [
{
"datum": {
"created_at": "2019-12-02T02:13:12Z",
"from_template": False,
"key": "sharkDeviceMobileData",
"updated_at": "2019-12-02T02:13:12Z",
"value": '{"vacModelNumber":"RV1001AE","vacSerialNumber":"S26xxxxxxxxx"}',
"dsn": "AC000Wxxxxxxxxx",
}
}
]
# Dummy shark.properties_full for testing. NB: this only includes those properties in the tests
SHARK_PROPERTIES_DICT = {
"Battery_Capacity": {"base_type": "integer", "read_only": True, "value": 50},
"Charging_Status": {"base_type": "boolean", "read_only": True, "value": 0},
"CleanComplete": {"base_type": "boolean", "read_only": True, "value": 0},
"Cleaning_Statistics": {"base_type": "file", "read_only": True, "value": None},
"DockedStatus": {"base_type": "boolean", "read_only": True, "value": 0},
"Error_Code": {"base_type": "integer", "read_only": True, "value": 7},
"Evacuating": {"base_type": "boolean", "read_only": True, "value": 1},
"Find_Device": {"base_type": "boolean", "read_only": False, "value": 0},
"LowLightMission": {"base_type": "boolean", "read_only": True, "value": 0},
"Nav_Module_FW_Version": {
"base_type": "string",
"read_only": True,
"value": "V3.4.11-20191015",
},
"Operating_Mode": {"base_type": "integer", "read_only": False, "value": 2},
"Power_Mode": {"base_type": "integer", "read_only": False, "value": 1},
"RSSI": {"base_type": "integer", "read_only": True, "value": -46},
"Recharge_Resume": {"base_type": "boolean", "read_only": False, "value": 1},
"Recharging_To_Resume": {"base_type": "boolean", "read_only": True, "value": 0},
"Robot_Firmware_Version": {
"base_type": "string",
"read_only": True,
"value": "Dummy Firmware 1.0",
},
}
TEST_USERNAME = "test-username"
TEST_PASSWORD = "test-password"
UNIQUE_ID = "foo@bar.com"
CONFIG = {CONF_USERNAME: TEST_USERNAME, CONF_PASSWORD: TEST_PASSWORD}

View File

@ -0,0 +1,138 @@
"""Test the Shark IQ config flow."""
import aiohttp
from sharkiqpy import SharkIqAuthError
from homeassistant import config_entries, setup
from homeassistant.components.sharkiq.const import DOMAIN
from .const import CONFIG, TEST_PASSWORD, TEST_USERNAME, UNIQUE_ID
from tests.async_mock import MagicMock, PropertyMock, patch
from tests.common import MockConfigEntry
def _create_mocked_ayla(connect=None):
"""Create a mocked AylaApi object."""
mocked_ayla = MagicMock()
type(mocked_ayla).sign_in = PropertyMock(side_effect=connect)
type(mocked_ayla).async_sign_in = PropertyMock(side_effect=connect)
return mocked_ayla
async def test_form(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["errors"] == {}
with patch("sharkiqpy.AylaApi.async_sign_in", return_value=True), patch(
"homeassistant.components.sharkiq.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.sharkiq.async_setup_entry", return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], CONFIG,
)
await hass.async_block_till_done()
assert result2["type"] == "create_entry"
assert result2["title"] == f"{TEST_USERNAME:s}"
assert result2["data"] == {
"username": TEST_USERNAME,
"password": TEST_PASSWORD,
}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass):
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mocked_ayla = _create_mocked_ayla(connect=SharkIqAuthError)
with patch(
"homeassistant.components.sharkiq.config_flow.get_ayla_api",
return_value=mocked_ayla,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], CONFIG,
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_cannot_connect(hass):
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mocked_ayla = _create_mocked_ayla(connect=aiohttp.ClientError)
with patch(
"homeassistant.components.sharkiq.config_flow.get_ayla_api",
return_value=mocked_ayla,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], CONFIG,
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_other_error(hass):
"""Test we handle other errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mocked_ayla = _create_mocked_ayla(connect=TypeError)
with patch(
"homeassistant.components.sharkiq.config_flow.get_ayla_api",
return_value=mocked_ayla,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], CONFIG
)
assert result2["type"] == "form"
assert result2["errors"] == {"base": "unknown"}
async def test_reauth(hass):
"""Test reauth flow."""
with patch(
"homeassistant.components.sharkiq.vacuum.async_setup_entry", return_value=True,
), patch("sharkiqpy.AylaApi.async_sign_in", return_value=True):
mock_config = MockConfigEntry(domain=DOMAIN, unique_id=UNIQUE_ID, data=CONFIG)
mock_config.add_to_hass(hass)
hass.config_entries.async_update_entry(mock_config, data=CONFIG)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth", "unique_id": UNIQUE_ID}, data=CONFIG
)
assert result["type"] == "abort"
assert result["reason"] == "reauth_successful"
with patch("sharkiqpy.AylaApi.async_sign_in", side_effect=SharkIqAuthError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth", "unique_id": UNIQUE_ID}, data=CONFIG,
)
assert result["type"] == "form"
assert result["errors"] == {"base": "invalid_auth"}
with patch("sharkiqpy.AylaApi.async_sign_in", side_effect=RuntimeError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth", "unique_id": UNIQUE_ID}, data=CONFIG,
)
assert result["type"] == "abort"
assert result["reason"] == "unknown"

View File

@ -0,0 +1,283 @@
"""Test the Shark IQ vacuum entity."""
from copy import deepcopy
import enum
from typing import Dict, List
from sharkiqpy import AylaApi, Properties, SharkIqAuthError, SharkIqVacuum, get_ayla_api
from homeassistant.components.sharkiq import SharkIqUpdateCoordinator
from homeassistant.components.sharkiq.vacuum import (
ATTR_ERROR_CODE,
ATTR_ERROR_MSG,
ATTR_LOW_LIGHT,
ATTR_RECHARGE_RESUME,
STATE_RECHARGING_TO_RESUME,
SharkVacuumEntity,
)
from homeassistant.components.vacuum import (
STATE_CLEANING,
STATE_DOCKED,
STATE_IDLE,
STATE_PAUSED,
STATE_RETURNING,
SUPPORT_BATTERY,
SUPPORT_FAN_SPEED,
SUPPORT_LOCATE,
SUPPORT_PAUSE,
SUPPORT_RETURN_HOME,
SUPPORT_START,
SUPPORT_STATE,
SUPPORT_STATUS,
SUPPORT_STOP,
)
from homeassistant.config_entries import ConfigEntriesFlowManager, ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import UpdateFailed
from .const import (
SHARK_DEVICE_DICT,
SHARK_METADATA_DICT,
SHARK_PROPERTIES_DICT,
TEST_PASSWORD,
TEST_USERNAME,
)
from tests.async_mock import MagicMock, patch
try:
import ujson as json
except ImportError:
import json
MockAyla = MagicMock(spec=AylaApi) # pylint: disable=invalid-name
def _set_property(self, property_name, value):
"""Set a property locally without hitting the API."""
if isinstance(property_name, enum.Enum):
property_name = property_name.value
if isinstance(value, enum.Enum):
value = value.value
self.properties_full[property_name]["value"] = value
async def _async_set_property(self, property_name, value):
"""Set a property locally without hitting the API."""
_set_property(self, property_name, value)
def _get_mock_shark_vac(ayla_api: AylaApi) -> SharkIqVacuum:
"""Create a crude sharkiq vacuum with mocked properties."""
shark = SharkIqVacuum(ayla_api, SHARK_DEVICE_DICT)
shark.properties_full = deepcopy(SHARK_PROPERTIES_DICT)
return shark
async def _async_list_devices(_) -> List[Dict]:
"""Generate a dummy of async_list_devices output."""
return [SHARK_DEVICE_DICT]
@patch.object(SharkIqVacuum, "set_property_value", new=_set_property)
@patch.object(SharkIqVacuum, "async_set_property_value", new=_async_set_property)
async def test_shark_operation_modes(hass: HomeAssistant) -> None:
"""Test all of the shark vacuum operation modes."""
ayla_api = MockAyla()
shark_vac = _get_mock_shark_vac(ayla_api)
coordinator = SharkIqUpdateCoordinator(hass, None, ayla_api, [shark_vac])
shark = SharkVacuumEntity(shark_vac, coordinator)
# These come from the setup
assert isinstance(shark.is_docked, bool) and not shark.is_docked
assert (
isinstance(shark.recharging_to_resume, bool) and not shark.recharging_to_resume
)
# Go through the operation modes while it's "off the dock"
await shark.async_start()
assert shark.operating_mode == shark.state == STATE_CLEANING
await shark.async_pause()
assert shark.operating_mode == shark.state == STATE_PAUSED
await shark.async_stop()
assert shark.operating_mode == shark.state == STATE_IDLE
await shark.async_return_to_base()
assert shark.operating_mode == shark.state == STATE_RETURNING
# Test the docked modes
await shark.async_stop()
shark.sharkiq.set_property_value(Properties.RECHARGING_TO_RESUME, 1)
shark.sharkiq.set_property_value(Properties.DOCKED_STATUS, 1)
assert isinstance(shark.is_docked, bool) and shark.is_docked
assert isinstance(shark.recharging_to_resume, bool) and shark.recharging_to_resume
assert shark.state == STATE_RECHARGING_TO_RESUME
shark.sharkiq.set_property_value(Properties.RECHARGING_TO_RESUME, 0)
assert shark.state == STATE_DOCKED
await shark.async_set_fan_speed("Eco")
assert shark.fan_speed == "Eco"
await shark.async_set_fan_speed("Max")
assert shark.fan_speed == "Max"
await shark.async_set_fan_speed("Normal")
assert shark.fan_speed == "Normal"
assert set(shark.fan_speed_list) == {"Normal", "Max", "Eco"}
@patch.object(SharkIqVacuum, "set_property_value", new=_set_property)
async def test_shark_vac_properties(hass: HomeAssistant) -> None:
"""Test all of the shark vacuum property accessors."""
ayla_api = MockAyla()
shark_vac = _get_mock_shark_vac(ayla_api)
coordinator = SharkIqUpdateCoordinator(hass, None, ayla_api, [shark_vac])
shark = SharkVacuumEntity(shark_vac, coordinator)
assert shark.name == "Sharknado"
assert shark.serial_number == "AC000Wxxxxxxxxx"
assert shark.model == "RV1000A"
assert shark.battery_level == 50
assert shark.fan_speed == "Eco"
shark.sharkiq.set_property_value(Properties.POWER_MODE, 0)
assert shark.fan_speed == "Normal"
assert isinstance(shark.recharge_resume, bool) and shark.recharge_resume
assert isinstance(shark.low_light, bool) and not shark.low_light
target_state_attributes = {
ATTR_ERROR_CODE: 7,
ATTR_ERROR_MSG: "Cliff sensor is blocked",
ATTR_RECHARGE_RESUME: True,
ATTR_LOW_LIGHT: False,
}
state_json = json.dumps(shark.device_state_attributes, sort_keys=True)
target_json = json.dumps(target_state_attributes, sort_keys=True)
assert state_json == target_json
assert not shark.should_poll
@patch.object(SharkIqVacuum, "set_property_value", new=_set_property)
@patch.object(SharkIqVacuum, "async_set_property_value", new=_async_set_property)
async def test_shark_metadata(hass: HomeAssistant) -> None:
"""Test shark properties coming from metadata."""
ayla_api = MockAyla()
shark_vac = _get_mock_shark_vac(ayla_api)
coordinator = SharkIqUpdateCoordinator(hass, None, ayla_api, [shark_vac])
shark = SharkVacuumEntity(shark_vac, coordinator)
shark.sharkiq._update_metadata( # pylint: disable=protected-access
SHARK_METADATA_DICT
)
target_device_info = {
"identifiers": {("sharkiq", "AC000Wxxxxxxxxx")},
"name": "Sharknado",
"manufacturer": "Shark",
"model": "RV1001AE",
"sw_version": "Dummy Firmware 1.0",
}
state_json = json.dumps(shark.device_info, sort_keys=True)
target_json = json.dumps(target_device_info, sort_keys=True)
assert state_json == target_json
def _get_async_update(err=None):
async def _async_update(_) -> bool:
if err is not None:
raise err
return True
return _async_update
@patch.object(AylaApi, "async_list_devices", new=_async_list_devices)
async def test_updates(hass: HomeAssistant) -> None:
"""Test the update coordinator update functions."""
ayla_api = get_ayla_api(TEST_USERNAME, TEST_PASSWORD)
shark_vac = _get_mock_shark_vac(ayla_api)
mock_config = MagicMock(spec=ConfigEntry)
coordinator = SharkIqUpdateCoordinator(hass, mock_config, ayla_api, [shark_vac])
with patch.object(SharkIqVacuum, "async_update", new=_get_async_update()):
update_called = (
await coordinator._async_update_data() # pylint: disable=protected-access
)
assert update_called
update_failed = False
with patch.object(
SharkIqVacuum, "async_update", new=_get_async_update(SharkIqAuthError)
), patch.object(HomeAssistant, "async_create_task"), patch.object(
ConfigEntriesFlowManager, "async_init"
):
try:
await coordinator._async_update_data() # pylint: disable=protected-access
except UpdateFailed:
update_failed = True
assert update_failed
async def test_coordinator_match(hass: HomeAssistant):
"""Test that sharkiq-coordinator references work."""
ayla_api = get_ayla_api(TEST_PASSWORD, TEST_USERNAME)
shark_vac1 = _get_mock_shark_vac(ayla_api)
shark_vac2 = _get_mock_shark_vac(ayla_api)
shark_vac2._dsn = "FOOBAR!" # pylint: disable=protected-access
coordinator = SharkIqUpdateCoordinator(hass, None, ayla_api, [shark_vac1])
# The first should succeed, the second should fail
api1 = SharkVacuumEntity(shark_vac1, coordinator)
try:
_ = SharkVacuumEntity(shark_vac2, coordinator)
except RuntimeError:
api2_failed = True
else:
api2_failed = False
assert api2_failed
coordinator.last_update_success = True
coordinator._online_dsns = set() # pylint: disable=protected-access
assert not api1.is_online
assert not api1.available
coordinator._online_dsns = { # pylint: disable=protected-access
shark_vac1.serial_number
}
assert api1.is_online
assert api1.available
coordinator.last_update_success = False
assert not api1.available
async def test_simple_properties(hass: HomeAssistant):
"""Test that simple properties work as intended."""
ayla_api = get_ayla_api(TEST_PASSWORD, TEST_USERNAME)
shark_vac1 = _get_mock_shark_vac(ayla_api)
coordinator = SharkIqUpdateCoordinator(hass, None, ayla_api, [shark_vac1])
entity = SharkVacuumEntity(shark_vac1, coordinator)
assert entity.unique_id == "AC000Wxxxxxxxxx"
assert entity.supported_features == (
SUPPORT_BATTERY
| SUPPORT_FAN_SPEED
| SUPPORT_PAUSE
| SUPPORT_RETURN_HOME
| SUPPORT_START
| SUPPORT_STATE
| SUPPORT_STATUS
| SUPPORT_STOP
| SUPPORT_LOCATE
)
assert entity.error_code == 7
assert entity.error_message == "Cliff sensor is blocked"
shark_vac1.properties_full[Properties.ERROR_CODE.value]["value"] = 0
assert entity.error_code == 0
assert entity.error_message is None
assert (
coordinator.online_dsns
is coordinator._online_dsns # pylint: disable=protected-access
)