[socket] Add an IPv4 lookup next to set_sockaddr (#19909)

Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>

tests/integration/fixtures/external_components/ipv4_resolve_test_component/__init__.py

@@ -0,0 +1,22 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.types import ConfigType
+
+AUTO_LOAD = ["socket"]
+
+ipv4_resolve_test_component_ns = cg.esphome_ns.namespace("ipv4_resolve_test_component")
+Ipv4ResolveTestComponent = ipv4_resolve_test_component_ns.class_(
+    "Ipv4ResolveTestComponent", cg.Component
+)
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(Ipv4ResolveTestComponent),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/ipv4_resolve_test_component/ipv4_resolve_test_component.cpp

@@ -0,0 +1,46 @@
+#include "ipv4_resolve_test_component.h"
+#include "esphome/components/socket/ipv4_resolve.h"
+#include "esphome/core/log.h"
+
+#include <cstring>
+
+namespace esphome::ipv4_resolve_test_component {
+
+static const char *const TAG = "ipv4_resolve_test";
+
+static bool check_sockaddr(socket::Ipv4Resolve &lookup, uint16_t port, uint32_t expected) {
+  struct sockaddr_storage addr {};
+  socklen_t len = lookup.to_sockaddr(reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr), port);
+  if (len != sizeof(sockaddr_in)) {
+    return false;
+  }
+  auto *in = reinterpret_cast<sockaddr_in *>(&addr);
+  return in->sin_family == AF_INET && ntohs(in->sin_port) == port && in->sin_addr.s_addr == htonl(expected);
+}
+
+void Ipv4ResolveTestComponent::setup() {
+  ESP_LOGI(TAG, "IPv4 resolve test starting");
+
+  socket::Ipv4Resolve lookup;
+  struct sockaddr_storage addr {};
+
+  lookup.start("192.168.1.1", 1, TAG);
+  bool ok = lookup.ready() && check_sockaddr(lookup, 6053, 0xC0A80101);
+  ESP_LOGI(TAG, "Literal resolve: %s", ok ? LOG_STR_LITERAL("PASSED") : LOG_STR_LITERAL("FAILED"));
+
+  lookup.forget();
+  ok = !lookup.ready() && lookup.to_sockaddr(reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr), 80) == 0;
+  ESP_LOGI(TAG, "Forget drops address: %s", ok ? LOG_STR_LITERAL("PASSED") : LOG_STR_LITERAL("FAILED"));
+
+  lookup.start("::1", 443, TAG);
+  ok = !lookup.ready() && lookup.consume_failure();
+  ESP_LOGI(TAG, "IPv6 literal rejected: %s", ok ? LOG_STR_LITERAL("PASSED") : LOG_STR_LITERAL("FAILED"));
+
+  lookup.start("localhost", 6053, TAG);
+  ok = lookup.ready() && check_sockaddr(lookup, 6053, 0x7F000001);
+  ESP_LOGI(TAG, "Hostname resolve: %s", ok ? LOG_STR_LITERAL("PASSED") : LOG_STR_LITERAL("FAILED"));
+
+  ESP_LOGI(TAG, "IPv4 resolve test complete");
+}
+
+}  // namespace esphome::ipv4_resolve_test_component

tests/integration/fixtures/external_components/ipv4_resolve_test_component/ipv4_resolve_test_component.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include "esphome/core/component.h"
+
+namespace esphome::ipv4_resolve_test_component {
+
+class Ipv4ResolveTestComponent : public Component {
+ public:
+  void setup() override;
+};
+
+}  // namespace esphome::ipv4_resolve_test_component

tests/integration/fixtures/socket_ipv4_resolve.yaml

@@ -0,0 +1,17 @@
+esphome:
+  name: socket-ipv4-resolve-test
+
+host:
+
+api:
+
+logger:
+  level: INFO
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [ipv4_resolve_test_component]
+
+ipv4_resolve_test_component:

tests/integration/test_socket_ipv4_resolve.py

@@ -0,0 +1,51 @@
+"""Integration test for the socket Ipv4Resolve helper on host."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+CHECKS = (
+    "Literal resolve",
+    "Forget drops address",
+    "IPv6 literal rejected",
+    "Hostname resolve",
+)
+
+
+@pytest.mark.asyncio
+async def test_socket_ipv4_resolve(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Exercise Ipv4Resolve literals, forget, failure, and getaddrinfo on host."""
+    test_complete = asyncio.Event()
+    results: dict[str, bool] = {}
+
+    def on_log_line(line: str) -> None:
+        if "IPv4 resolve test complete" in line:
+            test_complete.set()
+            return
+        for check in CHECKS:
+            if f"{check}:" in line:
+                results[check] = "PASSED" in line
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "socket-ipv4-resolve-test"
+
+        try:
+            await asyncio.wait_for(test_complete.wait(), timeout=10.0)
+        except TimeoutError:
+            pytest.fail("IPv4 resolve test timed out")
+
+    for check in CHECKS:
+        assert results.get(check), f"{check} check failed or never ran"