Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Remove neato integration (#154902)

Browse Source
This commit is contained in:
Michael
2025-10-29 14:05:59 +01:00
committed by GitHub
parent fbb07e16cb
commit a6c1ce86d8
30 changed files with 0 additions and 1574 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@@ -361,7 +361,6 @@ homeassistant.components.myuplink.*
homeassistant.components.nam.*
homeassistant.components.nanoleaf.*
homeassistant.components.nasweb.*
homeassistant.components.neato.*
homeassistant.components.nest.*
homeassistant.components.netatmo.*
homeassistant.components.network.*

76
homeassistant/components/neato/__init__.py
View File

@@ -1,76 +0,0 @@
"""Support for Neato botvac connected vacuum cleaners."""
import logging
import aiohttp
from pybotvac import Account
from pybotvac.exceptions import NeatoException
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_entry_oauth2_flow
from . import api
from .const import NEATO_DOMAIN, NEATO_LOGIN
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [
Platform.BUTTON,
Platform.CAMERA,
Platform.SENSOR,
Platform.SWITCH,
Platform.VACUUM,
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up config entry."""
hass.data.setdefault(NEATO_DOMAIN, {})
if CONF_TOKEN not in entry.data:
raise ConfigEntryAuthFailed
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
try:
await session.async_ensure_token_valid()
except aiohttp.ClientResponseError as ex:
_LOGGER.debug("API error: %s (%s)", ex.code, ex.message)
if ex.code in (401, 403):
raise ConfigEntryAuthFailed("Token not valid, trigger renewal") from ex
raise ConfigEntryNotReady from ex
neato_session = api.ConfigEntryAuth(hass, entry, implementation)
hass.data[NEATO_DOMAIN][entry.entry_id] = neato_session
hub = NeatoHub(hass, Account(neato_session))
await hub.async_update_entry_unique_id(entry)
try:
await hass.async_add_executor_job(hub.update_robots)
except NeatoException as ex:
_LOGGER.debug("Failed to connect to Neato API")
raise ConfigEntryNotReady from ex
hass.data[NEATO_LOGIN] = hub
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[NEATO_DOMAIN].pop(entry.entry_id)
return unload_ok

58
homeassistant/components/neato/api.py
View File

@@ -1,58 +0,0 @@
"""API for Neato Botvac bound to Home Assistant OAuth."""
from __future__ import annotations
from asyncio import run_coroutine_threadsafe
from typing import Any
import pybotvac
from homeassistant import config_entries, core
from homeassistant.components.application_credentials import AuthImplementation
from homeassistant.helpers import config_entry_oauth2_flow
class ConfigEntryAuth(pybotvac.OAuthSession): # type: ignore[misc]
"""Provide Neato Botvac authentication tied to an OAuth2 based config entry."""
def __init__(
self,
hass: core.HomeAssistant,
config_entry: config_entries.ConfigEntry,
implementation: config_entry_oauth2_flow.AbstractOAuth2Implementation,
) -> None:
"""Initialize Neato Botvac Auth."""
self.hass = hass
self.session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, implementation
)
super().__init__(self.session.token, vendor=pybotvac.Neato())
def refresh_tokens(self) -> str:
"""Refresh and return new Neato Botvac tokens."""
run_coroutine_threadsafe(
self.session.async_ensure_token_valid(), self.hass.loop
).result()
return self.session.token["access_token"] # type: ignore[no-any-return]
class NeatoImplementation(AuthImplementation):
"""Neato implementation of LocalOAuth2Implementation.
We need this class because we have to add client_secret
and scope to the authorization request.
"""
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
return {"client_secret": self.client_secret}
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Generate a url for the user to authorize.
We must make sure that the plus signs are not encoded.
"""
url = await super().async_generate_authorize_url(flow_id)
return f"{url}&scope=public_profile+control_robots+maps"

28
homeassistant/components/neato/application_credentials.py
View File

@@ -1,28 +0,0 @@
"""Application credentials platform for neato."""
from pybotvac import Neato
from homeassistant.components.application_credentials import (
AuthorizationServer,
ClientCredential,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from . import api
async def async_get_auth_implementation(
hass: HomeAssistant, auth_domain: str, credential: ClientCredential
) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
"""Return auth implementation for a custom auth implementation."""
vendor = Neato()
return api.NeatoImplementation(
hass,
auth_domain,
credential,
AuthorizationServer(
authorize_url=vendor.auth_endpoint,
token_url=vendor.token_endpoint,
),
)

44
homeassistant/components/neato/button.py
View File

@@ -1,44 +0,0 @@
"""Support for Neato buttons."""
from __future__ import annotations
from pybotvac import Robot
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_ROBOTS
from .entity import NeatoEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato button from config entry."""
entities = [NeatoDismissAlertButton(robot) for robot in hass.data[NEATO_ROBOTS]]
async_add_entities(entities, True)
class NeatoDismissAlertButton(NeatoEntity, ButtonEntity):
"""Representation of a dismiss_alert button entity."""
_attr_translation_key = "dismiss_alert"
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
robot: Robot,
) -> None:
"""Initialize a dismiss_alert Neato button entity."""
super().__init__(robot)
self._attr_unique_id = f"{robot.serial}_dismiss_alert"
async def async_press(self) -> None:
"""Press the button."""
await self.hass.async_add_executor_job(self.robot.dismiss_current_alert)

130
homeassistant/components/neato/camera.py
View File

@@ -1,130 +0,0 @@
"""Support for loading picture from Neato."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from urllib3.response import HTTPResponse
from homeassistant.components.camera import Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_MAP_DATA, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
ATTR_GENERATED_AT = "generated_at"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato camera with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
dev = [
NeatoCleaningMap(neato, robot, mapdata)
for robot in hass.data[NEATO_ROBOTS]
if "maps" in robot.traits
]
if not dev:
return
_LOGGER.debug("Adding robots for cleaning maps %s", dev)
async_add_entities(dev, True)
class NeatoCleaningMap(NeatoEntity, Camera):
"""Neato cleaning map for last clean."""
_attr_translation_key = "cleaning_map"
def __init__(
self, neato: NeatoHub, robot: Robot, mapdata: dict[str, Any] | None
) -> None:
"""Initialize Neato cleaning map."""
super().__init__(robot)
Camera.__init__(self)
self.neato = neato
self._mapdata = mapdata
self._available = neato is not None
self._robot_serial: str = self.robot.serial
self._attr_unique_id = self.robot.serial
self._generated_at: str | None = None
self._image_url: str | None = None
self._image: bytes | None = None
def camera_image(
self, width: int | None = None, height: int | None = None
) -> bytes | None:
"""Return image response."""
self.update()
return self._image
def update(self) -> None:
"""Check the contents of the map list."""
_LOGGER.debug("Running camera update for '%s'", self.entity_id)
try:
self.neato.update_robots()
except NeatoRobotException as ex:
if self._available: # Print only once when available
_LOGGER.error(
"Neato camera connection error for '%s': %s", self.entity_id, ex
)
self._image = None
self._image_url = None
self._available = False
return
if self._mapdata:
map_data: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
if (image_url := map_data["url"]) == self._image_url:
_LOGGER.debug(
"The map image_url for '%s' is the same as old", self.entity_id
)
return
try:
image: HTTPResponse = self.neato.download_map(image_url)
except NeatoRobotException as ex:
if self._available: # Print only once when available
_LOGGER.error(
"Neato camera connection error for '%s': %s", self.entity_id, ex
)
self._image = None
self._image_url = None
self._available = False
return
self._image = image.read()
self._image_url = image_url
self._generated_at = map_data.get("generated_at")
self._available = True
@property
def available(self) -> bool:
"""Return if the robot is available."""
return self._available
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the vacuum cleaner."""
data: dict[str, Any] = {}
if self._generated_at is not None:
data[ATTR_GENERATED_AT] = self._generated_at
return data

64
homeassistant/components/neato/config_flow.py
View File

@@ -1,64 +0,0 @@
"""Config flow for Neato Botvac."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from .const import NEATO_DOMAIN
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=NEATO_DOMAIN
):
"""Config flow to handle Neato Botvac OAuth2 authentication."""
DOMAIN = NEATO_DOMAIN
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return logging.getLogger(__name__)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Create an entry for the flow."""
current_entries = self._async_current_entries()
if self.source != SOURCE_REAUTH and current_entries:
# Already configured
return self.async_abort(reason="already_configured")
return await super().async_step_user(user_input=user_input)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Perform reauth upon migration of old entries."""
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm reauth upon migration of old entries."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()
async def async_oauth_create_entry(self, data: dict[str, Any]) -> ConfigFlowResult:
"""Create an entry for the flow. Update an entry if one already exist."""
current_entries = self._async_current_entries()
if self.source == SOURCE_REAUTH and current_entries:
# Update entry
self.hass.config_entries.async_update_entry(
current_entries[0], title=self.flow_impl.name, data=data
)
self.hass.async_create_task(
self.hass.config_entries.async_reload(current_entries[0].entry_id)
)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=self.flow_impl.name, data=data)

150
homeassistant/components/neato/const.py
View File

@@ -1,150 +0,0 @@
"""Constants for Neato integration."""
NEATO_DOMAIN = "neato"
CONF_VENDOR = "vendor"
NEATO_LOGIN = "neato_login"
NEATO_MAP_DATA = "neato_map_data"
NEATO_PERSISTENT_MAPS = "neato_persistent_maps"
NEATO_ROBOTS = "neato_robots"
SCAN_INTERVAL_MINUTES = 1
MODE = {1: "Eco", 2: "Turbo"}
ACTION = {
0: "Invalid",
1: "House Cleaning",
2: "Spot Cleaning",
3: "Manual Cleaning",
4: "Docking",
5: "User Menu Active",
6: "Suspended Cleaning",
7: "Updating",
8: "Copying logs",
9: "Recovering Location",
10: "IEC test",
11: "Map cleaning",
12: "Exploring map (creating a persistent map)",
13: "Acquiring Persistent Map IDs",
14: "Creating & Uploading Map",
15: "Suspended Exploration",
}
ERRORS = {
"ui_error_battery_battundervoltlithiumsafety": "Replace battery",
"ui_error_battery_critical": "Replace battery",
"ui_error_battery_invalidsensor": "Replace battery",
"ui_error_battery_lithiumadapterfailure": "Replace battery",
"ui_error_battery_mismatch": "Replace battery",
"ui_error_battery_nothermistor": "Replace battery",
"ui_error_battery_overtemp": "Replace battery",
"ui_error_battery_overvolt": "Replace battery",
"ui_error_battery_undercurrent": "Replace battery",
"ui_error_battery_undertemp": "Replace battery",
"ui_error_battery_undervolt": "Replace battery",
"ui_error_battery_unplugged": "Replace battery",
"ui_error_brush_stuck": "Brush stuck",
"ui_error_brush_overloaded": "Brush overloaded",
"ui_error_bumper_stuck": "Bumper stuck",
"ui_error_check_battery_switch": "Check battery",
"ui_error_corrupt_scb": "Call customer service corrupt board",
"ui_error_deck_debris": "Deck debris",
"ui_error_dflt_app": "Check Neato app",
"ui_error_disconnect_chrg_cable": "Disconnected charge cable",
"ui_error_disconnect_usb_cable": "Disconnected USB cable",
"ui_error_dust_bin_missing": "Dust bin missing",
"ui_error_dust_bin_full": "Dust bin full",
"ui_error_dust_bin_emptied": "Dust bin emptied",
"ui_error_hardware_failure": "Hardware failure",
"ui_error_ldrop_stuck": "Clear my path",
"ui_error_lds_jammed": "Clear my path",
"ui_error_lds_bad_packets": "Check Neato app",
"ui_error_lds_disconnected": "Check Neato app",
"ui_error_lds_missed_packets": "Check Neato app",
"ui_error_lwheel_stuck": "Clear my path",
"ui_error_navigation_backdrop_frontbump": "Clear my path",
"ui_error_navigation_backdrop_leftbump": "Clear my path",
"ui_error_navigation_backdrop_wheelextended": "Clear my path",
"ui_error_navigation_noprogress": "Clear my path",
"ui_error_navigation_origin_unclean": "Clear my path",
"ui_error_navigation_pathproblems": "Cannot return to base",
"ui_error_navigation_pinkycommsfail": "Clear my path",
"ui_error_navigation_falling": "Clear my path",
"ui_error_navigation_noexitstogo": "Clear my path",
"ui_error_navigation_nomotioncommands": "Clear my path",
"ui_error_navigation_rightdrop_leftbump": "Clear my path",
"ui_error_navigation_undockingfailed": "Clear my path",
"ui_error_picked_up": "Picked up",
"ui_error_qa_fail": "Check Neato app",
"ui_error_rdrop_stuck": "Clear my path",
"ui_error_reconnect_failed": "Reconnect failed",
"ui_error_rwheel_stuck": "Clear my path",
"ui_error_stuck": "Stuck!",
"ui_error_unable_to_return_to_base": "Unable to return to base",
"ui_error_unable_to_see": "Clean vacuum sensors",
"ui_error_vacuum_slip": "Clear my path",
"ui_error_vacuum_stuck": "Clear my path",
"ui_error_warning": "Error check app",
"batt_base_connect_fail": "Battery failed to connect to base",
"batt_base_no_power": "Battery base has no power",
"batt_low": "Battery low",
"batt_on_base": "Battery on base",
"clean_tilt_on_start": "Clean the tilt on start",
"dustbin_full": "Dust bin full",
"dustbin_missing": "Dust bin missing",
"gen_picked_up": "Picked up",
"hw_fail": "Hardware failure",
"hw_tof_sensor_sensor": "Hardware sensor disconnected",
"lds_bad_packets": "Bad packets",
"lds_deck_debris": "Debris on deck",
"lds_disconnected": "Disconnected",
"lds_jammed": "Jammed",
"lds_missed_packets": "Missed packets",
"maint_brush_stuck": "Brush stuck",
"maint_brush_overload": "Brush overloaded",
"maint_bumper_stuck": "Bumper stuck",
"maint_customer_support_qa": "Contact customer support",
"maint_vacuum_stuck": "Vacuum is stuck",
"maint_vacuum_slip": "Vacuum is stuck",
"maint_left_drop_stuck": "Vacuum is stuck",
"maint_left_wheel_stuck": "Vacuum is stuck",
"maint_right_drop_stuck": "Vacuum is stuck",
"maint_right_wheel_stuck": "Vacuum is stuck",
"not_on_charge_base": "Not on the charge base",
"nav_robot_falling": "Clear my path",
"nav_no_path": "Clear my path",
"nav_path_problem": "Clear my path",
"nav_backdrop_frontbump": "Clear my path",
"nav_backdrop_leftbump": "Clear my path",
"nav_backdrop_wheelextended": "Clear my path",
"nav_floorplan_zone_path_blocked": "Clear my path",
"nav_mag_sensor": "Clear my path",
"nav_no_exit": "Clear my path",
"nav_no_movement": "Clear my path",
"nav_rightdrop_leftbump": "Clear my path",
"nav_undocking_failed": "Clear my path",
}
ALERTS = {
"ui_alert_dust_bin_full": "Please empty dust bin",
"ui_alert_recovering_location": "Returning to start",
"ui_alert_battery_chargebasecommerr": "Battery error",
"ui_alert_busy_charging": "Busy charging",
"ui_alert_charging_base": "Base charging",
"ui_alert_charging_power": "Charging power",
"ui_alert_connect_chrg_cable": "Connect charge cable",
"ui_alert_info_thank_you": "Thank you",
"ui_alert_invalid": "Invalid check app",
"ui_alert_old_error": "Old error",
"ui_alert_swupdate_fail": "Update failed",
"dustbin_full": "Please empty dust bin",
"maint_brush_change": "Change the brush",
"maint_filter_change": "Change the filter",
"clean_completed_to_start": "Cleaning completed",
"nav_floorplan_not_created": "No floorplan found",
"nav_floorplan_load_fail": "Failed to load floorplan",
"nav_floorplan_localization_fail": "Failed to load floorplan",
"clean_incomplete_to_start": "Cleaning incomplete",
"log_upload_failed": "Logs failed to upload",
}

24
homeassistant/components/neato/entity.py
View File

@@ -1,24 +0,0 @@
"""Base entity for Neato."""
from __future__ import annotations
from pybotvac import Robot
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from .const import NEATO_DOMAIN
class NeatoEntity(Entity):
"""Base Neato entity."""
_attr_has_entity_name = True
def __init__(self, robot: Robot) -> None:
"""Initialize Neato entity."""
self.robot = robot
self._attr_device_info: DeviceInfo = DeviceInfo(
identifiers={(NEATO_DOMAIN, self.robot.serial)},
name=self.robot.name,
)

50
homeassistant/components/neato/hub.py
View File

@@ -1,50 +0,0 @@
"""Support for Neato botvac connected vacuum cleaners."""
from datetime import timedelta
import logging
from pybotvac import Account
from urllib3.response import HTTPResponse
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.util import Throttle
from .const import NEATO_MAP_DATA, NEATO_PERSISTENT_MAPS, NEATO_ROBOTS
_LOGGER = logging.getLogger(__name__)
class NeatoHub:
"""A My Neato hub wrapper class."""
def __init__(self, hass: HomeAssistant, neato: Account) -> None:
"""Initialize the Neato hub."""
self._hass = hass
self.my_neato: Account = neato
@Throttle(timedelta(minutes=1))
def update_robots(self) -> None:
"""Update the robot states."""
_LOGGER.debug("Running HUB.update_robots %s", self._hass.data.get(NEATO_ROBOTS))
self._hass.data[NEATO_ROBOTS] = self.my_neato.robots
self._hass.data[NEATO_PERSISTENT_MAPS] = self.my_neato.persistent_maps
self._hass.data[NEATO_MAP_DATA] = self.my_neato.maps
def download_map(self, url: str) -> HTTPResponse:
"""Download a new map image."""
map_image_data: HTTPResponse = self.my_neato.get_map_image(url)
return map_image_data
async def async_update_entry_unique_id(self, entry: ConfigEntry) -> str:
"""Update entry for unique_id."""
await self._hass.async_add_executor_job(self.my_neato.refresh_userdata)
unique_id: str = self.my_neato.unique_id
if entry.unique_id == unique_id:
return unique_id
_LOGGER.debug("Updating user unique_id for previous config entry")
self._hass.config_entries.async_update_entry(entry, unique_id=unique_id)
return unique_id

7
homeassistant/components/neato/icons.json
View File

@@ -1,7 +0,0 @@
{
"services": {
"custom_cleaning": {
"service": "mdi:broom"
}
}
}

11
homeassistant/components/neato/manifest.json
View File

@@ -1,11 +0,0 @@
{
"domain": "neato",
"name": "Neato Botvac",
"codeowners": [],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/neato",
"iot_class": "cloud_polling",
"loggers": ["pybotvac"],
"requirements": ["pybotvac==0.0.28"]
}

81
homeassistant/components/neato/sensor.py
View File

@@ -1,81 +0,0 @@
"""Support for Neato sensors."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
BATTERY = "Battery"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Neato sensor using config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
dev = [NeatoSensor(neato, robot) for robot in hass.data[NEATO_ROBOTS]]
if not dev:
return
_LOGGER.debug("Adding robots for sensors %s", dev)
async_add_entities(dev, True)
class NeatoSensor(NeatoEntity, SensorEntity):
"""Neato sensor."""
_attr_device_class = SensorDeviceClass.BATTERY
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_native_unit_of_measurement = PERCENTAGE
_attr_available: bool = False
def __init__(self, neato: NeatoHub, robot: Robot) -> None:
"""Initialize Neato sensor."""
super().__init__(robot)
self._robot_serial: str = self.robot.serial
self._attr_unique_id = self.robot.serial
self._state: dict[str, Any] | None = None
def update(self) -> None:
"""Update Neato Sensor."""
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available:
_LOGGER.error(
"Neato sensor connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
@property
def native_value(self) -> str | None:
"""Return the state."""
if self._state is not None:
return str(self._state["details"]["charge"])
return None

32
homeassistant/components/neato/services.yaml
View File

@@ -1,32 +0,0 @@
custom_cleaning:
target:
entity:
integration: neato
domain: vacuum
fields:
mode:
default: 2
selector:
number:
min: 1
max: 2
mode: box
navigation:
default: 1
selector:
number:
min: 1
max: 3
mode: box
category:
default: 4
selector:
number:
min: 2
max: 4
step: 2
mode: box
zone:
example: "Kitchen"
selector:
text:

73
homeassistant/components/neato/strings.json
View File

@@ -1,73 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"step": {
"pick_implementation": {
"data": {
"implementation": "[%key:common::config_flow::data::implementation%]"
},
"data_description": {
"implementation": "[%key:common::config_flow::description::implementation%]"
},
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]"
},
"reauth_confirm": {
"title": "[%key:common::config_flow::description::confirm_setup%]"
}
}
},
"entity": {
"button": {
"dismiss_alert": {
"name": "Dismiss alert"
}
},
"camera": {
"cleaning_map": {
"name": "Cleaning map"
}
},
"switch": {
"schedule": {
"name": "Schedule"
}
}
},
"services": {
"custom_cleaning": {
"description": "Starts a custom cleaning of your house.",
"fields": {
"category": {
"description": "Whether to use a persistent map or not for cleaning (i.e. No go lines): 2 for no map, 4 for map. Default to using map if not set (and fallback to no map if no map is found).",
"name": "Use cleaning map"
},
"mode": {
"description": "Sets the cleaning mode: 1 for eco and 2 for turbo. Defaults to turbo if not set.",
"name": "Cleaning mode"
},
"navigation": {
"description": "Sets the navigation mode: 1 for normal, 2 for extra care, 3 for deep. Defaults to normal if not set.",
"name": "Navigation mode"
},
"zone": {
"description": "Name of the zone to clean (only supported on the Botvac D7). Defaults to no zone i.e. complete house cleanup.",
"name": "Zone"
}
},
"name": "Custom cleaning"
}
}
}

118
homeassistant/components/neato/switch.py
View File

@@ -1,118 +0,0 @@
"""Support for Neato Connected Vacuums switches."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac.exceptions import NeatoRobotException
from pybotvac.robot import Robot
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_OFF, STATE_ON, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import NEATO_LOGIN, NEATO_ROBOTS, SCAN_INTERVAL_MINUTES
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
SWITCH_TYPE_SCHEDULE = "schedule"
SWITCH_TYPES = {SWITCH_TYPE_SCHEDULE: ["Schedule"]}
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato switch with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
dev = [
NeatoConnectedSwitch(neato, robot, type_name)
for robot in hass.data[NEATO_ROBOTS]
for type_name in SWITCH_TYPES
]
if not dev:
return
_LOGGER.debug("Adding switches %s", dev)
async_add_entities(dev, True)
class NeatoConnectedSwitch(NeatoEntity, SwitchEntity):
"""Neato Connected Switches."""
_attr_translation_key = "schedule"
_attr_available = False
_attr_entity_category = EntityCategory.CONFIG
def __init__(self, neato: NeatoHub, robot: Robot, switch_type: str) -> None:
"""Initialize the Neato Connected switches."""
super().__init__(robot)
self.type = switch_type
self._state: dict[str, Any] | None = None
self._schedule_state: str | None = None
self._clean_state = None
self._attr_unique_id = self.robot.serial
def update(self) -> None:
"""Update the states of Neato switches."""
_LOGGER.debug("Running Neato switch update for '%s'", self.entity_id)
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available: # Print only once when available
_LOGGER.error(
"Neato switch connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
if self.type == SWITCH_TYPE_SCHEDULE:
_LOGGER.debug("State: %s", self._state)
if self._state is not None and self._state["details"]["isScheduleEnabled"]:
self._schedule_state = STATE_ON
else:
self._schedule_state = STATE_OFF
_LOGGER.debug(
"Schedule state for '%s': %s", self.entity_id, self._schedule_state
)
@property
def is_on(self) -> bool:
"""Return true if switch is on."""
return bool(
self.type == SWITCH_TYPE_SCHEDULE and self._schedule_state == STATE_ON
)
def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.enable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato switch connection error '%s': %s", self.entity_id, ex
)
def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
if self.type == SWITCH_TYPE_SCHEDULE:
try:
self.robot.disable_schedule()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato switch connection error '%s': %s", self.entity_id, ex
)

