[tcp_uart] Report a failed flush when the flush drops the link (#20013)

esphome/components/tcp_uart/tcp_uart.cpp

@@ -107,8 +107,13 @@ bool TcpUart::read_array(uint8_t *data, size_t len) {
 }
 
 uart::UARTFlushResult TcpUart::flush() {
-  return this->link_.flush_tx() ? uart::UARTFlushResult::UART_FLUSH_RESULT_SUCCESS
-                                : uart::UARTFlushResult::UART_FLUSH_RESULT_TIMEOUT;
+  bool emptied = this->link_.flush_tx();
+  if (!this->link_.connected()) {
+    // A down link cannot have delivered anything, whether this flush dropped
+    // it or an earlier loop() write did.
+    return uart::UARTFlushResult::UART_FLUSH_RESULT_FAILED;
+  }
+  return emptied ? uart::UARTFlushResult::UART_FLUSH_RESULT_SUCCESS : uart::UARTFlushResult::UART_FLUSH_RESULT_TIMEOUT;
 }
 
 }  // namespace esphome::tcp_uart

tests/components/socket/test_tcp_client_link_host.cpp

@@ -0,0 +1,77 @@
+#include <gtest/gtest.h>
+
+#include <csignal>
+#include <cstring>
+#include <memory>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "esphome/components/socket/tcp_client_link.h"
+
+#ifdef USE_HOST
+
+namespace esphome::socket::testing {
+
+class LinkPeer {
+ public:
+  LinkPeer() {
+    // EPIPE must come back as an errno, not a signal.
+    signal(SIGPIPE, SIG_IGN);
+    int fds[2];
+    EXPECT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+    this->peer_fd_ = fds[1];
+    this->link_.set_host("peer");
+    this->link_.set_port(1);
+    this->link_.begin("link_test");
+    this->link_.adopt(std::make_unique<Socket>(fds[0]));
+  }
+  ~LinkPeer() {
+    if (this->peer_fd_ >= 0) {
+      ::close(this->peer_fd_);
+    }
+    this->link_.close();
+  }
+  void close_peer() {
+    ::close(this->peer_fd_);
+    this->peer_fd_ = -1;
+  }
+
+  TcpClientLink link_;
+  int peer_fd_{-1};
+};
+
+TEST(TcpClientLink, AdoptedSocketFlushesQueuedBytes) {
+  LinkPeer p;
+  ASSERT_TRUE(p.link_.connected());
+  EXPECT_EQ(p.link_.queue(reinterpret_cast<const uint8_t *>("ping"), 4), 4u);
+  EXPECT_TRUE(p.link_.flush_tx());
+  char buf[8];
+  EXPECT_EQ(::read(p.peer_fd_, buf, sizeof(buf)), 4);
+  EXPECT_EQ(std::memcmp(buf, "ping", 4), 0);
+}
+
+TEST(TcpClientLink, CloseClearsQueuedBytes) {
+  LinkPeer p;
+  EXPECT_EQ(p.link_.queue(reinterpret_cast<const uint8_t *>("MARKER"), 6), 6u);
+  p.link_.close();
+  EXPECT_FALSE(p.link_.connected());
+  EXPECT_EQ(p.link_.tx_free(), 0u);
+  // An uncleared buffer would make flush_tx() report it as still pending.
+  EXPECT_TRUE(p.link_.flush_tx());
+}
+
+TEST(TcpClientLink, FatalWriteInsideFlushDropsTheLink) {
+  LinkPeer p;
+  EXPECT_EQ(p.link_.queue(reinterpret_cast<const uint8_t *>("MARKER"), 6), 6u);
+  p.close_peer();
+  // Still connected from the link's point of view: the drop must happen
+  // inside this flush, the exact ordering TcpUart::flush() reports FAILED.
+  ASSERT_TRUE(p.link_.connected());
+  bool emptied = p.link_.flush_tx();
+  EXPECT_TRUE(emptied);
+  EXPECT_FALSE(p.link_.connected());
+}
+
+}  // namespace esphome::socket::testing
+
+#endif

tests/components/tcp_uart/__init__.py

@@ -0,0 +1,7 @@
+# This file's presence makes pytest treat this directory as a package named
+# "tcp_uart"; required for cpp unit testing.
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.multi_conf = False

tests/components/tcp_uart/test_flush_host.cpp

