images/ha_xiaomi_home
2
0
Fork 0
mirror of https://github.com/XiaoMi/ha_xiaomi_home.git synced 2025-04-04 16:55:30 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'XiaoMi:main' into main

Browse Source
This commit is contained in:
ted 2025-01-16 09:18:40 +08:00 committed by GitHub
commit fe3756db9b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 959 additions and 337 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
custom_components/xiaomi_home/__init__.py
View File

@ -308,3 +308,43 @@ async def async_remove_entry(
await miot_cert.remove_user_cert_async()
await miot_cert.remove_user_key_async()
return True
async def async_remove_config_entry_device(
hass: HomeAssistant,
config_entry: ConfigEntry,
device_entry: device_registry.DeviceEntry
) -> bool:
"""Remove the device."""
miot_client: MIoTClient = await get_miot_instance_async(
hass=hass, entry_id=config_entry.entry_id)
if len(device_entry.identifiers) != 1:
_LOGGER.error(
'remove device failed, invalid identifiers, %s, %s',
device_entry.id, device_entry.identifiers)
return False
identifiers = list(device_entry.identifiers)[0]
if identifiers[0] != DOMAIN:
_LOGGER.error(
'remove device failed, invalid domain, %s, %s',
device_entry.id, device_entry.identifiers)
return False
device_info = identifiers[1].split('_')
if len(device_info) != 2:
_LOGGER.error(
'remove device failed, invalid device info, %s, %s',
device_entry.id, device_entry.identifiers)
return False
did = device_info[1]
if did not in miot_client.device_list:
_LOGGER.error(
'remove device failed, device not found, %s, %s',
device_entry.id, device_entry.identifiers)
return False
# Remove device
await miot_client.remove_device_async(did)
device_registry.async_get(hass).async_remove_device(device_entry.id)
_LOGGER.info(
'remove device, %s, %s, %s', device_info[0], did, device_entry.id)
return True

55
custom_components/xiaomi_home/config_flow.py
View File

@ -91,7 +91,8 @@ from .miot.miot_cloud import MIoTHttpClient, MIoTOauthClient
from .miot.miot_storage import MIoTStorage, MIoTCert
from .miot.miot_mdns import MipsService
from .miot.web_pages import oauth_redirect_page
from .miot.miot_error import MIoTConfigError, MIoTError, MIoTOauthError
from .miot.miot_error import (
MIoTConfigError, MIoTError, MIoTErrorCode, MIoTOauthError)
from .miot.miot_i18n import MIoTI18n
from .miot.miot_network import MIoTNetwork
from .miot.miot_client import MIoTClient, get_miot_instance_async
@ -430,6 +431,8 @@ class XiaomiMihomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
redirect_url=self._oauth_redirect_url_full)
self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = (
miot_oauth.state)
self.hass.data[DOMAIN][self._virtual_did]['i18n'] = (
self._miot_i18n)
_LOGGER.info(
'async_step_oauth, oauth_url: %s', self._cc_oauth_auth_url)
webhook_async_unregister(
@ -1152,6 +1155,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
redirect_url=self._oauth_redirect_url_full)
self.hass.data[DOMAIN][self._virtual_did]['oauth_state'] = (
self._miot_oauth.state)
self.hass.data[DOMAIN][self._virtual_did]['i18n'] = (
self._miot_i18n)
_LOGGER.info(
'async_step_oauth, oauth_url: %s', self._cc_oauth_auth_url)
webhook_async_unregister(
@ -1967,29 +1972,61 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
async def _handle_oauth_webhook(hass, webhook_id, request):
"""Webhook to handle oauth2 callback."""
# pylint: disable=inconsistent-quotes
i18n: MIoTI18n = hass.data[DOMAIN][webhook_id].get('i18n', None)
try:
data = dict(request.query)
if data.get('code', None) is None or data.get('state', None) is None:
raise MIoTConfigError('invalid oauth code')
raise MIoTConfigError(
'invalid oauth code or state',
MIoTErrorCode.CODE_CONFIG_INVALID_INPUT)
if data['state'] != hass.data[DOMAIN][webhook_id]['oauth_state']:
raise MIoTConfigError(
f'invalid oauth state, '
f'{hass.data[DOMAIN][webhook_id]["oauth_state"]}, '
f'{data["state"]}')
f'inconsistent state, '
f'{hass.data[DOMAIN][webhook_id]["oauth_state"]}!='
f'{data["state"]}', MIoTErrorCode.CODE_CONFIG_INVALID_STATE)
fut_oauth_code: asyncio.Future = hass.data[DOMAIN][webhook_id].pop(
'fut_oauth_code', None)
fut_oauth_code.set_result(data['code'])
_LOGGER.info('webhook code: %s', data['code'])
success_trans: dict = {}
if i18n:
success_trans = i18n.translate(
'oauth2.success') or {} # type: ignore
# Delete
del hass.data[DOMAIN][webhook_id]['oauth_state']
del hass.data[DOMAIN][webhook_id]['i18n']
return web.Response(
body=oauth_redirect_page(
hass.config.language, 'success'), content_type='text/html')
body=await oauth_redirect_page(
title=success_trans.get('title', 'Success'),
content=success_trans.get(
'content', (
'Please close this page and return to the account '
'authentication page to click NEXT')),
button=success_trans.get('button', 'Close Page'),
success=True,
), content_type='text/html')
except MIoTConfigError:
except Exception as err: # pylint: disable=broad-exception-caught
fail_trans: dict = {}
err_msg: str = str(err)
if i18n:
if isinstance(err, MIoTConfigError):
err_msg = i18n.translate(
f'oauth2.error_msg.{err.code.value}'
) or err.message # type: ignore
fail_trans = i18n.translate('oauth2.fail') or {} # type: ignore
return web.Response(
body=oauth_redirect_page(hass.config.language, 'fail'),
body=await oauth_redirect_page(
title=fail_trans.get('title', 'Authentication Failed'),
content=str(fail_trans.get('content', (
'{error_msg}, Please close this page and return to the '
'account authentication page to click the authentication '
'link again.'))).replace('{error_msg}', err_msg),
button=fail_trans.get('button', 'Close Page'),
success=False),
content_type='text/html')

