jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-16 01:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump Roborock to 17.0 adding device specific support and bugfixes (#92547)

Browse Source 
* init commit

* use official version release

* remove options

* moved first refresh to gather

* add extra tests

* remove model_sepcification

* remove old mqtt test

* bump to 13.4

* fix dndtimer

* bump to 14.1

* add status back

* bump to 17.0

* remove error as it is not used

* addressing mr comments

* making enum access use get()

* add check for empty hass data
This commit is contained in:
Luke 2023-05-18 23:55:39 -04:00 committed by GitHub
parent aebded049b
commit 0ce1117287
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 161 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
homeassistant/components/roborock/__init__.py
View File

@ -7,8 +7,7 @@ import logging
from roborock.api import RoborockApiClient
from roborock.cloud_api import RoborockMqttClient
from roborock.containers import HomeDataDevice, RoborockDeviceInfo, UserData
from roborock.exceptions import RoborockException
from roborock.containers import DeviceData, HomeDataDevice, UserData
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_USERNAME
@ -32,39 +31,58 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
_LOGGER.debug("Getting home data")
home_data = await api_client.get_home_data(user_data)
_LOGGER.debug("Got home data %s", home_data)
devices: list[HomeDataDevice] = home_data.devices + home_data.received_devices
device_map: dict[str, HomeDataDevice] = {
device.duid: device for device in home_data.devices + home_data.received_devices
}
product_info = {product.id: product for product in home_data.products}
# Create a mqtt_client, which is needed to get the networking information of the device for local connection and in the future, get the map.
mqtt_client = RoborockMqttClient(
user_data, {device.duid: RoborockDeviceInfo(device) for device in devices}
mqtt_clients = [
RoborockMqttClient(
user_data, DeviceData(device, product_info[device.product_id].model)
)
for device in device_map.values()
]
network_results = await asyncio.gather(
*(mqtt_client.get_networking(device.duid) for device in devices)
*(mqtt_client.get_networking() for mqtt_client in mqtt_clients)
)
network_info = {
device.duid: result
for device, result in zip(devices, network_results)
for device, result in zip(device_map.values(), network_results)
if result is not None
}
try:
await mqtt_client.async_disconnect()
except RoborockException as err:
_LOGGER.warning("Failed disconnecting from the mqtt server %s", err)
await asyncio.gather(
*(mqtt_client.async_disconnect() for mqtt_client in mqtt_clients),
return_exceptions=True,
)
if not network_info:
raise ConfigEntryNotReady(
"Could not get network information about your devices"
)
product_info = {product.id: product for product in home_data.products}
coordinator = RoborockDataUpdateCoordinator(
coordinator_map: dict[str, RoborockDataUpdateCoordinator] = {}
for device_id, device in device_map.items():
coordinator_map[device_id] = RoborockDataUpdateCoordinator(
hass,
devices,
network_info,
product_info,
device,
network_info[device_id],
product_info[device.product_id],
)
# If one device update fails - we still want to set up other devices
await asyncio.gather(
*(
coordinator.async_config_entry_first_refresh()
for coordinator in coordinator_map.values()
),
return_exceptions=True,
)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
device_id: coordinator
for device_id, coordinator in coordinator_map.items()
if coordinator.last_update_success
} # Only add coordinators that succeeded
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
if not hass.data[DOMAIN][entry.entry_id]:
# Don't start if no coordinators succeeded.
raise ConfigEntryNotReady("There are no devices that can currently be reached.")
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -75,7 +93,12 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Handle removal of an entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
await hass.data[DOMAIN][entry.entry_id].release()
await asyncio.gather(
*(
coordinator.release()
for coordinator in hass.data[DOMAIN][entry.entry_id].values()
)
)
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

61
homeassistant/components/roborock/coordinator.py
View File

@ -1,19 +1,13 @@
"""Roborock Coordinator."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from roborock.containers import (
HomeDataDevice,
HomeDataProduct,
NetworkInfo,
RoborockLocalDeviceInfo,
)
from roborock.containers import DeviceData, HomeDataDevice, HomeDataProduct, NetworkInfo
from roborock.exceptions import RoborockException
from roborock.local_api import RoborockLocalClient
from roborock.typing import DeviceProp
from roborock.roborock_typing import DeviceProp
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -26,61 +20,44 @@ SCAN_INTERVAL = timedelta(seconds=30)
_LOGGER = logging.getLogger(__name__)
class RoborockDataUpdateCoordinator(DataUpdateCoordinator[dict[str, DeviceProp]]):
class RoborockDataUpdateCoordinator(DataUpdateCoordinator[DeviceProp]):
"""Class to manage fetching data from the API."""
def __init__(
self,
hass: HomeAssistant,
devices: list[HomeDataDevice],
devices_networking: dict[str, NetworkInfo],
product_info: dict[str, HomeDataProduct],
device: HomeDataDevice,
device_networking: NetworkInfo,
product_info: HomeDataProduct,
) -> None:
"""Initialize."""
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL)
local_devices_info: dict[str, RoborockLocalDeviceInfo] = {}
hass_devices_info: dict[str, RoborockHassDeviceInfo] = {}
for device in devices:
if not (networking := devices_networking.get(device.duid)):
_LOGGER.warning("Device %s is offline and cannot be setup", device.duid)
continue
hass_devices_info[device.duid] = RoborockHassDeviceInfo(
self.device_info = RoborockHassDeviceInfo(
device,
networking,
product_info[device.product_id],
device_networking,
product_info,
DeviceProp(),
)
local_devices_info[device.duid] = RoborockLocalDeviceInfo(
device, networking
)
self.api = RoborockLocalClient(local_devices_info)
self.devices_info = hass_devices_info
device_info = DeviceData(device, product_info.model, device_networking.ip)
self.api = RoborockLocalClient(device_info)
async def release(self) -> None:
"""Disconnect from API."""
await self.api.async_disconnect()
async def _update_device_prop(self, device_info: RoborockHassDeviceInfo) -> None:
async def _update_device_prop(self) -> None:
"""Update device properties."""
device_prop = await self.api.get_prop(device_info.device.duid)
device_prop = await self.api.get_prop()
if device_prop:
if device_info.props:
device_info.props.update(device_prop)
if self.device_info.props:
self.device_info.props.update(device_prop)
else:
device_info.props = device_prop
self.device_info.props = device_prop
async def _async_update_data(self) -> dict[str, DeviceProp]:
async def _async_update_data(self) -> DeviceProp:
"""Update data via library."""
try:
await asyncio.gather(
*(
self._update_device_prop(device_info)
for device_info in self.devices_info.values()
)
)
await self._update_device_prop()
except RoborockException as ex:
raise UpdateFailed(ex) from ex
return {
device_id: device_info.props
for device_id, device_info in self.devices_info.items()
}
return self.device_info.props

32
homeassistant/components/roborock/device.py
View File

@ -3,14 +3,15 @@
from typing import Any
from roborock.containers import Status
from roborock.typing import RoborockCommand
from roborock.exceptions import RoborockException
from roborock.roborock_typing import RoborockCommand
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import RoborockDataUpdateCoordinator
from .const import DOMAIN
from .models import RoborockHassDeviceInfo
class RoborockCoordinatedEntity(CoordinatorEntity[RoborockDataUpdateCoordinator]):
@ -21,25 +22,18 @@ class RoborockCoordinatedEntity(CoordinatorEntity[RoborockDataUpdateCoordinator]
def __init__(
self,
unique_id: str,
device_info: RoborockHassDeviceInfo,
coordinator: RoborockDataUpdateCoordinator,
) -> None:
"""Initialize the coordinated Roborock Device."""
super().__init__(coordinator)
self._attr_unique_id = unique_id
self._device_name = device_info.device.name
self._device_id = device_info.device.duid
self._device_model = device_info.product.model
self._fw_version = device_info.device.fv
@property
def _device_status(self) -> Status:
"""Return the status of the device."""
data = self.coordinator.data
if data:
device_data = data.get(self._device_id)
if device_data:
status = device_data.status
status = data.status
if status:
return status
return Status({})
@ -48,19 +42,23 @@ class RoborockCoordinatedEntity(CoordinatorEntity[RoborockDataUpdateCoordinator]
def device_info(self) -> DeviceInfo:
"""Return the device info."""
return DeviceInfo(
name=self._device_name,
identifiers={(DOMAIN, self._device_id)},
name=self.coordinator.device_info.device.name,
identifiers={(DOMAIN, self.coordinator.device_info.device.duid)},
manufacturer="Roborock",
model=self._device_model,
sw_version=self._fw_version,
model=self.coordinator.device_info.product.model,
sw_version=self.coordinator.device_info.device.fv,
)
async def send(
self, command: RoborockCommand, params: dict[str, Any] | list[Any] | None = None
) -> dict:
"""Send a command to a vacuum cleaner."""
response = await self.coordinator.api.send_command(
self._device_id, command, params
)
try:
response = await self.coordinator.api.send_command(command, params)
except RoborockException as err:
raise HomeAssistantError(
f"Error while calling {command.name} with {params}"
) from err
await self.coordinator.async_request_refresh()
return response

2
homeassistant/components/roborock/manifest.json
View File

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/roborock",
"iot_class": "local_polling",
"loggers": ["roborock"],
"requirements": ["python-roborock==0.8.3"]
"requirements": ["python-roborock==0.17.0"]
}

2
homeassistant/components/roborock/models.py
View File

@ -2,7 +2,7 @@
from dataclasses import dataclass
from roborock.containers import HomeDataDevice, HomeDataProduct, NetworkInfo
from roborock.typing import DeviceProp
from roborock.roborock_typing import DeviceProp
@dataclass

50
homeassistant/components/roborock/select.py
View File

@ -2,31 +2,32 @@
from collections.abc import Callable
from dataclasses import dataclass
from roborock.code_mappings import RoborockMopIntensityCode, RoborockMopModeCode
from roborock.containers import Status
from roborock.exceptions import RoborockException
from roborock.typing import RoborockCommand
from roborock.roborock_typing import RoborockCommand
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import slugify
from .const import DOMAIN
from .coordinator import RoborockDataUpdateCoordinator
from .device import RoborockCoordinatedEntity
from .models import RoborockHassDeviceInfo
@dataclass
class RoborockSelectDescriptionMixin:
"""Define an entity description mixin for select entities."""
# The command that the select entity will send to the api.
api_command: RoborockCommand
# Gets the current value of the select entity.
value_fn: Callable[[Status], str]
options_lambda: Callable[[str], list[int]]
# Gets all options of the select entity.
options_lambda: Callable[[Status], list[str]]
# Takes the value from the select entiy and converts it for the api.
parameter_lambda: Callable[[str, Status], list[int]]
@dataclass
@ -40,22 +41,20 @@ SELECT_DESCRIPTIONS: list[RoborockSelectDescription] = [
RoborockSelectDescription(
key="water_box_mode",
translation_key="mop_intensity",
options=RoborockMopIntensityCode.values(),
api_command=RoborockCommand.SET_WATER_BOX_CUSTOM_MODE,
value_fn=lambda data: data.water_box_mode,
options_lambda=lambda data: [
k for k, v in RoborockMopIntensityCode.items() if v == data
],
value_fn=lambda data: data.water_box_mode.name,
options_lambda=lambda data: data.water_box_mode.keys()
if data.water_box_mode
else None,
parameter_lambda=lambda key, status: [status.water_box_mode.as_dict().get(key)],
),
RoborockSelectDescription(
key="mop_mode",
translation_key="mop_mode",
options=RoborockMopModeCode.values(),
api_command=RoborockCommand.SET_MOP_MODE,
value_fn=lambda data: data.mop_mode,
options_lambda=lambda data: [
k for k, v in RoborockMopModeCode.items() if v == data
],
value_fn=lambda data: data.mop_mode.name,
options_lambda=lambda data: data.mop_mode.keys() if data.mop_mode else None,
parameter_lambda=lambda key, status: [status.mop_mode.as_dict().get(key)],
),
]
@ -67,18 +66,18 @@ async def async_setup_entry(
) -> None:
"""Set up Roborock select platform."""
coordinator: RoborockDataUpdateCoordinator = hass.data[DOMAIN][
coordinators: dict[str, RoborockDataUpdateCoordinator] = hass.data[DOMAIN][
config_entry.entry_id
]
async_add_entities(
RoborockSelectEntity(
f"{description.key}_{slugify(device_id)}",
device_info,
coordinator,
description,
)
for device_id, device_info in coordinator.devices_info.items()
for device_id, coordinator in coordinators.items()
for description in SELECT_DESCRIPTIONS
if description.options_lambda(coordinator.device_info.props.status) is not None
)
@ -90,25 +89,20 @@ class RoborockSelectEntity(RoborockCoordinatedEntity, SelectEntity):
def __init__(
self,
unique_id: str,
device_info: RoborockHassDeviceInfo,
coordinator: RoborockDataUpdateCoordinator,
entity_description: RoborockSelectDescription,
) -> None:
"""Create a select entity."""
self.entity_description = entity_description
super().__init__(unique_id, device_info, coordinator)
super().__init__(unique_id, coordinator)
self._attr_options = self.entity_description.options_lambda(self._device_status)
async def async_select_option(self, option: str) -> None:
"""Set the mop intensity."""
try:
"""Set the option."""
await self.send(
self.entity_description.api_command,
self.entity_description.options_lambda(option),
self.entity_description.parameter_lambda(option, self._device_status),
)
except RoborockException as err:
raise HomeAssistantError(
f"Error while setting {self.entity_description.key} to {option}"
) from err
@property
def current_option(self) -> str | None:

3
homeassistant/components/roborock/strings.json
View File

@ -34,7 +34,8 @@
"standard": "Standard",
"deep": "Deep",
"deep_plus": "Deep+",
"custom": "Custom"
"custom": "Custom",
"fast": "Fast"
}
},
"mop_intensity": {

83
homeassistant/components/roborock/vacuum.py
View File

@ -1,8 +1,8 @@
"""Support for Roborock vacuum class."""
from typing import Any
from roborock.code_mappings import RoborockFanPowerCode, RoborockStateCode
from roborock.typing import RoborockCommand
from roborock.code_mappings import RoborockStateCode
from roborock.roborock_typing import RoborockCommand
from homeassistant.components.vacuum import (
STATE_CLEANING,
@ -22,51 +22,46 @@ from homeassistant.util import slugify
from .const import DOMAIN
from .coordinator import RoborockDataUpdateCoordinator
from .device import RoborockCoordinatedEntity
from .models import RoborockHassDeviceInfo
STATE_CODE_TO_STATE = {
RoborockStateCode["1"]: STATE_IDLE, # "Starting"
RoborockStateCode["2"]: STATE_IDLE, # "Charger disconnected"
RoborockStateCode["3"]: STATE_IDLE, # "Idle"
RoborockStateCode["4"]: STATE_CLEANING, # "Remote control active"
RoborockStateCode["5"]: STATE_CLEANING, # "Cleaning"
RoborockStateCode["6"]: STATE_RETURNING, # "Returning home"
RoborockStateCode["7"]: STATE_CLEANING, # "Manual mode"
RoborockStateCode["8"]: STATE_DOCKED, # "Charging"
RoborockStateCode["9"]: STATE_ERROR, # "Charging problem"
RoborockStateCode["10"]: STATE_PAUSED, # "Paused"
RoborockStateCode["11"]: STATE_CLEANING, # "Spot cleaning"
RoborockStateCode["12"]: STATE_ERROR, # "Error"
RoborockStateCode["13"]: STATE_IDLE, # "Shutting down"
RoborockStateCode["14"]: STATE_DOCKED, # "Updating"
RoborockStateCode["15"]: STATE_RETURNING, # "Docking"
RoborockStateCode["16"]: STATE_CLEANING, # "Going to target"
RoborockStateCode["17"]: STATE_CLEANING, # "Zoned cleaning"
RoborockStateCode["18"]: STATE_CLEANING, # "Segment cleaning"
RoborockStateCode["22"]: STATE_DOCKED, # "Emptying the bin" on s7+
RoborockStateCode["23"]: STATE_DOCKED, # "Washing the mop" on s7maxV
RoborockStateCode["26"]: STATE_RETURNING, # "Going to wash the mop" on s7maxV
RoborockStateCode["100"]: STATE_DOCKED, # "Charging complete"
RoborockStateCode["101"]: STATE_ERROR, # "Device offline"
RoborockStateCode.starting: STATE_IDLE, # "Starting"
RoborockStateCode.charger_disconnected: STATE_IDLE, # "Charger disconnected"
RoborockStateCode.idle: STATE_IDLE, # "Idle"
RoborockStateCode.remote_control_active: STATE_CLEANING, # "Remote control active"
RoborockStateCode.cleaning: STATE_CLEANING, # "Cleaning"
RoborockStateCode.returning_home: STATE_RETURNING, # "Returning home"
RoborockStateCode.manual_mode: STATE_CLEANING, # "Manual mode"
RoborockStateCode.charging: STATE_DOCKED, # "Charging"
RoborockStateCode.charging_problem: STATE_ERROR, # "Charging problem"
RoborockStateCode.paused: STATE_PAUSED, # "Paused"
RoborockStateCode.spot_cleaning: STATE_CLEANING, # "Spot cleaning"
RoborockStateCode.error: STATE_ERROR, # "Error"
RoborockStateCode.shutting_down: STATE_IDLE, # "Shutting down"
RoborockStateCode.updating: STATE_DOCKED, # "Updating"
RoborockStateCode.docking: STATE_RETURNING, # "Docking"
RoborockStateCode.going_to_target: STATE_CLEANING, # "Going to target"
RoborockStateCode.zoned_cleaning: STATE_CLEANING, # "Zoned cleaning"
RoborockStateCode.segment_cleaning: STATE_CLEANING, # "Segment cleaning"
RoborockStateCode.emptying_the_bin: STATE_DOCKED, # "Emptying the bin" on s7+
RoborockStateCode.washing_the_mop: STATE_DOCKED, # "Washing the mop" on s7maxV
RoborockStateCode.going_to_wash_the_mop: STATE_RETURNING, # "Going to wash the mop" on s7maxV
RoborockStateCode.charging_complete: STATE_DOCKED, # "Charging complete"
RoborockStateCode.device_offline: STATE_ERROR, # "Device offline"
}
ATTR_STATUS = "status"
ATTR_ERROR = "error"
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Roborock sensor."""
coordinator: RoborockDataUpdateCoordinator = hass.data[DOMAIN][
coordinators: dict[str, RoborockDataUpdateCoordinator] = hass.data[DOMAIN][
config_entry.entry_id
]
async_add_entities(
RoborockVacuum(slugify(device_id), device_info, coordinator)
for device_id, device_info in coordinator.devices_info.items()
RoborockVacuum(slugify(device_id), coordinator)
for device_id, coordinator in coordinators.items()
)
@ -87,28 +82,22 @@ class RoborockVacuum(RoborockCoordinatedEntity, StateVacuumEntity):
| VacuumEntityFeature.STATE
| VacuumEntityFeature.START
)
_attr_fan_speed_list = RoborockFanPowerCode.values()
def __init__(
self,
unique_id: str,
device: RoborockHassDeviceInfo,
coordinator: RoborockDataUpdateCoordinator,
) -> None:
"""Initialize a vacuum."""
StateVacuumEntity.__init__(self)
RoborockCoordinatedEntity.__init__(self, unique_id, device, coordinator)
RoborockCoordinatedEntity.__init__(self, unique_id, coordinator)
self._attr_fan_speed_list = self._device_status.fan_power.keys()
@property
def state(self) -> str | None:
"""Return the status of the vacuum cleaner."""
return STATE_CODE_TO_STATE.get(self._device_status.state)
@property
def status(self) -> str | None:
"""Return the status of the vacuum cleaner."""
return self._device_status.status
@property
def battery_level(self) -> int | None:
"""Return the battery level of the vacuum cleaner."""
@ -117,12 +106,12 @@ class RoborockVacuum(RoborockCoordinatedEntity, StateVacuumEntity):
@property
def fan_speed(self) -> str | None:
"""Return the fan speed of the vacuum cleaner."""
return self._device_status.fan_power
return self._device_status.fan_power.name
@property
def error(self) -> str | None:
"""Get the error str if an error code exists."""
return self._device_status.error
def status(self) -> str | None:
"""Return the status of the vacuum cleaner."""
return self._device_status.state.name
async def async_start(self) -> None:
"""Start the vacuum."""
@ -152,11 +141,11 @@ class RoborockVacuum(RoborockCoordinatedEntity, StateVacuumEntity):
"""Set vacuum fan speed."""
await self.send(
RoborockCommand.SET_CUSTOM_MODE,
[k for k, v in RoborockFanPowerCode.items() if v == fan_speed],
[self._device_status.fan_power.as_dict().get(fan_speed)],
)
await self.coordinator.async_request_refresh()
async def async_start_pause(self):
async def async_start_pause(self) -> None:
"""Start, pause or resume the cleaning task."""
if self.state == STATE_CLEANING:
await self.async_pause()

2
requirements_all.txt
View File

@ -2111,7 +2111,7 @@ python-qbittorrent==0.4.2
python-ripple-api==0.0.3
# homeassistant.components.roborock
python-roborock==0.8.3
python-roborock==0.17.0
# homeassistant.components.smarttub
python-smarttub==0.0.33

2
requirements_test_all.txt
View File

@ -1531,7 +1531,7 @@ python-picnic-api==1.1.0
python-qbittorrent==0.4.2
# homeassistant.components.roborock
python-roborock==0.8.3
python-roborock==0.17.0
# homeassistant.components.smarttub
python-smarttub==0.0.33

11
tests/components/roborock/mock_data.py
View File

@ -5,13 +5,13 @@ from roborock.containers import (
CleanRecord,
CleanSummary,
Consumable,
DNDTimer,
DnDTimer,
HomeData,
NetworkInfo,
Status,
S7Status,
UserData,
)
from roborock.typing import DeviceProp
from roborock.roborock_typing import DeviceProp
# All data is based on a U.S. customer with a Roborock S7 MaxV Ultra
USER_EMAIL = "user@domain.com"
@ -311,7 +311,7 @@ CONSUMABLE = Consumable.from_dict(
}
)
DND_TIMER = DNDTimer.from_dict(
DND_TIMER = DnDTimer.from_dict(
{
"start_hour": 22,
"start_minute": 0,
@ -321,7 +321,7 @@ DND_TIMER = DNDTimer.from_dict(
}
)
STATUS = Status.from_dict(
STATUS = S7Status.from_dict(
{
"msg_ver": 2,
"msg_seq": 458,
@ -367,7 +367,6 @@ STATUS = Status.from_dict(
"unsave_map_flag": 0,
}
)
PROP = DeviceProp(STATUS, DND_TIMER, CLEAN_SUMMARY, CONSUMABLE, CLEAN_RECORD)
NETWORK_INFO = NetworkInfo(

23
tests/components/roborock/test_init.py
View File

@ -1,8 +1,6 @@
"""Test for Roborock init."""
from unittest.mock import patch
from roborock.exceptions import RoborockTimeout
from homeassistant.components.roborock.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@ -10,7 +8,6 @@ from homeassistant.helpers.update_coordinator import UpdateFailed
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.components.roborock.mock_data import HOME_DATA, NETWORK_INFO
async def test_unload_entry(
@ -41,23 +38,3 @@ async def test_config_entry_not_ready(
):
await async_setup_component(hass, DOMAIN, {})
assert mock_roborock_entry.state is ConfigEntryState.SETUP_RETRY
async def test_continue_setup_mqtt_disconnect_fail(
hass: HomeAssistant, mock_roborock_entry: MockConfigEntry
):
"""Test that if disconnect fails, we still continue setting up."""
with patch(
"homeassistant.components.roborock.RoborockApiClient.get_home_data",
return_value=HOME_DATA,
), patch(
"homeassistant.components.roborock.RoborockMqttClient.get_networking",
return_value=NETWORK_INFO,
), patch(
"homeassistant.components.roborock.RoborockMqttClient.async_disconnect",
side_effect=RoborockTimeout(),
), patch(
"homeassistant.components.roborock.RoborockDataUpdateCoordinator.async_config_entry_first_refresh"
):
await async_setup_component(hass, DOMAIN, {})
assert mock_roborock_entry.state is ConfigEntryState.LOADED

9
tests/components/roborock/test_vacuum.py
View File

@ -5,7 +5,7 @@ from typing import Any
from unittest.mock import patch
import pytest
from roborock.typing import RoborockCommand
from roborock.roborock_typing import RoborockCommand
from homeassistant.components.vacuum import (
SERVICE_CLEAN_SPOT,
@ -50,7 +50,7 @@ async def test_registry_entries(
(
SERVICE_SET_FAN_SPEED,
RoborockCommand.SET_CUSTOM_MODE,
{"fan_speed": "silent"},
{"fan_speed": "quiet"},
[101],
),
(
@ -86,6 +86,5 @@ async def test_commands(
blocking=True,
)
assert mock_send_command.call_count == 1
assert mock_send_command.call_args[0][0] == DEVICE_ID
assert mock_send_command.call_args[0][1] == command
assert mock_send_command.call_args[0][2] == called_params
assert mock_send_command.call_args[0][0] == command
assert mock_send_command.call_args[0][1] == called_params