@@ -0,0 +1,69 @@
+#include <gtest/gtest.h>
+
+#include <csignal>
+#include <memory>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include "esphome/components/tcp_uart/tcp_uart.h"
+
+#ifdef USE_HOST
+
+namespace esphome::tcp_uart::testing {
+
+class TcpUartUnderTest : public TcpUart {
+ public:
+  TcpUartUnderTest() : TcpUart("peer", 1) { this->link_.begin("flush_test"); }
+  socket::TcpClientLink &link() { return this->link_; }
+};
+
+class TcpUartFlush : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    // EPIPE must come back as an errno, not a signal.
+    signal(SIGPIPE, SIG_IGN);
+    int fds[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+    this->peer_fd_ = fds[1];
+    this->uart_.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+  }
+  void TearDown() override {
+    if (this->peer_fd_ >= 0) {
+      ::close(this->peer_fd_);
+    }
+    this->uart_.link().close();
+  }
+
+  TcpUartUnderTest uart_;
+  int peer_fd_{-1};
+};
+
+TEST_F(TcpUartFlush, SuccessWhenTheByteGoesOut) {
+  uint8_t b = 'x';
+  this->uart_.write_array(&b, 1);
+  EXPECT_EQ(this->uart_.flush(), uart::UARTFlushResult::UART_FLUSH_RESULT_SUCCESS);
+  char got;
+  EXPECT_EQ(::read(this->peer_fd_, &got, 1), 1);
+  EXPECT_EQ(got, 'x');
+}
+
+TEST_F(TcpUartFlush, FailedWhenTheFlushDropsTheLink) {
+  uint8_t b = 'x';
+  this->uart_.write_array(&b, 1);
+  ::close(this->peer_fd_);
+  this->peer_fd_ = -1;
+  // The drop happens inside this flush; checking connected() first would
+  // wrongly report success.
+  ASSERT_TRUE(this->uart_.is_connected());
+  EXPECT_EQ(this->uart_.flush(), uart::UARTFlushResult::UART_FLUSH_RESULT_FAILED);
+  EXPECT_FALSE(this->uart_.is_connected());
+}
+
+TEST_F(TcpUartFlush, FailedWhileTheLinkIsDown) {
+  this->uart_.link().close();
+  EXPECT_EQ(this->uart_.flush(), uart::UARTFlushResult::UART_FLUSH_RESULT_FAILED);
+}
+
+}  // namespace esphome::tcp_uart::testing
+
+#endif

tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp

