[emontx] Discard the rest of a line that overflows the buffer (#20515)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com>

esphome/components/emontx/emontx.cpp

@@ -22,11 +22,11 @@ void EmonTx::loop() {
     if (!this->read_byte(&received))
       break;
 
-    if (received == '\r') {
-      continue;  // Ignore CR
-    } else if (received == '\n') {
+    if (received == '\n') {
       // End of line - process the buffer
-      if (this->buffer_pos_ > 0) {
+      if (this->discarding_line_) {
+        this->discarding_line_ = false;
+      } else if (this->buffer_pos_ > 0) {
         // Null-terminate for safe logging and c_str() use
         size_t len = this->buffer_pos_;
         this->buffer_[len] = '\0';
@@ -44,9 +44,12 @@ void EmonTx::loop() {
           this->parse_json_(this->buffer_.data(), len);
         }
       }
+    } else if (received == '\r' || this->discarding_line_) {
+      // Ignore CR, and the rest of a line that overflowed the buffer
     } else if (this->buffer_pos_ >= MAX_LINE_LENGTH) {
-      ESP_LOGW(TAG, "Buffer overflow (>%zu bytes), discarding buffer", MAX_LINE_LENGTH);
+      ESP_LOGW(TAG, "Buffer overflow (>%zu bytes), discarding line", MAX_LINE_LENGTH);
       this->buffer_pos_ = 0;
+      this->discarding_line_ = true;
     } else {
       this->buffer_[this->buffer_pos_++] = static_cast<char>(received);
     }

esphome/components/emontx/emontx.h

@@ -55,6 +55,7 @@ class EmonTx final : public Component, public uart::UARTDevice {
   LazyCallbackManager<void(JsonObject, StringRef)> json_callbacks_;
   LazyCallbackManager<void(StringRef)> data_callbacks_;
   uint16_t buffer_pos_{0};
+  bool discarding_line_{false};  // Set after an overflow until the end of that line
   std::array<char, MAX_LINE_LENGTH + 1> buffer_{};
 };
 

tests/components/emontx/common.h

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <cstdint>
+#include <cstring>
+#include <vector>
+#include "esphome/components/uart/uart_component.h"
+#include "esphome/components/emontx/emontx.h"
+
+namespace esphome::emontx::testing {
+
+class MockUARTComponent : public uart::UARTComponent {
+ public:
+  std::vector<uint8_t> tx;
+  std::vector<uint8_t> rx;
+
+  void push_rx(const char *data) { this->rx.insert(this->rx.end(), data, data + strlen(data)); }
+  void push_rx(size_t count, uint8_t byte) { this->rx.insert(this->rx.end(), count, byte); }
+
+  // UARTComponent
+  void write_array(const uint8_t *data, size_t len) override { this->tx.insert(this->tx.end(), data, data + len); }
+
+  bool read_array(uint8_t *data, size_t len) override {
+    if (this->rx.size() < len) {
+      return false;
+    }
+
+    std::copy(this->rx.begin(), this->rx.begin() + len, data);
+    this->rx.erase(this->rx.begin(), this->rx.begin() + len);
+    return true;
+  }
+
+  size_t available() override { return this->rx.size(); }
+
+  MOCK_METHOD(bool, peek_byte, (uint8_t * data), (override));
+  MOCK_METHOD(uart::UARTFlushResult, flush, (), (override));
+  MOCK_METHOD(void, check_logger_conflict, (), (override));
+#if defined(USE_ESP8266) || defined(USE_ESP32)
+  void load_settings(bool dump_config) override {}
+#endif  // defined(USE_ESP8266) || defined(USE_ESP32)
+};
+
+}  // namespace esphome::emontx::testing

tests/components/emontx/emontx_test.cpp

@@ -0,0 +1,75 @@
+#include <gtest/gtest.h>
+
+#include <string>
+#include <vector>
+
+#include "esphome/components/emontx/emontx.h"
+#include "common.h"
+
+namespace esphome::emontx::testing {
+
+namespace {
+class EmonTxTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    this->sut_.set_uart_parent(&this->uart_);
+    this->sut_.add_on_data_callback([this](StringRef line) { this->lines_.emplace_back(line.c_str(), line.size()); });
+    this->sut_.add_on_json_callback([this](JsonObject, StringRef) { this->json_count_++; });
+  }
+
+  MockUARTComponent uart_;
+  EmonTx sut_;
+  std::vector<std::string> lines_;
+  int json_count_{0};
+};
+}  // namespace
+
+TEST_F(EmonTxTest, LinePassedToDataCallbacks) {
+  this->uart_.push_rx("MSG:1\r\n");
+  this->sut_.loop();
+
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "MSG:1");
+  EXPECT_EQ(this->json_count_, 0);
+}
+
+TEST_F(EmonTxTest, JsonLineParsed) {
+  this->uart_.push_rx("{\"P1\":100}\n");
+  this->sut_.loop();
+
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->json_count_, 1);
+}
+
+TEST_F(EmonTxTest, OverflowedLineDiscardedUntilLineFeed) {
+  // One byte more than the buffer holds, then more of the same line.
+  this->uart_.push_rx(MAX_LINE_LENGTH + 1, 'a');
+  this->uart_.push_rx("tail\n");
+  this->uart_.push_rx("MSG:2\n");
+  this->sut_.loop();
+
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "MSG:2");
+}
+
+TEST_F(EmonTxTest, OverflowedLineTailNotParsedAsJson) {
+  // The tail after the overflow looks like a JSON line and must not be parsed.
+  this->uart_.push_rx(MAX_LINE_LENGTH + 1, 'a');
+  this->uart_.push_rx("{\"P1\":100}\n");
+  this->sut_.loop();
+
+  EXPECT_TRUE(this->lines_.empty());
+  EXPECT_EQ(this->json_count_, 0);
+}
+
+TEST_F(EmonTxTest, OverflowAcrossLoopCalls) {
+  this->uart_.push_rx(MAX_LINE_LENGTH + 1, 'a');
+  this->sut_.loop();
+  this->uart_.push_rx("tail\nMSG:3\n");
+  this->sut_.loop();
+
+  ASSERT_EQ(this->lines_.size(), 1u);
+  EXPECT_EQ(this->lines_[0], "MSG:3");
+}
+
+}  // namespace esphome::emontx::testing