[api] Derive decode cases from the type's wire type

decode_case() reads wire_type instead of taking it at every call, the
expression and the store statement are two small hooks that repeated
fields override, and message fields build one body. No generated case
declares a local any more, so the braced form and its test go; the
compiler rejects a jump over a local if one ever appears. The wire
type test now proves a dropped frame with an ordering marker instead
of assuming it, and shares StateWaiter.

tests/integration/test_api_decode_wire_types.py

@@ -3,7 +3,6 @@ not, skip unknown fields, and handle two byte tags, varints and length prefixes.
 
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Callable
 import struct
 
@@ -18,7 +17,7 @@ from aioesphomeapi import (
 import pytest
 
 from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient, encode_varint
-from .state_utils import InitialStateHelper, require_entity
+from .state_utils import InitialStateHelper, StateWaiter, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 SWITCH_COMMAND = MESSAGE_TYPE_OF[api_pb2.SwitchCommandRequest]
@@ -29,10 +28,6 @@ def tag(field: int, wire_type: int) -> bytes:
     return encode_varint((field << 3) | wire_type)
 
 
-def key_field(key: int) -> bytes:
-    return tag(1, WIRE_FIXED32) + struct.pack("<I", key)
-
-
 @pytest.mark.asyncio
 async def test_api_decode_wire_types(
     yaml_config: str,
@@ -40,7 +35,6 @@ async def test_api_decode_wire_types(
     api_client_connected: APIClientConnectedFactory,
     unused_tcp_port: int,
 ) -> None:
-    loop = asyncio.get_running_loop()
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
@@ -51,104 +45,79 @@ async def test_api_decode_wire_types(
         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"
 
-        waiters: list[
-            tuple[Callable[[EntityState], bool], asyncio.Future[EntityState]]
-        ] = []
-        initial = InitialStateHelper(entities)
+        switch_states: list[bool] = []
+        waiter = StateWaiter()
 
         def on_state(state: EntityState) -> None:
-            for pred, fut in waiters:
-                if not fut.done() and pred(state):
-                    fut.set_result(state)
-
-        async def expect(pred: Callable[[EntityState], bool]) -> EntityState:
-            fut: asyncio.Future[EntityState] = loop.create_future()
-            waiters.append((pred, fut))
-            try:
-                return await asyncio.wait_for(fut, 5.0)
-            finally:
-                waiters.remove((pred, fut))
+            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_field(switch.key) + tag(2, WIRE_VARINT) + b"\x01"
-        )
-        await expect(switch_is(True))
-        await raw.send_raw(
-            SWITCH_COMMAND, key_field(switch.key) + tag(2, WIRE_VARINT) + b"\x00"
-        )
-        await expect(switch_is(False))
+        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: a length delimited or fixed32
-        # "state" must not turn the switch on, and a varint key never matches an entity
+        # 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_field(switch.key) + tag(2, WIRE_LENGTH) + b"\x01\x01"
+            SWITCH_COMMAND, key + tag(2, WIRE_FIXED32) + b"\x01\x00\x00\x00"
         )
         await raw.send_raw(
-            SWITCH_COMMAND,
-            key_field(switch.key) + tag(2, WIRE_FIXED32) + b"\x01\x00\x00\x00",
+            SWITCH_COMMAND, tag(1, WIRE_VARINT) + encode_varint(switch.key) + on
         )
+        # A later command on the same connection proves the bad frames were fully consumed;
+        # the number state arriving means any switch state from them would already be here
+        client.number_command(number.key, -77.5)
+        await waiter.expect(number_is(-77.5))
+        assert len(switch_states) == seen
+
+        # An unknown field ahead of the known ones is skipped; field 200 needs a two byte tag
         await raw.send_raw(
-            SWITCH_COMMAND,
-            tag(1, WIRE_VARINT)
-            + encode_varint(switch.key)
-            + tag(2, WIRE_VARINT)
-            + b"\x01",
+            SWITCH_COMMAND, tag(200, WIRE_VARINT) + encode_varint(300) + key + on
         )
-        # An unknown field ahead of the known ones is skipped and the rest still decodes;
-        # field 200 needs a two byte tag
-        await raw.send_raw(
-            SWITCH_COMMAND,
-            tag(200, WIRE_VARINT)
-            + encode_varint(300)
-            + key_field(switch.key)
-            + tag(2, WIRE_VARINT)
-            + b"\x01",
-        )
-        state = await expect(switch_is(True))
-        assert state.state is True
-        await raw.send_raw(
-            SWITCH_COMMAND, key_field(switch.key) + tag(2, WIRE_VARINT) + b"\x00"
-        )
-        await expect(switch_is(False))
+        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 expect(
+        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 expect(
+        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 expect(
+        await waiter.expect(
             lambda s: (
                 isinstance(s, TextState) and s.key == text.key and s.state == long_text
             )
         )
-
-        # A negative fixed32 float
-        client.number_command(number.key, -77.5)
-        await expect(
-            lambda s: (
-                isinstance(s, NumberState) and s.key == number.key and s.state == -77.5
-            )
-        )