starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 15:31:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add wrapper class for boolean values in Tuya models (#155905)

Browse Source
This commit is contained in:
epenet
2025-11-07 16:16:14 +01:00
committed by GitHub
parent 5eb5b93c0e
commit d52749c71a
5 changed files with 175 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/tuya/models.py
+14
View File
@@ -36,6 +36,20 @@ class DPCodeWrapper:
raise NotImplementedError("read_device_value must be implemented")
@dataclass
class DPCodeBooleanWrapper(DPCodeWrapper):
"""Simple wrapper for boolean values.
Supports True/False only.
"""
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read the device value for the dpcode."""
if (raw_value := self._read_device_status_raw(device)) in (True, False):
return raw_value
return None
@dataclass(kw_only=True)
class DPCodeEnumWrapper(DPCodeWrapper):
"""Simple wrapper for EnumTypeData values."""
homeassistant/components/tuya/switch.py
+13 -5
View File
@@ -27,6 +27,7 @@ from homeassistant.helpers.issue_registry import (
from . import TuyaConfigEntry
from .const import DOMAIN, TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
from .entity import TuyaEntity
from .models import DPCodeBooleanWrapper
@dataclass(frozen=True, kw_only=True)
@@ -938,7 +939,12 @@ async def async_setup_entry(
device = manager.device_map[device_id]
if descriptions := SWITCHES.get(device.category):
entities.extend(
TuyaSwitchEntity(device, manager, description)
TuyaSwitchEntity(
device,
manager,
description,
DPCodeBooleanWrapper(description.key),
)
for description in descriptions
if description.key in device.status
and _check_deprecation(
@@ -1015,21 +1021,23 @@ class TuyaSwitchEntity(TuyaEntity, SwitchEntity):
device: CustomerDevice,
device_manager: Manager,
description: SwitchEntityDescription,
dpcode_wrapper: DPCodeBooleanWrapper,
) -> None:
"""Init TuyaHaSwitch."""
super().__init__(device, device_manager)
self.entity_description = description
self._attr_unique_id = f"{super().unique_id}{description.key}"
self._dpcode_wrapper = dpcode_wrapper
@property
def is_on(self) -> bool:
def is_on(self) -> bool | None:
"""Return true if switch is on."""
return self.device.status.get(self.entity_description.key, False)
return self._dpcode_wrapper.read_device_status(self.device)
def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
self._send_command([{"code": self.entity_description.key, "value": True}])
self._send_command([{"code": self._dpcode_wrapper.dpcode, "value": True}])
def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
self._send_command([{"code": self.entity_description.key, "value": False}])
self._send_command([{"code": self._dpcode_wrapper.dpcode, "value": False}])
homeassistant/components/tuya/valve.py
+15 -5
View File
@@ -17,6 +17,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
from .entity import TuyaEntity
from .models import DPCodeBooleanWrapper
VALVES: dict[DeviceCategory, tuple[ValveEntityDescription, ...]] = {
DeviceCategory.SFKZQ: (
@@ -93,7 +94,12 @@ async def async_setup_entry(
device = manager.device_map[device_id]
if descriptions := VALVES.get(device.category):
entities.extend(
TuyaValveEntity(device, manager, description)
TuyaValveEntity(
device,
manager,
description,
DPCodeBooleanWrapper(description.key),
)
for description in descriptions
if description.key in device.status
)
@@ -117,25 +123,29 @@ class TuyaValveEntity(TuyaEntity, ValveEntity):
device: CustomerDevice,
device_manager: Manager,
description: ValveEntityDescription,
dpcode_wrapper: DPCodeBooleanWrapper,
) -> None:
"""Init TuyaValveEntity."""
super().__init__(device, device_manager)
self.entity_description = description
self._attr_unique_id = f"{super().unique_id}{description.key}"
self._dpcode_wrapper = dpcode_wrapper
@property
def is_closed(self) -> bool:
def is_closed(self) -> bool | None:
"""Return if the valve is closed."""
return not self.device.status.get(self.entity_description.key, False)
if (is_open := self._dpcode_wrapper.read_device_status(self.device)) is None:
return None
return not is_open
async def async_open_valve(self) -> None:
"""Open the valve."""
await self.hass.async_add_executor_job(
self._send_command, [{"code": self.entity_description.key, "value": True}]
self._send_command, [{"code": self._dpcode_wrapper.dpcode, "value": True}]
)
async def async_close_valve(self) -> None:
"""Close the valve."""
await self.hass.async_add_executor_job(
self._send_command, [{"code": self.entity_description.key, "value": False}]
self._send_command, [{"code": self._dpcode_wrapper.dpcode, "value": False}]
)
tests/components/tuya/test_switch.py
+99 -2
View File
@@ -2,15 +2,20 @@
from __future__ import annotations
from typing import Any
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.components.tuya import DOMAIN
from homeassistant.const import Platform
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, issue_registry as ir
@@ -83,3 +88,95 @@ async def test_sfkzq_deprecated_switch(
)
is not None
) is expected_issue
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.SWITCH])
@pytest.mark.parametrize(
"mock_device_code",
["cz_PGEkBctAbtzKOZng"],
)
async def test_turn_on(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
) -> None:
"""Test turning on a switch."""
entity_id = "switch.din_socket"
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: entity_id,
},
blocking=True,
)
mock_manager.send_commands.assert_called_once_with(
mock_device.id, [{"code": "switch", "value": True}]
)
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.SWITCH])
@pytest.mark.parametrize(
"mock_device_code",
["cz_PGEkBctAbtzKOZng"],
)
async def test_turn_off(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
) -> None:
"""Test turning off a switch."""
entity_id = "switch.din_socket"
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: entity_id,
},
blocking=True,
)
mock_manager.send_commands.assert_called_once_with(
mock_device.id, [{"code": "switch", "value": False}]
)
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.SWITCH])
@pytest.mark.parametrize(
"mock_device_code",
["cz_PGEkBctAbtzKOZng"],
)
@pytest.mark.parametrize(
("initial_status", "expected_state"),
[
(True, "on"),
(False, "off"),
(None, STATE_UNKNOWN),
("some string", STATE_UNKNOWN),
],
)
async def test_state(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
initial_status: Any,
expected_state: str,
) -> None:
"""Test switch state."""
entity_id = "switch.din_socket"
mock_device.status["switch"] = initial_status
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
assert state.state == expected_state
tests/components/tuya/test_valve.py
+34 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from typing import Any
from unittest.mock import patch
import pytest
@@ -13,7 +14,7 @@ from homeassistant.components.valve import (
SERVICE_CLOSE_VALVE,
SERVICE_OPEN_VALVE,
)
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@@ -95,3 +96,35 @@ async def test_close_valve(
mock_manager.send_commands.assert_called_once_with(
mock_device.id, [{"code": "switch_1", "value": False}]
)
@patch("homeassistant.components.tuya.PLATFORMS", [Platform.VALVE])
@pytest.mark.parametrize(
"mock_device_code",
["sfkzq_ed7frwissyqrejic"],
)
@pytest.mark.parametrize(
("initial_status", "expected_state"),
[
(True, "open"),
(False, "closed"),
(None, STATE_UNKNOWN),
("some string", STATE_UNKNOWN),
],
)
async def test_state(
hass: HomeAssistant,
mock_manager: Manager,
mock_config_entry: MockConfigEntry,
mock_device: CustomerDevice,
initial_status: Any,
expected_state: str,
) -> None:
"""Test valve state."""
entity_id = "valve.jie_hashui_fa_valve_1"
mock_device.status["switch_1"] = initial_status
await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
state = hass.states.get(entity_id)
assert state is not None, f"{entity_id} does not exist"
assert state.state == expected_state