[tcp] Fixes from code review

- Reserve the link's sockets, so the socket pool is sized for them
- Test the server role: accepting a client, receiving from it, and
  accepting the next after it disconnects
- Test a frame assembled across several reads, after a keep-alive

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/tcp/packet_interface/__init__.py

@@ -69,15 +69,19 @@ BASE_SCHEMA = packet_interface_schema(
     }
 )
 
-CONFIG_SCHEMA = cv.typed_schema(
-    {
-        "client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
-        "server": BASE_SCHEMA.extend(
-            {cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
-        ),
-    },
-    key=CONF_ROLE,
-    default_type="client",
+CONFIG_SCHEMA = cv.All(
+    cv.typed_schema(
+        {
+            "client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
+            "server": BASE_SCHEMA.extend(
+                {cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
+            ),
+        },
+        key=CONF_ROLE,
+        default_type="client",
+    ),
+    # A client uses one socket; a server one listener and one accepted socket
+    socket.consume_role_sockets("tcp.packet_interface"),
 )
 
 

tests/components/tcp/tcp_packet_interface.cpp

@@ -3,12 +3,14 @@
 #include <csignal>
 #include <cstdint>
 #include <memory>
+#include <netinet/in.h>
 #include <sys/socket.h>
 #include <unistd.h>
 #include <vector>
 
 #include "esphome/components/tcp/packet_interface/tcp_packet_interface.h"
 #include "esphome/core/application.h"
+#include "esphome/core/wake.h"
 
 #ifdef USE_HOST
 
@@ -156,6 +158,115 @@ TEST_F(TcpPacketInterfaceTest, NotSentWhileDown) {
   EXPECT_FALSE(this->a.send({1}));
 }
 
+TEST_F(TcpPacketInterfaceTest, FrameAssembledAcrossReads) {
+  End c;
+  int fds[2];
+  ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+  c.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+  c.link.loop();
+  // A keep-alive, then a 3 byte packet in three writes: half the length, the rest of it, then the payload
+  const uint8_t keep_alive[2] = {0x00, 0x00};
+  ASSERT_EQ(::write(fds[1], keep_alive, 2), 2);
+  c.link.loop();
+  const uint8_t length_low = 0x03;
+  ASSERT_EQ(::write(fds[1], &length_low, 1), 1);
+  c.link.loop();
+  const uint8_t length_high_and_start[2] = {0x00, 0xAA};
+  ASSERT_EQ(::write(fds[1], length_high_and_start, 2), 2);
+  c.link.loop();
+  EXPECT_TRUE(c.received.empty());
+  const uint8_t rest[2] = {0xBB, 0xCC};
+  ASSERT_EQ(::write(fds[1], rest, 2), 2);
+  c.link.loop();
+  EXPECT_EQ(c.received, (std::vector<Packet>{{0xAA, 0xBB, 0xCC}}));
+  EXPECT_TRUE(c.link.is_connected());
+  ::close(fds[1]);
+}
+
+// Waits in the main loop's select(), which marks the sockets that are ready to read
+static void wait_for_sockets() { esphome::internal::wakeable_delay(5); }
+
+// The server role, accepting connections on a loopback port
+class TcpServerTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    signal(SIGPIPE, SIG_IGN);
+    new (&App) Application;
+    App.pre_setup("tcp-tests", 9, "", 0);
+    set_time(1000);
+    this->port = free_port();
+    this->server.link.set_server(true);
+    this->server.link.set_port(this->port);
+  }
+  void TearDown() override {
+    this->server.link.on_shutdown();
+    for (int fd : this->clients)
+      ::close(fd);
+    App.~Application();
+  }
+  static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
+  static uint16_t free_port() {
+    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in addr {};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    socklen_t len = sizeof(addr);
+    EXPECT_EQ(::bind(fd, reinterpret_cast<struct sockaddr *>(&addr), len), 0);
+    EXPECT_EQ(::getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &len), 0);
+    ::close(fd);
+    return ntohs(addr.sin_port);
+  }
+  // Connects a client and runs the server's loop until it has accepted it
+  int connect_client() {
+    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in addr {};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = htons(this->port);
+    EXPECT_EQ(::connect(fd, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr)), 0);
+    this->clients.push_back(fd);
+    for (int i = 0; i != 100 && !this->server.link.is_connected(); i++) {
+      this->server.link.loop();
+      wait_for_sockets();
+    }
+    return fd;
+  }
+
+  End server;
+  uint16_t port{0};
+  std::vector<int> clients;
+};
+
+TEST_F(TcpServerTest, AcceptsClientAndReceives) {
+  this->server.link.loop();  // starts listening
+  int client = this->connect_client();
+  ASSERT_TRUE(this->server.link.is_connected());
+  EXPECT_EQ(this->server.states, std::vector<bool>{true});
+  const uint8_t frame[3] = {0x01, 0x00, 0x42};
+  ASSERT_EQ(::write(client, frame, sizeof(frame)), 3);
+  for (int i = 0; i != 100 && this->server.received.empty(); i++) {
+    this->server.link.loop();
+    wait_for_sockets();
+  }
+  EXPECT_EQ(this->server.received, std::vector<Packet>{{0x42}});
+}
+
+TEST_F(TcpServerTest, AcceptsNextClientAfterDisconnect) {
+  this->server.link.loop();
+  int first = this->connect_client();
+  ASSERT_TRUE(this->server.link.is_connected());
+  ::close(first);
+  this->clients.clear();
+  for (int i = 0; i != 100 && this->server.link.is_connected(); i++) {
+    this->server.link.loop();
+    wait_for_sockets();
+  }
+  EXPECT_EQ(this->server.states, (std::vector<bool>{true, false}));
+  this->connect_client();
+  EXPECT_TRUE(this->server.link.is_connected());
+  EXPECT_EQ(this->server.states, (std::vector<bool>{true, false, true}));
+}
+
 }  // namespace esphome::tcp::testing
 
 #endif

tests/unit_tests/test_tcp_packet_interface.py

@@ -3,8 +3,10 @@
 import pytest
 
 from esphome import config_validation as cv
+from esphome.components import socket
 from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
 from esphome.components.tcp import packet_interface as tcp_packet_interface
+from esphome.core import CORE
 
 
 def test_client_and_server(setup_core) -> None:
@@ -33,3 +35,19 @@ def test_auto_load_binary_sensor_for_connected(setup_core) -> None:
     assert tcp_packet_interface.AUTO_LOAD(config) == ["socket"]
     config["connected"] = {"name": "Link"}
     assert tcp_packet_interface.AUTO_LOAD(config) == ["socket", "binary_sensor"]
+
+
+@pytest.mark.parametrize(
+    ("role", "tcp", "listen"), [("client", 1, 0), ("server", 1, 1)]
+)
+def test_sockets_reserved(setup_core, role: str, tcp: int, listen: int) -> None:
+    """A client reserves one socket; a server also reserves a listener."""
+    config = {"role": role, "port": 6060}
+    if role == "client":
+        config["host"] = "192.168.1.50"
+    tcp_packet_interface.CONFIG_SCHEMA(config)
+    name = "tcp.packet_interface"
+    assert CORE.data[socket.KEY_SOCKET_CONSUMERS_TCP][name] == tcp
+    assert (
+        CORE.data.get(socket.KEY_SOCKET_CONSUMERS_TCP_LISTEN, {}).get(name, 0) == listen
+    )