Do not assume every PositionsEvent has a PositionType.DEEBOT inside (#955)

Co-authored-by: Robert Resch <robert@resch.dev>

Author: nanomad

deebot_client/device.py

@@ -77,18 +77,20 @@ async def on_pos(event: PositionsEvent) -> None:
             if self._state == StateEvent(State.DOCKED):
                 return
 
-            deebot = next(p for p in event.positions if p.type == PositionType.DEEBOT)
-
-            if deebot:
-                on_charger = filter(
-                    lambda p: p.type == PositionType.CHARGER
-                    and p.x == deebot.x
-                    and p.y == deebot.y,
-                    event.positions,
-                )
-                if on_charger:
-                    # deebot on charger so the status should be docked... Checking
-                    self.events.request_refresh(StateEvent)
+            deebot = next(
+                (p for p in event.positions if p.type == PositionType.DEEBOT), None
+            )
+
+            if deebot and any(
+                p
+                for p in event.positions
+                if p.type == PositionType.CHARGER
+                and p.x == deebot.x
+                and p.y == deebot.y
+            ):
+                # Deebot is on charger and the status is not docked
+                # Request refresh for the state
+                self.events.request_refresh(StateEvent)
 
         self.events.subscribe(PositionsEvent, on_pos)
 

tests/test_device.py

@@ -13,18 +13,21 @@
 from deebot_client.commands.xml import GetBatteryInfo
 from deebot_client.const import DataType
 from deebot_client.device import Device
-from deebot_client.events import AvailabilityEvent
+from deebot_client.events import AvailabilityEvent, StateEvent
+from deebot_client.events.map import Position, PositionsEvent
 from deebot_client.events.network import NetworkInfoEvent
 from deebot_client.hardware import get_static_device_info
 from deebot_client.messages.json import OnBattery
 from deebot_client.messages.xml import BatteryInfo
 from deebot_client.models import DeviceInfo, StaticDeviceInfo
 from deebot_client.mqtt_client import MqttClient, SubscriberInfo
+from deebot_client.rs.map import PositionType
 from tests.helpers import mock_static_device_info
 from tests.helpers.tasks import block_till_done
 
 if TYPE_CHECKING:
     from deebot_client.authentication import Authenticator
+    from deebot_client.event_bus import EventBus
     from deebot_client.message import Message
     from deebot_client.models import ApiDeviceInfo
 
@@ -277,3 +280,85 @@ async def on_status(event: AvailabilityEvent) -> None:
 
     # teardown bot
     await bot.teardown()
+
+
+@pytest.mark.parametrize(
+    ("pos_event", "expected_call"),
+    [
+        (PositionsEvent([]), False),
+        (PositionsEvent([Position(PositionType.CHARGER, 0, 0, 0)]), False),
+        (PositionsEvent([Position(PositionType.DEEBOT, 0, 0, 0)]), False),
+        (
+            PositionsEvent(
+                [
+                    Position(PositionType.DEEBOT, 0, 0, 0),
+                    Position(PositionType.CHARGER, 0, 0, 0),
+                ]
+            ),
+            True,
+        ),
+        (
+            PositionsEvent(
+                [
+                    Position(PositionType.CHARGER, 0, 0, 0),
+                    Position(PositionType.DEEBOT, 0, 0, 0),
+                ]
+            ),
+            True,
+        ),
+        (
+            PositionsEvent(
+                [
+                    Position(PositionType.CHARGER, 1, 0, 0),
+                    Position(PositionType.DEEBOT, 0, 0, 0),
+                ]
+            ),
+            False,
+        ),
+        (
+            PositionsEvent(
+                [
+                    Position(PositionType.DEEBOT, 0, 0, 0),
+                    Position(PositionType.CHARGER, 1, 0, 0),
+                ]
+            ),
+            False,
+        ),
+        (
+            PositionsEvent(
+                [
+                    Position(PositionType.DEEBOT, 0, 0, 0),
+                    Position(PositionType.CHARGER, 1, 0, 0),
+                    Position(PositionType.CHARGER, 0, 0, 0),
+                ]
+            ),
+            True,
+        ),
+    ],
+)
+async def test_onPos_device_handling(
+    authenticator: Authenticator,
+    device_info: DeviceInfo,
+    event_bus_mock: Mock,
+    event_bus: EventBus,
+    pos_event: PositionsEvent,
+    expected_call: bool,
+) -> None:
+    """Test the available check including if the status Event is fired correctly."""
+    with patch("deebot_client.device.EventBus", return_value=event_bus_mock):
+        bot = Device(device_info, authenticator)
+        mqtt_client = Mock(spec=MqttClient)
+        unsubscribe_mock = Mock(spec=Callable[[], None])
+        mqtt_client.subscribe.return_value = unsubscribe_mock
+        await bot.initialize(mqtt_client)
+
+    bot.events.notify(pos_event)
+    await block_till_done(event_bus._tasks)
+
+    if expected_call:
+        event_bus_mock.request_refresh.assert_called_once_with(StateEvent)
+    else:
+        event_bus_mock.request_refresh.assert_not_called()
+
+    # teardown bot
+    await bot.teardown()