Add wrappers for event (#44)

* Add wrappers for event

* Improve

Author: epenet

src/tuya_device_handlers/device_wrapper/event.py

@@ -0,0 +1,69 @@
+"""Tuya device wrapper."""
+
+from __future__ import annotations
+
+import base64
+from typing import TYPE_CHECKING, Any
+
+from .common import DPCodeEnumWrapper, DPCodeRawWrapper, DPCodeStringWrapper
+
+if TYPE_CHECKING:
+    from tuya_sharing import CustomerDevice  # type: ignore[import-untyped]
+
+
+class SimpleEventEnumWrapper(DPCodeEnumWrapper[tuple[str, None]]):
+    """Wrapper for event enum DP codes.
+
+    Does not provide attributes.
+    """
+
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> tuple[str, None] | None:
+        """Return the event details."""
+        if (raw_value := self._read_dpcode_value(device)) is None:
+            return None
+        return (raw_value, None)
+
+
+class Base64Utf8StringWrapper(DPCodeStringWrapper[tuple[str, dict[str, Any]]]):
+    """Wrapper for a string message received in a base64/UTF-8 STRING DPCode.
+
+    Raises 'triggered' event, with the message in the event attributes.
+    """
+
+    def __init__(self, dpcode: str, type_information: Any) -> None:
+        """Init Base64Utf8StringWrapper."""
+        super().__init__(dpcode, type_information)
+        self.options = ["triggered"]
+
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> tuple[str, dict[str, Any]] | None:
+        """Return the event attributes for the alarm message."""
+        if (raw_value := self._read_dpcode_value(device)) is None:
+            return None
+        return (
+            "triggered",
+            {"message": base64.b64decode(raw_value).decode("utf-8")},
+        )
+
+
+class Base64Utf8RawWrapper(DPCodeRawWrapper[tuple[str, dict[str, Any]]]):
+    """Wrapper for a string message received in a base64/UTF-8 RAW DPCode.
+
+    Raises 'triggered' event, with the message in the event attributes.
+    """
+
+    def __init__(self, dpcode: str, type_information: Any) -> None:
+        """Init _DoorbellPicWrapper."""
+        super().__init__(dpcode, type_information)
+        self.options = ["triggered"]
+
+    def read_device_status(
+        self, device: CustomerDevice
+    ) -> tuple[str, dict[str, Any]] | None:
+        """Return the event attributes for the doorbell picture."""
+        if (status := self._read_dpcode_value(device)) is None:
+            return None
+        return ("triggered", {"message": status.decode("utf-8")})

tests/device_wrapper/test_event.py

@@ -0,0 +1,74 @@
+"""Test DeviceWrapper classes"""
+
+from typing import Any
+
+import pytest
+from tuya_sharing import CustomerDevice  # type: ignore[import-untyped]
+
+from tuya_device_handlers.device_wrapper.common import (
+    DPCodeTypeInformationWrapper,
+)
+from tuya_device_handlers.device_wrapper.event import (
+    Base64Utf8RawWrapper,
+    Base64Utf8StringWrapper,
+    SimpleEventEnumWrapper,
+)
+
+try:
+    from typeguard import suppress_type_checks  # type: ignore[import-not-found]
+except ImportError:
+    from contextlib import nullcontext
+
+    suppress_type_checks = nullcontext
+
+
+@pytest.mark.parametrize(
+    ("wrapper_type", "dpcode", "status", "expected_device_status"),
+    [
+        (
+            Base64Utf8RawWrapper,
+            "demo_raw",
+            "aHR0cHM6Ly9zb21lLXBpY3R1cmUtdXJsLmNvbS9pbWFnZS5qcGc=",
+            (
+                "triggered",
+                {"message": "https://some-picture-url.com/image.jpg"},
+            ),
+        ),
+        (
+            Base64Utf8StringWrapper,
+            "demo_string",
+            "TXkgZG9nIGF0ZSBteSBkaW5uZXI=",
+            (
+                "triggered",
+                {"message": "My dog ate my dinner"},
+            ),
+        ),
+        (
+            SimpleEventEnumWrapper,
+            "demo_enum",
+            "scene",
+            ("scene", None),
+        ),
+    ],
+)
+def test_read_device_status(
+    dpcode: str,
+    wrapper_type: type[DPCodeTypeInformationWrapper[Any, Any]],
+    status: Any,
+    expected_device_status: Any,
+    mock_device: CustomerDevice,
+) -> None:
+    """Test read_device_status."""
+    mock_device.status[dpcode] = status
+    wrapper = wrapper_type.find_dpcode(mock_device, dpcode)
+
+    assert wrapper
+    assert wrapper.read_device_status(mock_device) == expected_device_status
+
+    # All wrappers return None if status is None
+    mock_device.status[dpcode] = None
+    assert wrapper.read_device_status(mock_device) is None
+
+    # All wrappers return None if status is missing
+    mock_device.status.pop(dpcode)
+    assert wrapper.read_device_status(mock_device) is None