Add XML Message BatteryInfo (#900)

Author: nanomad

deebot_client/messages/__init__.py

@@ -9,11 +9,13 @@
 from deebot_client.message import Message
 
 from .json import MESSAGES as JSON_MESSAGES
+from .xml import MESSAGES as XML_MESSAGES
 
 _LOGGER = get_logger(__name__)
 
 MESSAGES = {
     DataType.JSON: JSON_MESSAGES,
+    DataType.XML: XML_MESSAGES,
 }
 
 _LEGACY_USE_GET_COMMAND = [

deebot_client/messages/xml/__init__.py

@@ -0,0 +1,22 @@
+"""XML messages."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.messages.xml.battery import BatteryInfo
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from deebot_client.message import Message
+
+__all__: Sequence[str] = ["BatteryInfo"]
+# fmt: off
+# ordered by file asc
+_MESSAGES: list[type[Message]] = [
+    BatteryInfo
+]
+# fmt: on
+
+MESSAGES: dict[str, type[Message]] = {message.NAME: message for message in _MESSAGES}

deebot_client/messages/xml/battery.py

@@ -0,0 +1,41 @@
+"""Battery messages."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events import BatteryEvent
+from deebot_client.logging_filter import get_logger
+from deebot_client.message import HandlingResult
+from deebot_client.messages.xml.common import XmlMessage
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+_LOGGER = get_logger(__name__)
+
+
+class BatteryInfo(XmlMessage):
+    """BatteryInfo message."""
+
+    NAME = "BatteryInfo"
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        if (battery := xml.find("battery")) is None or (
+            power := battery.attrib.get("power")
+        ) is None:
+            return HandlingResult.analyse()
+
+        if power.isdecimal() and (power_int := int(power)) >= 0:
+            event_bus.notify(BatteryEvent(power_int))
+            return HandlingResult.success()
+
+        _LOGGER.error("Invalid battery power level received %s", power)
+        return HandlingResult.analyse()

deebot_client/messages/xml/common.py

@@ -0,0 +1,37 @@
+"""Common xml based messages."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from defusedxml import ElementTree  # type: ignore[import-untyped]
+
+from deebot_client.message import MessageStr
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+    from deebot_client.message import HandlingResult
+
+
+class XmlMessage(MessageStr, ABC):
+    """Xml message."""
+
+    @classmethod
+    def _handle_str(cls, event_bus: EventBus, message: str) -> HandlingResult:
+        """Handle string message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        xml = ElementTree.fromstring(message)
+        return cls._handle_xml(event_bus, xml)
+
+    @classmethod
+    @abstractmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """

tests/messages/__init__.py

@@ -11,11 +11,24 @@
 
 
 def assert_message(
-    message: type[Message], data: dict[str, Any], expected_event: Event
+    message: type[Message], data: dict[str, Any] | str, expected_event: Event
 ) -> None:
     event_bus = Mock(spec_set=EventBus)
 
     result = message.handle(event_bus, data)
 
     assert result.state == HandlingState.SUCCESS
     event_bus.notify.assert_called_once_with(expected_event)
+
+
+def assert_message_failure(
+    message: type[Message],
+    data: dict[str, Any] | str,
+    expected_result_state: HandlingState,
+) -> None:
+    event_bus = Mock(spec_set=EventBus)
+
+    result = message.handle(event_bus, data)
+
+    assert result.state == expected_result_state
+    event_bus.notify.assert_not_called()

tests/messages/xml/__init__.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import pytest
+
+from deebot_client.events import BatteryEvent
+from deebot_client.message import HandlingState
+from deebot_client.messages.xml import BatteryInfo
+from tests.messages import assert_message, assert_message_failure
+
+
+@pytest.mark.parametrize("percentage", [0, 49, 100])
+def test_BatteryInfo(percentage: int) -> None:
+    xml_message = f'<ctl ret="ok"><battery power="{percentage}" /></ctl>'
+
+    assert_message(BatteryInfo, xml_message, BatteryEvent(percentage))
+
+
+@pytest.mark.parametrize(
+    "xml_message",
+    [
+        '<ctl ret="ok"><battery /></ctl>',
+        '<ctl ret="ok"><battery power="-1" /></ctl>',
+        '<ctl ret="ok"><battery power="fake" /></ctl>',
+        '<ctl ret="ok"><wrong power="100" /></ctl>',
+    ],
+)
+def test_BatteryInfo_error(xml_message: str) -> None:
+    assert_message_failure(BatteryInfo, xml_message, HandlingState.ANALYSE_LOGGED)

tests/messages/xml/test_battery.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from deebot_client.events import Event
+from deebot_client.message import HandlingResult, HandlingState
+from deebot_client.messages.xml.common import XmlMessage
+from tests.messages import assert_message, assert_message_failure
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+
+@dataclass(frozen=True)
+class _TestEvent(Event):
+    payload: str
+
+
+class _TestXmlMessage(XmlMessage):
+    NAME = "Test"
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        if payload := xml.attrib.get("payload"):
+            event_bus.notify(_TestEvent(payload))
+            return HandlingResult.success()
+        return HandlingResult.analyse()
+
+
+def test_XmlMessageDecoding() -> None:
+    assert_message(
+        _TestXmlMessage, '<ctl ret="ok" payload="test" />', _TestEvent("test")
+    )
+
+
+def test_XmlMessageFailure() -> None:
+    assert_message_failure(
+        _TestXmlMessage, '<ctl ret="ok" />', HandlingState.ANALYSE_LOGGED
+    )