16
custom_components/xiaomi_home/miot/i18n/de.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Schnittstelle nicht verfügbar"
}
},
"oauth2": {
"success": {
"title": "Authentifizierung erfolgreich",
"content": "Bitte schließen Sie diese Seite und kehren Sie zur Kontoauthentifizierungsseite zurück, um auf „Weiter“ zu klicken.",
"button": "Schließen"
},
"fail": {
"title": "Authentifizierung fehlgeschlagen",
"content": "{error_msg}, bitte schließen Sie diese Seite und kehren Sie zur Kontoauthentifizierungsseite zurück, um den Authentifizierungslink erneut zu klicken.",
"button": "Schließen"
},
"error_msg": {
"-10100": "Ungültige Antwortparameter ('code' oder 'state' Feld ist leer)",
"-10101": "Übergebenes 'state' Feld stimmt nicht überein"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Ungültige Authentifizierungsinformationen, Cloud-Verbindung nicht verfügbar, bitte betreten Sie die Xiaomi Home-Integrationsseite und klicken Sie auf 'Optionen', um die Authentifizierung erneut durchzuführen",

16
custom_components/xiaomi_home/miot/i18n/en.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interface unavailable"
}
},
"oauth2": {
"success": {
"title": "Authentication Successful",
"content": "Please close this page and return to the account authentication page to click 'Next'.",
"button": "Close"
},
"fail": {
"title": "Authentication Failed",
"content": "{error_msg}, please close this page and return to the account authentication page to click the authentication link again.",
"button": "Close"
},
"error_msg": {
"-10100": "Invalid response parameters ('code' or 'state' field is empty)",
"-10101": "Passed-in 'state' field mismatch"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Authentication information is invalid, cloud link will be unavailable, please enter the Xiaomi Home integration page, click 'Options' to re-authenticate",

16
custom_components/xiaomi_home/miot/i18n/es.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interfaz no disponible"
}
},
"oauth2": {
"success": {
"title": "Autenticación exitosa",
"content": "Por favor, cierre esta página y regrese a la página de autenticación de la cuenta para hacer clic en 'Siguiente'.",
"button": "Cerrar"
},
"fail": {
"title": "Autenticación fallida",
"content": "{error_msg}, por favor, cierre esta página y regrese a la página de autenticación de la cuenta para hacer clic en el enlace de autenticación nuevamente.",
"button": "Cerrar"
},
"error_msg": {
"-10100": "Parámetros de respuesta inválidos ('code' o 'state' está vacío)",
"-10101": "El campo 'state' proporcionado no coincide"
}
},
"miot": {
"client": {
"invalid_oauth_info": "La información de autenticación es inválida, la conexión en la nube no estará disponible, por favor, vaya a la página de integración de Xiaomi Home, haga clic en 'Opciones' para volver a autenticar",

16
custom_components/xiaomi_home/miot/i18n/fr.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interface non disponible"
}
},
"oauth2": {
"success": {
"title": "Authentification réussie",
"content": "Veuillez fermer cette page et revenir à la page d'authentification du compte pour cliquer sur 'Suivant'.",
"button": "Fermer"
},
"fail": {
"title": "Échec de l'authentification",
"content": "{error_msg}, veuillez fermer cette page et revenir à la page d'authentification du compte pour cliquer à nouveau sur le lien d'authentification.",
"button": "Fermer"
},
"error_msg": {
"-10100": "Paramètres de réponse invalides ('code' ou 'state' est vide)",
"-10101": "Le champ 'state' transmis ne correspond pas"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Informations d'authentification non valides, le lien cloud ne sera pas disponible, veuillez accéder à la page d'intégration Xiaomi Home, cliquez sur \"Options\" pour vous réauthentifier",

16
custom_components/xiaomi_home/miot/i18n/ja.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "インターフェースが利用できません"
}
},
"oauth2": {
"success": {
"title": "認証成功",
"content": "このページを閉じて、アカウント認証ページに戻り、「次へ」をクリックしてください。",
"button": "閉じる"
},
"fail": {
"title": "認証失敗",
"content": "{error_msg}、このページを閉じて、アカウント認証ページに戻り、再度認証リンクをクリックしてください。",
"button": "閉じる"
},
"error_msg": {
"-10100": "無効な応答パラメータ（'code'または'state'フィールドが空です）",
"-10101": "渡された'state'フィールドが一致しません"
}
},
"miot": {
"client": {
"invalid_oauth_info": "認証情報が無効です。クラウドリンクは利用できません。Xiaomi Home統合ページに入り、[オプション]をクリックして再認証してください",

16
custom_components/xiaomi_home/miot/i18n/nl.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interface niet beschikbaar"
}
},
"oauth2": {
"success": {
"title": "Authenticatie geslaagd",
"content": "Sluit deze pagina en ga terug naar de accountauthenticatiepagina om op 'Volgende' te klikken.",
"button": "Sluiten"
},
"fail": {
"title": "Authenticatie mislukt",
"content": "{error_msg}, sluit deze pagina en ga terug naar de accountauthenticatiepagina om opnieuw op de authenticatielink te klikken.",
"button": "Sluiten"
},
"error_msg": {
"-10100": "Ongeldige antwoordparameters ('code' of 'state' veld is leeg)",
"-10101": "Doorgegeven 'state' veld komt niet overeen"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Authenticatie-informatie is ongeldig, cloudverbinding zal niet beschikbaar zijn. Ga naar de Xiaomi Home-integratiepagina en klik op 'Opties' om opnieuw te verifiëren.",

16
custom_components/xiaomi_home/miot/i18n/pt-BR.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interface indisponível"
}
},
"oauth2": {
"success": {
"title": "Autenticação bem-sucedida",
"content": "Por favor, feche esta página e volte para a página de autenticação da conta para clicar em 'Próximo'.",
"button": "Fechar"
},
"fail": {
"title": "Falha na autenticação",
"content": "{error_msg}, por favor, feche esta página e volte para a página de autenticação da conta para clicar no link de autenticação novamente.",
"button": "Fechar"
},
"error_msg": {
"-10100": "Parâmetros de resposta inválidos ('code' ou 'state' está vazio)",
"-10101": "O campo 'state' fornecido não corresponde"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Informações de autenticação inválidas, a conexão com a nuvem estará indisponível. Vá para a página de integração do Xiaomi Home e clique em 'Opções' para reautenticar.",

16
custom_components/xiaomi_home/miot/i18n/pt.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Interface indisponível"
}
},
"oauth2": {
"success": {
"title": "Autenticação bem-sucedida",
"content": "Por favor, feche esta página e volte para a página de autenticação da conta para clicar em 'Seguinte'.",
"button": "Fechar"
},
"fail": {
"title": "Falha na autenticação",
"content": "{error_msg}, por favor, feche esta página e volte para a página de autenticação da conta para clicar no link de autenticação novamente.",
"button": "Fechar"
},
"error_msg": {
"-10100": "Parâmetros de resposta inválidos ('code' ou 'state' está vazio)",
"-10101": "O campo 'state' fornecido não corresponde"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Informações de autenticação inválidas, a conexão na nuvem ficará indisponível. Por favor, acesse a página de integração do Xiaomi Home e clique em 'Opções' para autenticar novamente.",

16
custom_components/xiaomi_home/miot/i18n/ru.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "Интерфейс недоступен"
}
},
"oauth2": {
"success": {
"title": "Аутентификация успешна",
"content": "Пожалуйста, закройте эту страницу и вернитесь на страницу аутентификации учетной записи, чтобы нажать 'Далее'.",
"button": "Закрыть"
},
"fail": {
"title": "Аутентификация не удалась",
"content": "{error_msg}, пожалуйста, закройте эту страницу и вернитесь на страницу аутентификации учетной записи, чтобы снова нажать на ссылку аутентификации.",
"button": "Закрыть"
},
"error_msg": {
"-10100": "Недействительные параметры ответа ('code' или 'state' поле пусто)",
"-10101": "Переданное поле 'state' не совпадает"
}
},
"miot": {
"client": {
"invalid_oauth_info": "Информация об аутентификации недействительна, облако будет недоступно, пожалуйста, войдите на страницу интеграции Xiaomi Home, нажмите 'Опции' для повторной аутентификации",

16
custom_components/xiaomi_home/miot/i18n/zh-Hans.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "接口不可用"
}
},
"oauth2": {
"success": {
"title": "认证成功",
"content": "请关闭此页面，返回账号认证页面点击“下一步”",
"button": "关闭"
},
"fail": {
"title": "认证失败",
"content": "{error_msg}，请关闭此页面，返回账号认证页面重新点击认链接进行认证。",
"button": "关闭"
},
"error_msg": {
"-10100": "无效的响应参数（“code”或者“state”字段为空）",
"-10101": "传入“state”字段不一致"
}
},
"miot": {
"client": {
"invalid_oauth_info": "认证信息失效，云端链路将不可用，请进入 Xiaomi Home 集成页面，点击“选项”重新认证",

16
custom_components/xiaomi_home/miot/i18n/zh-Hant.json
View File

@ -64,6 +64,22 @@
"net_unavailable": "接口不可用"
}
},
"oauth2": {
"success": {
"title": "認證成功",
"content": "請關閉此頁面，返回帳號認證頁面點擊“下一步”",
"button": "關閉"
},
"fail": {
"title": "認證失敗",
"content": "{error_msg}，請關閉此頁面，返回帳號認證頁面重新點擊認鏈接進行認證。",
"button": "關閉"
},
"error_msg": {
"-10100": "無效的響應參數（“code”或者“state”字段為空）",
"-10101": "傳入的“state”字段不一致"
}
},
"miot": {
"client": {
"invalid_oauth_info": "認證信息失效，雲端鏈路將不可用，請進入 Xiaomi Home 集成頁面，點擊“選項”重新認證",

24
custom_components/xiaomi_home/miot/miot_client.py
View File

@ -848,6 +848,30 @@ class MIoTClient:
_LOGGER.debug('client unsub device state, %s', did)
return True
async def remove_device_async(self, did: str) -> None:
if did not in self._device_list_cache:
return
sub_from = self._sub_source_list.pop(did, None)
# Unsub
if sub_from:
if sub_from == 'cloud':
self._mips_cloud.unsub_prop(did=did)
self._mips_cloud.unsub_event(did=did)
elif sub_from == 'lan':
self._miot_lan.unsub_prop(did=did)
self._miot_lan.unsub_event(did=did)
elif sub_from in self._mips_local:
mips = self._mips_local[sub_from]
mips.unsub_prop(did=did)
mips.unsub_event(did=did)
# Storage
await self._storage.save_async(
domain='miot_devices',
name=f'{self._uid}_{self._cloud_server}',
data=self._device_list_cache)
# Update notify
self.__request_show_devices_changed_notify()
def __get_exec_error_with_rc(self, rc: int) -> str:
err_msg: str = self._i18n.translate(key=f'error.common.{rc}')
if not err_msg:

2
custom_components/xiaomi_home/miot/miot_error.py
View File

@ -72,6 +72,8 @@ class MIoTErrorCode(Enum):
# MIoT ev error code, -10080
# Mips service error code, -10090
# Config flow error code, -10100
CODE_CONFIG_INVALID_INPUT = -10100
CODE_CONFIG_INVALID_STATE = -10101
# Options flow error code , -10110
# MIoT lan error code, -10120
CODE_LAN_UNAVAILABLE = -10120

96
custom_components/xiaomi_home/miot/miot_mips.py
View File

@ -229,10 +229,9 @@ class _MipsClient(ABC):
_ca_file: Optional[str]
_cert_file: Optional[str]
_key_file: Optional[str]
_tls_done: bool
_mqtt_logger: Optional[logging.Logger]
_mqtt: Client
_mqtt: Optional[Client]
_mqtt_fd: int
_mqtt_timer: Optional[asyncio.TimerHandle]
_mqtt_state: bool
@ -272,16 +271,12 @@ class _MipsClient(ABC):
self._ca_file = ca_file
self._cert_file = cert_file
self._key_file = key_file
self._tls_done = False
self._mqtt_logger = None
self._mqtt_fd = -1
self._mqtt_timer = None
self._mqtt_state = False
# mqtt init for API_VERSION2,
# callback_api_version=CallbackAPIVersion.VERSION2,
self._mqtt = Client(client_id=self._client_id, protocol=MQTTv5)
self._mqtt.enable_logger(logger=self._mqtt_logger)
self._mqtt = None
# Mips init
self._event_connect = asyncio.Event()
@ -316,7 +311,9 @@ class _MipsClient(ABC):
Returns:
bool: True: connected, False: disconnected
"""
return self._mqtt and self._mqtt.is_connected()
if self._mqtt:
return self._mqtt.is_connected()
return False
def connect(self, thread_name: Optional[str] = None) -> None:
"""mips connect."""
@ -359,7 +356,22 @@ class _MipsClient(ABC):
self._ca_file = None
self._cert_file = None
self._key_file = None
self._tls_done = False
self._mqtt_logger = None
with self._mips_state_sub_map_lock:
self._mips_state_sub_map.clear()
self._mips_sub_pending_map.clear()
self._mips_sub_pending_timer = None
@final
async def deinit_async(self) -> None:
await self.disconnect_async()
self._logger = None
self._username = None
self._password = None
self._ca_file = None
self._cert_file = None
self._key_file = None
self._mqtt_logger = None
with self._mips_state_sub_map_lock:
self._mips_state_sub_map.clear()
@ -368,8 +380,9 @@ class _MipsClient(ABC):
def update_mqtt_password(self, password: str) -> None:
self._password = password
self._mqtt.username_pw_set(
username=self._username, password=self._password)
if self._mqtt:
self._mqtt.username_pw_set(
username=self._username, password=self._password)
def log_debug(self, msg, *args, **kwargs) -> None:
if self._logger:
@ -389,10 +402,12 @@ class _MipsClient(ABC):
def enable_mqtt_logger(
self, logger: Optional[logging.Logger] = None
) -> None:
if logger:
self._mqtt.enable_logger(logger=logger)
else:
self._mqtt.disable_logger()
self._mqtt_logger = logger
if self._mqtt:
if logger:
self._mqtt.enable_logger(logger=logger)
else:
self._mqtt.disable_logger()
@final
def sub_mips_state(
@ -587,25 +602,27 @@ class _MipsClient(ABC):
def __mips_loop_thread(self) -> None:
self.log_info('mips_loop_thread start')
# mqtt init for API_VERSION2,
# callback_api_version=CallbackAPIVersion.VERSION2,
self._mqtt = Client(client_id=self._client_id, protocol=MQTTv5)
self._mqtt.enable_logger(logger=self._mqtt_logger)
# Set mqtt config
if self._username:
self._mqtt.username_pw_set(
username=self._username, password=self._password)
if not self._tls_done:
if (
self._ca_file
and self._cert_file
and self._key_file
):
self._mqtt.tls_set(
tls_version=ssl.PROTOCOL_TLS_CLIENT,
ca_certs=self._ca_file,
certfile=self._cert_file,
keyfile=self._key_file)
else:
self._mqtt.tls_set(tls_version=ssl.PROTOCOL_TLS_CLIENT)
self._mqtt.tls_insecure_set(True)
self._tls_done = True
if (
self._ca_file
and self._cert_file
and self._key_file
):
self._mqtt.tls_set(
tls_version=ssl.PROTOCOL_TLS_CLIENT,
ca_certs=self._ca_file,
certfile=self._cert_file,
keyfile=self._key_file)
else:
self._mqtt.tls_set(tls_version=ssl.PROTOCOL_TLS_CLIENT)
self._mqtt.tls_insecure_set(True)
self._mqtt.on_connect = self.__on_connect
self._mqtt.on_connect_fail = self.__on_connect_failed
self._mqtt.on_disconnect = self.__on_disconnect
@ -617,6 +634,9 @@ class _MipsClient(ABC):
self.log_info('mips_loop_thread exit!')
def __on_connect(self, client, user_data, flags, rc, props) -> None:
if not self._mqtt:
_LOGGER.error('__on_connect, but mqtt is None')
return
if not self._mqtt.is_connected():
return
self.log_info(f'mips connect, {flags}, {rc}, {props}')
@ -685,6 +705,10 @@ class _MipsClient(ABC):
self._on_mips_message(topic=msg.topic, payload=msg.payload)
def __mips_sub_internal_pending_handler(self, ctx: Any) -> None:
if not self._mqtt or not self._mqtt.is_connected():
_LOGGER.error(
'mips sub internal pending, but mqtt is None or disconnected')
return
subbed_count = 1
for topic in list(self._mips_sub_pending_map.keys()):
if subbed_count > self.MIPS_SUB_PATCH:
@ -712,6 +736,9 @@ class _MipsClient(ABC):
self._mips_sub_pending_timer = None
def __mips_connect(self) -> None:
if not self._mqtt:
_LOGGER.error('__mips_connect, but mqtt is None')
return
result = MQTT_ERR_UNKNOWN
if self._mips_reconnect_timer:
self._mips_reconnect_timer.cancel()
@ -782,7 +809,14 @@ class _MipsClient(ABC):
self._internal_loop.remove_reader(self._mqtt_fd)
self._internal_loop.remove_writer(self._mqtt_fd)
self._mqtt_fd = -1
self._mqtt.disconnect()
# Clear retry sub
if self._mips_sub_pending_timer:
self._mips_sub_pending_timer.cancel()
self._mips_sub_pending_timer = None
self._mips_sub_pending_map = {}
if self._mqtt:
self._mqtt.disconnect()
self._mqtt = None
self._internal_loop.stop()
def __get_next_reconnect_time(self) -> float:

136
custom_components/xiaomi_home/miot/resource/oauth_redirect_page.html Normal file
View File

@ -0,0 +1,136 @@
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link rel="icon" href="https://cdn.web-global.fds.api.mi-img.com/mcfe--mi-account/static/favicon_new.ico">
<link as="style"
href="https://font.sec.miui.com/font/css?family=MiSans:300,400,500,600,700:Chinese_Simplify,Chinese_Traditional,Latin&amp;display=swap"
rel="preload">
<title>TITLE_PLACEHOLDER</title>
<style>
body {
background: white;
color: black;
display: flex;
justify-content: center;
align-items: center;
font-family: MiSans, -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Helvetica Neue, Arial, Noto Sans, sans-serif, Apple Color Emoji, Segoe UI Emoji, Segoe UI Symbol, Noto Color Emoji;
}
@media (prefers-color-scheme: dark) {
body {
background: black;
color: white;
}
}
.frame {
background: rgb(255 255 255 / 5%);
width: 360px;
padding: 40 45;
border-radius: 4px;
box-shadow: 0 20px 50px 0 hsla(0, 0%, 64%, .1);
text-align: center;
}
.logo-frame {
text-align: center;
}
.title-frame {
margin: 20px 0 20px 0;
font-size: 26px;
font-weight: 500;
line-height: 40px;
opacity: 0.8;
}
.content-frame {
font-size: 17px;
font-weight: 500;
line-height: 20px;
opacity: 0.8;
}
button {
margin-top: 20px;
background-color: #ff5c00;
border: none;
border-radius: 4px;
padding: 0 20px;
text-align: center;
width: 100%;
display: inline-block;
font-size: 18px;
font-weight: 400;
height: 60px;
line-height: 60px;
overflow: hidden;
position: relative;
text-overflow: ellipsis;
transition: all .3s cubic-bezier(.645, .045, .355, 1);
vertical-align: top;
white-space: nowrap;
cursor: pointer;
color: #fff;
}
</style>
</head>
<body>
<div class="frame">
<!-- XIAOMI LOGO-->
<div class="logo-frame">
<svg width="50" height="50" viewBox="0 0 193 193" xmlns="http://www.w3.org/2000/svg"
xmlns:xlink="http://www.w3.org/1999/xlink">
<title>编组</title>
<desc>Created with Sketch.</desc>
<defs>
<polygon id="path-1"
points="1.78097075e-14 0.000125324675 192.540685 0.000125324675 192.540685 192.540058 1.78097075e-14 192.540058">
</polygon>
</defs>
<g id="\u9875\u9762-1" stroke="none" stroke-width="1" fill="none" fill-rule="evenodd">
<g id="\u7F16\u7EC4">
<mask id="mask-2" fill="white">
<use xlink:href="#path-1"></use>
</mask>
<g id="Clip-2"></g>
<path
d="M172.473071,20.1164903 C154.306633,2.02148701 128.188344,-1.78097075e-14 96.2706558,-1.78097075e-14 C64.312237,-1.78097075e-14 38.155724,2.0452987 19.9974318,20.1872987 C1.84352597,38.3261656 1.78097075e-14,64.4406948 1.78097075e-14,96.3640227 C1.78097075e-14,128.286724 1.84352597,154.415039 20.0049513,172.556412 C38.1638701,190.704052 64.3141169,192.540058 96.2706558,192.540058 C128.225942,192.540058 154.376815,190.704052 172.53636,172.556412 C190.694653,154.409399 192.540685,128.286724 192.540685,96.3640227 C192.540685,64.3999643 190.672721,38.2553571 172.473071,20.1164903"
id="Fill-1" fill="#FF6900" mask="url(#mask-2)"></path>
<path
d="M89.1841721,131.948836 C89.1841721,132.594885 88.640263,133.130648 87.9779221,133.130648 L71.5585097,133.130648 C70.8848896,133.130648 70.338474,132.594885 70.338474,131.948836 L70.338474,89.0100961 C70.338474,88.3584078 70.8848896,87.8251513 71.5585097,87.8251513 L87.9779221,87.8251513 C88.640263,87.8251513 89.1841721,88.3584078 89.1841721,89.0100961 L89.1841721,131.948836 Z"
id="Fill-3" fill="#FFFFFF" mask="url(#mask-2)"></path>
<path
d="M121.332896,131.948836 C121.332896,132.594885 120.786481,133.130648 120.121633,133.130648 L104.492393,133.130648 C103.821906,133.130648 103.275491,132.594885 103.275491,131.948836 L103.275491,131.788421 L103.275491,94.9022357 C103.259198,88.4342292 102.889491,81.7863818 99.5502146,78.445226 C96.6790263,75.5652649 91.3251562,74.9054305 85.7557276,74.7669468 L57.4242049,74.7669468 C56.7555977,74.7669468 56.2154484,75.3045896 56.2154484,75.9512649 L56.2154484,128.074424 L56.2154484,131.948836 C56.2154484,132.594885 55.6640198,133.130648 54.9954127,133.130648 L39.3555198,133.130648 C38.6875393,133.130648 38.1498964,132.594885 38.1498964,131.948836 L38.1498964,60.5996188 C38.1498964,59.9447974 38.6875393,59.4121675 39.3555198,59.4121675 L84.4786692,59.4121675 C96.2717211,59.4121675 108.599909,59.9498104 114.680036,66.0380831 C120.786481,72.1533006 121.332896,84.4595571 121.332896,96.2657682 L121.332896,131.948836 Z"
id="Fill-5" fill="#FFFFFF" mask="url(#mask-2)"></path>
<path
d="M153.53056,131.948836 C153.53056,132.594885 152.978505,133.130648 152.316164,133.130648 L136.678778,133.130648 C136.010797,133.130648 135.467515,132.594885 135.467515,131.948836 L135.467515,60.5996188 C135.467515,59.9447974 136.010797,59.4121675 136.678778,59.4121675 L152.316164,59.4121675 C152.978505,59.4121675 153.53056,59.9447974 153.53056,60.5996188 L153.53056,131.948836 Z"
id="Fill-7" fill="#FFFFFF" mask="url(#mask-2)"></path>
</g>
</g>
</svg>
</div>
<!-- TITLE -->
<div class="title-frame">
<a id="titleArea">TITLE_PLACEHOLDER</a>
</div>
<!-- CONTENT -->
<div class="content-frame">
<a id="contentArea">CONTENT_PLACEHOLDER</a>
</div>
<!-- BUTTON -->
<button onClick="window.close();" id="buttonArea">BUTTON_PLACEHOLDER</button>
</div>
<script>
if (STATUS_PLACEHOLDER) {
window.opener = null;
window.open('', '_self');
window.close();
}
</script>
</body>
</html>

8
custom_components/xiaomi_home/miot/specs/specv2entity.py
View File

@ -431,6 +431,14 @@ SPEC_PROP_TRANS_MAP: dict[str, dict | str] = {
'unit_of_measurement': UnitOfEnergy.KILO_WATT_HOUR
}
},
'power': {
'device_class': SensorDeviceClass.POWER,
'entity': 'sensor',
'optional': {
'state_class': SensorStateClass.MEASUREMENT,
'unit_of_measurement': UnitOfPower.WATT
}
},
'total-battery': {
'device_class': SensorDeviceClass.ENERGY,
'entity': 'sensor',

258
custom_components/xiaomi_home/miot/web_pages.py
View File

@ -46,237 +46,31 @@ off Xiaomi or its affiliates' products.
MIoT redirect web pages.
"""
# pylint: disable=line-too-long
import os
import asyncio
def oauth_redirect_page(lang: str, status: str) -> str:
_template = ''
def _load_page_template():
path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
'resource/oauth_redirect_page.html')
with open(path, 'r', encoding='utf-8') as f:
global _template
_template = f.read()
async def oauth_redirect_page(
title: str, content: str, button: str, success: bool
) -> str:
"""Return oauth redirect page."""
return '''
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link rel="icon" href="https://cdn.web-global.fds.api.mi-img.com/mcfe--mi-account/static/favicon_new.ico">
<link as="style"
href="https://font.sec.miui.com/font/css?family=MiSans:300,400,500,600,700:Chinese_Simplify,Chinese_Traditional,Latin&amp;display=swap"
rel="preload">
<title></title>
<style>
body {
background: white;
color: black;
display: flex;
justify-content: center;
align-items: center;
font-family: MiSans, -apple-system, BlinkMacSystemFont, Segoe UI, Roboto, Helvetica Neue, Arial, Noto Sans, sans-serif, Apple Color Emoji, Segoe UI Emoji, Segoe UI Symbol, Noto Color Emoji;
}
@media (prefers-color-scheme: dark) {
body {
background: black;
color: white;
}
}
.frame {
background: rgb(255 255 255 / 5%);
width: 360px;
padding: 40 45;
border-radius: 4px;
box-shadow: 0 20px 50px 0 hsla(0, 0%, 64%, .1);
text-align: center;
}
.logo-frame {
text-align: center;
}
.title-frame {
margin: 20px 0 20px 0;
font-size: 26px;
font-weight: 500;
line-height: 40px;
opacity: 0.8;
}
.content-frame {
font-size: 17px;
font-weight: 500;
line-height: 20px;
opacity: 0.8;
}
button {
margin-top: 20px;
background-color: #ff5c00;
border: none;
border-radius: 4px;
padding: 0 20px;
text-align: center;
width: 100%;
display: inline-block;
font-size: 18px;
font-weight: 400;
height: 60px;
line-height: 60px;
overflow: hidden;
position: relative;
text-overflow: ellipsis;
transition: all .3s cubic-bezier(.645, .045, .355, 1);
vertical-align: top;
white-space: nowrap;
cursor: pointer;
color: #fff;
}
</style>
</head>
<body>
<div class="frame">
<!-- XIAOMI LOGO-->
<div class="logo-frame">
<svg width="50" height="50" viewBox="0 0 193 193" xmlns="http://www.w3.org/2000/svg"
xmlns:xlink="http://www.w3.org/1999/xlink"><title>编组</title>
<desc>Created with Sketch.</desc>
<defs>
<polygon id="path-1"
points="1.78097075e-14 0.000125324675 192.540685 0.000125324675 192.540685 192.540058 1.78097075e-14 192.540058"></polygon>
</defs>
<g id="\u9875\u9762-1" stroke="none" stroke-width="1" fill="none" fill-rule="evenodd">
<g id="\u7F16\u7EC4">
<mask id="mask-2" fill="white">
<use xlink:href="#path-1"></use>
</mask>
<g id="Clip-2"></g>
<path d="M172.473071,20.1164903 C154.306633,2.02148701 128.188344,-1.78097075e-14 96.2706558,-1.78097075e-14 C64.312237,-1.78097075e-14 38.155724,2.0452987 19.9974318,20.1872987 C1.84352597,38.3261656 1.78097075e-14,64.4406948 1.78097075e-14,96.3640227 C1.78097075e-14,128.286724 1.84352597,154.415039 20.0049513,172.556412 C38.1638701,190.704052 64.3141169,192.540058 96.2706558,192.540058 C128.225942,192.540058 154.376815,190.704052 172.53636,172.556412 C190.694653,154.409399 192.540685,128.286724 192.540685,96.3640227 C192.540685,64.3999643 190.672721,38.2553571 172.473071,20.1164903"
id="Fill-1" fill="#FF6900" mask="url(#mask-2)"></path>
<path d="M89.1841721,131.948836 C89.1841721,132.594885 88.640263,133.130648 87.9779221,133.130648 L71.5585097,133.130648 C70.8848896,133.130648 70.338474,132.594885 70.338474,131.948836 L70.338474,89.0100961 C70.338474,88.3584078 70.8848896,87.8251513 71.5585097,87.8251513 L87.9779221,87.8251513 C88.640263,87.8251513 89.1841721,88.3584078 89.1841721,89.0100961 L89.1841721,131.948836 Z"
id="Fill-3" fill="#FFFFFF" mask="url(#mask-2)"></path>
<path d="M121.332896,131.948836 C121.332896,132.594885 120.786481,133.130648 120.121633,133.130648 L104.492393,133.130648 C103.821906,133.130648 103.275491,132.594885 103.275491,131.948836 L103.275491,131.788421 L103.275491,94.9022357 C103.259198,88.4342292 102.889491,81.7863818 99.5502146,78.445226 C96.6790263,75.5652649 91.3251562,74.9054305 85.7557276,74.7669468 L57.4242049,74.7669468 C56.7555977,74.7669468 56.2154484,75.3045896 56.2154484,75.9512649 L56.2154484,128.074424 L56.2154484,131.948836 C56.2154484,132.594885 55.6640198,133.130648 54.9954127,133.130648 L39.3555198,133.130648 C38.6875393,133.130648 38.1498964,132.594885 38.1498964,131.948836 L38.1498964,60.5996188 C38.1498964,59.9447974 38.6875393,59.4121675 39.3555198,59.4121675 L84.4786692,59.4121675 C96.2717211,59.4121675 108.599909,59.9498104 114.680036,66.0380831 C120.786481,72.1533006 121.332896,84.4595571 121.332896,96.2657682 L121.332896,131.948836 Z"
id="Fill-5" fill="#FFFFFF" mask="url(#mask-2)"></path>
<path d="M153.53056,131.948836 C153.53056,132.594885 152.978505,133.130648 152.316164,133.130648 L136.678778,133.130648 C136.010797,133.130648 135.467515,132.594885 135.467515,131.948836 L135.467515,60.5996188 C135.467515,59.9447974 136.010797,59.4121675 136.678778,59.4121675 L152.316164,59.4121675 C152.978505,59.4121675 153.53056,59.9447974 153.53056,60.5996188 L153.53056,131.948836 Z"
id="Fill-7" fill="#FFFFFF" mask="url(#mask-2)"></path>
</g>
</g>
</svg>
</div>
<!-- TITLE -->
<div class="title-frame">
<a id="titleArea"></a>
</div>
<!-- CONTENT -->
<div class="content-frame">
<a id="contentArea"></a>
</div>
<!-- BUTTON -->
<button onClick="window.close();" id="buttonArea"></button>
</div>
<script>
// get language (user language -> system language)
const locale = (localStorage.getItem('selectedLanguage')?? "''' + lang + '''").replaceAll('"','');
const language = locale.includes("-") ? locale.substring(0, locale.indexOf("-")).trim() : locale;
const status = "''' + status + '''";
console.log(locale);
// translation
let translation = {
zh: {
success: {
title: "认证完成",
content: "请关闭此页面，返回账号认证页面点击“下一步”",
button: "关闭页面"
},
fail: {
title: "认证失败",
content: "请关闭此页面，返回账号认证页面重新点击认链接进行认证。",
button: "关闭页面"
}
},
'zh-Hant': {
success: {
title: "認證完成",
content: "請關閉此頁面，返回帳號認證頁面點擊「下一步」。",
button: "關閉頁面"
},
fail: {
title: "認證失敗",
content: "請關閉此頁面，返回帳號認證頁面重新點擊認鏈接進行認證。",
button: "關閉頁面"
}
},
en: {
success: {
title: "Authentication Completed",
content: "Please close this page and return to the account authentication page to click NEXT",
button: "Close Page"
},
fail: {
title: "Authentication Failed",
content: "Please close this page and return to the account authentication page to click the authentication link again.",
button: "Close Page"
}
},
fr: {
success: {
title: "Authentification Terminée",
content: "Veuillez fermer cette page et revenir à la page d'authentification du compte pour cliquer sur « SUIVANT »",
button: "Fermer la page"
},
fail: {
title: "Échec de l'Authentification",
content: "Veuillez fermer cette page et revenir à la page d'authentification du compte pour cliquer de nouveau sur le lien d'authentification.",
button: "Fermer la page"
}
},
ru: {
success: {
title: "Подтверждение завершено",
content: "Пожалуйста, закройте эту страницу, вернитесь на страницу аутентификации учетной записи и нажмите кнопку «Далее».",
button: "Закрыть страницу"
},
fail: {
title: "Ошибка аутентификации",
content: "Пожалуйста, закройте эту страницу, вернитесь на страницу аутентификации учетной записи и повторите процесс аутентификации, щелкнув ссылку.",
button: "Закрыть страницу"
}
},
de: {
success: {
title: "Authentifizierung abgeschlossen",
content: "Bitte schließen Sie diese Seite, kehren Sie zur Kontobestätigungsseite zurück und klicken Sie auf „WEITER“.",
button: "Seite schließen"
},
fail: {
title: "Authentifizierung fehlgeschlagen",
content: "Bitte schließen Sie diese Seite, kehren Sie zur Kontobestätigungsseite zurück und wiederholen Sie den Authentifizierungsprozess, indem Sie auf den Link klicken.",
button: "Seite schließen"
}
},
es: {
success: {
title: "Autenticación completada",
content: "Por favor, cierre esta página, regrese a la página de autenticación de la cuenta y haga clic en 'SIGUIENTE'.",
button: "Cerrar página"
},
fail: {
title: "Error de autenticación",
content: "Por favor, cierre esta página, regrese a la página de autenticación de la cuenta y vuelva a hacer clic en el enlace de autenticación.",
button: "Cerrar página"
}
},
ja: {
success: {
title: "認証完了",
content: "このページを閉じて、アカウント認証ページに戻り、「次」をクリックしてください。",
button: "ページを閉じる"
},
fail: {
title: "認証失敗",
content: "このページを閉じて、アカウント認証ページに戻り、認証リンクを再度クリックしてください。",
button: "ページを閉じる"
}
}
}
// insert translate into page / match order: locale > language > english
document.title = translation[locale]?.[status]?.title ?? translation[language]?.[status]?.title ?? translation["en"]?.[status]?.title;
document.getElementById("titleArea").innerText = translation[locale]?.[status]?.title ?? translation[language]?.[status]?.title ?? translation["en"]?.[status]?.title;
document.getElementById("contentArea").innerText = translation[locale]?.[status]?.content ?? translation[language]?.[status]?.content ?? translation["en"]?.[status]?.content;
document.getElementById("buttonArea").innerText = translation[locale]?.[status]?.button ?? translation[language]?.[status]?.button ?? translation["en"]?.[status]?.button;
window.opener=null;
window.open('','_self');
window.close();
</script>
</body>
</html>
'''
if _template == '':
await asyncio.get_running_loop().run_in_executor(
None, _load_page_template)
web_page = _template.replace('TITLE_PLACEHOLDER', title)
web_page = web_page.replace('CONTENT_PLACEHOLDER', content)
web_page = web_page.replace('BUTTON_PLACEHOLDER', button)
web_page = web_page.replace(
'STATUS_PLACEHOLDER', 'true' if success else 'false')
return web_page

31
test/conftest.py
View File

@ -15,8 +15,7 @@ TEST_LANG: str = 'zh-Hans'
TEST_UID: str = '123456789'
TEST_CLOUD_SERVER: str = 'cn'
DOMAIN_OAUTH2: str = 'oauth2_info'
DOMAIN_USER_INFO: str = 'user_info'
DOMAIN_CLOUD_CACHE: str = 'cloud_cache'
_LOGGER = logging.getLogger(__name__)
@ -139,8 +138,18 @@ def test_cloud_server() -> str:
@pytest.fixture(scope='session')
def test_domain_oauth2() -> str:
return DOMAIN_OAUTH2
def test_domain_cloud_cache() -> str:
return DOMAIN_CLOUD_CACHE
@pytest.fixture(scope='session')
def test_name_oauth2_info() -> str:
return f'{TEST_CLOUD_SERVER}_oauth2_info'
@pytest.fixture(scope='session')
def test_name_uid() -> str:
return f'{TEST_CLOUD_SERVER}_uid'
@pytest.fixture(scope='session')
@ -149,5 +158,15 @@ def test_name_uuid() -> str:
@pytest.fixture(scope='session')
def test_domain_user_info() -> str:
return DOMAIN_USER_INFO
def test_name_rd_did() -> str:
return f'{TEST_CLOUD_SERVER}_rd_did'
@pytest.fixture(scope='session')
def test_name_homes() -> str:
return f'{TEST_CLOUD_SERVER}_homes'
@pytest.fixture(scope='session')
def test_name_devices() -> str:
return f'{TEST_CLOUD_SERVER}_devices'

197
test/test_cloud.py
View File

@ -16,8 +16,9 @@ async def test_miot_oauth_async(
test_cache_path: str,
test_cloud_server: str,
test_oauth2_redirect_url: str,
test_domain_oauth2: str,
test_uuid: str,
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uuid: str
) -> dict:
from miot.const import OAUTH2_CLIENT_ID
@ -26,7 +27,7 @@ async def test_miot_oauth_async(
miot_storage = MIoTStorage(test_cache_path)
local_uuid = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_name_uuid, type_=str)
domain=test_domain_cloud_cache, name=test_name_uuid, type_=str)
uuid = str(local_uuid or test_uuid)
_LOGGER.info('uuid: %s', uuid)
miot_oauth = MIoTOauthClient(
@ -37,7 +38,7 @@ async def test_miot_oauth_async(
oauth_info = None
load_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
if (
isinstance(load_info, dict)
and 'access_token' in load_info
@ -61,11 +62,11 @@ async def test_miot_oauth_async(
oauth_info = res_obj
_LOGGER.info('get_access_token result: %s', res_obj)
rc = await miot_storage.save_async(
test_domain_oauth2, test_cloud_server, oauth_info)
test_domain_cloud_cache, test_name_oauth2_info, oauth_info)
assert rc
_LOGGER.info('save oauth info')
rc = await miot_storage.save_async(
test_domain_oauth2, test_name_uuid, uuid)
test_domain_cloud_cache, test_name_uuid, uuid)
assert rc
_LOGGER.info('save uuid')
@ -86,7 +87,8 @@ async def test_miot_oauth_refresh_token(
test_cache_path: str,
test_cloud_server: str,
test_oauth2_redirect_url: str,
test_domain_oauth2: str,
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uuid: str
):
from miot.const import OAUTH2_CLIENT_ID
@ -95,10 +97,10 @@ async def test_miot_oauth_refresh_token(
miot_storage = MIoTStorage(test_cache_path)
uuid = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_name_uuid, type_=str)
domain=test_domain_cloud_cache, name=test_name_uuid, type_=str)
assert isinstance(uuid, str)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict)
assert 'access_token' in oauth_info
assert 'refresh_token' in oauth_info
@ -122,9 +124,9 @@ async def test_miot_oauth_refresh_token(
remaining_time = update_info['expires_ts'] - int(time.time())
assert remaining_time > 0
_LOGGER.info('refresh token, remaining valid time: %ss', remaining_time)
# Save token
# Save oauth2 info
rc = await miot_storage.save_async(
test_domain_oauth2, test_cloud_server, update_info)
test_domain_cloud_cache, test_name_oauth2_info, update_info)
assert rc
_LOGGER.info('refresh token success, %s', update_info)
@ -136,7 +138,8 @@ async def test_miot_oauth_refresh_token(
async def test_miot_cloud_get_nickname_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -144,7 +147,7 @@ async def test_miot_cloud_get_nickname_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -164,8 +167,9 @@ async def test_miot_cloud_get_nickname_async(
async def test_miot_cloud_get_uid_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uid: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -173,7 +177,7 @@ async def test_miot_cloud_get_uid_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -184,8 +188,7 @@ async def test_miot_cloud_get_uid_async(
_LOGGER.info('your uid: %s', uid)
# Save uid
rc = await miot_storage.save_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', data=uid)
domain=test_domain_cloud_cache, name=test_name_uid, data=uid)
assert rc
await miot_http.deinit_async()
@ -196,8 +199,9 @@ async def test_miot_cloud_get_uid_async(
async def test_miot_cloud_get_homeinfos_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uid: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -205,7 +209,7 @@ async def test_miot_cloud_get_homeinfos_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -223,8 +227,7 @@ async def test_miot_cloud_get_homeinfos_async(
uid = homeinfos.get('uid', '')
# Compare uid with uid in storage
uid2 = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', type_=str)
domain=test_domain_cloud_cache, name=test_name_uid, type_=str)
assert uid == uid2
_LOGGER.info('your uid: %s', uid)
# Get homes
@ -242,8 +245,11 @@ async def test_miot_cloud_get_homeinfos_async(
async def test_miot_cloud_get_devices_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uid: str,
test_name_homes: str,
test_name_devices: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -251,7 +257,7 @@ async def test_miot_cloud_get_devices_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -266,8 +272,7 @@ async def test_miot_cloud_get_devices_async(
# Compare uid with uid in storage
uid = devices.get('uid', '')
uid2 = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'uid_{test_cloud_server}', type_=str)
domain=test_domain_cloud_cache, name=test_name_uid, type_=str)
assert uid == uid2
_LOGGER.info('your uid: %s', uid)
# Get homes
@ -278,12 +283,10 @@ async def test_miot_cloud_get_devices_async(
_LOGGER.info('your devices count: %s', len(devices))
# Storage homes and devices
rc = await miot_storage.save_async(
domain=test_domain_user_info,
name=f'homes_{test_cloud_server}', data=homes)
domain=test_domain_cloud_cache, name=test_name_homes, data=homes)
assert rc
rc = await miot_storage.save_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', data=devices)
domain=test_domain_cloud_cache, name=test_name_devices, data=devices)
assert rc
await miot_http.deinit_async()
@ -294,8 +297,9 @@ async def test_miot_cloud_get_devices_async(
async def test_miot_cloud_get_devices_with_dids_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -303,7 +307,7 @@ async def test_miot_cloud_get_devices_with_dids_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -311,8 +315,7 @@ async def test_miot_cloud_get_devices_with_dids_async(
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', type_=dict)
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
did_list = list(local_devices.keys())
assert len(did_list) > 0
@ -328,13 +331,96 @@ async def test_miot_cloud_get_devices_with_dids_async(
await miot_http.deinit_async()
@pytest.mark.asyncio
async def test_miot_cloud_get_cert(
test_cache_path: str,
test_cloud_server: str,
test_random_did: str,
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_uid: str,
test_name_rd_did: str
):
"""
NOTICE: Currently, only certificate acquisition in the CN region is
supported.
"""
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
from miot.miot_storage import MIoTCert, MIoTStorage
if test_cloud_server.lower() != 'cn':
_LOGGER.info('only support CN region')
return
miot_storage = MIoTStorage(test_cache_path)
uid = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_uid, type_=str)
assert isinstance(uid, str)
_LOGGER.info('your uid: %s', uid)
random_did = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_rd_did, type_=str)
if not random_did:
random_did = test_random_did
rc = await miot_storage.save_async(
domain=test_domain_cloud_cache, name=test_name_rd_did,
data=random_did)
assert rc
assert isinstance(random_did, str)
_LOGGER.info('your random_did: %s', random_did)
oauth_info = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict)
assert 'access_token' in oauth_info
access_token = oauth_info['access_token']
# Get certificates
miot_cert = MIoTCert(storage=miot_storage, uid=uid, cloud_server='CN')
assert await miot_cert.verify_ca_cert_async(), 'invalid ca cert'
remaining_time: int = await miot_cert.user_cert_remaining_time_async()
if remaining_time > 0:
_LOGGER.info(
'user cert is valid, remaining time, %ss', remaining_time)
_LOGGER.info((
'if you want to obtain it again, please delete the '
'key, csr, and cert files in %s.'), test_cache_path)
return
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server,
client_id=OAUTH2_CLIENT_ID,
access_token=access_token)
user_key = miot_cert.gen_user_key()
assert isinstance(user_key, str)
_LOGGER.info('user_key str, %s', user_key)
user_csr = miot_cert.gen_user_csr(user_key=user_key, did=random_did)
assert isinstance(user_csr, str)
_LOGGER.info('user_csr str, %s', user_csr)
cert_str = await miot_http.get_central_cert_async(csr=user_csr)
assert isinstance(cert_str, str)
_LOGGER.info('user_cert str, %s', cert_str)
rc = await miot_cert.update_user_key_async(key=user_key)
assert rc
rc = await miot_cert.update_user_cert_async(cert=cert_str)
assert rc
# verify user certificates
remaining_time = await miot_cert.user_cert_remaining_time_async(
cert_data=cert_str.encode('utf-8'), did=random_did)
assert remaining_time > 0
_LOGGER.info('user cert remaining time, %ss', remaining_time)
await miot_http.deinit_async()
@pytest.mark.asyncio
@pytest.mark.dependency()
async def test_miot_cloud_get_prop_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -342,7 +428,7 @@ async def test_miot_cloud_get_prop_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -350,8 +436,7 @@ async def test_miot_cloud_get_prop_async(
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', type_=dict)
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
did_list = list(local_devices.keys())
assert len(did_list) > 0
@ -370,8 +455,9 @@ async def test_miot_cloud_get_prop_async(
async def test_miot_cloud_get_props_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_cloud import MIoTHttpClient
@ -379,7 +465,7 @@ async def test_miot_cloud_get_props_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -387,8 +473,7 @@ async def test_miot_cloud_get_props_async(
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', type_=dict)
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
did_list = list(local_devices.keys())
assert len(did_list) > 0
@ -409,8 +494,9 @@ async def test_miot_cloud_get_props_async(
async def test_miot_cloud_set_prop_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
"""
WARNING: This test case will control the actual device and is not enabled
@ -422,7 +508,7 @@ async def test_miot_cloud_set_prop_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -430,8 +516,7 @@ async def test_miot_cloud_set_prop_async(
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', type_=dict)
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
assert len(local_devices) > 0
# Set prop
@ -460,8 +545,9 @@ async def test_miot_cloud_set_prop_async(
async def test_miot_cloud_action_async(
test_cache_path: str,
test_cloud_server: str,
test_domain_oauth2: str,
test_domain_user_info: str
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
"""
WARNING: This test case will control the actual device and is not enabled
@ -473,7 +559,7 @@ async def test_miot_cloud_action_async(
miot_storage = MIoTStorage(test_cache_path)
oauth_info = await miot_storage.load_async(
domain=test_domain_oauth2, name=test_cloud_server, type_=dict)
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server, client_id=OAUTH2_CLIENT_ID,
@ -481,8 +567,7 @@ async def test_miot_cloud_action_async(
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_user_info,
name=f'devices_{test_cloud_server}', type_=dict)
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
assert len(local_devices) > 0
# Action

9
test/test_mdns.py
View File

@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_mdns.py."""
import asyncio
import logging
import pytest
from zeroconf import IPVersion
@ -12,10 +13,10 @@ _LOGGER = logging.getLogger(__name__)
@pytest.mark.asyncio
async def test_service_loop_async():
from miot.miot_mdns import MipsService, MipsServiceData, MipsServiceState
from miot.miot_mdns import MipsService, MipsServiceState
async def on_service_state_change(
group_id: str, state: MipsServiceState, data: MipsServiceData):
group_id: str, state: MipsServiceState, data: dict):
_LOGGER.info(
'on_service_state_change, %s, %s, %s', group_id, state, data)
@ -23,8 +24,10 @@ async def test_service_loop_async():
mips_service = MipsService(aiozc)
mips_service.sub_service_change('test', '*', on_service_state_change)
await mips_service.init_async()
# Wait for service to discover
await asyncio.sleep(3)
services_detail = mips_service.get_services()
_LOGGER.info('get all service, %s', services_detail.keys())
_LOGGER.info('get all service, %s', list(services_detail.keys()))
for name, data in services_detail.items():
_LOGGER.info(
'\tinfo, %s, %s, %s, %s',

264
test/test_mips.py Normal file
View File

@ -0,0 +1,264 @@
# -*- coding: utf-8 -*-
"""Unit test for miot_mips.py.
NOTICE: When running this test case, you need to run test_cloud.py first to
obtain the token and certificate information, and at the same time avoid data
deletion.
"""
import ipaddress
from typing import Any, Tuple
import pytest
import asyncio
import logging
_LOGGER = logging.getLogger(__name__)
# pylint: disable = import-outside-toplevel, unused-argument
@pytest.mark.parametrize('central_info', [
('<Group id>', 'Gateway did', 'Gateway ip', 8883),
])
@pytest.mark.asyncio
async def test_mips_local_async(
test_cache_path: str,
test_domain_cloud_cache: str,
test_name_uid: str,
test_name_rd_did: str,
central_info: Tuple[str, str, str, int]
):
"""
NOTICE:
- Mips local is used to connect to the central gateway and is only
supported in the Chinese mainland region.
- Before running this test case, you need to run test_mdns.py first to
obtain the group_id, did, ip, and port of the hub, and then fill in this
information in the parametrize. you can enter multiple central connection
information items for separate tests.
- This test case requires running test_cloud.py first to obtain the
central connection certificate.
- This test case will control the indicator light switch of the central
gateway.
"""
from miot.miot_storage import MIoTStorage, MIoTCert
from miot.miot_mips import MipsLocalClient
central_group_id: str = central_info[0]
assert isinstance(central_group_id, str)
central_did: str = central_info[1]
assert central_did.isdigit()
central_ip: str = central_info[2]
assert ipaddress.ip_address(central_ip)
central_port: int = central_info[3]
assert isinstance(central_port, int)
miot_storage = MIoTStorage(test_cache_path)
uid = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_uid, type_=str)
assert isinstance(uid, str)
random_did = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_rd_did, type_=str)
assert isinstance(random_did, str)
miot_cert = MIoTCert(storage=miot_storage, uid=uid, cloud_server='CN')
assert miot_cert.ca_file
assert miot_cert.cert_file
assert miot_cert.key_file
_LOGGER.info(
'cert info, %s, %s, %s', miot_cert.ca_file, miot_cert.cert_file,
miot_cert.key_file)
mips_local = MipsLocalClient(
did=random_did,
host=central_ip,
group_id=central_group_id,
ca_file=miot_cert.ca_file,
cert_file=miot_cert.cert_file,
key_file=miot_cert.key_file,
port=central_port,
home_name='mips local test')
mips_local.enable_logger(logger=_LOGGER)
mips_local.enable_mqtt_logger(logger=_LOGGER)
async def on_mips_state_changed_async(key: str, state: bool):
_LOGGER.info('on mips state changed, %s, %s', key, state)
async def on_dev_list_changed_async(
mips: MipsLocalClient, did_list: list[str]
):
_LOGGER.info('dev list changed, %s', did_list)
def on_prop_changed(payload: dict, ctx: Any):
_LOGGER.info('prop changed, %s=%s', ctx, payload)
def on_event_occurred(payload: dict, ctx: Any):
_LOGGER.info('event occurred, %s=%s', ctx, payload)
# Reg mips state
mips_local.sub_mips_state(
key='mips_local', handler=on_mips_state_changed_async)
mips_local.on_dev_list_changed = on_dev_list_changed_async
# Connect
await mips_local.connect_async()
await asyncio.sleep(0.5)
# Get device list
device_list = await mips_local.get_dev_list_async()
assert isinstance(device_list, dict)
_LOGGER.info(
'get_dev_list, %d, %s', len(device_list), list(device_list.keys()))
# Sub Prop
mips_local.sub_prop(
did=central_did, handler=on_prop_changed,
handler_ctx=f'{central_did}.*')
# Sub Event
mips_local.sub_event(
did=central_did, handler=on_event_occurred,
handler_ctx=f'{central_did}.*')
# Get/set prop
test_siid = 3
test_piid = 1
# mips_local.sub_prop(
# did=central_did, siid=test_siid, piid=test_piid,
# handler=on_prop_changed,
# handler_ctx=f'{central_did}.{test_siid}.{test_piid}')
result1 = await mips_local.get_prop_async(
did=central_did, siid=test_siid, piid=test_piid)
assert isinstance(result1, bool)
_LOGGER.info('get prop.%s.%s, value=%s', test_siid, test_piid, result1)
result2 = await mips_local.set_prop_async(
did=central_did, siid=test_siid, piid=test_piid, value=not result1)
_LOGGER.info(
'set prop.%s.%s=%s, result=%s',
test_siid, test_piid, not result1, result2)
assert isinstance(result2, dict)
result3 = await mips_local.get_prop_async(
did=central_did, siid=test_siid, piid=test_piid)
assert isinstance(result3, bool)
_LOGGER.info('get prop.%s.%s, value=%s', test_siid, test_piid, result3)
# Action
test_siid = 4
test_aiid = 1
in_list = [{'piid': 1, 'value': 'hello world.'}]
result4 = await mips_local.action_async(
did=central_did, siid=test_siid, aiid=test_aiid,
in_list=in_list)
assert isinstance(result4, dict)
_LOGGER.info(
'action.%s.%s=%s, result=%s', test_siid, test_piid, in_list, result4)
# Disconnect
await mips_local.disconnect_async()
await mips_local.deinit_async()
@pytest.mark.asyncio
async def test_mips_cloud_async(
test_cache_path: str,
test_name_uuid: str,
test_cloud_server: str,
test_domain_cloud_cache: str,
test_name_oauth2_info: str,
test_name_devices: str
):
"""
NOTICE:
- This test case requires running test_cloud.py first to obtain the
central connection certificate.
- This test case will control the indicator light switch of the central
gateway.
"""
from miot.const import OAUTH2_CLIENT_ID
from miot.miot_storage import MIoTStorage
from miot.miot_mips import MipsCloudClient
from miot.miot_cloud import MIoTHttpClient
miot_storage = MIoTStorage(test_cache_path)
uuid = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_uuid, type_=str)
assert isinstance(uuid, str)
oauth_info = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_oauth2_info, type_=dict)
assert isinstance(oauth_info, dict) and 'access_token' in oauth_info
access_token = oauth_info['access_token']
_LOGGER.info('connect info, %s, %s', uuid, access_token)
mips_cloud = MipsCloudClient(
uuid=uuid,
cloud_server=test_cloud_server,
app_id=OAUTH2_CLIENT_ID,
token=access_token)
mips_cloud.enable_logger(logger=_LOGGER)
mips_cloud.enable_mqtt_logger(logger=_LOGGER)
miot_http = MIoTHttpClient(
cloud_server=test_cloud_server,
client_id=OAUTH2_CLIENT_ID,
access_token=access_token)
async def on_mips_state_changed_async(key: str, state: bool):
_LOGGER.info('on mips state changed, %s, %s', key, state)
def on_prop_changed(payload: dict, ctx: Any):
_LOGGER.info('prop changed, %s=%s', ctx, payload)
def on_event_occurred(payload: dict, ctx: Any):
_LOGGER.info('event occurred, %s=%s', ctx, payload)
await mips_cloud.connect_async()
await asyncio.sleep(0.5)
# Sub mips state
mips_cloud.sub_mips_state(
key='mips_cloud', handler=on_mips_state_changed_async)
# Load devices
local_devices = await miot_storage.load_async(
domain=test_domain_cloud_cache, name=test_name_devices, type_=dict)
assert isinstance(local_devices, dict)
central_did = ''
for did, info in local_devices.items():
if info['model'] != 'xiaomi.gateway.hub1':
continue
central_did = did
break
if central_did:
# Sub Prop
mips_cloud.sub_prop(
did=central_did, handler=on_prop_changed,
handler_ctx=f'{central_did}.*')
# Sub Event
mips_cloud.sub_event(
did=central_did, handler=on_event_occurred,
handler_ctx=f'{central_did}.*')
# Get/set prop
test_siid = 3
test_piid = 1
# mips_cloud.sub_prop(
# did=central_did, siid=test_siid, piid=test_piid,
# handler=on_prop_changed,
# handler_ctx=f'{central_did}.{test_siid}.{test_piid}')
result1 = await miot_http.get_prop_async(
did=central_did, siid=test_siid, piid=test_piid)
assert isinstance(result1, bool)
_LOGGER.info('get prop.%s.%s, value=%s', test_siid, test_piid, result1)
result2 = await miot_http.set_prop_async(params=[{
'did': central_did, 'siid': test_siid, 'piid': test_piid,
'value': not result1}])
_LOGGER.info(
'set prop.%s.%s=%s, result=%s',
test_siid, test_piid, not result1, result2)
assert isinstance(result2, list)
result3 = await miot_http.get_prop_async(
did=central_did, siid=test_siid, piid=test_piid)
assert isinstance(result3, bool)
_LOGGER.info('get prop.%s.%s, value=%s', test_siid, test_piid, result3)
# Action
test_siid = 4
test_aiid = 1
in_list = [{'piid': 1, 'value': 'hello world.'}]
result4 = await miot_http.action_async(
did=central_did, siid=test_siid, aiid=test_aiid,
in_list=in_list)
assert isinstance(result4, dict)
_LOGGER.info(
'action.%s.%s=%s, result=%s',
test_siid, test_piid, in_list, result4)
await asyncio.sleep(1)
# Disconnect
await mips_cloud.disconnect_async()
await mips_cloud.deinit_async()
await miot_http.deinit_async()