Select the dial-back target with an explicit hello flag instead of guessing from client info

tests/component_tests/api/test_outgoing_connection.py

@@ -9,6 +9,7 @@ from esphome.components.api import CONFIG_SCHEMA
 from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
 import esphome.config_validation as cv
 from esphome.const import PlatformFramework
+from esphome.core import CORE
 from esphome.types import ConfigType
 from tests.component_tests.types import SetCoreConfigCallable
 
@@ -23,15 +24,17 @@ def _api_config(outgoing: ConfigType, *, encryption: bool = True) -> ConfigType:
     return config
 
 
-def test_outgoing_connection_generates_setters(
+def test_outgoing_connection_generates_defines(
     generate_main: Callable[[str | Path], str],
 ) -> None:
-    """A valid config emits the setters with defaults applied."""
-    main_cpp = generate_main("tests/component_tests/api/test_outgoing_connection.yaml")
+    """A valid config emits the compile-time defines with defaults applied."""
+    generate_main("tests/component_tests/api/test_outgoing_connection.yaml")
 
-    assert 'set_outgoing_connection_host("192.168.1.2")' in main_cpp
-    assert "set_outgoing_connection_port(6054)" in main_cpp
-    assert "set_outgoing_connection_delay(60000)" in main_cpp
+    defines = {define.name: define.value for define in CORE.defines}
+    assert "USE_API_OUTGOING_CONNECTION" in defines
+    assert str(defines["API_OUTGOING_CONNECTION_HOST"]) == '"192.168.1.2"'
+    assert str(defines["API_OUTGOING_CONNECTION_PORT"]) == "6054"
+    assert str(defines["API_OUTGOING_CONNECTION_DELAY"]) == "60000"
 
 
 def test_outgoing_connection_defaults(
@@ -69,5 +72,5 @@ def test_outgoing_connection_rejects_hostnames(
     set_core_config: SetCoreConfigCallable,
 ) -> None:
     set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
-    with pytest.raises(cv.Invalid, match="must be an IP address"):
+    with pytest.raises(cv.Invalid, match="not a valid IP address"):
         CONFIG_SCHEMA(_api_config({"host": "homeassistant.local"}))

tests/integration/test_api_outgoing_connection.py

@@ -1,11 +1,13 @@
 """Integration tests for the api outgoing_connection option.
 
-The device dials out to the test's listener when no client with a state
-subscription is connected. The listener plays the Home Assistant side over the
-accepted socket using aioesphomeapi's sans-IO Noise handshake: the device
-sends its server hello first so the listener could pick the right key, and the
-NNpsk0 handshake then verifies both sides. Protocol roles stay unchanged, so
-the client speaks exactly the same frames as over a normal connection.
+The device dials out to the test's listener when no dial-back target client is
+connected. The listener plays the Home Assistant side over the accepted socket
+using aioesphomeapi's sans-IO Noise handshake: the device sends its server
+hello first so the listener could pick the right key, and the NNpsk0 handshake
+then verifies both sides. Protocol roles stay unchanged, so the client speaks
+exactly the same frames as over a normal connection. A client becomes the
+remembered dial-back target by setting the outgoing_connection_target flag in
+its hello.
 """
 
 from __future__ import annotations
@@ -14,15 +16,18 @@ import asyncio
 import socket
 from typing import Any
 
-from aioesphomeapi import APIClient, api_pb2
+from aioesphomeapi import api_pb2
 import pytest
 
 from .raw_api_client import MESSAGE_TYPE_OF
-from .types import APIClientConnectedFactory, RunCompiledFunction
+from .types import RunCompiledFunction
 
 KEY = "bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU="
 DEVICE_NAME = "outgoing-conn-test"
 HA_CLIENT_INFO = "Home Assistant 2026.8.0"
+# HelloRequest field 4 (outgoing_connection_target) as raw protobuf bytes; the
+# installed aioesphomeapi's api_pb2 predates the field, so append it manually.
+HELLO_TARGET_FLAG = b"\x20\x01"
 
 
 @pytest.fixture(autouse=True)
@@ -43,76 +48,78 @@ async def _read_frame(reader: asyncio.StreamReader, timeout: float = 10.0) -> by
     )
 
 
-async def _serve_home_assistant(
-    listener: socket.socket, *, subscribe_states: bool = False
+def _check_server_hello(server_hello: bytes) -> None:
+    assert server_hello[0] == 0x01, "Bad chosen proto in server hello"
+    name, mac, _rest = server_hello[1:].split(b"\x00", 2)
+    assert name.decode() == DEVICE_NAME
+    assert len(mac) == 12, f"Expected bare MAC, got {mac!r}"
+
+
+async def _run_ha_session(
+    reader: asyncio.StreamReader,
+    writer: asyncio.StreamWriter,
+    *,
+    device_dialed_out: bool,
 ) -> None:
-    """Accept one dial-in from the device and run the client side over it."""
+    """Handshake and exchange the usual first messages as Home Assistant would."""
     # Lazy import per the module's own contract (pulls in the noise stack)
     from aioesphomeapi.noise import NoiseHandshake
 
+    if device_dialed_out:
+        # On an outgoing connection the device announces itself first so the
+        # peer can pick the matching key before its PSK-mixed first message.
+        _check_server_hello(await _read_frame(reader))
+
+    handshake = NoiseHandshake(KEY, b"NoiseAPIInit\x00\x00")
+    writer.write(b"\x01\x00\x00" + _frame(b"\x00" + handshake.write_message()))
+    await writer.drain()
+
+    if not device_dialed_out:
+        _check_server_hello(await _read_frame(reader))
+
+    reply = await _read_frame(reader)
+    assert reply[0] == 0, f"Handshake rejected: {reply[1:].decode(errors='replace')}"
+    handshake.read_message(reply[1:])
+    encrypt_cipher, decrypt_cipher = handshake.get_ciphers()
+
+    async def transact(msg: Any, response_cls: Any, extra_payload: bytes = b"") -> Any:
+        msg_type = MESSAGE_TYPE_OF[type(msg)]
+        payload = msg.SerializeToString() + extra_payload
+        plaintext = (
+            bytes(
+                (msg_type >> 8, msg_type & 0xFF, len(payload) >> 8, len(payload) & 0xFF)
+            )
+            + payload
+        )
+        writer.write(_frame(encrypt_cipher.encrypt(plaintext)))
+        await writer.drain()
+        want = MESSAGE_TYPE_OF[response_cls]
+        while True:
+            plain = decrypt_cipher.decrypt(await _read_frame(reader))
+            if ((plain[0] << 8) | plain[1]) == want:
+                response = response_cls()
+                response.ParseFromString(bytes(plain[4:]))
+                return response
+
+    # Declare this client a dial-back target in the hello
+    await transact(
+        api_pb2.HelloRequest(client_info=HA_CLIENT_INFO),
+        api_pb2.HelloResponse,
+        extra_payload=HELLO_TARGET_FLAG,
+    )
+    device_info = await transact(
+        api_pb2.DeviceInfoRequest(), api_pb2.DeviceInfoResponse
+    )
+    assert device_info.name == DEVICE_NAME
+
+
+async def _serve_home_assistant(listener: socket.socket) -> None:
+    """Accept one dial-in from the device and run the client side over it."""
     loop = asyncio.get_running_loop()
     conn, _ = await asyncio.wait_for(loop.sock_accept(listener), timeout=30)
     reader, writer = await asyncio.open_connection(sock=conn)
     try:
-        # On an outgoing connection the device announces itself first so the
-        # peer can pick the matching key before its PSK-mixed first message.
-        server_hello = await _read_frame(reader)
-        assert server_hello[0] == 0x01, "Bad chosen proto in server hello"
-        name, mac, _rest = server_hello[1:].split(b"\x00", 2)
-        assert name.decode() == DEVICE_NAME
-        assert len(mac) == 12, f"Expected bare MAC, got {mac!r}"
-
-        # Normal NNpsk0 handshake: client hello plus PSK-mixed message one,
-        # then the device's response completes it and proves the key matches.
-        handshake = NoiseHandshake(KEY, b"NoiseAPIInit\x00\x00")
-        writer.write(b"\x01\x00\x00" + _frame(b"\x00" + handshake.write_message()))
-        await writer.drain()
-        reply = await _read_frame(reader)
-        assert reply[0] == 0, (
-            f"Handshake rejected: {reply[1:].decode(errors='replace')}"
-        )
-        handshake.read_message(reply[1:])
-        encrypt_cipher, decrypt_cipher = handshake.get_ciphers()
-
-        async def transact(msg: Any, response_cls: Any | None) -> Any:
-            msg_type = MESSAGE_TYPE_OF[type(msg)]
-            payload = msg.SerializeToString()
-            plaintext = (
-                bytes(
-                    (
-                        msg_type >> 8,
-                        msg_type & 0xFF,
-                        len(payload) >> 8,
-                        len(payload) & 0xFF,
-                    )
-                )
-                + payload
-            )
-            writer.write(_frame(encrypt_cipher.encrypt(plaintext)))
-            await writer.drain()
-            if response_cls is None:
-                return None
-            want = MESSAGE_TYPE_OF[response_cls]
-            while True:
-                plain = decrypt_cipher.decrypt(await _read_frame(reader))
-                if ((plain[0] << 8) | plain[1]) == want:
-                    response = response_cls()
-                    response.ParseFromString(bytes(plain[4:]))
-                    return response
-
-        await transact(
-            api_pb2.HelloRequest(client_info=HA_CLIENT_INFO), api_pb2.HelloResponse
-        )
-        device_info = await transact(
-            api_pb2.DeviceInfoRequest(), api_pb2.DeviceInfoResponse
-        )
-        assert device_info.name == DEVICE_NAME
-
-        if subscribe_states:
-            await transact(api_pb2.SubscribeStatesRequest(), None)
-            # No entities are configured, so there is nothing to wait for;
-            # give the device a moment to process the subscription.
-            await asyncio.sleep(0.5)
+        await _run_ha_session(reader, writer, device_dialed_out=True)
     finally:
         writer.close()
 
@@ -132,7 +139,7 @@ async def test_api_outgoing_connection(
     try:
         yaml = yaml_config.replace("OUTGOING_PORT", str(port))
         async with run_compiled(yaml):
-            await _serve_home_assistant(listener, subscribe_states=True)
+            await _serve_home_assistant(listener)
     finally:
         listener.close()
 
@@ -141,10 +148,10 @@ async def test_api_outgoing_connection(
 async def test_api_outgoing_connection_remembered(
     yaml_config: str,
     run_compiled: RunCompiledFunction,
-    api_client_connected: APIClientConnectedFactory,
+    unused_tcp_port: int,
 ) -> None:
-    """No host configured: the device remembers the Home Assistant client that
-    connected inbound and dials that address after a restart."""
+    """No host configured: the device remembers the client whose hello carried
+    the dial-back flag and dials that address after a restart."""
     listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     # Bound but not yet listening: dial attempts in the first phase are
     # refused, exercising the retry path without queueing stale connections.
@@ -154,17 +161,14 @@ async def test_api_outgoing_connection_remembered(
     try:
         yaml = yaml_config.replace("OUTGOING_PORT", str(port))
 
-        async with (
-            run_compiled(yaml),
-            api_client_connected(noise_psk=KEY, client_info=HA_CLIENT_INFO) as client,
-        ):
-            client: APIClient
-            device_info = await client.device_info()
-            assert device_info.name == DEVICE_NAME
-            # Subscribing to states marks this client as Home Assistant;
-            # the device persists the peer address for dial-back.
-            client.subscribe_states(lambda state: None)
-            await asyncio.sleep(1.0)
+        async with run_compiled(yaml):
+            # Connect inbound with the dial-back flag; the device persists the
+            # peer address during the hello.
+            reader, writer = await asyncio.open_connection("127.0.0.1", unused_tcp_port)
+            try:
+                await _run_ha_session(reader, writer, device_dialed_out=False)
+            finally:
+                writer.close()
 
         # Restart with the same preferences: the device now dials the
         # remembered address on its own.