388
homeassistant/components/neato/vacuum.py
View File

@@ -1,388 +0,0 @@
"""Support for Neato Connected Vacuums."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from pybotvac import Robot
from pybotvac.exceptions import NeatoRobotException
import voluptuous as vol
from homeassistant.components.vacuum import (
ATTR_STATUS,
StateVacuumEntity,
VacuumActivity,
VacuumEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_MODE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
ACTION,
ALERTS,
ERRORS,
MODE,
NEATO_LOGIN,
NEATO_MAP_DATA,
NEATO_PERSISTENT_MAPS,
NEATO_ROBOTS,
SCAN_INTERVAL_MINUTES,
)
from .entity import NeatoEntity
from .hub import NeatoHub
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=SCAN_INTERVAL_MINUTES)
ATTR_CLEAN_START = "clean_start"
ATTR_CLEAN_STOP = "clean_stop"
ATTR_CLEAN_AREA = "clean_area"
ATTR_CLEAN_BATTERY_START = "battery_level_at_clean_start"
ATTR_CLEAN_BATTERY_END = "battery_level_at_clean_end"
ATTR_CLEAN_SUSP_COUNT = "clean_suspension_count"
ATTR_CLEAN_SUSP_TIME = "clean_suspension_time"
ATTR_CLEAN_PAUSE_TIME = "clean_pause_time"
ATTR_CLEAN_ERROR_TIME = "clean_error_time"
ATTR_LAUNCHED_FROM = "launched_from"
ATTR_NAVIGATION = "navigation"
ATTR_CATEGORY = "category"
ATTR_ZONE = "zone"
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Neato vacuum with config entry."""
neato: NeatoHub = hass.data[NEATO_LOGIN]
mapdata: dict[str, Any] | None = hass.data.get(NEATO_MAP_DATA)
persistent_maps: dict[str, Any] | None = hass.data.get(NEATO_PERSISTENT_MAPS)
dev = [
NeatoConnectedVacuum(neato, robot, mapdata, persistent_maps)
for robot in hass.data[NEATO_ROBOTS]
]
if not dev:
return
_LOGGER.debug("Adding vacuums %s", dev)
async_add_entities(dev, True)
platform = entity_platform.async_get_current_platform()
assert platform is not None
platform.async_register_entity_service(
"custom_cleaning",
{
vol.Optional(ATTR_MODE, default=2): cv.positive_int,
vol.Optional(ATTR_NAVIGATION, default=1): cv.positive_int,
vol.Optional(ATTR_CATEGORY, default=4): cv.positive_int,
vol.Optional(ATTR_ZONE): cv.string,
},
"neato_custom_cleaning",
)
class NeatoConnectedVacuum(NeatoEntity, StateVacuumEntity):
"""Representation of a Neato Connected Vacuum."""
_attr_supported_features = (
VacuumEntityFeature.BATTERY
| VacuumEntityFeature.PAUSE
| VacuumEntityFeature.RETURN_HOME
| VacuumEntityFeature.STOP
| VacuumEntityFeature.START
| VacuumEntityFeature.CLEAN_SPOT
| VacuumEntityFeature.STATE
| VacuumEntityFeature.MAP
| VacuumEntityFeature.LOCATE
)
_attr_name = None
def __init__(
self,
neato: NeatoHub,
robot: Robot,
mapdata: dict[str, Any] | None,
persistent_maps: dict[str, Any] | None,
) -> None:
"""Initialize the Neato Connected Vacuum."""
super().__init__(robot)
self._attr_available: bool = neato is not None
self._mapdata = mapdata
self._robot_has_map: bool = self.robot.has_persistent_maps
self._robot_maps = persistent_maps
self._robot_serial: str = self.robot.serial
self._attr_unique_id: str = self.robot.serial
self._status_state: str | None = None
self._state: dict[str, Any] | None = None
self._clean_time_start: str | None = None
self._clean_time_stop: str | None = None
self._clean_area: float | None = None
self._clean_battery_start: int | None = None
self._clean_battery_end: int | None = None
self._clean_susp_charge_count: int | None = None
self._clean_susp_time: int | None = None
self._clean_pause_time: int | None = None
self._clean_error_time: int | None = None
self._launched_from: str | None = None
self._robot_boundaries: list = []
self._robot_stats: dict[str, Any] | None = None
def update(self) -> None:
"""Update the states of Neato Vacuums."""
_LOGGER.debug("Running Neato Vacuums update for '%s'", self.entity_id)
try:
if self._robot_stats is None:
self._robot_stats = self.robot.get_general_info().json().get("data")
except NeatoRobotException:
_LOGGER.warning("Couldn't fetch robot information of %s", self.entity_id)
try:
self._state = self.robot.state
except NeatoRobotException as ex:
if self._attr_available: # print only once when available
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
self._state = None
self._attr_available = False
return
if self._state is None:
return
self._attr_available = True
_LOGGER.debug("self._state=%s", self._state)
if "alert" in self._state:
robot_alert = ALERTS.get(self._state["alert"])
else:
robot_alert = None
if self._state["state"] == 1:
if self._state["details"]["isCharging"]:
self._attr_activity = VacuumActivity.DOCKED
self._status_state = "Charging"
elif (
self._state["details"]["isDocked"]
and not self._state["details"]["isCharging"]
):
self._attr_activity = VacuumActivity.DOCKED
self._status_state = "Docked"
else:
self._attr_activity = VacuumActivity.IDLE
self._status_state = "Stopped"
if robot_alert is not None:
self._status_state = robot_alert
elif self._state["state"] == 2:
if robot_alert is None:
self._attr_activity = VacuumActivity.CLEANING
self._status_state = (
f"{MODE.get(self._state['cleaning']['mode'])} "
f"{ACTION.get(self._state['action'])}"
)
if (
"boundary" in self._state["cleaning"]
and "name" in self._state["cleaning"]["boundary"]
):
self._status_state += (
f" {self._state['cleaning']['boundary']['name']}"
)
else:
self._status_state = robot_alert
elif self._state["state"] == 3:
self._attr_activity = VacuumActivity.PAUSED
self._status_state = "Paused"
elif self._state["state"] == 4:
self._attr_activity = VacuumActivity.ERROR
self._status_state = ERRORS.get(self._state["error"])
self._attr_battery_level = self._state["details"]["charge"]
if self._mapdata is None or not self._mapdata.get(self._robot_serial, {}).get(
"maps", []
):
return
mapdata: dict[str, Any] = self._mapdata[self._robot_serial]["maps"][0]
self._clean_time_start = mapdata["start_at"]
self._clean_time_stop = mapdata["end_at"]
self._clean_area = mapdata["cleaned_area"]
self._clean_susp_charge_count = mapdata["suspended_cleaning_charging_count"]
self._clean_susp_time = mapdata["time_in_suspended_cleaning"]
self._clean_pause_time = mapdata["time_in_pause"]
self._clean_error_time = mapdata["time_in_error"]
self._clean_battery_start = mapdata["run_charge_at_start"]
self._clean_battery_end = mapdata["run_charge_at_end"]
self._launched_from = mapdata["launched_from"]
if (
self._robot_has_map
and self._state
and self._state["availableServices"]["maps"] != "basic-1"
and self._robot_maps
):
allmaps: dict = self._robot_maps[self._robot_serial]
_LOGGER.debug(
"Found the following maps for '%s': %s", self.entity_id, allmaps
)
self._robot_boundaries = [] # Reset boundaries before refreshing boundaries
for maps in allmaps:
try:
robot_boundaries = self.robot.get_map_boundaries(maps["id"]).json()
except NeatoRobotException as ex:
_LOGGER.error(
"Could not fetch map boundaries for '%s': %s",
self.entity_id,
ex,
)
return
_LOGGER.debug(
"Boundaries for robot '%s' in map '%s': %s",
self.entity_id,
maps["name"],
robot_boundaries,
)
if "boundaries" in robot_boundaries["data"]:
self._robot_boundaries += robot_boundaries["data"]["boundaries"]
_LOGGER.debug(
"List of boundaries for '%s': %s",
self.entity_id,
self._robot_boundaries,
)
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the vacuum cleaner."""
data: dict[str, Any] = {}
if self._status_state is not None:
data[ATTR_STATUS] = self._status_state
if self._clean_time_start is not None:
data[ATTR_CLEAN_START] = self._clean_time_start
if self._clean_time_stop is not None:
data[ATTR_CLEAN_STOP] = self._clean_time_stop
if self._clean_area is not None:
data[ATTR_CLEAN_AREA] = self._clean_area
if self._clean_susp_charge_count is not None:
data[ATTR_CLEAN_SUSP_COUNT] = self._clean_susp_charge_count
if self._clean_susp_time is not None:
data[ATTR_CLEAN_SUSP_TIME] = self._clean_susp_time
if self._clean_pause_time is not None:
data[ATTR_CLEAN_PAUSE_TIME] = self._clean_pause_time
if self._clean_error_time is not None:
data[ATTR_CLEAN_ERROR_TIME] = self._clean_error_time
if self._clean_battery_start is not None:
data[ATTR_CLEAN_BATTERY_START] = self._clean_battery_start
if self._clean_battery_end is not None:
data[ATTR_CLEAN_BATTERY_END] = self._clean_battery_end
if self._launched_from is not None:
data[ATTR_LAUNCHED_FROM] = self._launched_from
return data
@property
def device_info(self) -> DeviceInfo:
"""Device info for neato robot."""
device_info = self._attr_device_info
if self._robot_stats:
device_info["manufacturer"] = self._robot_stats["battery"]["vendor"]
device_info["model"] = self._robot_stats["model"]
device_info["sw_version"] = self._robot_stats["firmware"]
return device_info
def start(self) -> None:
"""Start cleaning or resume cleaning."""
if self._state:
try:
if self._state["state"] == 1:
self.robot.start_cleaning()
elif self._state["state"] == 3:
self.robot.resume_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def pause(self) -> None:
"""Pause the vacuum."""
try:
self.robot.pause_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def return_to_base(self, **kwargs: Any) -> None:
"""Set the vacuum cleaner to return to the dock."""
try:
if self._attr_activity == VacuumActivity.CLEANING:
self.robot.pause_cleaning()
self._attr_activity = VacuumActivity.RETURNING
self.robot.send_to_base()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def stop(self, **kwargs: Any) -> None:
"""Stop the vacuum cleaner."""
try:
self.robot.stop_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def locate(self, **kwargs: Any) -> None:
"""Locate the robot by making it emit a sound."""
try:
self.robot.locate()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def clean_spot(self, **kwargs: Any) -> None:
"""Run a spot cleaning starting from the base."""
try:
self.robot.start_spot_cleaning()
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)
def neato_custom_cleaning(
self, mode: str, navigation: str, category: str, zone: str | None = None
) -> None:
"""Zone cleaning service call."""
boundary_id = None
if zone is not None:
for boundary in self._robot_boundaries:
if zone in boundary["name"]:
boundary_id = boundary["id"]
if boundary_id is None:
_LOGGER.error(
"Zone '%s' was not found for the robot '%s'", zone, self.entity_id
)
return
_LOGGER.debug(
"Start cleaning zone '%s' with robot %s", zone, self.entity_id
)
self._attr_activity = VacuumActivity.CLEANING
try:
self.robot.start_cleaning(mode, navigation, category, boundary_id)
except NeatoRobotException as ex:
_LOGGER.error(
"Neato vacuum connection error for '%s': %s", self.entity_id, ex
)

1
homeassistant/generated/application_credentials.py generated
View File

@@ -27,7 +27,6 @@ APPLICATION_CREDENTIALS = [
"miele",
"monzo",
"myuplink",
"neato",
"nest",
"netatmo",
"ondilo_ico",

1
homeassistant/generated/config_flows.py generated
View File

@@ -428,7 +428,6 @@ FLOWS = {
"nam",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"nest",
"netatmo",

6
homeassistant/generated/integrations.json
View File

@@ -4339,12 +4339,6 @@
"integration_type": "virtual",
"supported_by": "opower"
},
"neato": {
"name": "Neato Botvac",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"nederlandse_spoorwegen": {
"name": "Nederlandse Spoorwegen (NS)",
"integration_type": "service",

10
mypy.ini generated
View File

@@ -3366,16 +3366,6 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.neato.*]
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.nest.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt generated
View File

@@ -1911,9 +1911,6 @@ pyblackbird==0.6
# homeassistant.components.bluesound
pyblu==2.0.5
# homeassistant.components.neato
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.3.4

3
requirements_test_all.txt generated
View File

@@ -1616,9 +1616,6 @@ pyblackbird==0.6
# homeassistant.components.bluesound
pyblu==2.0.5
# homeassistant.components.neato
pybotvac==0.0.28
# homeassistant.components.braviatv
pybravia==0.3.4

2
script/hassfest/quality_scale.py
View File

@@ -673,7 +673,6 @@ INTEGRATIONS_WITHOUT_QUALITY_SCALE_FILE = [
"namecheapdns",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"ness_alarm",
"netatmo",
@@ -1706,7 +1705,6 @@ INTEGRATIONS_WITHOUT_SCALE = [
"namecheapdns",
"nanoleaf",
"nasweb",
"neato",
"nederlandse_spoorwegen",
"nest",
"ness_alarm",

1
tests/components/analytics_insights/fixtures/current_data.json
View File

@@ -577,7 +577,6 @@
"ourgroceries": 469,
"vicare": 1495,
"thermopro": 639,
"neato": 935,
"roon": 405,
"renault": 1287,
"bthome": 4166,

16
tests/components/homeassistant_alerts/fixtures/alerts_1.json
View File

@@ -125,22 +125,6 @@
"filename": "logi_circle.markdown",
"alert_url": "https://alerts.home-assistant.io/#logi_circle.markdown"
},
{
"id": "neato",
"title": "New Neato Botvacs Do Not Support Existing API",
"created": "2021-12-20T13:27:00.000Z",
"integrations": [
{
"package": "neato"
}
],
"homeassistant": {
"package": "homeassistant",
"affected_from_version": "0.30"
},
"filename": "neato.markdown",
"alert_url": "https://alerts.home-assistant.io/#neato.markdown"
},
{
"id": "nest",
"title": "Nest Desktop Auth Deprecation",

16
tests/components/homeassistant_alerts/fixtures/alerts_2.json
View File

@@ -88,22 +88,6 @@
"filename": "logi_circle.markdown",
"alert_url": "https://alerts.home-assistant.io/#logi_circle.markdown"
},
{
"id": "neato",
"title": "New Neato Botvacs Do Not Support Existing API",
"created": "2021-12-20T13:27:00.000Z",
"integrations": [
{
"package": "neato"
}
],
"homeassistant": {
"package": "homeassistant",
"affected_from_version": "0.30"
},
"filename": "neato.markdown",
"alert_url": "https://alerts.home-assistant.io/#neato.markdown"
},
{
"id": "nest",
"title": "Nest Desktop Auth Deprecation",

15
tests/components/homeassistant_alerts/test_init.py
View File

@@ -55,7 +55,6 @@ async def setup_repairs(hass: HomeAssistant) -> None:
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -71,7 +70,6 @@ async def setup_repairs(hass: HomeAssistant) -> None:
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -87,7 +85,6 @@ async def setup_repairs(hass: HomeAssistant) -> None:
("hikvision", "hikvisioncam"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -121,7 +118,6 @@ async def test_alerts(
"hive",
"homematicip_cloud",
"logi_circle",
"neato",
"nest",
"senseme",
"sochain",
@@ -198,7 +194,6 @@ async def test_alerts(
"hive",
"homematicip_cloud",
"logi_circle",
"neato",
"nest",
"senseme",
"sochain",
@@ -216,7 +211,6 @@ async def test_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -233,7 +227,6 @@ async def test_alerts(
"hive",
"homematicip_cloud",
"logi_circle",
"neato",
"nest",
"senseme",
"sochain",
@@ -248,7 +241,6 @@ async def test_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -264,7 +256,6 @@ async def test_alerts(
"hive",
"homematicip_cloud",
"logi_circle",
"neato",
"nest",
"senseme",
"sochain",
@@ -280,7 +271,6 @@ async def test_alerts(
("hikvision", "hikvisioncam"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -527,7 +517,6 @@ async def test_no_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -540,7 +529,6 @@ async def test_no_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -556,7 +544,6 @@ async def test_no_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -570,7 +557,6 @@ async def test_no_alerts(
("hive_us", "hive"),
("homematicip_cloud", "homematicip_cloud"),
("logi_circle", "logi_circle"),
("neato", "neato"),
("nest", "nest"),
("senseme", "senseme"),
("sochain", "sochain"),
@@ -606,7 +592,6 @@ async def test_alerts_change(
"hive",
"homematicip_cloud",
"logi_circle",
"neato",
"nest",
"senseme",
"sochain",

1
tests/components/neato/__init__.py
View File

@@ -1 +0,0 @@
"""Tests for the Neato component."""

164
tests/components/neato/test_config_flow.py
View File

@@ -1,164 +0,0 @@
"""Test the Neato Botvac config flow."""
from unittest.mock import patch
from pybotvac.neato import Neato
import pytest
from homeassistant import config_entries, setup
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.neato.const import NEATO_DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
VENDOR = Neato()
OAUTH2_AUTHORIZE = VENDOR.auth_endpoint
OAUTH2_TOKEN = VENDOR.token_endpoint
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Check full flow."""
assert await setup.async_setup_component(hass, "neato", {})
await async_import_client_credential(
hass, NEATO_DOMAIN, ClientCredential(CLIENT_ID, CLIENT_SECRET)
)
result = await hass.config_entries.flow.async_init(
"neato", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
f"&client_secret={CLIENT_SECRET}"
"&scope=public_profile+control_robots+maps"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
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,
},
)
with patch(
"homeassistant.components.neato.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(NEATO_DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
async def test_abort_if_already_setup(hass: HomeAssistant) -> None:
"""Test we abort if Neato is already setup."""
entry = MockConfigEntry(
domain=NEATO_DOMAIN,
data={"auth_implementation": "neato", "token": {"some": "data"}},
)
entry.add_to_hass(hass)
# Should fail
result = await hass.config_entries.flow.async_init(
"neato", context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauth(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test initialization of the reauth flow."""
assert await setup.async_setup_component(hass, "neato", {})
await async_import_client_credential(
hass, NEATO_DOMAIN, ClientCredential(CLIENT_ID, CLIENT_SECRET)
)
entry = MockConfigEntry(
entry_id="my_entry",
domain=NEATO_DOMAIN,
data={"username": "abcdef", "password": "123456", "vendor": "neato"},
)
entry.add_to_hass(hass)
# Should show form
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
# Confirm reauth flow
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
# Update entry
with patch(
"homeassistant.components.neato.async_setup_entry", return_value=True
) as mock_setup:
result3 = await hass.config_entries.flow.async_configure(result2["flow_id"])
await hass.async_block_till_done()
new_entry = hass.config_entries.async_get_entry("my_entry")
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reauth_successful"
assert new_entry.state is ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(NEATO_DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1