diff --git a/esphome/components/tcp_uart/tcp_uart.cpp b/esphome/components/tcp_uart/tcp_uart.cpp index 00b59d82ba..b24280c8b1 100644 --- a/esphome/components/tcp_uart/tcp_uart.cpp +++ b/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 diff --git a/tests/components/socket/test_tcp_client_link_host.cpp b/tests/components/socket/test_tcp_client_link_host.cpp new file mode 100644 index 0000000000..9a6d388b72 --- /dev/null +++ b/tests/components/socket/test_tcp_client_link_host.cpp @@ -0,0 +1,77 @@ +#include + +#include +#include +#include +#include +#include + +#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(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("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("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("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 diff --git a/tests/components/tcp_uart/__init__.py b/tests/components/tcp_uart/__init__.py new file mode 100644 index 0000000000..1e4de448b7 --- /dev/null +++ b/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 diff --git a/tests/components/tcp_uart/test_flush_host.cpp b/tests/components/tcp_uart/test_flush_host.cpp new file mode 100644 index 0000000000..9c6c63110c --- /dev/null +++ b/tests/components/tcp_uart/test_flush_host.cpp @@ -0,0 +1,69 @@ +#include + +#include +#include +#include +#include + +#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(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 diff --git a/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp b/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp index bd45e60a00..0f6de0ee9f 100644 --- a/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp +++ b/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]; diff --git a/tests/integration/fixtures/tcp_uart_bridge.yaml b/tests/integration/fixtures/tcp_uart_bridge.yaml new file mode 100644 index 0000000000..12791e7dd4 --- /dev/null +++ b/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); + } diff --git a/tests/integration/test_socket_tcp_client_link.py b/tests/integration/test_socket_tcp_client_link.py index 31cc344f67..16b5f544f7 100644 --- a/tests/integration/test_socket_tcp_client_link.py +++ 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() diff --git a/tests/integration/test_tcp_uart_bridge.py b/tests/integration/test_tcp_uart_bridge.py new file mode 100644 index 0000000000..fb22a26b55 --- /dev/null +++ b/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()