Fix mqtt valve is not resetting opening or closing state (#106240)

* Fix mqtt valve is not resetting opening or closing state

* Require state or position attr in JSON state update

* Do not change `_attr_is_closed` if valve reports a position

* Add comment, use tuple

* Call _update_state

homeassistant/components/mqtt/valve.py

@@ -95,6 +95,8 @@ DEFAULTS = {
     CONF_STATE_CLOSED: STATE_CLOSED,
 }
 
+RESET_CLOSING_OPENING = "reset_opening_closing"
+
 
 def _validate_and_add_defaults(config: ConfigType) -> ConfigType:
     """Validate config options and set defaults."""
@@ -218,10 +220,12 @@ class MqttValve(MqttEntity, ValveEntity):
 
     @callback
     def _update_state(self, state: str) -> None:
-        """Update the valve state based on static payload."""
+        """Update the valve state properties."""
-        self._attr_is_closed = state == STATE_CLOSED
         self._attr_is_opening = state == STATE_OPENING
         self._attr_is_closing = state == STATE_CLOSING
+        if self.reports_position:
+            return
+        self._attr_is_closed = state == STATE_CLOSED
 
     @callback
     def _process_binary_valve_update(
@@ -270,7 +274,11 @@ class MqttValve(MqttEntity, ValveEntity):
                     msg.topic,
                 )
             else:
-                self._attr_current_valve_position = min(max(percentage_payload, 0), 100)
+                percentage_payload = min(max(percentage_payload, 0), 100)
+                self._attr_current_valve_position = percentage_payload
+                # Reset closing and opening if the valve is fully opened or fully closed
+                if state is None and percentage_payload in (0, 100):
+                    state = RESET_CLOSING_OPENING
                 position_set = True
         if state_payload and state is None and not position_set:
             _LOGGER.warning(
@@ -301,10 +309,10 @@ class MqttValve(MqttEntity, ValveEntity):
         )
         def state_message_received(msg: ReceiveMessage) -> None:
             """Handle new MQTT state messages."""
-            payload_dict: Any = None
-            position_payload: Any = None
-            state_payload: Any = None
             payload = self._value_template(msg.payload)
+            payload_dict: Any = None
+            position_payload: Any = payload
+            state_payload: Any = payload
 
             if not payload:
                 _LOGGER.debug("Ignoring empty state message from '%s'", msg.topic)
@@ -312,12 +320,25 @@ class MqttValve(MqttEntity, ValveEntity):
 
             with suppress(*JSON_DECODE_EXCEPTIONS):
                 payload_dict = json_loads(payload)
-            if isinstance(payload_dict, dict) and "position" in payload_dict:
+                if isinstance(payload_dict, dict):
-                position_payload = payload_dict["position"]
+                    if self.reports_position and "position" not in payload_dict:
-            if isinstance(payload_dict, dict) and "state" in payload_dict:
+                        _LOGGER.warning(
-                state_payload = payload_dict["state"]
+                            "Missing required `position` attribute in json payload "
-            state_payload = payload if state_payload is None else state_payload
+                            "on topic '%s', got: %s",
-            position_payload = payload if position_payload is None else position_payload
+                            msg.topic,
+                            payload,
+                        )
+                        return
+                    if not self.reports_position and "state" not in payload_dict:
+                        _LOGGER.warning(
+                            "Missing required `state` attribute in json payload "
+                            " on topic '%s', got: %s",
+                            msg.topic,
+                            payload,
+                        )
+                        return
+                    position_payload = payload_dict.get("position")
+                    state_payload = payload_dict.get("state")
 
             if self._config[CONF_REPORTS_POSITION]:
                 self._process_position_valve_update(

tests/components/mqtt/test_valve.py

@@ -291,6 +291,113 @@ async def test_state_via_state_topic_through_position(
     assert state.attributes.get(ATTR_CURRENT_POSITION) == valve_position
 
 
+@pytest.mark.parametrize(
+    "hass_config",
+    [
+        {
+            mqtt.DOMAIN: {
+                valve.DOMAIN: {
+                    "name": "test",
+                    "state_topic": "state-topic",
+                    "command_topic": "command-topic",
+                    "reports_position": True,
+                }
+            }
+        }
+    ],
+)
+async def test_opening_closing_state_is_reset(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+) -> None:
+    """Test the controlling state via topic through position.
+
+    Test  a `opening` or `closing` state update is reset correctly after sequential updates.
+    """
+    await mqtt_mock_entry()
+
+    state = hass.states.get("valve.test")
+    assert state.state == STATE_UNKNOWN
+    assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    messages = [
+        ('{"position": 0, "state": "opening"}', STATE_OPENING, 0),
+        ('{"position": 50, "state": "opening"}', STATE_OPENING, 50),
+        ('{"position": 60}', STATE_OPENING, 60),
+        ('{"position": 100, "state": "opening"}', STATE_OPENING, 100),
+        ('{"position": 100, "state": null}', STATE_OPEN, 100),
+        ('{"position": 90, "state": "closing"}', STATE_CLOSING, 90),
+        ('{"position": 40}', STATE_CLOSING, 40),
+        ('{"position": 0}', STATE_CLOSED, 0),
+        ('{"position": 10}', STATE_OPEN, 10),
+        ('{"position": 0, "state": "opening"}', STATE_OPENING, 0),
+        ('{"position": 0, "state": "closing"}', STATE_CLOSING, 0),
+        ('{"position": 0}', STATE_CLOSED, 0),
+    ]
+
+    for message, asserted_state, valve_position in messages:
+        async_fire_mqtt_message(hass, "state-topic", message)
+
+        state = hass.states.get("valve.test")
+        assert state.state == asserted_state
+        assert state.attributes.get(ATTR_CURRENT_POSITION) == valve_position
+
+
+@pytest.mark.parametrize(
+    ("hass_config", "message", "err_message"),
+    [
+        (
+            {
+                mqtt.DOMAIN: {
+                    valve.DOMAIN: {
+                        "name": "test",
+                        "state_topic": "state-topic",
+                        "command_topic": "command-topic",
+                        "reports_position": False,
+                    }
+                }
+            },
+            '{"position": 0}',
+            "Missing required `state` attribute in json payload",
+        ),
+        (
+            {
+                mqtt.DOMAIN: {
+                    valve.DOMAIN: {
+                        "name": "test",
+                        "state_topic": "state-topic",
+                        "command_topic": "command-topic",
+                        "reports_position": True,
+                    }
+                }
+            },
+            '{"state": "opening"}',
+            "Missing required `position` attribute in json payload",
+        ),
+    ],
+)
+async def test_invalid_state_updates(
+    hass: HomeAssistant,
+    mqtt_mock_entry: MqttMockHAClientGenerator,
+    caplog: pytest.LogCaptureFixture,
+    message: str,
+    err_message: str,
+) -> None:
+    """Test the controlling state via topic through position.
+
+    Test  a `opening` or `closing` state update is reset correctly after sequential updates.
+    """
+    await mqtt_mock_entry()
+
+    state = hass.states.get("valve.test")
+    assert state.state == STATE_UNKNOWN
+    assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "state-topic", message)
+    state = hass.states.get("valve.test")
+    assert err_message in caplog.text
+
+
 @pytest.mark.parametrize(
     "hass_config",
     [