Add diagnostics tests to tuya (#148489)

This commit is contained in:
epenet 2025-07-09 15:47:48 +02:00 committed by GitHub
parent f6e2b962fd
commit e1cdc1af1c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 282 additions and 3 deletions

View File

@ -6,7 +6,7 @@ from collections.abc import Generator
from unittest.mock import MagicMock, patch
import pytest
from tuya_sharing import CustomerDevice
from tuya_sharing import CustomerApi, CustomerDevice, DeviceFunction, DeviceStatusRange
from homeassistant.components.tuya import ManagerCompat
from homeassistant.components.tuya.const import (
@ -19,6 +19,7 @@ from homeassistant.components.tuya.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.json import json_dumps
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, async_load_json_object_fixture
@ -116,6 +117,12 @@ def mock_manager() -> ManagerCompat:
manager = MagicMock(spec=ManagerCompat)
manager.device_map = {}
manager.mq = MagicMock()
manager.mq.client = MagicMock()
manager.mq.client.is_connected = MagicMock(return_value=True)
manager.customer_api = MagicMock(spec=CustomerApi)
# Meaningless URL / UUIDs
manager.customer_api.endpoint = "https://apigw.tuyaeu.com"
manager.terminal_id = "7cd96aff-6ec8-4006-b093-3dbff7947591"
return manager
@ -142,12 +149,34 @@ async def mock_device(hass: HomeAssistant, mock_device_code: str) -> CustomerDev
device.product_id = details["product_id"]
device.product_name = details["product_name"]
device.online = details["online"]
device.sub = details.get("sub")
device.time_zone = details.get("time_zone")
device.active_time = details.get("active_time")
if device.active_time:
device.active_time = int(dt_util.as_timestamp(device.active_time))
device.create_time = details.get("create_time")
if device.create_time:
device.create_time = int(dt_util.as_timestamp(device.create_time))
device.update_time = details.get("update_time")
if device.update_time:
device.update_time = int(dt_util.as_timestamp(device.update_time))
device.support_local = details.get("support_local")
device.mqtt_connected = details.get("mqtt_connected")
device.function = {
key: MagicMock(type=value["type"], values=json_dumps(value["value"]))
key: DeviceFunction(
code=value.get("code"),
type=value["type"],
values=json_dumps(value["value"]),
)
for key, value in details["function"].items()
}
device.status_range = {
key: MagicMock(type=value["type"], values=json_dumps(value["value"]))
key: DeviceStatusRange(
code=value.get("code"),
type=value["type"],
values=json_dumps(value["value"]),
)
for key, value in details["status_range"].items()
}
device.status = details["status"]

View File

@ -0,0 +1,183 @@
# serializer version: 1
# name: test_device_diagnostics[rqbj_gas_sensor]
dict({
'active_time': '2025-06-24T20:33:10+00:00',
'category': 'rqbj',
'create_time': '2025-06-24T20:33:10+00:00',
'disabled_by': None,
'disabled_polling': False,
'endpoint': 'https://apigw.tuyaeu.com',
'function': dict({
'null': dict({
'type': 'Boolean',
'value': dict({
}),
}),
}),
'home_assistant': dict({
'disabled': False,
'disabled_by': None,
'entities': list([
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': None,
'icon': None,
'original_device_class': 'gas',
'original_icon': None,
'state': dict({
'attributes': dict({
'device_class': 'gas',
'friendly_name': 'Gas sensor Gas',
}),
'entity_id': 'binary_sensor.gas_sensor_gas',
'state': 'off',
}),
'unit_of_measurement': None,
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': None,
'icon': None,
'original_device_class': None,
'original_icon': None,
'state': dict({
'attributes': dict({
'friendly_name': 'Gas sensor Gas',
'state_class': 'measurement',
'unit_of_measurement': 'ppm',
}),
'entity_id': 'sensor.gas_sensor_gas',
'state': '0.0',
}),
'unit_of_measurement': 'ppm',
}),
]),
'name': 'Gas sensor',
'name_by_user': None,
}),
'id': 'ebb9d0eb5014f98cfboxbz',
'mqtt_connected': True,
'name': 'Gas sensor',
'online': True,
'product_id': '4iqe2hsfyd86kwwc',
'product_name': 'Gas sensor',
'set_up': True,
'status': dict({
'alarm_time': 300,
'checking_result': 'check_success',
'gas_sensor_status': 'normal',
'gas_sensor_value': 0,
'muffling': True,
'self_checking': False,
}),
'status_range': dict({
'null': dict({
'type': 'Boolean',
'value': dict({
}),
}),
}),
'sub': False,
'support_local': True,
'terminal_id': '7cd96aff-6ec8-4006-b093-3dbff7947591',
'time_zone': '-04:00',
'update_time': '2025-06-24T20:33:10+00:00',
})
# ---
# name: test_entry_diagnostics[rqbj_gas_sensor]
dict({
'devices': list([
dict({
'active_time': '2025-06-24T20:33:10+00:00',
'category': 'rqbj',
'create_time': '2025-06-24T20:33:10+00:00',
'function': dict({
'null': dict({
'type': 'Boolean',
'value': dict({
}),
}),
}),
'home_assistant': dict({
'disabled': False,
'disabled_by': None,
'entities': list([
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': None,
'icon': None,
'original_device_class': 'gas',
'original_icon': None,
'state': dict({
'attributes': dict({
'device_class': 'gas',
'friendly_name': 'Gas sensor Gas',
}),
'entity_id': 'binary_sensor.gas_sensor_gas',
'state': 'off',
}),
'unit_of_measurement': None,
}),
dict({
'device_class': None,
'disabled': False,
'disabled_by': None,
'entity_category': None,
'icon': None,
'original_device_class': None,
'original_icon': None,
'state': dict({
'attributes': dict({
'friendly_name': 'Gas sensor Gas',
'state_class': 'measurement',
'unit_of_measurement': 'ppm',
}),
'entity_id': 'sensor.gas_sensor_gas',
'state': '0.0',
}),
'unit_of_measurement': 'ppm',
}),
]),
'name': 'Gas sensor',
'name_by_user': None,
}),
'id': 'ebb9d0eb5014f98cfboxbz',
'name': 'Gas sensor',
'online': True,
'product_id': '4iqe2hsfyd86kwwc',
'product_name': 'Gas sensor',
'set_up': True,
'status': dict({
'alarm_time': 300,
'checking_result': 'check_success',
'gas_sensor_status': 'normal',
'gas_sensor_value': 0,
'muffling': True,
'self_checking': False,
}),
'status_range': dict({
'null': dict({
'type': 'Boolean',
'value': dict({
}),
}),
}),
'sub': False,
'support_local': True,
'time_zone': '-04:00',
'update_time': '2025-06-24T20:33:10+00:00',
}),
]),
'disabled_by': None,
'disabled_polling': False,
'endpoint': 'https://apigw.tuyaeu.com',
'mqtt_connected': True,
'terminal_id': '7cd96aff-6ec8-4006-b093-3dbff7947591',
})
# ---

