268 lines
9.1 KiB
Python
268 lines
9.1 KiB
Python
"""Test the Shark IQ vacuum entity."""
|
|
from copy import deepcopy
|
|
import enum
|
|
import json
|
|
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,
|
|
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
|
|
|
|
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_DOCKED
|
|
|
|
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",
|
|
}
|
|
|
|
assert shark.device_info == target_device_info
|
|
|
|
|
|
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])
|
|
|
|
api = SharkVacuumEntity(shark_vac1, coordinator)
|
|
coordinator.last_update_success = True
|
|
coordinator._online_dsns = set() # pylint: disable=protected-access
|
|
assert not api.is_online
|
|
assert not api.available
|
|
|
|
coordinator._online_dsns = { # pylint: disable=protected-access
|
|
shark_vac1.serial_number
|
|
}
|
|
assert api.is_online
|
|
assert api.available
|
|
|
|
coordinator.last_update_success = False
|
|
assert not api.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
|
|
)
|