Implement ResetLifeSpan

Author: nanomad

deebot_client/commands/xml/__init__.py

@@ -12,7 +12,7 @@
 from .clean_logs import GetCleanLogs
 from .error import GetError
 from .fan_speed import GetFanSpeed
-from .life_span import GetLifeSpan
+from .life_span import GetLifeSpan, ResetLifeSpan
 from .play_sound import PlaySound
 from .pos import GetPos
 from .stats import GetCleanSum
@@ -31,6 +31,7 @@
     "GetLifeSpan",
     "GetPos",
     "PlaySound",
+    "ResetLifeSpan",
 ]
 
 # fmt: off
@@ -46,6 +47,7 @@
     GetLifeSpan,
     GetPos,
     PlaySound,
+    ResetLifeSpan,
 ]
 # fmt: on
 

deebot_client/commands/xml/life_span.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import LifeSpan, LifeSpanEvent
-from deebot_client.message import HandlingResult
+from deebot_client.message import HandlingResult, HandlingState
 
-from .common import XmlCommandWithMessageHandling
+from .common import ExecuteCommand, XmlCommandMqttP2P, XmlCommandWithMessageHandling
 
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
@@ -20,8 +22,11 @@ class GetLifeSpan(XmlCommandWithMessageHandling):
 
     NAME = "GetLifeSpan"
 
-    def __init__(self, life_span: LifeSpan) -> None:
-        super().__init__({"type": life_span.xml_value})
+    def __init__(self, life_span: LifeSpan | str) -> None:
+        xml_value = (
+            life_span.xml_value if isinstance(life_span, LifeSpan) else life_span
+        )
+        super().__init__({"type": xml_value})
 
     @classmethod
     def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
@@ -47,3 +52,24 @@ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
             LifeSpanEvent(LifeSpan.from_xml(component_type), percent, left)
         )
         return HandlingResult.success()
+
+
+class ResetLifeSpan(ExecuteCommand, XmlCommandMqttP2P):
+    """ResetLifeSpan command."""
+
+    NAME = "ResetLifeSpan"
+    _mqtt_params = MappingProxyType({"type": InitParam(LifeSpan, "life_span")})
+
+    def __init__(self, life_span: LifeSpan | str) -> None:
+        xml_value = (
+            life_span.xml_value if isinstance(life_span, LifeSpan) else life_span
+        )
+        super().__init__({"type": xml_value})
+
+    def _handle_mqtt_p2p(
+        self, event_bus: EventBus, response: dict[str, Any] | str
+    ) -> None:
+        """Handle response received over the mqtt channel "p2p"."""
+        result = self.handle(event_bus, response)
+        if result.state == HandlingState.SUCCESS:
+            event_bus.request_refresh(LifeSpanEvent)

tests/commands/xml/test_life_span.py

@@ -1,20 +1,38 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
 import pytest
 
 from deebot_client.command import CommandResult
-from deebot_client.commands.xml import GetLifeSpan
+from deebot_client.commands.xml import GetLifeSpan, ResetLifeSpan
 from deebot_client.events import LifeSpan, LifeSpanEvent
 from deebot_client.message import HandlingState
 from tests.commands import assert_command
 
-from . import get_request_xml
+from . import (
+    assert_execute_command,
+    get_failure_body,
+    get_request_xml,
+    get_success_body,
+)
+
+if TYPE_CHECKING:
+    from deebot_client.event_bus import EventBus
 
 
 @pytest.mark.parametrize(
     ("component_type", "lifespan_type", "left", "total", "expected_event"),
     [
         ("Brush", LifeSpan.BRUSH, 50, 100, LifeSpanEvent(LifeSpan.BRUSH, 50, 50)),
+        (
+            "Brush",
+            LifeSpan.BRUSH.xml_value,
+            50,
+            100,
+            LifeSpanEvent(LifeSpan.BRUSH, 50, 50),
+        ),
         (
             "DustCaseHeap",
             LifeSpan.DUST_CASE_HEAP,
@@ -33,15 +51,15 @@
 )
 async def test_get_life_span(
     component_type: str,
-    lifespan_type: LifeSpan,
+    lifespan_type: LifeSpan | str,
     left: int,
     total: int,
     expected_event: LifeSpanEvent,
 ) -> None:
-    json = get_request_xml(
+    xml = get_request_xml(
         f"<ctl ret='ok' type='{component_type}' left='{left}' total='{total}'/>"
     )
-    await assert_command(GetLifeSpan(lifespan_type), json, expected_event)
+    await assert_command(GetLifeSpan(lifespan_type), xml, expected_event)
 
 
 @pytest.mark.parametrize(
@@ -57,3 +75,42 @@ async def test_get_life_span_error(xml: str) -> None:
         None,
         command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
     )
+
+
+@pytest.mark.parametrize(
+    ("command", "args"),
+    [
+        (ResetLifeSpan(LifeSpan.FILTER), {"type": LifeSpan.FILTER.xml_value}),
+        (ResetLifeSpan(LifeSpan.FILTER.xml_value), {"type": LifeSpan.FILTER.xml_value}),
+        (
+            ResetLifeSpan.create_from_mqtt(b'<ctl type="Brush" />'),
+            {"type": LifeSpan.BRUSH.xml_value},
+        ),
+    ],
+)
+async def test_ResetLifeSpan(command: ResetLifeSpan, args: dict[str, str]) -> None:
+    await assert_execute_command(command, args)
+
+
+def test_ResetLifeSpan_invokes_refresh(event_bus: EventBus) -> None:
+    command = ResetLifeSpan(LifeSpan.FILTER)
+    success_response = get_success_body()
+
+    with patch.object(
+        event_bus, "request_refresh", return_value=None
+    ) as mock_request_refresh:
+        command.handle_mqtt_p2p(event_bus, success_response)
+
+    mock_request_refresh.assert_called_with(LifeSpanEvent)
+
+
+def test_ResetLifeSpan_not_invokes_refresh(event_bus: EventBus) -> None:
+    command = ResetLifeSpan(LifeSpan.FILTER)
+    failure_response = get_failure_body()
+
+    with patch.object(
+        event_bus, "request_refresh", return_value=None
+    ) as mock_request_refresh:
+        command.handle_mqtt_p2p(event_bus, failure_response)
+
+    mock_request_refresh.assert_not_called()