View File

@ -0,0 +1,67 @@
"""Test Tuya diagnostics platform."""
from __future__ import annotations
import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
from tuya_sharing import CustomerDevice
from homeassistant.components.tuya import ManagerCompat
from homeassistant.components.tuya.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import initialize_entry
from tests.common import MockConfigEntry
from tests.components.diagnostics import (
get_diagnostics_for_config_entry,
get_diagnostics_for_device,
)
from tests.typing import ClientSessionGenerator
@pytest.mark.parametrize("mock_device_code", ["rqbj_gas_sensor"])
async def test_entry_diagnostics(
hass: HomeAssistant,
mock_manager: ManagerCompat,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test config entry diagnostics."""
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
result = await get_diagnostics_for_config_entry(
hass, hass_client, mock_config_entry
)
assert result == snapshot(
exclude=props("last_changed", "last_reported", "last_updated")
)
@pytest.mark.parametrize("mock_device_code", ["rqbj_gas_sensor"])
async def test_device_diagnostics(
hass: HomeAssistant,
mock_manager: ManagerCompat,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
hass_client: ClientSessionGenerator,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test device diagnostics."""
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
device = device_registry.async_get_device(identifiers={(DOMAIN, mock_device.id)})
assert device, repr(device_registry.devices)
result = await get_diagnostics_for_device(
hass, hass_client, mock_config_entry, device
)
assert result == snapshot(
exclude=props("last_changed", "last_reported", "last_updated")
)