[api] Collapse the three protobuf decode virtuals into one (#19016)

tests/integration/fixtures/api_decode_wire_types.yaml

@@ -0,0 +1,43 @@
+esphome:
+  name: api-decode-wire-types-test
+host:
+api:
+logger:
+  level: DEBUG
+
+switch:
+  - platform: template
+    name: "Wire Switch"
+    optimistic: true
+
+output:
+  - platform: template
+    id: wire_dim
+    type: float
+    write_action:
+      - lambda: ""
+
+light:
+  - platform: monochromatic
+    name: "Wire Light"
+    output: wire_dim
+    default_transition_length: 0s
+    effects:
+      - pulse:
+          name: Pulse
+
+text:
+  - platform: template
+    name: "Wire Text"
+    optimistic: true
+    mode: text
+    min_length: 0
+    max_length: 255
+
+number:
+  - platform: template
+    name: "Wire Number"
+    optimistic: true
+    min_value: -1000
+    max_value: 1000
+    step: 0.5

tests/integration/raw_api_client.py

@@ -125,11 +125,12 @@ class RawApiClient:
         await self.read_until_frame(MESSAGE_TYPE_OF[api_pb2.HelloResponse])
 
     async def send_message(self, msg: message.Message) -> None:
+        await self.send_raw(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString())
+
+    async def send_raw(self, msg_type: int, payload: bytes) -> None:
+        """Send a frame with a hand built payload, for shapes protobuf will not serialize."""
         loop = asyncio.get_running_loop()
-        await loop.sock_sendall(
-            self._sock,
-            encode_frame(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString()),
-        )
+        await loop.sock_sendall(self._sock, encode_frame(msg_type, payload))
 
     async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None:
         """Read until at least one frame of msg_type has been received."""

tests/integration/test_api_decode_wire_types.py

@@ -0,0 +1,142 @@
+"""decode_field() must take fields that match their declared wire type, drop the ones that do
+not, skip unknown fields, and handle two byte tags, varints and length prefixes."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import struct
+
+from aioesphomeapi import (
+    EntityState,
+    LightState,
+    NumberState,
+    SwitchState,
+    TextState,
+    api_pb2,
+)
+import pytest
+
+from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient, encode_varint
+from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+SWITCH_COMMAND = MESSAGE_TYPE_OF[api_pb2.SwitchCommandRequest]
+WIRE_VARINT, WIRE_LENGTH, WIRE_FIXED32 = 0, 2, 5
+
+
+def tag(field: int, wire_type: int) -> bytes:
+    return encode_varint((field << 3) | wire_type)
+
+
+@pytest.mark.asyncio
+async def test_api_decode_wire_types(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+    unused_tcp_port: int,
+) -> None:
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+        RawApiClient(unused_tcp_port) as raw,
+    ):
+        entities, _ = await client.list_entities_services()
+        switch = require_entity(entities, "wire_switch")
+        light = require_entity(entities, "wire_light")
+        text = require_entity(entities, "wire_text")
+        number = require_entity(entities, "wire_number")
+        key = tag(1, WIRE_FIXED32) + struct.pack("<I", switch.key)
+        on, off = tag(2, WIRE_VARINT) + b"\x01", tag(2, WIRE_VARINT) + b"\x00"
+
+        switch_states: list[bool] = []
+        waiter = StateWaiter()
+
+        def on_state(state: EntityState) -> None:
+            if isinstance(state, SwitchState) and state.key == switch.key:
+                switch_states.append(state.state)
+            waiter.on_state(state)
+
+        def switch_is(value: bool) -> Callable[[EntityState], bool]:
+            return lambda s: (
+                isinstance(s, SwitchState) and s.key == switch.key and s.state is value
+            )
+
+        def number_is(value: float) -> Callable[[EntityState], bool]:
+            return lambda s: (
+                isinstance(s, NumberState) and s.key == number.key and s.state == value
+            )
+
+        initial = InitialStateHelper(entities)
+        client.subscribe_states(initial.on_state_wrapper(on_state))
+        await initial.wait_for_initial_states()
+        await raw.connect()
+
+        # A well formed command: fixed32 key, varint state
+        await raw.send_raw(SWITCH_COMMAND, key + on)
+        await waiter.expect(switch_is(True))
+        await raw.send_raw(SWITCH_COMMAND, key + off)
+        await waiter.expect(switch_is(False))
+
+        # The same field with the wrong wire type is dropped, and a varint key never matches an
+        # entity; each of these would turn the switch on if the payload were read as a varint
+        seen = len(switch_states)
+        await raw.send_raw(SWITCH_COMMAND, key + tag(2, WIRE_LENGTH) + b"\x01\x01")
+        await raw.send_raw(
+            SWITCH_COMMAND, key + tag(2, WIRE_FIXED32) + b"\x01\x00\x00\x00"
+        )
+        await raw.send_raw(
+            SWITCH_COMMAND, tag(1, WIRE_VARINT) + encode_varint(switch.key) + on
+        )
+        # Ordered on the raw socket itself: this frame cannot be parsed before the bad ones, so
+        # the only switch state since the marker must be the one it produces
+        await raw.send_raw(SWITCH_COMMAND, key + on)
+        await waiter.expect(switch_is(True), label="switch on after wrong wire types")
+        assert switch_states[seen:] == [True]
+        await raw.send_raw(SWITCH_COMMAND, key + off)
+        await waiter.expect(switch_is(False))
+
+        # Truncated bodies stop the decode loop without taking the connection down: a tag with its
+        # continuation bit set and nothing after it, a length prefix past the end of the payload,
+        # and a fixed32 with two of its four bytes
+        seen = len(switch_states)
+        await raw.send_raw(SWITCH_COMMAND, key + b"\x80")
+        await raw.send_raw(SWITCH_COMMAND, key + tag(2, WIRE_LENGTH) + b"\x7f" + b"ab")
+        await raw.send_raw(SWITCH_COMMAND, tag(1, WIRE_FIXED32) + b"\x01\x02")
+        await raw.send_raw(SWITCH_COMMAND, key + on)
+        await waiter.expect(switch_is(True), label="switch on after truncated frames")
+        assert switch_states[seen:] == [True]
+        await raw.send_raw(SWITCH_COMMAND, key + off)
+        await waiter.expect(switch_is(False))
+
+        # A negative number goes through the fixed32 float path of a normal client
+        client.number_command(number.key, -77.5)
+        await waiter.expect(number_is(-77.5))
+
+        # An unknown field ahead of the known ones is skipped; field 200 needs a two byte tag
+        await raw.send_raw(
+            SWITCH_COMMAND, tag(200, WIRE_VARINT) + encode_varint(300) + key + on
+        )
+        await waiter.expect(switch_is(True))
+
+        # Two byte tags (effect fields 18 and 19) and a two byte varint (300 ms transition)
+        client.light_command(
+            light.key, state=True, brightness=0.5, transition_length=0.3, effect="Pulse"
+        )
+        await waiter.expect(
+            lambda s: (
+                isinstance(s, LightState) and s.key == light.key and s.effect == "Pulse"
+            )
+        )
+        client.light_command(light.key, effect="None", state=False)
+        await waiter.expect(
+            lambda s: isinstance(s, LightState) and s.key == light.key and not s.state
+        )
+
+        # A string whose length prefix needs two varint bytes
+        long_text = "w" * 200
+        client.text_command(text.key, long_text)
+        await waiter.expect(
+            lambda s: (
+                isinstance(s, TextState) and s.key == text.key and s.state == long_text
+            )
+        )