From 7a9025449c9cac1614405b4c8378d321cef77c52 Mon Sep 17 00:00:00 2001 From: Luke Date: Sat, 31 Oct 2026 22:43:05 -0400 Subject: [PATCH 1/2] refactor: split web_api into per-resource modules Turn roborock/web_api.py into a roborock.web_api package with one mixin per resource group (auth, homes, devices, scenes, jobs, ota, products) composed into RoborockApiClient. The package re-exports every name the old module exposed, so imports and existing patch targets keep working. Co-Authored-By: Claude Opus 5.5 --- roborock/web_api.py | 977 ---------------------------------- roborock/web_api/__init__.py | 51 ++ roborock/web_api/_base.py | 168 ++++++ roborock/web_api/_hawk.py | 58 ++ roborock/web_api/_request.py | 44 ++ roborock/web_api/auth.py | 278 ++++++++++ roborock/web_api/client.py | 80 +++ roborock/web_api/devices.py | 84 +++ roborock/web_api/homes.py | 81 +++ roborock/web_api/jobs.py | 50 ++ roborock/web_api/ota.py | 44 ++ roborock/web_api/products.py | 49 ++ roborock/web_api/scenes.py | 28 + tests/test_web_api.py | 683 ------------------------ tests/web_api/__init__.py | 5 + tests/web_api/conftest.py | 8 + tests/web_api/test_auth.py | 234 ++++++++ tests/web_api/test_base.py | 72 +++ tests/web_api/test_client.py | 55 ++ tests/web_api/test_devices.py | 13 + tests/web_api/test_hawk.py | 23 + tests/web_api/test_homes.py | 81 +++ tests/web_api/test_jobs.py | 174 ++++++ tests/web_api/test_ota.py | 33 ++ tests/web_api/test_request.py | 31 ++ tests/web_api/test_scenes.py | 27 + 26 files changed, 1771 insertions(+), 1660 deletions(-) delete mode 100644 roborock/web_api.py create mode 100644 roborock/web_api/__init__.py create mode 100644 roborock/web_api/_base.py create mode 100644 roborock/web_api/_hawk.py create mode 100644 roborock/web_api/_request.py create mode 100644 roborock/web_api/auth.py create mode 100644 roborock/web_api/client.py create mode 100644 roborock/web_api/devices.py create mode 100644 roborock/web_api/homes.py create mode 100644 roborock/web_api/jobs.py create mode 100644 roborock/web_api/ota.py create mode 100644 roborock/web_api/products.py create mode 100644 roborock/web_api/scenes.py delete mode 100644 tests/test_web_api.py create mode 100644 tests/web_api/__init__.py create mode 100644 tests/web_api/conftest.py create mode 100644 tests/web_api/test_auth.py create mode 100644 tests/web_api/test_base.py create mode 100644 tests/web_api/test_client.py create mode 100644 tests/web_api/test_devices.py create mode 100644 tests/web_api/test_hawk.py create mode 100644 tests/web_api/test_homes.py create mode 100644 tests/web_api/test_jobs.py create mode 100644 tests/web_api/test_ota.py create mode 100644 tests/web_api/test_request.py create mode 100644 tests/web_api/test_scenes.py diff --git a/roborock/web_api.py b/roborock/web_api.py deleted file mode 100644 index cdadb904..00000000 --- a/roborock/web_api.py +++ /dev/null @@ -1,977 +0,0 @@ -import base64 -import hashlib -import hmac -import json -import logging -import math -import secrets -import string -import time -from collections.abc import Callable -from dataclasses import dataclass - -import aiohttp -from aiohttp import ContentTypeError, FormData -from pyrate_limiter import Duration, Limiter, Rate - -from roborock import HomeDataSchedule -from roborock.data import FirmwareInfo, HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData -from roborock.exceptions import ( - RoborockAccountDoesNotExist, - RoborockException, - RoborockInvalidCode, - RoborockInvalidCredentials, - RoborockInvalidEmail, - RoborockInvalidUserAgreement, - RoborockMissingParameters, - RoborockNoResponseFromBaseURL, - RoborockNoUserAgreement, - RoborockRateLimit, - RoborockTooFrequentCodeRequests, -) - -_LOGGER = logging.getLogger(__name__) -BASE_URLS = [ - "https://usiot.roborock.com", - "https://euiot.roborock.com", - "https://cniot.roborock.com", - "https://ruiot.roborock.com", -] - -# Fallback user agreement version, used only if the latest version cannot be fetched. -DEFAULT_AGREEMENT_MAJOR_VERSION = 14 -DEFAULT_AGREEMENT_MINOR_VERSION = 0 - - -@dataclass -class IotLoginInfo: - """Information about the login to the iot server.""" - - base_url: str - country_code: str - country: str - - -class RoborockApiClient: - _LOGIN_RATES = [ - Rate(1, Duration.SECOND), - Rate(3, Duration.MINUTE), - Rate(10, Duration.HOUR), - Rate(20, Duration.DAY), - ] - _HOME_DATA_RATES = [ - Rate(1, Duration.SECOND), - Rate(3, Duration.MINUTE), - Rate(5, Duration.HOUR), - Rate(40, Duration.DAY), - ] - - _login_limiter = Limiter(_LOGIN_RATES) - _home_data_limiter = Limiter(_HOME_DATA_RATES) - - def __init__( - self, username: str, base_url: str | None = None, session: aiohttp.ClientSession | None = None - ) -> None: - """Sample API Client.""" - self._username = username - self._base_url = base_url - self._device_identifier = secrets.token_urlsafe(16) - self.session = session - self._iot_login_info: IotLoginInfo | None = None - self._base_urls = BASE_URLS if base_url is None else [base_url] - - async def _get_iot_login_info(self) -> IotLoginInfo: - if self._iot_login_info is None: - for iot_url in self._base_urls: - url_request = PreparedRequest(iot_url, self.session) - response = await url_request.request( - "post", - "/api/v1/getUrlByEmail", - params={"email": self._username, "needtwostepauth": "false"}, - ) - if response is None: - continue - response_code = response.get("code") - if response_code != 200: - if response_code == 2003: - raise RoborockInvalidEmail("Your email was incorrectly formatted.") - elif response_code == 1001: - raise RoborockMissingParameters( - "You are missing parameters for this request, are you sure you entered your username?" - ) - else: - raise RoborockException(f"{response.get('msg')} - response code: {response_code}") - country_code = response["data"]["countrycode"] - country = response["data"]["country"] - if country_code is not None or country is not None: - self._iot_login_info = IotLoginInfo( - base_url=response["data"]["url"], - country=country, - country_code=country_code, - ) - _LOGGER.debug("Country determined to be %s and code is %s", country, country_code) - return self._iot_login_info - raise RoborockNoResponseFromBaseURL( - "No account was found for any base url we tried. Either your email is incorrect or we do not have a" - " record of the roborock server your device is on." - ) - return self._iot_login_info - - @property - async def base_url(self): - if self._base_url is not None: - return self._base_url - return (await self._get_iot_login_info()).base_url - - @property - async def country(self): - return (await self._get_iot_login_info()).country - - @property - async def country_code(self): - return (await self._get_iot_login_info()).country_code - - def _get_header_client_id(self): - md5 = hashlib.md5() - md5.update(self._username.encode()) - md5.update(self._device_identifier.encode()) - return base64.b64encode(md5.digest()).decode() - - async def nc_prepare(self, user_data: UserData, timezone: str) -> dict: - """This gets a few critical parameters for adding a device to your account.""" - if ( - user_data.rriot is None - or user_data.rriot.r is None - or user_data.rriot.u is None - or user_data.rriot.r.a is None - ): - raise RoborockException("Your userdata is missing critical attributes.") - base_url = user_data.rriot.r.a - prepare_request = PreparedRequest(base_url, self.session) - hid = await self._get_home_id(user_data) - - data = FormData() - data.add_field("hid", hid) - data.add_field("tzid", timezone) - - prepare_response = await prepare_request.request( - "post", - "/nc/prepare", - headers={ - "Authorization": _get_hawk_authentication( - user_data.rriot, "/nc/prepare", {"hid": hid, "tzid": timezone} - ), - }, - data=data, - ) - - if prepare_response is None: - raise RoborockException("prepare_response is None") - if not prepare_response.get("success"): - raise RoborockException(f"{prepare_response.get('msg')} - response code: {prepare_response.get('code')}") - - return prepare_response["result"] - - async def add_device(self, user_data: UserData, s: str, t: str) -> dict: - """This will add a new device to your account - it is recommended to only use this during a pairing cycle with a device. - Please see here: https://github.com/Python-roborock/Roborockmitmproxy/blob/main/handshake_protocol.md - """ - if ( - user_data.rriot is None - or user_data.rriot.r is None - or user_data.rriot.u is None - or user_data.rriot.r.a is None - ): - raise RoborockException("Your userdata is missing critical attributes.") - base_url = user_data.rriot.r.a - add_device_request = PreparedRequest(base_url, self.session) - - add_device_response = await add_device_request.request( - "GET", - "/user/devices/newadd", - headers={ - "Authorization": _get_hawk_authentication( - user_data.rriot, "/user/devices/newadd", params={"s": s, "t": t} - ), - }, - params={"s": s, "t": t}, - ) - - if add_device_response is None: - raise RoborockException("add_device is None") - if not add_device_response.get("success"): - raise RoborockException( - f"{add_device_response.get('msg')} - response code: {add_device_response.get('code')}" - ) - - return add_device_response["result"] - - async def request_code(self) -> None: - if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): - raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") - base_url = await self.base_url - header_clientid = self._get_header_client_id() - code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - - code_response = await code_request.request( - "post", - "/api/v1/sendEmailCode", - params={ - "username": self._username, - "type": "auth", - }, - ) - if code_response is None: - raise RoborockException("Failed to get a response from send email code") - response_code = code_response.get("code") - if response_code != 200: - _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) - if response_code == 2008: - raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.") - elif response_code == 9002: - raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later") - else: - raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") - - async def request_code_v4(self) -> None: - """Request a code using the v4 endpoint.""" - if await self.country_code is None or await self.country is None: - _LOGGER.info("No country code or country found, trying old version of request code.") - return await self.request_code() - if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): - raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") - base_url = await self.base_url - header_clientid = self._get_header_client_id() - code_request = PreparedRequest( - base_url, - self.session, - { - "header_clientid": header_clientid, - "Content-Type": "application/x-www-form-urlencoded", - "header_clientlang": "en", - }, - ) - - code_response = await code_request.request( - "post", - "/api/v4/email/code/send", - data={"email": self._username, "type": "login", "platform": ""}, - ) - if code_response is None: - raise RoborockException("Failed to get a response from send email code") - response_code = code_response.get("code") - if response_code != 200: - _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) - if response_code == 2008: - raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.") - elif response_code == 9002: - raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later") - elif response_code == 3030 and len(self._base_urls) > 1: - self._base_urls = self._base_urls[1:] - self._iot_login_info = None - return await self.request_code_v4() - else: - raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") - - async def _sign_key_v3(self, s: str) -> str: - """Sign a randomly generated string.""" - base_url = await self.base_url - header_clientid = self._get_header_client_id() - code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - - code_response = await code_request.request( - "post", - "/api/v3/key/sign", - params={"s": s}, - ) - - if not code_response or "data" not in code_response or "k" not in code_response["data"]: - raise RoborockException("Failed to get a response from sign key") - response_code = code_response.get("code") - - if response_code != 200: - _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) - raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") - - return code_response["data"]["k"] - - async def _get_agreement_version(self, country: str) -> dict[str, int]: - """Get the latest user agreement version for the given country. - - The login endpoint rejects a stale agreement version with code 3006. The - current version differs per server and per country, so it must be looked - up rather than hardcoded. Falls back to the previously hardcoded values - if the lookup fails so that login is never blocked by this request. - """ - try: - base_url = await self.base_url - agreement_request = PreparedRequest(base_url, self.session, {"header_clientlang": "en"}) - response = await agreement_request.request( - "get", - "/api/v3/app/agreement/latest", - params={"country": country}, - ) - if response is not None and response.get("code") == 200: - data = response.get("data") or {} - major = data.get("majorVersion") - minor = data.get("minorVersion") - if isinstance(major, int) and isinstance(minor, int): - _LOGGER.debug("Using user agreement version %s.%s for %s", major, minor, country) - return {"majorVersion": major, "minorVersion": minor} - _LOGGER.debug("Unexpected agreement version response: %s", response) - except RoborockException as err: - _LOGGER.debug("Could not fetch latest user agreement version: %s", err) - return {"majorVersion": DEFAULT_AGREEMENT_MAJOR_VERSION, "minorVersion": DEFAULT_AGREEMENT_MINOR_VERSION} - - async def code_login_v4( - self, code: int | str, country: str | None = None, country_code: int | None = None - ) -> UserData: - """ - Login via code authentication. - :param code: The code from the email. - :param country: The two-character representation of the country, i.e. "US" - :param country_code: the country phone number code i.e. 1 for US. - """ - base_url = await self.base_url - if country is None: - country = await self.country - if country_code is None: - country_code = await self.country_code - if country_code is None or country is None: - _LOGGER.info("No country code or country found, trying old version of code login.") - return await self.code_login(code) - header_clientid = self._get_header_client_id() - x_mercy_ks = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(16)) - x_mercy_k = await self._sign_key_v3(x_mercy_ks) - login_request = PreparedRequest( - base_url, - self.session, - { - "header_clientid": header_clientid, - "x-mercy-ks": x_mercy_ks, - "x-mercy-k": x_mercy_k, - "Content-Type": "application/x-www-form-urlencoded", - "header_clientlang": "en", - "header_appversion": "4.54.02", - "header_phonesystem": "iOS", - "header_phonemodel": "iPhone16,1", - }, - ) - login_response = await login_request.request( - "post", - "/api/v4/auth/email/login/code", - data={ - "country": country, - "countryCode": country_code, - "email": self._username, - "code": code, - **await self._get_agreement_version(country), - }, - ) - if login_response is None: - raise RoborockException("Login request response is None") - response_code = login_response.get("code") - if response_code != 200: - _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) - if response_code == 2018: - raise RoborockInvalidCode("Invalid code - check your code and try again.") - if response_code == 3009: - raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.") - if response_code == 3006: - raise RoborockInvalidUserAgreement( - "User agreement must be accepted again - or you are attempting to use the Mi Home app account." - ) - if response_code == 3039: - raise RoborockAccountDoesNotExist( - "This account does not exist - please ensure that you selected the right region and email." - ) - raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}") - user_data = login_response.get("data") - if not isinstance(user_data, dict): - raise RoborockException("Got unexpected data type for user_data") - return UserData.from_dict(user_data) - - async def pass_login(self, password: str) -> UserData: - if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): - raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") - base_url = await self.base_url - header_clientid = self._get_header_client_id() - - login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - login_response = await login_request.request( - "post", - "/api/v1/login", - params={ - "username": self._username, - "password": password, - "needtwostepauth": "false", - }, - ) - if login_response is None: - raise RoborockException("Login response is none") - if login_response.get("code") != 200: - _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) - raise RoborockException(f"{login_response.get('msg')} - response code: {login_response.get('code')}") - user_data = login_response.get("data") - if not isinstance(user_data, dict): - raise RoborockException("Got unexpected data type for user_data") - return UserData.from_dict(user_data) - - async def pass_login_v3(self, password: str) -> UserData: - """Seemingly it follows the format below, but password is encrypted in some manner. - # login_response = await login_request.request( - # "post", - # "/api/v3/auth/email/login", - # params={ - # "email": self._username, - # "password": password, - # "twoStep": 1, - # "version": 0 - # }, - # ) - """ - raise NotImplementedError("Pass_login_v3 has not yet been implemented") - - async def code_login(self, code: int | str) -> UserData: - base_url = await self.base_url - header_clientid = self._get_header_client_id() - - login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - login_response = await login_request.request( - "post", - "/api/v1/loginWithCode", - params={ - "username": self._username, - "verifycode": code, - "verifycodetype": "AUTH_EMAIL_CODE", - }, - ) - if login_response is None: - raise RoborockException("Login request response is None") - response_code = login_response.get("code") - if response_code != 200: - _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) - if response_code == 2018: - raise RoborockInvalidCode("Invalid code - check your code and try again.") - if response_code == 3009: - raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.") - if response_code == 3006: - raise RoborockInvalidUserAgreement( - "User agreement must be accepted again - or you are attempting to use the Mi Home app account." - ) - raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}") - user_data = login_response.get("data") - if not isinstance(user_data, dict): - raise RoborockException("Got unexpected data type for user_data") - return UserData.from_dict(user_data) - - async def _get_home_id(self, user_data: UserData): - base_url = await self.base_url - header_clientid = self._get_header_client_id() - home_id_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - home_id_response = await home_id_request.request( - "get", - "/api/v1/getHomeDetail", - headers={"Authorization": user_data.token}, - ) - if home_id_response is None: - raise RoborockException("home_id_response is None") - if home_id_response.get("code") != 200: - _LOGGER.info("Get Home Id failed with the following context: %s", home_id_response) - if home_id_response.get("code") == 2010: - raise RoborockInvalidCredentials( - f"Invalid credentials ({home_id_response.get('msg')}) - check your login and try again." - ) - raise RoborockException(f"{home_id_response.get('msg')} - response code: {home_id_response.get('code')}") - - return home_id_response["data"]["rrHomeId"] - - async def get_home_data(self, user_data: UserData) -> HomeData: - if not self._home_data_limiter.try_acquire("home_data", blocking=False): - raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - home_id = await self._get_home_id(user_data) - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - home_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, f"/user/homes/{str(home_id)}"), - }, - ) - home_response = await home_request.request("get", "/user/homes/" + str(home_id)) - if not home_response.get("success"): - raise RoborockException(home_response) - home_data = home_response.get("result") - if isinstance(home_data, dict): - return HomeData.from_dict(home_data) - else: - raise RoborockException("home_response result was an unexpected type") - - async def get_home_data_v2(self, user_data: UserData) -> HomeData: - """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums.""" - if not self._home_data_limiter.try_acquire("home_data", blocking=False): - raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - home_id = await self._get_home_id(user_data) - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - home_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, "/v2/user/homes/" + str(home_id)), - }, - ) - home_response = await home_request.request("get", "/v2/user/homes/" + str(home_id)) - if not home_response.get("success"): - raise RoborockException(home_response) - home_data = home_response.get("result") - if isinstance(home_data, dict): - return HomeData.from_dict(home_data) - else: - raise RoborockException("home_response result was an unexpected type") - - async def get_home_data_v3(self, user_data: UserData) -> HomeData: - """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums.""" - if not self._home_data_limiter.try_acquire("home_data", blocking=False): - raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") - rriot = user_data.rriot - home_id = await self._get_home_id(user_data) - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - home_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, "/v3/user/homes/" + str(home_id)), - }, - ) - home_response = await home_request.request("get", "/v3/user/homes/" + str(home_id)) - if not home_response.get("success"): - raise RoborockException(home_response) - home_data = home_response.get("result") - if isinstance(home_data, dict): - return HomeData.from_dict(home_data) - raise RoborockException(f"home_response result was an unexpected type: {home_data}") - - async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> list[HomeDataRoom]: - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if home_id is None: - home_id = await self._get_home_id(user_data) - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - room_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, f"/user/homes/{home_id}/rooms"), - }, - ) - room_response = await room_request.request("get", f"/user/homes/{home_id}/rooms") - if not room_response.get("success"): - raise RoborockException(room_response) - rooms = room_response.get("result") - if isinstance(rooms, list): - output_list = [] - for room in rooms: - output_list.append(HomeDataRoom.from_dict(room)) - return output_list - else: - raise RoborockException("home_response result was an unexpected type") - - async def get_shared_device_rooms(self, user_data: UserData, device_id: str) -> list[HomeDataRoom]: - """Fetch room names for a shared (received) device.""" - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/user/deviceshare/query/{device_id}/rooms" - room_request = PreparedRequest( - rriot.r.a, - self.session, - {"Authorization": _get_hawk_authentication(rriot, path)}, - ) - room_response = await room_request.request("get", path) - if not room_response.get("success"): - raise RoborockException(room_response) - rooms = room_response.get("result") - if isinstance(rooms, list): - output_list = [] - for room in rooms: - normalized_room = room - if isinstance(room, dict) and "id" not in room and "roomId" in room: - normalized_room = {**room, "id": room["roomId"]} - output_list.append(HomeDataRoom.from_dict(normalized_room)) - return output_list - raise RoborockException("get_shared_device_rooms result was an unexpected type") - - async def get_scenes(self, user_data: UserData, device_id: str) -> list[HomeDataScene]: - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - scenes_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, f"/user/scene/device/{str(device_id)}"), - }, - ) - scenes_response = await scenes_request.request("get", f"/user/scene/device/{str(device_id)}") - if not scenes_response.get("success"): - raise RoborockException(scenes_response) - scenes = scenes_response.get("result") - if isinstance(scenes, list): - return [HomeDataScene.from_dict(scene) for scene in scenes] - else: - raise RoborockException("scene_response result was an unexpected type") - - async def get_firmware_info(self, user_data: UserData, device_id: str) -> FirmwareInfo: - """Get firmware/OTA info for a device (latest version + updatable flag).""" - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/ota/firmware/{device_id}/updatev2" - params = {"lang": "en"} - firmware_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, path, params=params), - }, - ) - firmware_response = await firmware_request.request("get", path, params=params) - if not firmware_response or not firmware_response.get("success"): - raise RoborockException(firmware_response) - return FirmwareInfo.from_dict(firmware_response.get("result") or {}) - - async def start_firmware_update(self, user_data: UserData, device_id: str) -> None: - """Trigger the (irreversible) firmware update for a device. - - The device downloads and flashes the latest firmware reported by - :meth:`get_firmware_info`. - """ - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/ota/device/{device_id}/upgrade" - upgrade_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, path), - }, - ) - upgrade_response = await upgrade_request.request("post", path) - if not upgrade_response or not upgrade_response.get("success"): - raise RoborockException(upgrade_response) - - async def set_silent_ota(self, user_data: UserData, device_id: str, enabled: bool) -> None: - """Enable/disable automatic (silent) firmware updates for a device. - - The current value is reported as ``silent_ota_switch`` on the home-data device. - """ - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/user/devices/{device_id}" - formdata = {"silentOtaSwitch": "true" if enabled else "false"} - silent_ota_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, path, formdata=formdata), - }, - ) - silent_ota_response = await silent_ota_request.request("put", path, data=formdata) - if not silent_ota_response or not silent_ota_response.get("success"): - raise RoborockException(silent_ota_response) - - async def execute_scene(self, user_data: UserData, scene_id: int) -> None: - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - execute_scene_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, f"/user/scene/{str(scene_id)}/execute"), - }, - ) - execute_scene_response = await execute_scene_request.request("POST", f"/user/scene/{str(scene_id)}/execute") - if not execute_scene_response.get("success"): - raise RoborockException(execute_scene_response) - - async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]: - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - schedules_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, f"/user/devices/{device_id}/jobs"), - }, - ) - schedules_response = await schedules_request.request("get", f"/user/devices/{str(device_id)}/jobs") - if not schedules_response.get("success"): - raise RoborockException(schedules_response) - schedules = schedules_response.get("result") - if isinstance(schedules, list): - return [HomeDataSchedule.from_dict(schedule) for schedule in schedules] - else: - raise RoborockException(f"schedule_response result was an unexpected type: {schedules}") - - async def create_job(self, user_data: UserData, device_id: str, job: dict) -> dict: - """Create a /jobs entry (schedule or one-time room clean) on a B01 device. - - Body-bearing writes must sign the request body in the Hawk payload slot and send those same - compact bytes via ``data=``; ``json=`` would re-serialize with spaces and break the MAC. - """ - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/user/devices/{device_id}/jobs" - job_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, path, body=job), - "Content-Type": "application/json", - }, - ) - response = await job_request.request("post", path, data=_compact_json(job).encode()) - if not response.get("success"): - raise RoborockException(response) - return response - - async def update_job(self, user_data: UserData, device_id: str, job: HomeDataSchedule) -> None: - """Replace an existing cloud job with the full supplied schedule. - - Pass a schedule returned by :meth:`get_schedules` with the desired changes. - The job ID identifies the URL; all other fields are sent as the replacement body. - """ - rriot = user_data.rriot - if rriot is None: - raise RoborockException("rriot is none") - if rriot.r.a is None: - raise RoborockException("Missing field 'a' in rriot reference") - path = f"/user/devices/{device_id}/jobs/{job.id}" - body = job.as_dict(exclude={"id"}) - job_request = PreparedRequest( - rriot.r.a, - self.session, - { - "Authorization": _get_hawk_authentication(rriot, path, body=body), - "Content-Type": "application/json", - }, - ) - response = await job_request.request("put", path, data=_compact_json(body).encode()) - if not response or not response.get("success"): - raise RoborockException(response) - - async def get_products(self, user_data: UserData) -> ProductResponse: - """Gets all products and their schemas, good for determining status codes and model numbers.""" - base_url = await self.base_url - header_clientid = self._get_header_client_id() - product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - product_response = await product_request.request( - "get", - "/api/v4/product", - headers={"Authorization": user_data.token}, - ) - if product_response is None: - raise RoborockException("home_id_response is None") - if product_response.get("code") != 200: - raise RoborockException(f"{product_response.get('msg')} - response code: {product_response.get('code')}") - result = product_response.get("data") - if isinstance(result, dict): - return ProductResponse.from_dict(result) - raise RoborockException("product result was an unexpected type") - - async def download_code(self, user_data: UserData, product_id: int): - base_url = await self.base_url - header_clientid = self._get_header_client_id() - product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - request = {"apilevel": 99999, "productids": [product_id], "type": 2} - response = await product_request.request( - "post", - "/api/v1/appplugin", - json=request, - headers={"Authorization": user_data.token, "Content-Type": "application/json"}, - ) - return response["data"][0]["url"] - - async def download_category_code(self, user_data: UserData): - base_url = await self.base_url - header_clientid = self._get_header_client_id() - product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid}) - response = await product_request.request( - "get", - "api/v1/plugins?apiLevel=99999&type=2", - headers={ - "Authorization": user_data.token, - }, - ) - return {r["category"]: r["url"] for r in response["data"]["categoryPluginList"]} - - -class PreparedRequest: - def __init__( - self, base_url: str, session: aiohttp.ClientSession | None = None, base_headers: dict | None = None - ) -> None: - self.base_url = base_url - self.base_headers = base_headers or {} - self.session = session - - async def request(self, method: str, url: str, params=None, data=None, headers=None, json=None) -> dict: - _url = "/".join(s.strip("/") for s in [self.base_url, url]) - _headers = {**self.base_headers, **(headers or {})} - close_session = self.session is None - session = self.session if self.session is not None else aiohttp.ClientSession() - try: - async with session.request(method, _url, params=params, data=data, headers=_headers, json=json) as resp: - return await resp.json() - except ContentTypeError as err: - """If we get an error, lets log everything for debugging.""" - try: - resp_json = await resp.json(content_type=None) - _LOGGER.info("Resp: %s", resp_json) - except ContentTypeError as err_2: - _LOGGER.info(err_2) - resp_raw = await resp.read() - _LOGGER.info("Resp raw: %s", resp_raw) - # Still raise the err so that it's clear it failed. - raise err - except (aiohttp.ClientError, TimeoutError, OSError) as err: - raise RoborockException(f"Network error contacting {_url}: {err}") from err - finally: - if close_session: - await session.close() - - -def _process_extra_hawk_values(values: dict | None) -> str: - if values is None: - return "" - else: - sorted_keys = sorted(values.keys()) - result = [] - for key in sorted_keys: - value = values.get(key) - result.append(f"{key}={value}") - return hashlib.md5("&".join(result).encode()).hexdigest() - - -def _compact_json(body: dict) -> str: - """Serialize a JSON body to the exact compact bytes that are both signed and sent.""" - return json.dumps(body, separators=(",", ":")) - - -def _get_hawk_authentication( - rriot: RRiot, - url: str, - formdata: dict | None = None, - params: dict | None = None, - body: dict | None = None, -) -> str: - timestamp = math.floor(time.time()) - nonce = secrets.token_urlsafe(6) - params_str = _process_extra_hawk_values(params) - if body is not None: - payload_str = hashlib.md5(_compact_json(body).encode()).hexdigest() - else: - payload_str = _process_extra_hawk_values(formdata) - - prestr = ":".join( - [ - rriot.u, - rriot.s, - nonce, - str(timestamp), - hashlib.md5(url.encode()).hexdigest(), - params_str, - payload_str, - ] - ) - mac = base64.b64encode(hmac.new(rriot.h.encode(), prestr.encode(), hashlib.sha256).digest()).decode() - return f'Hawk id="{rriot.u}",s="{rriot.s}",ts="{timestamp}",nonce="{nonce}",mac="{mac}"' - - -class UserWebApiClient: - """Wrapper around RoborockApiClient to provide information for a specific user. - - This binds a RoborockApiClient to a specific user context with the - provided UserData. This allows for easier access to user-specific data, - to avoid needing to pass UserData around and mock out the web API. - """ - - def __init__( - self, web_api: RoborockApiClient, user_data: UserData, unauthorized_hook: Callable[[], None] | None = None - ) -> None: - """Initialize the wrapper with the API client and user data.""" - self._web_api = web_api - self._user_data = user_data - self._unauthorized_hook = unauthorized_hook - - async def get_home_data(self) -> HomeData: - """Fetch home data using the API client.""" - try: - return await self._web_api.get_home_data_v3(self._user_data) - except RoborockInvalidCredentials: - if self._unauthorized_hook: - self._unauthorized_hook() - raise - - async def get_routines(self, device_id: str) -> list[HomeDataScene]: - """Fetch routines (scenes) for a specific device.""" - try: - return await self._web_api.get_scenes(self._user_data, device_id) - except RoborockInvalidCredentials: - if self._unauthorized_hook: - self._unauthorized_hook() - raise - - async def get_rooms(self) -> list[HomeDataRoom]: - """Fetch rooms using the API client.""" - try: - return await self._web_api.get_rooms(self._user_data) - except RoborockInvalidCredentials: - if self._unauthorized_hook: - self._unauthorized_hook() - raise - - async def get_shared_device_rooms(self, device_id: str) -> list[HomeDataRoom]: - """Fetch shared-device rooms using the API client.""" - return await self._web_api.get_shared_device_rooms(self._user_data, device_id) - - async def execute_routine(self, scene_id: int) -> None: - """Execute a specific routine (scene) by its ID.""" - try: - await self._web_api.execute_scene(self._user_data, scene_id) - except RoborockInvalidCredentials: - if self._unauthorized_hook: - self._unauthorized_hook() - raise diff --git a/roborock/web_api/__init__.py b/roborock/web_api/__init__.py new file mode 100644 index 00000000..1a260899 --- /dev/null +++ b/roborock/web_api/__init__.py @@ -0,0 +1,51 @@ +"""Roborock cloud REST API clients. + +The public entry points are :class:`RoborockApiClient` and :class:`UserWebApiClient`. +Endpoints are grouped by resource into submodules (``auth``, ``homes``, ``devices``, +``scenes``, ``jobs``, ``ota``, ``products``); each exposes a mixin that +``RoborockApiClient`` composes. Everything that was importable from the former +``roborock.web_api`` module is re-exported here for backwards compatibility. +""" + +# Modules and the logger the former single-file module exposed. Re-exported so that +# existing patch targets such as ``roborock.web_api.time.time`` keep resolving. +import base64 +import hashlib +import hmac +import json +import math +import secrets +import string +import time + +import aiohttp + +from ._base import _LOGGER, BASE_URLS, IotLoginInfo, RoborockApiBase +from ._hawk import _compact_json, _get_hawk_authentication, _process_extra_hawk_values +from ._request import PreparedRequest +from .auth import DEFAULT_AGREEMENT_MAJOR_VERSION, DEFAULT_AGREEMENT_MINOR_VERSION, AuthApi +from .client import RoborockApiClient, UserWebApiClient +from .devices import DevicesApi +from .homes import HomesApi +from .jobs import JobsApi +from .ota import OtaApi +from .products import ProductsApi +from .scenes import ScenesApi + +__all__ = [ + "BASE_URLS", + "DEFAULT_AGREEMENT_MAJOR_VERSION", + "DEFAULT_AGREEMENT_MINOR_VERSION", + "AuthApi", + "DevicesApi", + "HomesApi", + "IotLoginInfo", + "JobsApi", + "OtaApi", + "PreparedRequest", + "ProductsApi", + "RoborockApiBase", + "RoborockApiClient", + "ScenesApi", + "UserWebApiClient", +] diff --git a/roborock/web_api/_base.py b/roborock/web_api/_base.py new file mode 100644 index 00000000..25a4fac8 --- /dev/null +++ b/roborock/web_api/_base.py @@ -0,0 +1,168 @@ +"""Shared state and request-building helpers for the web API mixins. + +Every resource mixin (``AuthApi``, ``HomesApi``, ``JobsApi`` ...) inherits from +``RoborockApiBase`` so that it can rely on the username, session, regional base +URL resolution and the two request builders defined here. ``RoborockApiClient`` +composes the mixins into the single public client. +""" + +import base64 +import hashlib +import logging +import secrets +from dataclasses import dataclass + +import aiohttp + +from roborock.data import UserData +from roborock.exceptions import ( + RoborockException, + RoborockInvalidCredentials, + RoborockInvalidEmail, + RoborockMissingParameters, + RoborockNoResponseFromBaseURL, +) + +from ._hawk import _get_hawk_authentication +from ._request import PreparedRequest + +_LOGGER = logging.getLogger(__name__) + +BASE_URLS = [ + "https://usiot.roborock.com", + "https://euiot.roborock.com", + "https://cniot.roborock.com", + "https://ruiot.roborock.com", +] + + +@dataclass +class IotLoginInfo: + """Information about the login to the iot server.""" + + base_url: str + country_code: str + country: str + + +class RoborockApiBase: + """State and helpers shared by all web API resource mixins.""" + + def __init__( + self, username: str, base_url: str | None = None, session: aiohttp.ClientSession | None = None + ) -> None: + """Sample API Client.""" + self._username = username + self._base_url = base_url + self._device_identifier = secrets.token_urlsafe(16) + self.session = session + self._iot_login_info: IotLoginInfo | None = None + self._base_urls = BASE_URLS if base_url is None else [base_url] + + async def _get_iot_login_info(self) -> IotLoginInfo: + if self._iot_login_info is None: + for iot_url in self._base_urls: + url_request = PreparedRequest(iot_url, self.session) + response = await url_request.request( + "post", + "/api/v1/getUrlByEmail", + params={"email": self._username, "needtwostepauth": "false"}, + ) + if response is None: + continue + response_code = response.get("code") + if response_code != 200: + if response_code == 2003: + raise RoborockInvalidEmail("Your email was incorrectly formatted.") + elif response_code == 1001: + raise RoborockMissingParameters( + "You are missing parameters for this request, are you sure you entered your username?" + ) + else: + raise RoborockException(f"{response.get('msg')} - response code: {response_code}") + country_code = response["data"]["countrycode"] + country = response["data"]["country"] + if country_code is not None or country is not None: + self._iot_login_info = IotLoginInfo( + base_url=response["data"]["url"], + country=country, + country_code=country_code, + ) + _LOGGER.debug("Country determined to be %s and code is %s", country, country_code) + return self._iot_login_info + raise RoborockNoResponseFromBaseURL( + "No account was found for any base url we tried. Either your email is incorrect or we do not have a" + " record of the roborock server your device is on." + ) + return self._iot_login_info + + @property + async def base_url(self): + if self._base_url is not None: + return self._base_url + return (await self._get_iot_login_info()).base_url + + @property + async def country(self): + return (await self._get_iot_login_info()).country + + @property + async def country_code(self): + return (await self._get_iot_login_info()).country_code + + def _get_header_client_id(self): + md5 = hashlib.md5() + md5.update(self._username.encode()) + md5.update(self._device_identifier.encode()) + return base64.b64encode(md5.digest()).decode() + + async def _iot_request(self, extra_headers: dict | None = None) -> PreparedRequest: + """Build a request against the regional Base IoT server with the standard client headers.""" + base_url = await self.base_url + headers = {"header_clientid": self._get_header_client_id(), **(extra_headers or {})} + return PreparedRequest(base_url, self.session, headers) + + def _rriot_request( + self, + user_data: UserData, + path: str, + *, + formdata: dict | None = None, + params: dict | None = None, + body: dict | None = None, + extra_headers: dict | None = None, + ) -> PreparedRequest: + """Build a Hawk-authenticated request against the RRIoT server for ``path``. + + The same ``formdata``/``params``/``body`` passed here must be sent on the request so + that the signed MAC matches what the server receives. + """ + rriot = user_data.rriot + if rriot is None: + raise RoborockException("rriot is none") + if rriot.r.a is None: + raise RoborockException("Missing field 'a' in rriot reference") + headers = { + "Authorization": _get_hawk_authentication(rriot, path, formdata=formdata, params=params, body=body), + **(extra_headers or {}), + } + return PreparedRequest(rriot.r.a, self.session, headers) + + async def _get_home_id(self, user_data: UserData): + home_id_request = await self._iot_request() + home_id_response = await home_id_request.request( + "get", + "/api/v1/getHomeDetail", + headers={"Authorization": user_data.token}, + ) + if home_id_response is None: + raise RoborockException("home_id_response is None") + if home_id_response.get("code") != 200: + _LOGGER.info("Get Home Id failed with the following context: %s", home_id_response) + if home_id_response.get("code") == 2010: + raise RoborockInvalidCredentials( + f"Invalid credentials ({home_id_response.get('msg')}) - check your login and try again." + ) + raise RoborockException(f"{home_id_response.get('msg')} - response code: {home_id_response.get('code')}") + + return home_id_response["data"]["rrHomeId"] diff --git a/roborock/web_api/_hawk.py b/roborock/web_api/_hawk.py new file mode 100644 index 00000000..cc568aba --- /dev/null +++ b/roborock/web_api/_hawk.py @@ -0,0 +1,58 @@ +"""Hawk request signing for the RRIoT (``user_data.rriot.r.a``) API surface.""" + +import base64 +import hashlib +import hmac +import json +import math +import secrets +import time + +from roborock.data import RRiot + + +def _process_extra_hawk_values(values: dict | None) -> str: + if values is None: + return "" + else: + sorted_keys = sorted(values.keys()) + result = [] + for key in sorted_keys: + value = values.get(key) + result.append(f"{key}={value}") + return hashlib.md5("&".join(result).encode()).hexdigest() + + +def _compact_json(body: dict) -> str: + """Serialize a JSON body to the exact compact bytes that are both signed and sent.""" + return json.dumps(body, separators=(",", ":")) + + +def _get_hawk_authentication( + rriot: RRiot, + url: str, + formdata: dict | None = None, + params: dict | None = None, + body: dict | None = None, +) -> str: + timestamp = math.floor(time.time()) + nonce = secrets.token_urlsafe(6) + params_str = _process_extra_hawk_values(params) + if body is not None: + payload_str = hashlib.md5(_compact_json(body).encode()).hexdigest() + else: + payload_str = _process_extra_hawk_values(formdata) + + prestr = ":".join( + [ + rriot.u, + rriot.s, + nonce, + str(timestamp), + hashlib.md5(url.encode()).hexdigest(), + params_str, + payload_str, + ] + ) + mac = base64.b64encode(hmac.new(rriot.h.encode(), prestr.encode(), hashlib.sha256).digest()).decode() + return f'Hawk id="{rriot.u}",s="{rriot.s}",ts="{timestamp}",nonce="{nonce}",mac="{mac}"' diff --git a/roborock/web_api/_request.py b/roborock/web_api/_request.py new file mode 100644 index 00000000..de97b4b3 --- /dev/null +++ b/roborock/web_api/_request.py @@ -0,0 +1,44 @@ +"""Low-level HTTP request helper shared by every web API module.""" + +import logging + +import aiohttp +from aiohttp import ContentTypeError + +from roborock.exceptions import RoborockException + +_LOGGER = logging.getLogger(__name__) + + +class PreparedRequest: + def __init__( + self, base_url: str, session: aiohttp.ClientSession | None = None, base_headers: dict | None = None + ) -> None: + self.base_url = base_url + self.base_headers = base_headers or {} + self.session = session + + async def request(self, method: str, url: str, params=None, data=None, headers=None, json=None) -> dict: + _url = "/".join(s.strip("/") for s in [self.base_url, url]) + _headers = {**self.base_headers, **(headers or {})} + close_session = self.session is None + session = self.session if self.session is not None else aiohttp.ClientSession() + try: + async with session.request(method, _url, params=params, data=data, headers=_headers, json=json) as resp: + return await resp.json() + except ContentTypeError as err: + """If we get an error, lets log everything for debugging.""" + try: + resp_json = await resp.json(content_type=None) + _LOGGER.info("Resp: %s", resp_json) + except ContentTypeError as err_2: + _LOGGER.info(err_2) + resp_raw = await resp.read() + _LOGGER.info("Resp raw: %s", resp_raw) + # Still raise the err so that it's clear it failed. + raise err + except (aiohttp.ClientError, TimeoutError, OSError) as err: + raise RoborockException(f"Network error contacting {_url}: {err}") from err + finally: + if close_session: + await session.close() diff --git a/roborock/web_api/auth.py b/roborock/web_api/auth.py new file mode 100644 index 00000000..5492a32f --- /dev/null +++ b/roborock/web_api/auth.py @@ -0,0 +1,278 @@ +"""Login, verification-code and account-discovery endpoints on the Base IoT API.""" + +import logging +import secrets +import string + +from pyrate_limiter import Duration, Limiter, Rate + +from roborock.data import UserData +from roborock.exceptions import ( + RoborockAccountDoesNotExist, + RoborockException, + RoborockInvalidCode, + RoborockInvalidUserAgreement, + RoborockNoUserAgreement, + RoborockRateLimit, + RoborockTooFrequentCodeRequests, +) + +from ._base import RoborockApiBase +from ._request import PreparedRequest + +_LOGGER = logging.getLogger(__name__) + +# Fallback user agreement version, used only if the latest version cannot be fetched. +DEFAULT_AGREEMENT_MAJOR_VERSION = 14 +DEFAULT_AGREEMENT_MINOR_VERSION = 0 + + +class AuthApi(RoborockApiBase): + """Email/password and email-code login flows.""" + + _LOGIN_RATES = [ + Rate(1, Duration.SECOND), + Rate(3, Duration.MINUTE), + Rate(10, Duration.HOUR), + Rate(20, Duration.DAY), + ] + _login_limiter = Limiter(_LOGIN_RATES) + + async def request_code(self) -> None: + if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): + raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") + code_request = await self._iot_request() + + code_response = await code_request.request( + "post", + "/api/v1/sendEmailCode", + params={ + "username": self._username, + "type": "auth", + }, + ) + if code_response is None: + raise RoborockException("Failed to get a response from send email code") + response_code = code_response.get("code") + if response_code != 200: + _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) + if response_code == 2008: + raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.") + elif response_code == 9002: + raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later") + else: + raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") + + async def request_code_v4(self) -> None: + """Request a code using the v4 endpoint.""" + if await self.country_code is None or await self.country is None: + _LOGGER.info("No country code or country found, trying old version of request code.") + return await self.request_code() + if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): + raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") + code_request = await self._iot_request( + { + "Content-Type": "application/x-www-form-urlencoded", + "header_clientlang": "en", + } + ) + + code_response = await code_request.request( + "post", + "/api/v4/email/code/send", + data={"email": self._username, "type": "login", "platform": ""}, + ) + if code_response is None: + raise RoborockException("Failed to get a response from send email code") + response_code = code_response.get("code") + if response_code != 200: + _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) + if response_code == 2008: + raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.") + elif response_code == 9002: + raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later") + elif response_code == 3030 and len(self._base_urls) > 1: + self._base_urls = self._base_urls[1:] + self._iot_login_info = None + return await self.request_code_v4() + else: + raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") + + async def _sign_key_v3(self, s: str) -> str: + """Sign a randomly generated string.""" + code_request = await self._iot_request() + + code_response = await code_request.request( + "post", + "/api/v3/key/sign", + params={"s": s}, + ) + + if not code_response or "data" not in code_response or "k" not in code_response["data"]: + raise RoborockException("Failed to get a response from sign key") + response_code = code_response.get("code") + + if response_code != 200: + _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response) + raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}") + + return code_response["data"]["k"] + + async def _get_agreement_version(self, country: str) -> dict[str, int]: + """Get the latest user agreement version for the given country. + + The login endpoint rejects a stale agreement version with code 3006. The + current version differs per server and per country, so it must be looked + up rather than hardcoded. Falls back to the previously hardcoded values + if the lookup fails so that login is never blocked by this request. + """ + try: + base_url = await self.base_url + agreement_request = PreparedRequest(base_url, self.session, {"header_clientlang": "en"}) + response = await agreement_request.request( + "get", + "/api/v3/app/agreement/latest", + params={"country": country}, + ) + if response is not None and response.get("code") == 200: + data = response.get("data") or {} + major = data.get("majorVersion") + minor = data.get("minorVersion") + if isinstance(major, int) and isinstance(minor, int): + _LOGGER.debug("Using user agreement version %s.%s for %s", major, minor, country) + return {"majorVersion": major, "minorVersion": minor} + _LOGGER.debug("Unexpected agreement version response: %s", response) + except RoborockException as err: + _LOGGER.debug("Could not fetch latest user agreement version: %s", err) + return {"majorVersion": DEFAULT_AGREEMENT_MAJOR_VERSION, "minorVersion": DEFAULT_AGREEMENT_MINOR_VERSION} + + async def code_login_v4( + self, code: int | str, country: str | None = None, country_code: int | None = None + ) -> UserData: + """ + Login via code authentication. + :param code: The code from the email. + :param country: The two-character representation of the country, i.e. "US" + :param country_code: the country phone number code i.e. 1 for US. + """ + if country is None: + country = await self.country + if country_code is None: + country_code = await self.country_code + if country_code is None or country is None: + _LOGGER.info("No country code or country found, trying old version of code login.") + return await self.code_login(code) + x_mercy_ks = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(16)) + x_mercy_k = await self._sign_key_v3(x_mercy_ks) + login_request = await self._iot_request( + { + "x-mercy-ks": x_mercy_ks, + "x-mercy-k": x_mercy_k, + "Content-Type": "application/x-www-form-urlencoded", + "header_clientlang": "en", + "header_appversion": "4.54.02", + "header_phonesystem": "iOS", + "header_phonemodel": "iPhone16,1", + } + ) + login_response = await login_request.request( + "post", + "/api/v4/auth/email/login/code", + data={ + "country": country, + "countryCode": country_code, + "email": self._username, + "code": code, + **await self._get_agreement_version(country), + }, + ) + if login_response is None: + raise RoborockException("Login request response is None") + response_code = login_response.get("code") + if response_code != 200: + _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) + if response_code == 2018: + raise RoborockInvalidCode("Invalid code - check your code and try again.") + if response_code == 3009: + raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.") + if response_code == 3006: + raise RoborockInvalidUserAgreement( + "User agreement must be accepted again - or you are attempting to use the Mi Home app account." + ) + if response_code == 3039: + raise RoborockAccountDoesNotExist( + "This account does not exist - please ensure that you selected the right region and email." + ) + raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}") + user_data = login_response.get("data") + if not isinstance(user_data, dict): + raise RoborockException("Got unexpected data type for user_data") + return UserData.from_dict(user_data) + + async def pass_login(self, password: str) -> UserData: + if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1): + raise RoborockRateLimit("Reached maximum requests for login. Please try again later.") + login_request = await self._iot_request() + login_response = await login_request.request( + "post", + "/api/v1/login", + params={ + "username": self._username, + "password": password, + "needtwostepauth": "false", + }, + ) + if login_response is None: + raise RoborockException("Login response is none") + if login_response.get("code") != 200: + _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) + raise RoborockException(f"{login_response.get('msg')} - response code: {login_response.get('code')}") + user_data = login_response.get("data") + if not isinstance(user_data, dict): + raise RoborockException("Got unexpected data type for user_data") + return UserData.from_dict(user_data) + + async def pass_login_v3(self, password: str) -> UserData: + """Seemingly it follows the format below, but password is encrypted in some manner. + # login_response = await login_request.request( + # "post", + # "/api/v3/auth/email/login", + # params={ + # "email": self._username, + # "password": password, + # "twoStep": 1, + # "version": 0 + # }, + # ) + """ + raise NotImplementedError("Pass_login_v3 has not yet been implemented") + + async def code_login(self, code: int | str) -> UserData: + login_request = await self._iot_request() + login_response = await login_request.request( + "post", + "/api/v1/loginWithCode", + params={ + "username": self._username, + "verifycode": code, + "verifycodetype": "AUTH_EMAIL_CODE", + }, + ) + if login_response is None: + raise RoborockException("Login request response is None") + response_code = login_response.get("code") + if response_code != 200: + _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response) + if response_code == 2018: + raise RoborockInvalidCode("Invalid code - check your code and try again.") + if response_code == 3009: + raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.") + if response_code == 3006: + raise RoborockInvalidUserAgreement( + "User agreement must be accepted again - or you are attempting to use the Mi Home app account." + ) + raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}") + user_data = login_response.get("data") + if not isinstance(user_data, dict): + raise RoborockException("Got unexpected data type for user_data") + return UserData.from_dict(user_data) diff --git a/roborock/web_api/client.py b/roborock/web_api/client.py new file mode 100644 index 00000000..257070d6 --- /dev/null +++ b/roborock/web_api/client.py @@ -0,0 +1,80 @@ +"""Public web API clients composed from the per-resource mixins.""" + +from collections.abc import Callable + +from roborock.data import HomeData, HomeDataRoom, HomeDataScene, UserData +from roborock.exceptions import RoborockInvalidCredentials + +from .auth import AuthApi +from .devices import DevicesApi +from .homes import HomesApi +from .jobs import JobsApi +from .ota import OtaApi +from .products import ProductsApi +from .scenes import ScenesApi + + +class RoborockApiClient(AuthApi, HomesApi, DevicesApi, ScenesApi, JobsApi, OtaApi, ProductsApi): + """Client for the Roborock cloud REST APIs. + + Each group of endpoints lives in its own mixin (``roborock.web_api.auth``, + ``roborock.web_api.homes`` ...); this class simply composes them so callers keep a + single entry point. + """ + + +class UserWebApiClient: + """Wrapper around RoborockApiClient to provide information for a specific user. + + This binds a RoborockApiClient to a specific user context with the + provided UserData. This allows for easier access to user-specific data, + to avoid needing to pass UserData around and mock out the web API. + """ + + def __init__( + self, web_api: RoborockApiClient, user_data: UserData, unauthorized_hook: Callable[[], None] | None = None + ) -> None: + """Initialize the wrapper with the API client and user data.""" + self._web_api = web_api + self._user_data = user_data + self._unauthorized_hook = unauthorized_hook + + async def get_home_data(self) -> HomeData: + """Fetch home data using the API client.""" + try: + return await self._web_api.get_home_data_v3(self._user_data) + except RoborockInvalidCredentials: + if self._unauthorized_hook: + self._unauthorized_hook() + raise + + async def get_routines(self, device_id: str) -> list[HomeDataScene]: + """Fetch routines (scenes) for a specific device.""" + try: + return await self._web_api.get_scenes(self._user_data, device_id) + except RoborockInvalidCredentials: + if self._unauthorized_hook: + self._unauthorized_hook() + raise + + async def get_rooms(self) -> list[HomeDataRoom]: + """Fetch rooms using the API client.""" + try: + return await self._web_api.get_rooms(self._user_data) + except RoborockInvalidCredentials: + if self._unauthorized_hook: + self._unauthorized_hook() + raise + + async def get_shared_device_rooms(self, device_id: str) -> list[HomeDataRoom]: + """Fetch shared-device rooms using the API client.""" + return await self._web_api.get_shared_device_rooms(self._user_data, device_id) + + async def execute_routine(self, scene_id: int) -> None: + """Execute a specific routine (scene) by its ID.""" + try: + await self._web_api.execute_scene(self._user_data, scene_id) + except RoborockInvalidCredentials: + if self._unauthorized_hook: + self._unauthorized_hook() + raise diff --git a/roborock/web_api/devices.py b/roborock/web_api/devices.py new file mode 100644 index 00000000..9a043975 --- /dev/null +++ b/roborock/web_api/devices.py @@ -0,0 +1,84 @@ +"""Device pairing and device-record endpoints (``/nc/prepare``, ``/user/devices``).""" + +from aiohttp import FormData + +from roborock.data import UserData +from roborock.exceptions import RoborockException + +from ._base import RoborockApiBase +from ._hawk import _get_hawk_authentication +from ._request import PreparedRequest + + +class DevicesApi(RoborockApiBase): + """Adding devices to an account and editing per-device cloud settings.""" + + async def nc_prepare(self, user_data: UserData, timezone: str) -> dict: + """This gets a few critical parameters for adding a device to your account.""" + if ( + user_data.rriot is None + or user_data.rriot.r is None + or user_data.rriot.u is None + or user_data.rriot.r.a is None + ): + raise RoborockException("Your userdata is missing critical attributes.") + base_url = user_data.rriot.r.a + prepare_request = PreparedRequest(base_url, self.session) + hid = await self._get_home_id(user_data) + + data = FormData() + data.add_field("hid", hid) + data.add_field("tzid", timezone) + + prepare_response = await prepare_request.request( + "post", + "/nc/prepare", + headers={ + "Authorization": _get_hawk_authentication( + user_data.rriot, "/nc/prepare", {"hid": hid, "tzid": timezone} + ), + }, + data=data, + ) + + if prepare_response is None: + raise RoborockException("prepare_response is None") + if not prepare_response.get("success"): + raise RoborockException(f"{prepare_response.get('msg')} - response code: {prepare_response.get('code')}") + + return prepare_response["result"] + + async def add_device(self, user_data: UserData, s: str, t: str) -> dict: + """This will add a new device to your account + it is recommended to only use this during a pairing cycle with a device. + Please see here: https://github.com/Python-roborock/Roborockmitmproxy/blob/main/handshake_protocol.md + """ + if ( + user_data.rriot is None + or user_data.rriot.r is None + or user_data.rriot.u is None + or user_data.rriot.r.a is None + ): + raise RoborockException("Your userdata is missing critical attributes.") + base_url = user_data.rriot.r.a + add_device_request = PreparedRequest(base_url, self.session) + + add_device_response = await add_device_request.request( + "GET", + "/user/devices/newadd", + headers={ + "Authorization": _get_hawk_authentication( + user_data.rriot, "/user/devices/newadd", params={"s": s, "t": t} + ), + }, + params={"s": s, "t": t}, + ) + + if add_device_response is None: + raise RoborockException("add_device is None") + if not add_device_response.get("success"): + raise RoborockException( + f"{add_device_response.get('msg')} - response code: {add_device_response.get('code')}" + ) + + return add_device_response["result"] diff --git a/roborock/web_api/homes.py b/roborock/web_api/homes.py new file mode 100644 index 00000000..4ca67e30 --- /dev/null +++ b/roborock/web_api/homes.py @@ -0,0 +1,81 @@ +"""Home and room endpoints (``/user/homes``, ``/user/deviceshare/.../rooms``).""" + +from pyrate_limiter import Duration, Limiter, Rate + +from roborock.data import HomeData, HomeDataRoom, UserData +from roborock.exceptions import RoborockException, RoborockRateLimit + +from ._base import RoborockApiBase + + +class HomesApi(RoborockApiBase): + """Home data (devices, products, rooms) lookups.""" + + _HOME_DATA_RATES = [ + Rate(1, Duration.SECOND), + Rate(3, Duration.MINUTE), + Rate(5, Duration.HOUR), + Rate(40, Duration.DAY), + ] + _home_data_limiter = Limiter(_HOME_DATA_RATES) + + async def _get_home_data_version(self, user_data: UserData, path_prefix: str) -> HomeData: + """Fetch home data from ``{path_prefix}/user/homes/{home_id}`` with rate limiting.""" + if not self._home_data_limiter.try_acquire("home_data", blocking=False): + raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.") + if user_data.rriot is None: + raise RoborockException("rriot is none") + home_id = await self._get_home_id(user_data) + path = f"{path_prefix}/user/homes/{home_id}" + home_request = self._rriot_request(user_data, path) + home_response = await home_request.request("get", path) + if not home_response.get("success"): + raise RoborockException(home_response) + home_data = home_response.get("result") + if isinstance(home_data, dict): + return HomeData.from_dict(home_data) + raise RoborockException(f"home_response result was an unexpected type: {home_data}") + + async def get_home_data(self, user_data: UserData) -> HomeData: + return await self._get_home_data_version(user_data, "") + + async def get_home_data_v2(self, user_data: UserData) -> HomeData: + """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums.""" + return await self._get_home_data_version(user_data, "/v2") + + async def get_home_data_v3(self, user_data: UserData) -> HomeData: + """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums.""" + return await self._get_home_data_version(user_data, "/v3") + + async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> list[HomeDataRoom]: + if user_data.rriot is None: + raise RoborockException("rriot is none") + if home_id is None: + home_id = await self._get_home_id(user_data) + path = f"/user/homes/{home_id}/rooms" + room_request = self._rriot_request(user_data, path) + room_response = await room_request.request("get", path) + if not room_response.get("success"): + raise RoborockException(room_response) + rooms = room_response.get("result") + if isinstance(rooms, list): + return [HomeDataRoom.from_dict(room) for room in rooms] + raise RoborockException("home_response result was an unexpected type") + + async def get_shared_device_rooms(self, user_data: UserData, device_id: str) -> list[HomeDataRoom]: + """Fetch room names for a shared (received) device.""" + path = f"/user/deviceshare/query/{device_id}/rooms" + room_request = self._rriot_request(user_data, path) + room_response = await room_request.request("get", path) + if not room_response.get("success"): + raise RoborockException(room_response) + rooms = room_response.get("result") + if isinstance(rooms, list): + output_list = [] + for room in rooms: + normalized_room = room + if isinstance(room, dict) and "id" not in room and "roomId" in room: + normalized_room = {**room, "id": room["roomId"]} + output_list.append(HomeDataRoom.from_dict(normalized_room)) + return output_list + raise RoborockException("get_shared_device_rooms result was an unexpected type") diff --git a/roborock/web_api/jobs.py b/roborock/web_api/jobs.py new file mode 100644 index 00000000..f5925c33 --- /dev/null +++ b/roborock/web_api/jobs.py @@ -0,0 +1,50 @@ +"""Scheduled job / timer endpoints (``/user/devices/{duid}/jobs``).""" + +from roborock.data import HomeDataSchedule, UserData +from roborock.exceptions import RoborockException + +from ._base import RoborockApiBase +from ._hawk import _compact_json + + +class JobsApi(RoborockApiBase): + """Cloud-side cleaning schedules and one-time jobs.""" + + async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]: + path = f"/user/devices/{device_id}/jobs" + schedules_request = self._rriot_request(user_data, path) + schedules_response = await schedules_request.request("get", path) + if not schedules_response.get("success"): + raise RoborockException(schedules_response) + schedules = schedules_response.get("result") + if isinstance(schedules, list): + return [HomeDataSchedule.from_dict(schedule) for schedule in schedules] + raise RoborockException(f"schedule_response result was an unexpected type: {schedules}") + + async def create_job(self, user_data: UserData, device_id: str, job: dict) -> dict: + """Create a /jobs entry (schedule or one-time room clean) on a B01 device. + + Body-bearing writes must sign the request body in the Hawk payload slot and send those same + compact bytes via ``data=``; ``json=`` would re-serialize with spaces and break the MAC. + """ + path = f"/user/devices/{device_id}/jobs" + job_request = self._rriot_request(user_data, path, body=job, extra_headers={"Content-Type": "application/json"}) + response = await job_request.request("post", path, data=_compact_json(job).encode()) + if not response.get("success"): + raise RoborockException(response) + return response + + async def update_job(self, user_data: UserData, device_id: str, job: HomeDataSchedule) -> None: + """Replace an existing cloud job with the full supplied schedule. + + Pass a schedule returned by :meth:`get_schedules` with the desired changes. + The job ID identifies the URL; all other fields are sent as the replacement body. + """ + path = f"/user/devices/{device_id}/jobs/{job.id}" + body = job.as_dict(exclude={"id"}) + job_request = self._rriot_request( + user_data, path, body=body, extra_headers={"Content-Type": "application/json"} + ) + response = await job_request.request("put", path, data=_compact_json(body).encode()) + if not response or not response.get("success"): + raise RoborockException(response) diff --git a/roborock/web_api/ota.py b/roborock/web_api/ota.py new file mode 100644 index 00000000..19f301bf --- /dev/null +++ b/roborock/web_api/ota.py @@ -0,0 +1,44 @@ +"""Firmware / OTA endpoints (``/ota/...`` and the silent-OTA device switch).""" + +from roborock.data import FirmwareInfo, UserData +from roborock.exceptions import RoborockException + +from ._base import RoborockApiBase + + +class OtaApi(RoborockApiBase): + """Firmware update discovery and control.""" + + async def get_firmware_info(self, user_data: UserData, device_id: str) -> FirmwareInfo: + """Get firmware/OTA info for a device (latest version + updatable flag).""" + path = f"/ota/firmware/{device_id}/updatev2" + params = {"lang": "en"} + firmware_request = self._rriot_request(user_data, path, params=params) + firmware_response = await firmware_request.request("get", path, params=params) + if not firmware_response or not firmware_response.get("success"): + raise RoborockException(firmware_response) + return FirmwareInfo.from_dict(firmware_response.get("result") or {}) + + async def start_firmware_update(self, user_data: UserData, device_id: str) -> None: + """Trigger the (irreversible) firmware update for a device. + + The device downloads and flashes the latest firmware reported by + :meth:`get_firmware_info`. + """ + path = f"/ota/device/{device_id}/upgrade" + upgrade_request = self._rriot_request(user_data, path) + upgrade_response = await upgrade_request.request("post", path) + if not upgrade_response or not upgrade_response.get("success"): + raise RoborockException(upgrade_response) + + async def set_silent_ota(self, user_data: UserData, device_id: str, enabled: bool) -> None: + """Enable/disable automatic (silent) firmware updates for a device. + + The current value is reported as ``silent_ota_switch`` on the home-data device. + """ + path = f"/user/devices/{device_id}" + formdata = {"silentOtaSwitch": "true" if enabled else "false"} + silent_ota_request = self._rriot_request(user_data, path, formdata=formdata) + silent_ota_response = await silent_ota_request.request("put", path, data=formdata) + if not silent_ota_response or not silent_ota_response.get("success"): + raise RoborockException(silent_ota_response) diff --git a/roborock/web_api/products.py b/roborock/web_api/products.py new file mode 100644 index 00000000..019bd364 --- /dev/null +++ b/roborock/web_api/products.py @@ -0,0 +1,49 @@ +"""Product catalog and plugin download endpoints on the Base IoT API.""" + +from roborock.data import ProductResponse, UserData +from roborock.exceptions import RoborockException + +from ._base import RoborockApiBase + + +class ProductsApi(RoborockApiBase): + """Product schemas and device plugin bundles.""" + + async def get_products(self, user_data: UserData) -> ProductResponse: + """Gets all products and their schemas, good for determining status codes and model numbers.""" + product_request = await self._iot_request() + product_response = await product_request.request( + "get", + "/api/v4/product", + headers={"Authorization": user_data.token}, + ) + if product_response is None: + raise RoborockException("home_id_response is None") + if product_response.get("code") != 200: + raise RoborockException(f"{product_response.get('msg')} - response code: {product_response.get('code')}") + result = product_response.get("data") + if isinstance(result, dict): + return ProductResponse.from_dict(result) + raise RoborockException("product result was an unexpected type") + + async def download_code(self, user_data: UserData, product_id: int): + product_request = await self._iot_request() + request = {"apilevel": 99999, "productids": [product_id], "type": 2} + response = await product_request.request( + "post", + "/api/v1/appplugin", + json=request, + headers={"Authorization": user_data.token, "Content-Type": "application/json"}, + ) + return response["data"][0]["url"] + + async def download_category_code(self, user_data: UserData): + product_request = await self._iot_request() + response = await product_request.request( + "get", + "api/v1/plugins?apiLevel=99999&type=2", + headers={ + "Authorization": user_data.token, + }, + ) + return {r["category"]: r["url"] for r in response["data"]["categoryPluginList"]} diff --git a/roborock/web_api/scenes.py b/roborock/web_api/scenes.py new file mode 100644 index 00000000..ca9b698f --- /dev/null +++ b/roborock/web_api/scenes.py @@ -0,0 +1,28 @@ +"""Scene (routine) endpoints (``/user/scene``).""" + +from roborock.data import HomeDataScene, UserData +from roborock.exceptions import RoborockException + +from ._base import RoborockApiBase + + +class ScenesApi(RoborockApiBase): + """Listing and executing app routines.""" + + async def get_scenes(self, user_data: UserData, device_id: str) -> list[HomeDataScene]: + path = f"/user/scene/device/{str(device_id)}" + scenes_request = self._rriot_request(user_data, path) + scenes_response = await scenes_request.request("get", path) + if not scenes_response.get("success"): + raise RoborockException(scenes_response) + scenes = scenes_response.get("result") + if isinstance(scenes, list): + return [HomeDataScene.from_dict(scene) for scene in scenes] + raise RoborockException("scene_response result was an unexpected type") + + async def execute_scene(self, user_data: UserData, scene_id: int) -> None: + path = f"/user/scene/{str(scene_id)}/execute" + execute_scene_request = self._rriot_request(user_data, path) + execute_scene_response = await execute_scene_request.request("POST", path) + if not execute_scene_response.get("success"): + raise RoborockException(execute_scene_response) diff --git a/tests/test_web_api.py b/tests/test_web_api.py deleted file mode 100644 index 72038435..00000000 --- a/tests/test_web_api.py +++ /dev/null @@ -1,683 +0,0 @@ -import json -import re -from typing import Any -from unittest.mock import AsyncMock, Mock - -import aiohttp -import pytest -from aioresponses.compat import normalize_url - -from roborock import HomeData, HomeDataRoom, HomeDataScene, HomeDataSchedule, UserData -from roborock.exceptions import RoborockAccountDoesNotExist, RoborockException, RoborockInvalidCredentials -from roborock.web_api import ( - DEFAULT_AGREEMENT_MAJOR_VERSION, - DEFAULT_AGREEMENT_MINOR_VERSION, - IotLoginInfo, - PreparedRequest, - RoborockApiClient, - UserWebApiClient, - _compact_json, - _get_hawk_authentication, -) -from tests.mock_data import HOME_DATA_RAW, USER_DATA - -pytest_plugins = [ - "tests.fixtures.web_api_fixtures", -] - - -@pytest.fixture(autouse=True) -def auto_mock_rest_fixture(mock_rest: Any) -> None: - """Auto use the mock rest fixture for all tests in this module.""" - - -async def test_pass_login_flow() -> None: - """Test that we can login with a password and we get back the correct userdata object.""" - my_session = aiohttp.ClientSession() - api = RoborockApiClient(username="test_user@gmail.com", session=my_session) - ud = await api.pass_login("password") - assert ud == UserData.from_dict(USER_DATA) - assert not my_session.closed - - -async def test_code_login_flow() -> None: - """Test that we can login with a code and we get back the correct userdata object.""" - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code() - ud = await api.code_login(4123) - assert ud == UserData.from_dict(USER_DATA) - - -async def test_get_home_data_v2(): - """Test a full standard flow where we get the home data to end it off. - This matches what HA does""" - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code() - ud = await api.code_login(4123) - hd = await api.get_home_data_v2(ud) - assert hd == HomeData.from_dict(HOME_DATA_RAW) - - -async def test_nc_prepare(): - """Test adding a device and that nothing breaks""" - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code() - ud = await api.code_login(4123) - prepare = await api.nc_prepare(ud, "America/New_York") - new_device = await api.add_device(ud, prepare["s"], prepare["t"]) - assert new_device["duid"] == "rand_duid" - - -async def test_get_scenes(): - """Test that we can get scenes""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - sc = await api.get_scenes(ud, "123456") - assert sc == [ - HomeDataScene.from_dict( - { - "id": 1234567, - "name": "My plan", - } - ) - ] - - -async def test_get_firmware_info(): - """Test that we can get firmware/OTA info for a device.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - fw = await api.get_firmware_info(ud, "abc123") - assert fw.version == "1.2.3" - assert fw.current_version == "1.2.0" - assert fw.updatable is True - assert fw.desc == "Bug fixes and improvements" - assert fw.release_time == "2026-05-01" - assert fw.force_update is False - - -async def test_start_firmware_update(): - """Test that we can trigger a firmware update for a device.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - # Should not raise when the API reports success. - await api.start_firmware_update(ud, "abc123") - - -async def test_set_silent_ota(): - """Test that we can toggle automatic firmware updates for a device.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - # Should not raise when the API reports success. - await api.set_silent_ota(ud, "abc123", True) - await api.set_silent_ota(ud, "abc123", False) - - -async def test_execute_scene(mock_rest): - """Test that we can execute a scene""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - await api.execute_scene(ud, 123456) - mock_rest.assert_any_call("https://api-us.roborock.com/user/scene/123456/execute", "post") - - -async def test_code_login_v4_flow(mock_rest) -> None: - """Test that we can login with a code and we get back the correct userdata object.""" - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code_v4() - ud = await api.code_login_v4(4123, "US", 1) - assert ud == UserData.from_dict(USER_DATA) - - -async def test_code_login_v4_uses_latest_agreement_version(mock_rest) -> None: - """Test that the agreement version sent on login is fetched, not hardcoded. - - The login endpoint rejects a stale agreement version with code 3006, and the - current version differs per server and country, so it cannot be hardcoded. - """ - mock_rest.get( - re.compile(r"https://.*iot\.roborock\.com/api/v3/app/agreement/latest.*"), - status=200, - payload={"code": 200, "data": {"majorVersion": 19, "minorVersion": 1}, "msg": "success"}, - ) - - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code_v4() - await api.code_login_v4(4123, "US", 1) - - login_calls = [ - (key, call) - for key, calls in mock_rest.requests.items() - for call in calls - if "api/v4/auth/email/login/code" in str(key[1]) - ] - assert login_calls, "expected a v4 login request" - data = login_calls[-1][1].kwargs["data"] - assert data["majorVersion"] == 19 - assert data["minorVersion"] == 1 - - -async def test_code_login_v4_agreement_version_fallback(mock_rest) -> None: - """Test that login still proceeds when the agreement version lookup fails.""" - mock_rest.get( - re.compile(r"https://.*iot\.roborock\.com/api/v3/app/agreement/latest.*"), - exception=aiohttp.ClientError("boom"), - ) - - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code_v4() - ud = await api.code_login_v4(4123, "US", 1) - assert ud == UserData.from_dict(USER_DATA) - - login_calls = [ - (key, call) - for key, calls in mock_rest.requests.items() - for call in calls - if "api/v4/auth/email/login/code" in str(key[1]) - ] - data = login_calls[-1][1].kwargs["data"] - assert data["majorVersion"] == DEFAULT_AGREEMENT_MAJOR_VERSION - assert data["minorVersion"] == DEFAULT_AGREEMENT_MINOR_VERSION - - -async def test_code_login_v4_account_does_not_exist(mock_rest) -> None: - """Test that response code 3039 raises RoborockAccountDoesNotExist.""" - mock_rest.clear() - - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"country": "US", "countrycode": "1", "url": "https://usiot.roborock.com"}, - "msg": "success", - }, - ) - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v4/email/code/send.*"), - status=200, - payload={"code": 200, "data": None, "msg": "success"}, - ) - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v3/key/sign.*"), - status=200, - payload={"code": 200, "data": {"k": "mock_k"}, "msg": "success"}, - ) - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v4/auth/email/login/code.*"), - status=200, - payload={"code": 3039, "data": None, "msg": "account does not exist"}, - ) - - api = RoborockApiClient(username="test_user@gmail.com") - await api.request_code_v4() - with pytest.raises(RoborockAccountDoesNotExist) as exc_info: - await api.code_login_v4(4123, "US", 1) - assert "This account does not exist" in str(exc_info.value) - - -async def test_url_cycling(mock_rest) -> None: - """Test that we cycle through the URLs correctly.""" - # Clear mock rest so that we can override the patches. - mock_rest.clear() - - # 1. Mock US URL to return valid status but None for countrycode - mock_rest.post( - re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://usiot.roborock.com", "country": None, "countrycode": None}, - "msg": "Success", - }, - ) - - # 2. Mock EU URL to return valid status but None for countrycode - mock_rest.post( - re.compile("https://euiot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://euiot.roborock.com", "country": None, "countrycode": None}, - "msg": "Success", - }, - ) - - # 3. Mock CN URL to return the correct, valid data - mock_rest.post( - re.compile("https://cniot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://cniot.roborock.com", "country": "CN", "countrycode": "86"}, - "msg": "Success", - }, - ) - - # The RU URL should not be called, but we can mock it just in case - # to catch unexpected behavior. - mock_rest.post(re.compile("https://ruiot.roborock.com/api/v1/getUrlByEmail.*"), status=500) - - client = RoborockApiClient("test@example.com") - result = await client._get_iot_login_info() - - assert result is not None - assert isinstance(result, IotLoginInfo) - assert result.base_url == "https://cniot.roborock.com" - assert result.country == "CN" - assert result.country_code == "86" - - assert client._iot_login_info == result - # Check that all three urls were called. We have to do this kind of weirdly as aioresponses seems to have a bug. - assert ( - len( - mock_rest.requests[ - ( - "post", - normalize_url( - "https://usiot.roborock.com/api/v1/getUrlByEmail?email=test%2540example.com&needtwostepauth=false" - ), - ) - ] - ) - == 1 - ) - assert ( - len( - mock_rest.requests[ - ( - "post", - normalize_url( - "https://euiot.roborock.com/api/v1/getUrlByEmail?email=test%2540example.com&needtwostepauth=false" - ), - ) - ] - ) - == 1 - ) - assert ( - len( - mock_rest.requests[ - ( - "post", - normalize_url( - "https://cniot.roborock.com/api/v1/getUrlByEmail?email=test%2540example.com&needtwostepauth=false" - ), - ) - ] - ) - == 1 - ) - # Make sure we just have the three we tested for above. - assert len(mock_rest.requests) == 3 - - -async def test_thirty_thirty_cycling(mock_rest) -> None: - """Test that we cycle through the URLs correctly when users have deleted accounts in higher prio regions.""" - # Clear mock rest so that we can override the patches. - mock_rest.clear() - - mock_rest.post( - re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://usiot.roborock.com", "country": "US", "countrycode": 1}, - "msg": "Account in deletion", - }, - ) - - mock_rest.post( - re.compile("https://euiot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://euiot.roborock.com", "country": "EU", "countrycode": 49}, - "msg": "Success", - }, - ) - - mock_rest.post( - re.compile("https://usiot.roborock.com/api/v4/email/code/send.*"), - status=200, - payload={ - "code": 3030, - }, - ) - mock_rest.post( - re.compile("https://euiot.roborock.com/api/v4/email/code/send.*"), - status=200, - payload={ - "code": 200, - }, - ) - - mock_rest.post(re.compile("https://ruiot.roborock.com/api/v1/getUrlByEmail.*"), status=500) - mock_rest.post(re.compile("https://cniot.roborock.com/api/v1/getUrlByEmail.*"), status=500) - - client = RoborockApiClient("test@example.com") - await client.request_code_v4() - - assert ( - len( - mock_rest.requests[ - ( - "post", - normalize_url("https://euiot.roborock.com/api/v4/email/code/send"), - ) - ] - ) - == 1 - ) - assert ( - len( - mock_rest.requests[ - ( - "post", - normalize_url("https://usiot.roborock.com/api/v4/email/code/send"), - ) - ] - ) - == 1 - ) - # Assert that we didn't try on the Russian or Chinese regions - assert "https://ruiot.roborock.com/api/v4/email/code/send" not in mock_rest.requests - assert "https://cniot.roborock.com/api/v4/email/code/send" not in mock_rest.requests - - -async def test_missing_country_login(mock_rest) -> None: - """Test that we cycle through the URLs correctly.""" - mock_rest.clear() - # Make country None, but country code set. - mock_rest.post( - re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), - status=200, - payload={ - "code": 200, - "data": {"url": "https://usiot.roborock.com", "country": None, "countrycode": 1}, - "msg": "Success", - }, - ) - # v4 is not mocked, so it would fail it were called. - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v1/loginWithCode.*"), - status=200, - payload={"code": 200, "data": USER_DATA, "msg": "success"}, - ) - mock_rest.post( - re.compile(r"https://.*iot\.roborock\.com/api/v1/sendEmailCode.*"), - status=200, - payload={"code": 200, "data": None, "msg": "success"}, - ) - - client = RoborockApiClient("test@example.com") - await client.request_code_v4() - ud = await client.code_login_v4(4123) - assert ud is not None - # Ensure we have no surprise REST calls. - assert len(mock_rest.requests) == 3 - - -async def test_get_schedules(mock_rest) -> None: - """Test that we can get schedules.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - - # Mock the response - mock_rest.get( - "https://api-us.roborock.com/user/devices/123456/jobs", - status=200, - payload={ - "api": None, - "result": [ - { - "id": 3878757, - "cron": "03 13 15 12 ?", - "repeated": False, - "enabled": True, - "param": { - "id": 1, - "method": "server_scheduled_start", - "params": [ - { - "repeat": 1, - "water_box_mode": 202, - "segments": "0", - "fan_power": 102, - "mop_mode": 300, - "clean_mop": 1, - "map_index": -1, - "name": "1765735413736", - } - ], - }, - } - ], - "status": "ok", - "success": True, - }, - ) - - schedules = await api.get_schedules(ud, "123456") - assert len(schedules) == 1 - schedule = schedules[0] - assert schedule.id == 3878757 - assert schedule.cron == "03 13 15 12 ?" - assert schedule.repeated is False - assert schedule.enabled is True - - -async def test_create_job(mock_rest) -> None: - """A /jobs write (schedule or one-time clean) POSTs the body and returns the parsed result.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - - mock_rest.post( - "https://api-us.roborock.com/user/devices/123456/jobs", - status=200, - payload={"api": None, "result": "done", "status": "ok", "success": True}, - ) - - job = {"cron": "05 10 * * ?", "repeated": True, "enabled": True, "param": {"rooms": [1, 2], "roomCount": 2}} - response = await api.create_job(ud, "123456", job) - assert response["success"] is True - assert response["result"] == "done" - - -@pytest.mark.parametrize( - ("enabled", "mac"), - [ - (True, "fCO2ogG/WjxJjWEZw6l3ncC7cFChl/95jjjzu/YNRV0="), - (False, "GsGuHY1XF375VJRdhMmoIYI/AacILSNCFBU1+3G/O0Y="), - ], -) -async def test_update_job(mock_rest, monkeypatch, enabled: bool, mac: str) -> None: - """Updates send the full schedule as compact JSON with a matching body-aware Hawk MAC.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = UserData.from_dict(USER_DATA) - monkeypatch.setattr("roborock.web_api.time.time", lambda: 1_700_000_000) - monkeypatch.setattr("roborock.web_api.secrets.token_urlsafe", lambda n: "fixednonce") - url = "https://api-us.roborock.com/user/devices/123456/jobs/3878757" - mock_rest.put(url, payload={"result": "done", "status": "ok", "success": True}) - job = HomeDataSchedule( - id=3878757, - cron="05 10 * * ?", - repeated=True, - enabled=enabled, - param={"mapId": 0, "rooms": [1, 2], "roomCount": 2, "custom_key": {"id": 7, "value": None}}, - ) - original_job = job.as_dict() - - await api.update_job(ud, "123456", job) - - enabled_json = b"true" if enabled else b"false" - expected_body = ( - b'{"cron":"05 10 * * ?","repeated":true,"enabled":' - + enabled_json - + b',"param":{"mapId":0,"rooms":[1,2],"roomCount":2,"custom_key":{"id":7,"value":null}}}' - ) - mock_rest.assert_called_once() - request = mock_rest.requests[("put", normalize_url(url))][0] - assert request.kwargs["data"] == expected_body - assert request.kwargs["headers"] == { - "Authorization": f'Hawk id="user123",s="pass123",ts="1700000000",nonce="fixednonce",mac="{mac}"', - "Content-Type": "application/json", - } - assert request.kwargs["params"] is None - assert request.kwargs["json"] is None - assert job.as_dict() == original_job - - -@pytest.mark.parametrize("response", [{"success": False, "error": "Job not found"}, {}, None]) -async def test_update_job_failed_response(mock_rest, response: dict | None) -> None: - """Rejected and missing success responses raise the library's exception.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = UserData.from_dict(USER_DATA) - mock_rest.put( - "https://api-us.roborock.com/user/devices/123456/jobs/3878757", - body=json.dumps(response), - content_type="application/json", - ) - job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) - - with pytest.raises(RoborockException) as exc_info: - await api.update_job(ud, "123456", job) - - assert exc_info.value.args == (response,) - - -@pytest.mark.parametrize("missing_field", ["rriot", "a"]) -async def test_update_job_missing_credentials(mock_rest, missing_field: str) -> None: - """Missing cloud credentials are rejected before making a request.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = UserData.from_dict(USER_DATA) - if missing_field == "rriot": - ud.rriot = None - message = "rriot is none" - else: - assert ud.rriot is not None - ud.rriot.r.a = None - message = "Missing field 'a' in rriot reference" - job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) - - with pytest.raises(RoborockException, match=message): - await api.update_job(ud, "123456", job) - - assert not mock_rest.requests - - -async def test_update_job_network_error(mock_rest) -> None: - """Network failures surface as RoborockException through the shared request layer.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = UserData.from_dict(USER_DATA) - mock_rest.put( - "https://api-us.roborock.com/user/devices/123456/jobs/3878757", - exception=aiohttp.ClientConnectionError("connection lost"), - ) - job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) - - with pytest.raises(RoborockException, match="Network error contacting .*: connection lost"): - await api.update_job(ud, "123456", job) - - -def test_hawk_authentication_signs_body(monkeypatch) -> None: - """Body-bearing writes sign md5(compact JSON body) in the Hawk payload slot; GET is unchanged.""" - from roborock import web_api - - monkeypatch.setattr(web_api.time, "time", lambda: 1_700_000_000) - monkeypatch.setattr(web_api.secrets, "token_urlsafe", lambda n: "fixednonce") - rriot = UserData.from_dict(USER_DATA).rriot - assert rriot is not None - - body = {"b": 2, "a": 1} - signed = _get_hawk_authentication(rriot, "/p", body=body) - unsigned = _get_hawk_authentication(rriot, "/p") - formdata_signed = _get_hawk_authentication(rriot, "/p", formdata=body) - - assert signed != unsigned # the body is actually covered by the MAC - assert signed != formdata_signed # body-signing differs from formdata-signing - assert _compact_json(body) == '{"b":2,"a":1}' - - -@pytest.mark.parametrize( - "result_payload", - [ - # roomId field (deviceshare endpoint convention) - [ - {"roomId": "2362048", "name": "Living Room"}, - {"roomId": 2362044, "name": "Kitchen"}, - ], - # id field (matches /user/homes/{id}/rooms — defensive in case the API normalizes) - [ - {"id": 2362048, "name": "Living Room"}, - {"id": 2362044, "name": "Kitchen"}, - ], - ], -) -async def test_get_shared_device_rooms(mock_rest, result_payload) -> None: - """Test that shared-device rooms are fetched from the deviceshare query path.""" - api = RoborockApiClient(username="test_user@gmail.com") - ud = await api.pass_login("password") - - mock_rest.get( - "https://api-us.roborock.com/user/deviceshare/query/device-id-q7/rooms", - status=200, - payload={ - "api": None, - "code": 200, - "result": result_payload, - "status": "ok", - "success": True, - }, - ) - - rooms = await api.get_shared_device_rooms(ud, "device-id-q7") - - assert rooms == [ - HomeDataRoom(id=2362048, name="Living Room"), - HomeDataRoom(id=2362044, name="Kitchen"), - ] - - -async def test_user_web_api_client_unauthorized_hook() -> None: - """Test that UserWebApiClient triggers unauthorized hook on RoborockInvalidCredentials.""" - mock_hook = Mock() - mock_api = AsyncMock(spec=RoborockApiClient) - - # Setup mock to raise RoborockInvalidCredentials - mock_api.get_home_data_v3.side_effect = RoborockInvalidCredentials("Unauthorized") - - client = UserWebApiClient(mock_api, UserData.from_dict(USER_DATA), unauthorized_hook=mock_hook) - - with pytest.raises(RoborockInvalidCredentials): - await client.get_home_data() - - mock_hook.assert_called_once() - - # Test another method - mock_hook.reset_mock() - mock_api.get_rooms.side_effect = RoborockInvalidCredentials("Unauthorized") - with pytest.raises(RoborockInvalidCredentials): - await client.get_rooms() - mock_hook.assert_called_once() - - -@pytest.mark.parametrize( - "exception", - [ - aiohttp.ClientError("connection failed"), - TimeoutError("timed out"), - OSError("dns failure"), - ], - ids=["client_error", "timeout", "os_error"], -) -async def test_prepared_request_wraps_network_errors(exception: Exception, mock_rest: Any) -> None: - """Network errors from aiohttp must be wrapped as RoborockException so - consumers can rely on the typed exception hierarchy.""" - url_pattern = re.compile(r"https://example\.com/api/test.*") - mock_rest.post(url_pattern, exception=exception) - - req = PreparedRequest("https://example.com") - with pytest.raises(RoborockException) as exc_info: - await req.request("post", "/api/test") - assert exc_info.value.__cause__ is exception diff --git a/tests/web_api/__init__.py b/tests/web_api/__init__.py new file mode 100644 index 00000000..bfc0a86b --- /dev/null +++ b/tests/web_api/__init__.py @@ -0,0 +1,5 @@ +"""Tests for the roborock.web_api package.""" + +pytest_plugins = [ + "tests.fixtures.web_api_fixtures", +] diff --git a/tests/web_api/conftest.py b/tests/web_api/conftest.py new file mode 100644 index 00000000..67fdbe1f --- /dev/null +++ b/tests/web_api/conftest.py @@ -0,0 +1,8 @@ +from typing import Any + +import pytest + + +@pytest.fixture(autouse=True) +def auto_mock_rest_fixture(mock_rest: Any) -> None: + """Auto use the mock rest fixture for all tests in this package.""" diff --git a/tests/web_api/test_auth.py b/tests/web_api/test_auth.py new file mode 100644 index 00000000..8af3f95c --- /dev/null +++ b/tests/web_api/test_auth.py @@ -0,0 +1,234 @@ +"""Tests for login and verification-code flows.""" + +import re + +import aiohttp +import pytest +from aioresponses.compat import normalize_url + +from roborock import UserData +from roborock.exceptions import RoborockAccountDoesNotExist +from roborock.web_api import ( + DEFAULT_AGREEMENT_MAJOR_VERSION, + DEFAULT_AGREEMENT_MINOR_VERSION, + RoborockApiClient, +) +from tests.mock_data import USER_DATA + + +async def test_pass_login_flow() -> None: + """Test that we can login with a password and we get back the correct userdata object.""" + my_session = aiohttp.ClientSession() + api = RoborockApiClient(username="test_user@gmail.com", session=my_session) + ud = await api.pass_login("password") + assert ud == UserData.from_dict(USER_DATA) + assert not my_session.closed + + +async def test_code_login_flow() -> None: + """Test that we can login with a code and we get back the correct userdata object.""" + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code() + ud = await api.code_login(4123) + assert ud == UserData.from_dict(USER_DATA) + + +async def test_code_login_v4_flow(mock_rest) -> None: + """Test that we can login with a code and we get back the correct userdata object.""" + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code_v4() + ud = await api.code_login_v4(4123, "US", 1) + assert ud == UserData.from_dict(USER_DATA) + + +async def test_code_login_v4_uses_latest_agreement_version(mock_rest) -> None: + """Test that the agreement version sent on login is fetched, not hardcoded. + + The login endpoint rejects a stale agreement version with code 3006, and the + current version differs per server and country, so it cannot be hardcoded. + """ + mock_rest.get( + re.compile(r"https://.*iot\.roborock\.com/api/v3/app/agreement/latest.*"), + status=200, + payload={"code": 200, "data": {"majorVersion": 19, "minorVersion": 1}, "msg": "success"}, + ) + + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code_v4() + await api.code_login_v4(4123, "US", 1) + + login_calls = [ + (key, call) + for key, calls in mock_rest.requests.items() + for call in calls + if "api/v4/auth/email/login/code" in str(key[1]) + ] + assert login_calls, "expected a v4 login request" + data = login_calls[-1][1].kwargs["data"] + assert data["majorVersion"] == 19 + assert data["minorVersion"] == 1 + + +async def test_code_login_v4_agreement_version_fallback(mock_rest) -> None: + """Test that login still proceeds when the agreement version lookup fails.""" + mock_rest.get( + re.compile(r"https://.*iot\.roborock\.com/api/v3/app/agreement/latest.*"), + exception=aiohttp.ClientError("boom"), + ) + + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code_v4() + ud = await api.code_login_v4(4123, "US", 1) + assert ud == UserData.from_dict(USER_DATA) + + login_calls = [ + (key, call) + for key, calls in mock_rest.requests.items() + for call in calls + if "api/v4/auth/email/login/code" in str(key[1]) + ] + data = login_calls[-1][1].kwargs["data"] + assert data["majorVersion"] == DEFAULT_AGREEMENT_MAJOR_VERSION + assert data["minorVersion"] == DEFAULT_AGREEMENT_MINOR_VERSION + + +async def test_code_login_v4_account_does_not_exist(mock_rest) -> None: + """Test that response code 3039 raises RoborockAccountDoesNotExist.""" + mock_rest.clear() + + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"country": "US", "countrycode": "1", "url": "https://usiot.roborock.com"}, + "msg": "success", + }, + ) + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v4/email/code/send.*"), + status=200, + payload={"code": 200, "data": None, "msg": "success"}, + ) + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v3/key/sign.*"), + status=200, + payload={"code": 200, "data": {"k": "mock_k"}, "msg": "success"}, + ) + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v4/auth/email/login/code.*"), + status=200, + payload={"code": 3039, "data": None, "msg": "account does not exist"}, + ) + + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code_v4() + with pytest.raises(RoborockAccountDoesNotExist) as exc_info: + await api.code_login_v4(4123, "US", 1) + assert "This account does not exist" in str(exc_info.value) + + +async def test_thirty_thirty_cycling(mock_rest) -> None: + """Test that we cycle through the URLs correctly when users have deleted accounts in higher prio regions.""" + # Clear mock rest so that we can override the patches. + mock_rest.clear() + + mock_rest.post( + re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://usiot.roborock.com", "country": "US", "countrycode": 1}, + "msg": "Account in deletion", + }, + ) + + mock_rest.post( + re.compile("https://euiot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://euiot.roborock.com", "country": "EU", "countrycode": 49}, + "msg": "Success", + }, + ) + + mock_rest.post( + re.compile("https://usiot.roborock.com/api/v4/email/code/send.*"), + status=200, + payload={ + "code": 3030, + }, + ) + mock_rest.post( + re.compile("https://euiot.roborock.com/api/v4/email/code/send.*"), + status=200, + payload={ + "code": 200, + }, + ) + + mock_rest.post(re.compile("https://ruiot.roborock.com/api/v1/getUrlByEmail.*"), status=500) + mock_rest.post(re.compile("https://cniot.roborock.com/api/v1/getUrlByEmail.*"), status=500) + + client = RoborockApiClient("test@example.com") + await client.request_code_v4() + + assert ( + len( + mock_rest.requests[ + ( + "post", + normalize_url("https://euiot.roborock.com/api/v4/email/code/send"), + ) + ] + ) + == 1 + ) + assert ( + len( + mock_rest.requests[ + ( + "post", + normalize_url("https://usiot.roborock.com/api/v4/email/code/send"), + ) + ] + ) + == 1 + ) + # Assert that we didn't try on the Russian or Chinese regions + assert "https://ruiot.roborock.com/api/v4/email/code/send" not in mock_rest.requests + assert "https://cniot.roborock.com/api/v4/email/code/send" not in mock_rest.requests + + +async def test_missing_country_login(mock_rest) -> None: + """Test that we cycle through the URLs correctly.""" + mock_rest.clear() + # Make country None, but country code set. + mock_rest.post( + re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://usiot.roborock.com", "country": None, "countrycode": 1}, + "msg": "Success", + }, + ) + # v4 is not mocked, so it would fail it were called. + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v1/loginWithCode.*"), + status=200, + payload={"code": 200, "data": USER_DATA, "msg": "success"}, + ) + mock_rest.post( + re.compile(r"https://.*iot\.roborock\.com/api/v1/sendEmailCode.*"), + status=200, + payload={"code": 200, "data": None, "msg": "success"}, + ) + + client = RoborockApiClient("test@example.com") + await client.request_code_v4() + ud = await client.code_login_v4(4123) + assert ud is not None + # Ensure we have no surprise REST calls. + assert len(mock_rest.requests) == 3 diff --git a/tests/web_api/test_base.py b/tests/web_api/test_base.py new file mode 100644 index 00000000..48da1d0d --- /dev/null +++ b/tests/web_api/test_base.py @@ -0,0 +1,72 @@ +"""Tests for regional base URL discovery shared by every web API mixin.""" + +import re + +from aioresponses.compat import normalize_url + +from roborock.web_api import IotLoginInfo, RoborockApiClient + + +async def test_url_cycling(mock_rest) -> None: + """Test that we cycle through the URLs correctly.""" + # Clear mock rest so that we can override the patches. + mock_rest.clear() + + # 1. Mock US URL to return valid status but None for countrycode + mock_rest.post( + re.compile("https://usiot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://usiot.roborock.com", "country": None, "countrycode": None}, + "msg": "Success", + }, + ) + + # 2. Mock EU URL to return valid status but None for countrycode + mock_rest.post( + re.compile("https://euiot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://euiot.roborock.com", "country": None, "countrycode": None}, + "msg": "Success", + }, + ) + + # 3. Mock CN URL to return the correct, valid data + mock_rest.post( + re.compile("https://cniot.roborock.com/api/v1/getUrlByEmail.*"), + status=200, + payload={ + "code": 200, + "data": {"url": "https://cniot.roborock.com", "country": "CN", "countrycode": "86"}, + "msg": "Success", + }, + ) + + # The RU URL should not be called, but we can mock it just in case + # to catch unexpected behavior. + mock_rest.post(re.compile("https://ruiot.roborock.com/api/v1/getUrlByEmail.*"), status=500) + + client = RoborockApiClient("test@example.com") + result = await client._get_iot_login_info() + + assert result is not None + assert isinstance(result, IotLoginInfo) + assert result.base_url == "https://cniot.roborock.com" + assert result.country == "CN" + assert result.country_code == "86" + + assert client._iot_login_info == result + # Check that all three urls were called. We have to do this kind of weirdly as aioresponses seems to have a bug. + for region in ("us", "eu", "cn"): + key = ( + "post", + normalize_url( + f"https://{region}iot.roborock.com/api/v1/getUrlByEmail?email=test%2540example.com&needtwostepauth=false" + ), + ) + assert len(mock_rest.requests[key]) == 1 + # Make sure we just have the three we tested for above. + assert len(mock_rest.requests) == 3 diff --git a/tests/web_api/test_client.py b/tests/web_api/test_client.py new file mode 100644 index 00000000..40546b44 --- /dev/null +++ b/tests/web_api/test_client.py @@ -0,0 +1,55 @@ +"""Tests for the composed public clients and the backwards-compatible module surface.""" + +from unittest.mock import AsyncMock, Mock + +import pytest + +from roborock import UserData +from roborock.exceptions import RoborockInvalidCredentials +from roborock.web_api import RoborockApiClient, UserWebApiClient +from tests.mock_data import USER_DATA + + +def test_legacy_module_exports() -> None: + """Names previously importable from the ``roborock.web_api`` module keep resolving.""" + from roborock import web_api + + for name in ( + "BASE_URLS", + "DEFAULT_AGREEMENT_MAJOR_VERSION", + "DEFAULT_AGREEMENT_MINOR_VERSION", + "IotLoginInfo", + "PreparedRequest", + "RoborockApiClient", + "UserWebApiClient", + "_compact_json", + "_get_hawk_authentication", + "_LOGGER", + "aiohttp", + "secrets", + "time", + ): + assert hasattr(web_api, name), name + + +async def test_user_web_api_client_unauthorized_hook() -> None: + """Test that UserWebApiClient triggers unauthorized hook on RoborockInvalidCredentials.""" + mock_hook = Mock() + mock_api = AsyncMock(spec=RoborockApiClient) + + # Setup mock to raise RoborockInvalidCredentials + mock_api.get_home_data_v3.side_effect = RoborockInvalidCredentials("Unauthorized") + + client = UserWebApiClient(mock_api, UserData.from_dict(USER_DATA), unauthorized_hook=mock_hook) + + with pytest.raises(RoborockInvalidCredentials): + await client.get_home_data() + + mock_hook.assert_called_once() + + # Test another method + mock_hook.reset_mock() + mock_api.get_rooms.side_effect = RoborockInvalidCredentials("Unauthorized") + with pytest.raises(RoborockInvalidCredentials): + await client.get_rooms() + mock_hook.assert_called_once() diff --git a/tests/web_api/test_devices.py b/tests/web_api/test_devices.py new file mode 100644 index 00000000..bab69879 --- /dev/null +++ b/tests/web_api/test_devices.py @@ -0,0 +1,13 @@ +"""Tests for device pairing endpoints.""" + +from roborock.web_api import RoborockApiClient + + +async def test_nc_prepare(): + """Test adding a device and that nothing breaks""" + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code() + ud = await api.code_login(4123) + prepare = await api.nc_prepare(ud, "America/New_York") + new_device = await api.add_device(ud, prepare["s"], prepare["t"]) + assert new_device["duid"] == "rand_duid" diff --git a/tests/web_api/test_hawk.py b/tests/web_api/test_hawk.py new file mode 100644 index 00000000..2c374a8a --- /dev/null +++ b/tests/web_api/test_hawk.py @@ -0,0 +1,23 @@ +"""Tests for Hawk request signing.""" + +from roborock import UserData +from roborock.web_api import _hawk +from roborock.web_api._hawk import _compact_json, _get_hawk_authentication +from tests.mock_data import USER_DATA + + +def test_hawk_authentication_signs_body(monkeypatch) -> None: + """Body-bearing writes sign md5(compact JSON body) in the Hawk payload slot; GET is unchanged.""" + monkeypatch.setattr(_hawk.time, "time", lambda: 1_700_000_000) + monkeypatch.setattr(_hawk.secrets, "token_urlsafe", lambda n: "fixednonce") + rriot = UserData.from_dict(USER_DATA).rriot + assert rriot is not None + + body = {"b": 2, "a": 1} + signed = _get_hawk_authentication(rriot, "/p", body=body) + unsigned = _get_hawk_authentication(rriot, "/p") + formdata_signed = _get_hawk_authentication(rriot, "/p", formdata=body) + + assert signed != unsigned # the body is actually covered by the MAC + assert signed != formdata_signed # body-signing differs from formdata-signing + assert _compact_json(body) == '{"b":2,"a":1}' diff --git a/tests/web_api/test_homes.py b/tests/web_api/test_homes.py new file mode 100644 index 00000000..a8d562ca --- /dev/null +++ b/tests/web_api/test_homes.py @@ -0,0 +1,81 @@ +"""Tests for home data and room endpoints.""" + +import pytest +from aioresponses.compat import normalize_url + +from roborock import HomeData, HomeDataRoom, UserData +from roborock.exceptions import RoborockException +from roborock.web_api import RoborockApiClient +from tests.mock_data import HOME_DATA_RAW, USER_DATA + + +async def test_get_home_data_v2(): + """Test a full standard flow where we get the home data to end it off. + This matches what HA does""" + api = RoborockApiClient(username="test_user@gmail.com") + await api.request_code() + ud = await api.code_login(4123) + hd = await api.get_home_data_v2(ud) + assert hd == HomeData.from_dict(HOME_DATA_RAW) + + +async def test_get_home_data_v3(mock_rest): + """Test that the v3 home data endpoint is requested and parsed.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + hd = await api.get_home_data_v3(ud) + assert hd == HomeData.from_dict(HOME_DATA_RAW) + assert ("get", normalize_url("https://api-us.roborock.com/v3/user/homes/123456")) in mock_rest.requests + + +@pytest.mark.parametrize( + "result_payload", + [ + # roomId field (deviceshare endpoint convention) + [ + {"roomId": "2362048", "name": "Living Room"}, + {"roomId": 2362044, "name": "Kitchen"}, + ], + # id field (matches /user/homes/{id}/rooms — defensive in case the API normalizes) + [ + {"id": 2362048, "name": "Living Room"}, + {"id": 2362044, "name": "Kitchen"}, + ], + ], +) +async def test_get_shared_device_rooms(mock_rest, result_payload) -> None: + """Test that shared-device rooms are fetched from the deviceshare query path.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + + mock_rest.get( + "https://api-us.roborock.com/user/deviceshare/query/device-id-q7/rooms", + status=200, + payload={ + "api": None, + "code": 200, + "result": result_payload, + "status": "ok", + "success": True, + }, + ) + + rooms = await api.get_shared_device_rooms(ud, "device-id-q7") + + assert rooms == [ + HomeDataRoom(id=2362048, name="Living Room"), + HomeDataRoom(id=2362044, name="Kitchen"), + ] + + +@pytest.mark.parametrize("method", ["get_home_data", "get_home_data_v2", "get_home_data_v3", "get_rooms"]) +async def test_missing_rriot_fails_before_network(mock_rest, method: str) -> None: + """Missing rriot credentials are rejected before the home-detail lookup is made.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + ud.rriot = None + + with pytest.raises(RoborockException, match="rriot is none"): + await getattr(api, method)(ud) + + assert not mock_rest.requests diff --git a/tests/web_api/test_jobs.py b/tests/web_api/test_jobs.py new file mode 100644 index 00000000..0a32e1b6 --- /dev/null +++ b/tests/web_api/test_jobs.py @@ -0,0 +1,174 @@ +"""Tests for scheduled job endpoints.""" + +import json + +import aiohttp +import pytest +from aioresponses.compat import normalize_url + +from roborock import HomeDataSchedule, UserData +from roborock.exceptions import RoborockException +from roborock.web_api import RoborockApiClient +from tests.mock_data import USER_DATA + + +async def test_get_schedules(mock_rest) -> None: + """Test that we can get schedules.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + + # Mock the response + mock_rest.get( + "https://api-us.roborock.com/user/devices/123456/jobs", + status=200, + payload={ + "api": None, + "result": [ + { + "id": 3878757, + "cron": "03 13 15 12 ?", + "repeated": False, + "enabled": True, + "param": { + "id": 1, + "method": "server_scheduled_start", + "params": [ + { + "repeat": 1, + "water_box_mode": 202, + "segments": "0", + "fan_power": 102, + "mop_mode": 300, + "clean_mop": 1, + "map_index": -1, + "name": "1765735413736", + } + ], + }, + } + ], + "status": "ok", + "success": True, + }, + ) + + schedules = await api.get_schedules(ud, "123456") + assert len(schedules) == 1 + schedule = schedules[0] + assert schedule.id == 3878757 + assert schedule.cron == "03 13 15 12 ?" + assert schedule.repeated is False + assert schedule.enabled is True + + +async def test_create_job(mock_rest) -> None: + """A /jobs write (schedule or one-time clean) POSTs the body and returns the parsed result.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + + mock_rest.post( + "https://api-us.roborock.com/user/devices/123456/jobs", + status=200, + payload={"api": None, "result": "done", "status": "ok", "success": True}, + ) + + job = {"cron": "05 10 * * ?", "repeated": True, "enabled": True, "param": {"rooms": [1, 2], "roomCount": 2}} + response = await api.create_job(ud, "123456", job) + assert response["success"] is True + assert response["result"] == "done" + + +@pytest.mark.parametrize( + ("enabled", "mac"), + [ + (True, "fCO2ogG/WjxJjWEZw6l3ncC7cFChl/95jjjzu/YNRV0="), + (False, "GsGuHY1XF375VJRdhMmoIYI/AacILSNCFBU1+3G/O0Y="), + ], +) +async def test_update_job(mock_rest, monkeypatch, enabled: bool, mac: str) -> None: + """Updates send the full schedule as compact JSON with a matching body-aware Hawk MAC.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + monkeypatch.setattr("roborock.web_api._hawk.time.time", lambda: 1_700_000_000) + monkeypatch.setattr("roborock.web_api._hawk.secrets.token_urlsafe", lambda n: "fixednonce") + url = "https://api-us.roborock.com/user/devices/123456/jobs/3878757" + mock_rest.put(url, payload={"result": "done", "status": "ok", "success": True}) + job = HomeDataSchedule( + id=3878757, + cron="05 10 * * ?", + repeated=True, + enabled=enabled, + param={"mapId": 0, "rooms": [1, 2], "roomCount": 2, "custom_key": {"id": 7, "value": None}}, + ) + original_job = job.as_dict() + + await api.update_job(ud, "123456", job) + + enabled_json = b"true" if enabled else b"false" + expected_body = ( + b'{"cron":"05 10 * * ?","repeated":true,"enabled":' + + enabled_json + + b',"param":{"mapId":0,"rooms":[1,2],"roomCount":2,"custom_key":{"id":7,"value":null}}}' + ) + mock_rest.assert_called_once() + request = mock_rest.requests[("put", normalize_url(url))][0] + assert request.kwargs["data"] == expected_body + assert request.kwargs["headers"] == { + "Authorization": f'Hawk id="user123",s="pass123",ts="1700000000",nonce="fixednonce",mac="{mac}"', + "Content-Type": "application/json", + } + assert request.kwargs["params"] is None + assert request.kwargs["json"] is None + assert job.as_dict() == original_job + + +@pytest.mark.parametrize("response", [{"success": False, "error": "Job not found"}, {}, None]) +async def test_update_job_failed_response(mock_rest, response: dict | None) -> None: + """Rejected and missing success responses raise the library's exception.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + mock_rest.put( + "https://api-us.roborock.com/user/devices/123456/jobs/3878757", + body=json.dumps(response), + content_type="application/json", + ) + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException) as exc_info: + await api.update_job(ud, "123456", job) + + assert exc_info.value.args == (response,) + + +@pytest.mark.parametrize("missing_field", ["rriot", "a"]) +async def test_update_job_missing_credentials(mock_rest, missing_field: str) -> None: + """Missing cloud credentials are rejected before making a request.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + if missing_field == "rriot": + ud.rriot = None + message = "rriot is none" + else: + assert ud.rriot is not None + ud.rriot.r.a = None + message = "Missing field 'a' in rriot reference" + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException, match=message): + await api.update_job(ud, "123456", job) + + assert not mock_rest.requests + + +async def test_update_job_network_error(mock_rest) -> None: + """Network failures surface as RoborockException through the shared request layer.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = UserData.from_dict(USER_DATA) + mock_rest.put( + "https://api-us.roborock.com/user/devices/123456/jobs/3878757", + exception=aiohttp.ClientConnectionError("connection lost"), + ) + job = HomeDataSchedule(id=3878757, cron="05 10 * * ?", repeated=True, enabled=False) + + with pytest.raises(RoborockException, match="Network error contacting .*: connection lost"): + await api.update_job(ud, "123456", job) diff --git a/tests/web_api/test_ota.py b/tests/web_api/test_ota.py new file mode 100644 index 00000000..e7dda2ad --- /dev/null +++ b/tests/web_api/test_ota.py @@ -0,0 +1,33 @@ +"""Tests for firmware / OTA endpoints.""" + +from roborock.web_api import RoborockApiClient + + +async def test_get_firmware_info(): + """Test that we can get firmware/OTA info for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + fw = await api.get_firmware_info(ud, "abc123") + assert fw.version == "1.2.3" + assert fw.current_version == "1.2.0" + assert fw.updatable is True + assert fw.desc == "Bug fixes and improvements" + assert fw.release_time == "2026-05-01" + assert fw.force_update is False + + +async def test_start_firmware_update(): + """Test that we can trigger a firmware update for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + # Should not raise when the API reports success. + await api.start_firmware_update(ud, "abc123") + + +async def test_set_silent_ota(): + """Test that we can toggle automatic firmware updates for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + # Should not raise when the API reports success. + await api.set_silent_ota(ud, "abc123", True) + await api.set_silent_ota(ud, "abc123", False) diff --git a/tests/web_api/test_request.py b/tests/web_api/test_request.py new file mode 100644 index 00000000..a4596ca9 --- /dev/null +++ b/tests/web_api/test_request.py @@ -0,0 +1,31 @@ +"""Tests for the low-level PreparedRequest helper.""" + +import re +from typing import Any + +import aiohttp +import pytest + +from roborock.exceptions import RoborockException +from roborock.web_api import PreparedRequest + + +@pytest.mark.parametrize( + "exception", + [ + aiohttp.ClientError("connection failed"), + TimeoutError("timed out"), + OSError("dns failure"), + ], + ids=["client_error", "timeout", "os_error"], +) +async def test_prepared_request_wraps_network_errors(exception: Exception, mock_rest: Any) -> None: + """Network errors from aiohttp must be wrapped as RoborockException so + consumers can rely on the typed exception hierarchy.""" + url_pattern = re.compile(r"https://example\.com/api/test.*") + mock_rest.post(url_pattern, exception=exception) + + req = PreparedRequest("https://example.com") + with pytest.raises(RoborockException) as exc_info: + await req.request("post", "/api/test") + assert exc_info.value.__cause__ is exception diff --git a/tests/web_api/test_scenes.py b/tests/web_api/test_scenes.py new file mode 100644 index 00000000..8a8aef81 --- /dev/null +++ b/tests/web_api/test_scenes.py @@ -0,0 +1,27 @@ +"""Tests for scene (routine) endpoints.""" + +from roborock import HomeDataScene +from roborock.web_api import RoborockApiClient + + +async def test_get_scenes(): + """Test that we can get scenes""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + sc = await api.get_scenes(ud, "123456") + assert sc == [ + HomeDataScene.from_dict( + { + "id": 1234567, + "name": "My plan", + } + ) + ] + + +async def test_execute_scene(mock_rest): + """Test that we can execute a scene""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + await api.execute_scene(ud, 123456) + mock_rest.assert_any_call("https://api-us.roborock.com/user/scene/123456/execute", "post") From 7f90403b7b01cdacf88367c67d34bd2e2e993d1a Mon Sep 17 00:00:00 2001 From: Luke Date: Sun, 1 Nov 2026 18:02:34 -0500 Subject: [PATCH 2/2] refactor(web_api): remove unused stdlib and private re-exports from __init__ --- roborock/web_api/__init__.py | 16 +--------------- tests/web_api/test_client.py | 6 ------ 2 files changed, 1 insertion(+), 21 deletions(-) diff --git a/roborock/web_api/__init__.py b/roborock/web_api/__init__.py index 1a260899..a3af05da 100644 --- a/roborock/web_api/__init__.py +++ b/roborock/web_api/__init__.py @@ -7,21 +7,7 @@ ``roborock.web_api`` module is re-exported here for backwards compatibility. """ -# Modules and the logger the former single-file module exposed. Re-exported so that -# existing patch targets such as ``roborock.web_api.time.time`` keep resolving. -import base64 -import hashlib -import hmac -import json -import math -import secrets -import string -import time - -import aiohttp - -from ._base import _LOGGER, BASE_URLS, IotLoginInfo, RoborockApiBase -from ._hawk import _compact_json, _get_hawk_authentication, _process_extra_hawk_values +from ._base import BASE_URLS, IotLoginInfo, RoborockApiBase from ._request import PreparedRequest from .auth import DEFAULT_AGREEMENT_MAJOR_VERSION, DEFAULT_AGREEMENT_MINOR_VERSION, AuthApi from .client import RoborockApiClient, UserWebApiClient diff --git a/tests/web_api/test_client.py b/tests/web_api/test_client.py index 40546b44..ac5f595b 100644 --- a/tests/web_api/test_client.py +++ b/tests/web_api/test_client.py @@ -22,12 +22,6 @@ def test_legacy_module_exports() -> None: "PreparedRequest", "RoborockApiClient", "UserWebApiClient", - "_compact_json", - "_get_hawk_authentication", - "_LOGGER", - "aiohttp", - "secrets", - "time", ): assert hasattr(web_api, name), name