@@ -17,9 +17,8 @@ void TcpClientLinkTestComponent::loop() {
   if (!up) {
     return;
   }
-  // Retries a partial echo; inline no-op when nothing is queued.
-  this->link_.flush_tx();
-  if (!this->link_.ready()) {
+  // Echo only once the previous echo is fully sent, so it never drops bytes.
+  if (!this->link_.flush_tx() || !this->link_.ready()) {
     return;
   }
   uint8_t buf[64];

tests/integration/fixtures/tcp_uart_bridge.yaml

@@ -0,0 +1,33 @@
+esphome:
+  name: tcp-uart-bridge-test
+
+host:
+
+api:
+  services:
+    - service: send_byte
+      then:
+        - lambda: |-
+            id(tcp_uart_1).write_byte('X');
+            auto r = id(tcp_uart_1).flush();
+            ESP_LOGI("bridge_test", "Flush result %d", (int) r);
+
+logger:
+  level: INFO
+
+tcp_uart:
+  - id: tcp_uart_1
+    host: 127.0.0.1
+    port: 18125
+    reconnect_interval: 3s
+    connected:
+      name: Bridge Connected
+
+interval:
+  - interval: 50ms
+    then:
+      - lambda: |-
+          uint8_t b;
+          while (id(tcp_uart_1).read_byte(&b)) {
+            id(tcp_uart_1).write_byte(b);
+          }

tests/integration/test_socket_tcp_client_link.py

@@ -1,7 +1,8 @@
 """Integration test for socket::TcpClientLink on host.
 
 Pytest runs a real TCP server; the device echoes through the link.
-Covers connect, read, write, a server-initiated drop and the reconnect.
+Covers connect, read, write, a server-initiated drop, the reconnect and
+that no bytes from the first session leak into the second.
 """
 
 from __future__ import annotations
@@ -14,6 +15,7 @@ import pytest
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 PAYLOAD = b"hello link"
+SECOND_PAYLOAD = b"second session"
 
 
 @pytest.mark.asyncio
@@ -27,6 +29,7 @@ async def test_socket_tcp_client_link(
     yaml_config = yaml_config.replace("port: 18123", f"port: {server_port}")
 
     echoed: list[bytes] = []
+    second_echoed: list[bytes] = []
     echo_done = asyncio.Event()
     reconnected = asyncio.Event()
     link_down = asyncio.Event()
@@ -56,6 +59,14 @@ async def test_socket_tcp_client_link(
             # Drop the connection so the link has to reconnect.
             writer.close()
             return
+        # Second session: the first bytes back must be this session's echo;
+        # anything left over from the first session would arrive ahead of it.
+        writer.write(SECOND_PAYLOAD)
+        await writer.drain()
+        with contextlib.suppress(TimeoutError, asyncio.IncompleteReadError):
+            second_echoed.append(
+                await asyncio.wait_for(reader.readexactly(len(SECOND_PAYLOAD)), 10)
+            )
         reconnected.set()
 
     server = await asyncio.start_server(handle, "127.0.0.1", server_port)
@@ -83,6 +94,9 @@ async def test_socket_tcp_client_link(
                 await asyncio.wait_for(second_link_up.wait(), timeout=15.0)
             except TimeoutError:
                 pytest.fail("Link did not reconnect after the server dropped it")
+            assert second_echoed == [SECOND_PAYLOAD], (
+                "Second session echo wrong; stale bytes from the first session?"
+            )
     finally:
         server.close()
         await server.wait_closed()

tests/integration/test_tcp_uart_bridge.py

@@ -0,0 +1,102 @@
+"""Integration test for the tcp_uart bridge on host.
+
+Pytest runs a real TCP server; a lambda echoes UART bytes back to the socket.
+Covers the UART read and write paths, flush results for a live and a dropped
+link, the offline drop warning and the automatic reconnect.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+import pytest
+
+from .log_utils import LineWaiter
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+GREETING = b"hi"
+RECONNECT_GREETING = b"yo"
+
+
+async def _wait(event: asyncio.Event, timeout: float, message: str) -> None:
+    try:
+        await asyncio.wait_for(event.wait(), timeout)
+    except TimeoutError:
+        pytest.fail(message)
+
+
+@pytest.mark.asyncio
+async def test_tcp_uart_bridge(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+    unused_tcp_port_factory,
+) -> None:
+    server_port = unused_tcp_port_factory()
+    yaml_config = yaml_config.replace("port: 18125", f"port: {server_port}")
+
+    lines = LineWaiter()
+    echo_ok = asyncio.Event()
+    service_byte_ok = asyncio.Event()
+    reconnect_echo_ok = asyncio.Event()
+    sessions = 0
+
+    async def handle(
+        reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        nonlocal sessions
+        sessions += 1
+        if sessions == 1:
+            writer.write(GREETING)
+            await writer.drain()
+            with contextlib.suppress(TimeoutError, asyncio.IncompleteReadError):
+                if (
+                    await asyncio.wait_for(reader.readexactly(len(GREETING)), 10)
+                    == GREETING
+                ):
+                    echo_ok.set()
+                if await asyncio.wait_for(reader.readexactly(1), 10) == b"X":
+                    service_byte_ok.set()
+            writer.close()
+            return
+        writer.write(RECONNECT_GREETING)
+        await writer.drain()
+        with contextlib.suppress(TimeoutError, asyncio.IncompleteReadError):
+            if (
+                await asyncio.wait_for(reader.readexactly(len(RECONNECT_GREETING)), 10)
+                == RECONNECT_GREETING
+            ):
+                reconnect_echo_ok.set()
+
+    server = await asyncio.start_server(handle, "127.0.0.1", server_port)
+    try:
+        async with (
+            run_compiled(yaml_config, line_callback=lines.callback),
+            api_client_connected() as client,
+        ):
+            device_info = await client.device_info()
+            assert device_info is not None
+            assert device_info.name == "tcp-uart-bridge-test"
+            _, services = await client.list_entities_services()
+            send_byte = next(s for s in services if s.name == "send_byte")
+
+            await _wait(echo_ok, 15.0, "UART echo through the bridge never arrived")
+
+            await client.execute_service(send_byte, {})
+            await _wait(service_byte_ok, 10.0, "Service byte never reached the server")
+            await lines.wait_for("Flush result 0")
+
+            # The server closed session one; a write while down must warn and
+            # a flush on the down link must report FAILED (2).
+            await lines.wait_for("Connection lost")
+            await client.execute_service(send_byte, {})
+            await lines.wait_for("Not connected, dropped")
+            await lines.wait_for("Flush result 2")
+
+            await _wait(
+                reconnect_echo_ok, 15.0, "Bridge did not reconnect and echo again"
+            )
+    finally:
+        server.close()
+        await server.wait_closed()