Merge branch 'cron-trigger-interval' into integration

tests/benchmarks/components/button/__init__.py

@@ -0,0 +1,5 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.enable_codegen()

tests/benchmarks/components/button/bench_button.cpp

@@ -0,0 +1,55 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/button/button.h"
+
+namespace esphome::button::benchmarks {
+
+static constexpr int kInnerIterations = 2000;
+
+// Minimal Button for benchmarking — press_action() is a no-op.
+class BenchButton : public Button {
+ public:
+  void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
+
+ protected:
+  void press_action() override {}
+};
+
+// --- Button::press() ---
+// Measures: ESP_LOGD + press_action() + callback dispatch.
+
+static void ButtonPress(benchmark::State &state) {
+  BenchButton button;
+  button.configure("test_button");
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      button.press();
+    }
+    benchmark::DoNotOptimize(&button);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(ButtonPress);
+
+// --- Button::press() with callback ---
+// Measures callback dispatch overhead.
+
+static void ButtonPress_WithCallback(benchmark::State &state) {
+  BenchButton button;
+  button.configure("test_button");
+
+  uint64_t callback_count = 0;
+  button.add_on_press_callback([&callback_count]() { callback_count++; });
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      button.press();
+    }
+    benchmark::DoNotOptimize(callback_count);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(ButtonPress_WithCallback);
+
+}  // namespace esphome::button::benchmarks

tests/benchmarks/components/button/benchmark.yaml

@@ -0,0 +1 @@
+button:

tests/benchmarks/components/number/__init__.py

@@ -0,0 +1,5 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.enable_codegen()

tests/benchmarks/components/number/bench_number.cpp

@@ -0,0 +1,121 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/number/number.h"
+
+namespace esphome::benchmarks {
+
+// Inner iteration count to amortize CodSpeed instrumentation overhead.
+static constexpr int kInnerIterations = 2000;
+
+// Minimal Number for benchmarking — control() publishes the value back.
+class BenchNumber : public number::Number {
+ public:
+  void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
+
+ protected:
+  void control(float value) override { this->publish_state(value); }
+};
+
+// Helper to create a typical number entity for benchmarks.
+static void setup_number(BenchNumber &number) {
+  number.configure("test_number");
+  number.traits.set_min_value(0.0f);
+  number.traits.set_max_value(100.0f);
+  number.traits.set_step(1.0f);
+  number.traits.set_mode(number::NUMBER_MODE_SLIDER);
+}
+
+// --- Number::publish_state() ---
+// Measures the publish path: set_has_state, store value, callback dispatch.
+
+static void NumberPublish_State(benchmark::State &state) {
+  BenchNumber number;
+  setup_number(number);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      number.publish_state(static_cast<float>(i % 100));
+    }
+    benchmark::DoNotOptimize(number.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(NumberPublish_State);
+
+// --- Number::publish_state() with callback ---
+// Measures callback dispatch overhead.
+
+static void NumberPublish_WithCallback(benchmark::State &state) {
+  BenchNumber number;
+  setup_number(number);
+
+  uint64_t callback_count = 0;
+  number.add_on_state_callback([&callback_count](float) { callback_count++; });
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      number.publish_state(static_cast<float>(i % 100));
+    }
+    benchmark::DoNotOptimize(callback_count);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(NumberPublish_WithCallback);
+
+// --- NumberCall::perform() set value ---
+// The most common number call — setting an absolute value.
+// Exercises: validation against min/max, control() dispatch.
+
+static void NumberCall_SetValue(benchmark::State &state) {
+  BenchNumber number;
+  setup_number(number);
+  number.publish_state(50.0f);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      float val = static_cast<float>(i % 100);
+      number.make_call().set_value(val).perform();
+    }
+    benchmark::DoNotOptimize(number.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(NumberCall_SetValue);
+
+// --- NumberCall::perform() increment ---
+// Exercises: state read, step arithmetic, max clamping.
+
+static void NumberCall_Increment(benchmark::State &state) {
+  BenchNumber number;
+  setup_number(number);
+  number.publish_state(0.0f);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      number.make_call().number_increment(true).perform();
+    }
+    benchmark::DoNotOptimize(number.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(NumberCall_Increment);
+
+// --- NumberCall::perform() decrement ---
+// Exercises: state read, step arithmetic, min clamping.
+
+static void NumberCall_Decrement(benchmark::State &state) {
+  BenchNumber number;
+  setup_number(number);
+  number.publish_state(100.0f);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      number.make_call().number_decrement(true).perform();
+    }
+    benchmark::DoNotOptimize(number.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(NumberCall_Decrement);
+
+}  // namespace esphome::benchmarks

tests/benchmarks/components/number/benchmark.yaml

@@ -0,0 +1 @@
+number:

tests/benchmarks/components/select/__init__.py

@@ -0,0 +1,5 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.enable_codegen()

tests/benchmarks/components/select/bench_select.cpp

@@ -0,0 +1,157 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/select/select.h"
+
+namespace esphome::benchmarks {
+
+// Inner iteration count to amortize CodSpeed instrumentation overhead.
+static constexpr int kInnerIterations = 2000;
+
+// Minimal Select for benchmarking — control() publishes directly by index.
+class BenchSelect : public select::Select {
+ public:
+  void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
+
+ protected:
+  void control(size_t index) override { this->publish_state(index); }
+};
+
+// Helper to create a select with the given options.
+static void setup_select(BenchSelect &select, const char *name, std::initializer_list<const char *> options) {
+  select.configure(name);
+  select.traits.set_options(options);
+  select.publish_state(size_t(0));
+}
+
+// --- Select::publish_state(size_t) ---
+// The fast path: publish by index, no string lookup.
+
+static void SelectPublish_ByIndex(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.publish_state(static_cast<size_t>(i % 4));
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectPublish_ByIndex);
+
+// --- Select::publish_state(const char *) ---
+// The string path: requires index_of() lookup via strncmp.
+
+static void SelectPublish_ByString(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  const char *options[] = {"off", "still", "move", "still+move"};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.publish_state(options[i % 4]);
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectPublish_ByString);
+
+// --- Select::publish_state() with callback ---
+// Measures callback dispatch overhead on the index path.
+
+static void SelectPublish_WithCallback(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  uint64_t callback_count = 0;
+  select.add_on_state_callback([&callback_count](size_t) { callback_count++; });
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.publish_state(static_cast<size_t>(i % 4));
+    }
+    benchmark::DoNotOptimize(callback_count);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectPublish_WithCallback);
+
+// --- SelectCall::perform() set by index ---
+// The fast call path — no string matching needed.
+
+static void SelectCall_SetByIndex(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.make_call().set_index(i % 4).perform();
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectCall_SetByIndex);
+
+// --- SelectCall::perform() set by option string ---
+// Exercises the string lookup path through index_of().
+
+static void SelectCall_SetByOption(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  const char *options[] = {"off", "still", "move", "still+move"};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.make_call().set_option(options[i % 4]).perform();
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectCall_SetByOption);
+
+// --- SelectCall::perform() next with cycling ---
+// Exercises the navigation path through active_index_.
+
+static void SelectCall_NextCycle(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(select, "test_select", {"off", "still", "move", "still+move"});
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.make_call().select_next(true).perform();
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectCall_NextCycle);
+
+// --- SelectCall with 10 options (string lookup) ---
+// Worst-case string matching with more options.
+
+static void SelectCall_SetByOption_10Options(benchmark::State &state) {
+  BenchSelect select;
+  setup_select(
+      select, "test_select",
+      {"off", "still", "move", "still+move", "custom1", "custom2", "custom3", "custom4", "custom5", "custom6"});
+
+  // Pick options spread across the list to exercise different search depths
+  const char *picks[] = {"off", "custom3", "custom6", "move"};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      select.make_call().set_option(picks[i % 4]).perform();
+    }
+    benchmark::DoNotOptimize(select.active_index());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SelectCall_SetByOption_10Options);
+
+}  // namespace esphome::benchmarks

tests/benchmarks/components/select/benchmark.yaml

@@ -0,0 +1 @@
+select:

tests/benchmarks/components/switch/__init__.py

@@ -0,0 +1,5 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.enable_codegen()

tests/benchmarks/components/switch/bench_switch.cpp

@@ -0,0 +1,137 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/switch/switch.h"
+
+namespace esphome::benchmarks {
+
+// Inner iteration count to amortize CodSpeed instrumentation overhead.
+static constexpr int kInnerIterations = 2000;
+
+// Minimal Switch for benchmarking — write_state() publishes directly.
+class BenchSwitch : public switch_::Switch {
+ public:
+  void configure(const char *name) { this->configure_entity_(name, 0x12345678, 0); }
+
+ protected:
+  void write_state(bool state) override { this->publish_state(state); }
+};
+
+// --- Switch::publish_state() alternating ---
+// Forces state change every call, exercising the full publish path.
+
+static void SwitchPublish_Alternating(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+  sw.publish_state(false);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.publish_state(i % 2 == 0);
+    }
+    benchmark::DoNotOptimize(sw.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchPublish_Alternating);
+
+// --- Switch::publish_state() no change ---
+// Tests the deduplication fast path in publish_dedup_.
+
+static void SwitchPublish_NoChange(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+  sw.publish_state(true);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.publish_state(true);
+    }
+    benchmark::DoNotOptimize(sw.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchPublish_NoChange);
+
+// --- Switch::publish_state() with callback ---
+// Measures callback dispatch overhead on state changes.
+
+static void SwitchPublish_WithCallback(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+
+  uint64_t callback_count = 0;
+  sw.add_on_state_callback([&callback_count](bool) { callback_count++; });
+  sw.publish_state(false);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.publish_state(i % 2 == 0);
+    }
+    benchmark::DoNotOptimize(callback_count);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchPublish_WithCallback);
+
+// --- Switch::turn_on() / turn_off() ---
+// The front-end call path: turn_on → write_state → publish_state.
+
+static void SwitchTurnOn(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+  sw.publish_state(false);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.turn_on();
+    }
+    benchmark::DoNotOptimize(sw.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchTurnOn);
+
+// --- Switch::toggle() alternating ---
+// Exercises the toggle path which reads current state to determine target.
+
+static void SwitchToggle(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+  sw.publish_state(false);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.toggle();
+    }
+    benchmark::DoNotOptimize(sw.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchToggle);
+
+// --- Switch::publish_state() inverted ---
+// Verifies the inversion path doesn't add significant overhead.
+
+static void SwitchPublish_Inverted(benchmark::State &state) {
+  BenchSwitch sw;
+  sw.configure("test_switch");
+  sw.set_restore_mode(switch_::SWITCH_ALWAYS_OFF);
+  sw.set_inverted(true);
+  sw.publish_state(false);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sw.publish_state(i % 2 == 0);
+    }
+    benchmark::DoNotOptimize(sw.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(SwitchPublish_Inverted);
+
+}  // namespace esphome::benchmarks

tests/benchmarks/components/switch/benchmark.yaml

@@ -0,0 +1 @@
+switch:

tests/benchmarks/components/text_sensor/__init__.py

@@ -0,0 +1,5 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    manifest.enable_codegen()

tests/benchmarks/components/text_sensor/bench_text_sensor.cpp

@@ -0,0 +1,108 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/text_sensor/text_sensor.h"
+
+namespace esphome::text_sensor::benchmarks {
+
+static constexpr int kInnerIterations = 2000;
+
+// --- publish_state(const char *) with short string, value changes each time ---
+// Exercises: memcmp check (mismatch), string assign, callback dispatch.
+
+static void TextSensorPublish_Short_Changing(benchmark::State &state) {
+  TextSensor sensor;
+
+  // Pre-populate with different short strings
+  const char *values[] = {"192.168.1.1", "192.168.1.2", "192.168.1.3", "192.168.1.4"};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sensor.publish_state(values[i % 4]);
+    }
+    benchmark::DoNotOptimize(sensor.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(TextSensorPublish_Short_Changing);
+
+// --- publish_state(const char *) with short string, same value (dedup path) ---
+// Exercises: memcmp check (match), skips string assign.
+
+static void TextSensorPublish_Short_NoChange(benchmark::State &state) {
+  TextSensor sensor;
+  sensor.publish_state("192.168.1.100");
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sensor.publish_state("192.168.1.100");
+    }
+    benchmark::DoNotOptimize(sensor.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(TextSensorPublish_Short_NoChange);
+
+// --- publish_state with longer string (firmware version, MAC address) ---
+// Exercises: memcmp on longer strings, string assign with potential realloc.
+
+static void TextSensorPublish_Long_Changing(benchmark::State &state) {
+  TextSensor sensor;
+
+  const char *values[] = {
+      "2025.12.0-dev (Jan 15 2025, 10:30:00)",
+      "2025.12.1-dev (Feb 20 2025, 14:45:00)",
+      "2025.12.2-dev (Mar 10 2025, 08:15:00)",
+      "2025.12.3-dev (Apr  5 2025, 16:00:00)",
+  };
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sensor.publish_state(values[i % 4]);
+    }
+    benchmark::DoNotOptimize(sensor.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(TextSensorPublish_Long_Changing);
+
+// --- publish_state with callback ---
+// Measures callback dispatch overhead for text sensors.
+
+static void TextSensorPublish_WithCallback(benchmark::State &state) {
+  TextSensor sensor;
+
+  uint64_t callback_count = 0;
+  sensor.add_on_state_callback([&callback_count](const std::string &) { callback_count++; });
+
+  const char *values[] = {"192.168.1.1", "192.168.1.2", "192.168.1.3", "192.168.1.4"};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sensor.publish_state(values[i % 4]);
+    }
+    benchmark::DoNotOptimize(callback_count);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(TextSensorPublish_WithCallback);
+
+// --- publish_state(const char *, size_t) direct ---
+// The lowest-level overload, avoids strlen.
+
+static void TextSensorPublish_WithLen(benchmark::State &state) {
+  TextSensor sensor;
+
+  static constexpr const char *values[] = {"192.168.1.1", "192.168.1.2", "192.168.1.3", "192.168.1.4"};
+  static constexpr size_t lens[] = {11, 11, 11, 11};
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      sensor.publish_state(values[i % 4], lens[i % 4]);
+    }
+    benchmark::DoNotOptimize(sensor.state);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(TextSensorPublish_WithLen);
+
+}  // namespace esphome::text_sensor::benchmarks

tests/benchmarks/components/text_sensor/benchmark.yaml

@@ -0,0 +1 @@
+text_sensor:

tests/component_tests/display/test_display_metadata.py

@@ -0,0 +1,81 @@
+"""Tests for display component metadata functions."""
+
+from unittest.mock import patch
+
+from esphome.components.display import (
+    DisplayMetaData,
+    add_metadata,
+    get_all_display_metadata,
+    get_display_metadata,
+)
+from esphome.cpp_generator import MockObj
+
+
+def test_add_metadata_with_string_id():
+    """Test adding metadata with a plain string ID."""
+    with patch("esphome.components.display.CORE.data", {}):
+        add_metadata("my_display", 320, 240, True)
+        meta = get_display_metadata("my_display")
+        assert meta == DisplayMetaData(
+            width=320, height=240, has_writer=True, has_hardware_rotation=False
+        )
+
+
+def test_add_metadata_with_mockobj_id():
+    """Test adding metadata with a MockObj ID (converted via str())."""
+    with patch("esphome.components.display.CORE.data", {}):
+        mock_id = MockObj("my_display_obj")
+        add_metadata(mock_id, 480, 320, False, has_hardware_rotation=True)
+        meta = get_display_metadata("my_display_obj")
+        assert meta == DisplayMetaData(
+            width=480, height=320, has_writer=False, has_hardware_rotation=True
+        )
+
+
+def test_add_metadata_hardware_rotation_default():
+    """Test that has_hardware_rotation defaults to False."""
+    with patch("esphome.components.display.CORE.data", {}):
+        add_metadata("disp", 128, 64, False)
+        meta = get_display_metadata("disp")
+        assert meta.has_hardware_rotation is False
+
+
+def test_get_display_metadata_missing_returns_none():
+    """Test that querying a non-existent ID returns None."""
+    with patch("esphome.components.display.CORE.data", {}):
+        data = get_display_metadata("no_such_display")
+        assert data.width == 0
+        assert data.height == 0
+        assert data.has_writer is False
+        assert data.has_hardware_rotation is False
+
+
+def test_add_multiple_displays():
+    """Test adding metadata for multiple displays."""
+    with patch("esphome.components.display.CORE.data", {}):
+        add_metadata("disp_a", 320, 240, True)
+        add_metadata("disp_b", 128, 64, False, has_hardware_rotation=True)
+
+        all_meta = get_all_display_metadata()
+        assert len(all_meta) == 2
+        assert all_meta["disp_a"] == DisplayMetaData(320, 240, True, False)
+        assert all_meta["disp_b"] == DisplayMetaData(128, 64, False, True)
+
+
+def test_add_metadata_overwrites_existing():
+    """Test that adding metadata for the same ID overwrites the previous entry."""
+    with patch("esphome.components.display.CORE.data", {}):
+        add_metadata("disp", 320, 240, True)
+        add_metadata("disp", 640, 480, False, has_hardware_rotation=True)
+        meta = get_display_metadata("disp")
+        assert meta == DisplayMetaData(640, 480, False, True)
+
+
+def test_metadata_is_frozen():
+    """Test that DisplayMetaData instances are immutable (frozen dataclass)."""
+    meta = DisplayMetaData(320, 240, True, False)
+    try:
+        meta.width = 640
+        assert False, "Expected FrozenInstanceError"
+    except AttributeError:
+        pass

tests/component_tests/mipi_spi/test_display_metadata.py

@@ -0,0 +1,200 @@
+"""Tests for display metadata created by mipi_spi component."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+from esphome.components.display import (
+    DisplayMetaData,
+    get_all_display_metadata,
+    get_display_metadata,
+)
+from esphome.components.esp32 import (
+    KEY_BOARD,
+    KEY_VARIANT,
+    VARIANT_ESP32,
+    VARIANT_ESP32S3,
+)
+from esphome.components.mipi_spi.display import (
+    CONFIG_SCHEMA,
+    FINAL_VALIDATE_SCHEMA,
+    get_instance,
+)
+from esphome.const import PlatformFramework
+from tests.component_tests.types import SetCoreConfigCallable
+
+
+def validated_config(config):
+    """Run schema + final validation and return the validated config."""
+    return FINAL_VALIDATE_SCHEMA(CONFIG_SCHEMA(config))
+
+
+def test_metadata_native_quad_default_test_card(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """A quad-mode display with no explicit drawing gets a test card from final validation."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32-s3-devkitc-1", KEY_VARIANT: VARIANT_ESP32S3},
+    )
+    config = validated_config({"model": "JC3636W518"})
+    get_instance(config)
+    meta = get_display_metadata(str(config["id"]))
+    assert meta is not None
+    assert meta.width == 360
+    assert meta.height == 360
+    # final validation auto-enables show_test_card when no drawing methods are configured
+    assert meta.has_writer is True
+    assert meta.has_hardware_rotation is True
+
+
+def test_metadata_single_mode_with_dc_pin(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """A single-mode display with no explicit drawing gets a test card from final validation."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
+    )
+    config = validated_config(
+        {
+            "model": "ST7735",
+            "dc_pin": 18,
+        }
+    )
+    get_instance(config)
+    meta = get_display_metadata(str(config["id"]))
+    assert meta is not None
+    assert meta.width == 128
+    assert meta.height == 160
+    assert meta.has_writer is True
+    assert meta.has_hardware_rotation is True
+
+
+def test_metadata_custom_dimensions(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """A custom model picks up explicit dimensions."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
+    )
+    config = validated_config(
+        {
+            "model": "custom",
+            "dc_pin": 18,
+            "dimensions": {"width": 480, "height": 320},
+            "init_sequence": [[0xA0, 0x01]],
+        }
+    )
+    get_instance(config)
+    meta = get_display_metadata(str(config["id"]))
+    assert meta is not None
+    assert meta.width == 480
+    assert meta.height == 320
+    # final validation auto-enables show_test_card
+    assert meta.has_writer is True
+    assert meta.has_hardware_rotation is True
+
+
+def test_metadata_with_test_card_has_writer(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """When show_test_card is enabled, has_writer should be True."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
+    )
+    config = validated_config(
+        {
+            "model": "custom",
+            "dc_pin": 18,
+            "dimensions": {"width": 240, "height": 240},
+            "init_sequence": [[0xA0, 0x01]],
+            "show_test_card": True,
+        }
+    )
+    get_instance(config)
+    meta = get_display_metadata(str(config["id"]))
+    assert meta is not None
+    assert meta.has_writer is True
+
+
+def test_metadata_no_swap_xy_not_full_hardware_rotation(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """A model that disables swap_xy should report has_hardware_rotation=False."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32-s3-devkitc-1", KEY_VARIANT: VARIANT_ESP32S3},
+    )
+    # JC3248W535 has swap_xy=cv.UNDEFINED -> transforms={mirror_x, mirror_y} only
+    config = validated_config({"model": "JC3248W535"})
+    get_instance(config)
+    meta = get_display_metadata(str(config["id"]))
+    assert meta is not None
+    assert meta.has_hardware_rotation is False
+
+
+def test_metadata_multiple_displays_independent(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """Multiple displays each get their own metadata entry."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
+    )
+    config_a = validated_config(
+        {
+            "id": "disp_a",
+            "model": "custom",
+            "dc_pin": 18,
+            "dimensions": {"width": 320, "height": 240},
+            "init_sequence": [[0xA0, 0x01]],
+        }
+    )
+    config_b = validated_config(
+        {
+            "id": "disp_b",
+            "model": "custom",
+            "dc_pin": 19,
+            "dimensions": {"width": 128, "height": 64},
+            "init_sequence": [[0xA0, 0x01]],
+        }
+    )
+    get_instance(config_a)
+    get_instance(config_b)
+
+    all_meta = get_all_display_metadata()
+    # final validation auto-enables show_test_card for both
+    assert all_meta["disp_a"] == DisplayMetaData(320, 240, True, True)
+    assert all_meta["disp_b"] == DisplayMetaData(128, 64, True, True)
+
+
+def test_metadata_via_code_generation_native(
+    generate_main: Callable[[str | Path], str],
+    component_fixture_path: Callable[[str], Path],
+) -> None:
+    """Full code generation for native.yaml should produce correct metadata."""
+    generate_main(component_fixture_path("native.yaml"))
+    all_meta = get_all_display_metadata()
+    # native.yaml: model JC3636W518 -> 360x360, no writer, full hardware rotation
+    assert len(all_meta) == 1
+    meta = next(iter(all_meta.values()))
+    assert meta == DisplayMetaData(
+        width=360, height=360, has_writer=True, has_hardware_rotation=True
+    )
+
+
+def test_metadata_via_code_generation_lvgl(
+    generate_main: Callable[[str | Path], str],
+    component_fixture_path: Callable[[str], Path],
+) -> None:
+    """Full code generation for lvgl.yaml should produce correct metadata."""
+    generate_main(component_fixture_path("lvgl.yaml"))
+    all_meta = get_all_display_metadata()
+    # lvgl.yaml: model ST7735 -> 128x160, no writer (lvgl draws directly), full hw rotation
+    assert len(all_meta) == 1
+    meta = next(iter(all_meta.values()))
+    assert meta == DisplayMetaData(
+        width=128, height=160, has_writer=False, has_hardware_rotation=True
+    )

tests/component_tests/mipi_spi/test_init.py

@@ -204,11 +204,6 @@ def test_transform_and_init_sequence_errors(
             r"extra keys not allowed @ data\['brightness'\]",
             id="brightness_not_supported",
         ),
-        pytest.param(
-            {"model": "T-DISPLAY-S3-PRO"},
-            "PSRAM is required for this display",
-            id="psram_required",
-        ),
     ],
 )
 def test_esp32s3_specific_errors(
@@ -319,7 +314,7 @@ def test_native_generation(
 
     main_cpp = generate_main(component_fixture_path("native.yaml"))
     assert (
-        "mipi_spi::MipiSpiBuffer<uint16_t, mipi_spi::PIXEL_MODE_16, true, mipi_spi::PIXEL_MODE_16, mipi_spi::BUS_TYPE_QUAD, 360, 360, 0, 1, display::DISPLAY_ROTATION_0_DEGREES, 1, 1>()"
+        "mipi_spi::MipiSpiBuffer<uint16_t, mipi_spi::PIXEL_MODE_16, true, mipi_spi::PIXEL_MODE_16, mipi_spi::BUS_TYPE_QUAD, 360, 360, 0, 1, 0, true, 1, 1>()"
         in main_cpp
     )
     assert "set_init_sequence({240, 1, 8, 242" in main_cpp
@@ -335,7 +330,7 @@ def test_lvgl_generation(
 
     main_cpp = generate_main(component_fixture_path("lvgl.yaml"))
     assert (
-        "mipi_spi::MipiSpi<uint16_t, mipi_spi::PIXEL_MODE_16, true, mipi_spi::PIXEL_MODE_16, mipi_spi::BUS_TYPE_SINGLE, 128, 160, 0, 0>();"
+        "mipi_spi::MipiSpi<uint16_t, mipi_spi::PIXEL_MODE_16, true, mipi_spi::PIXEL_MODE_16, mipi_spi::BUS_TYPE_SINGLE, 128, 160, 0, 0, 0, true>();"
         in main_cpp
     )
     assert "set_init_sequence({1, 0, 10, 255, 177" in main_cpp

tests/components/time/common.yaml

@@ -6,5 +6,9 @@ api:
 
 time:
   - platform: homeassistant
+    on_time:
+      - seconds: "0,10,20,30,40,50"
+        then:
+          - logger.log: "CronTrigger fired (every 10 seconds)"
   - platform: sntp
     id: sntp_time

tests/integration/conftest.py

@@ -198,6 +198,13 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
             '      - "-g"       # Add debug symbols',
         )
 
+    # Replace external component path placeholder if present
+    if "EXTERNAL_COMPONENT_PATH" in content:
+        external_components_path = str(
+            Path(__file__).parent / "fixtures" / "external_components"
+        )
+        content = content.replace("EXTERNAL_COMPONENT_PATH", external_components_path)
+
     return content
 
 

tests/integration/fixtures/uart_mock_modbus.yaml

@@ -22,6 +22,8 @@ uart_mock:
     baud_rate: 9600
     rx_full_threshold: 120
     rx_timeout: 2
+    # auto_start must be false to avoid races: the test presses the
+    # "Start Scenario" button only after subscribing to states.
     auto_start: false
     debug:
     responses:
@@ -46,7 +48,7 @@ modbus:
 modbus_controller:
   - address: 1
     id: modbus_controller_ok
-    max_cmd_retries: 0
+    max_cmd_retries: 2
     update_interval: 1s
   - address: 2
     id: modbus_controller_slow
@@ -89,4 +91,4 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: 'id(virtual_uart_dev).start_scenario();'
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml

@@ -22,6 +22,8 @@ uart:
 uart_mock:
   - id: virtual_uart_dev
     baud_rate: 9600
+    # auto_start must be false to avoid races: the test presses the
+    # "Start Scenario" button only after subscribing to states.
     auto_start: false
     debug:
     on_tx:
@@ -40,7 +42,8 @@ uart_mock:
                     0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
                 - uart_mock.inject_rx: # Second USB packet: rest of response (staged with 40ms latency)
                     delay: 40ms
-                    data: !lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    data:
+                      !lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
                       0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x42,0x6F,0xCC,0xCD,0x43,0x7C,0xB8,0x10,0x3D,0x38,0x51,0xEC,
                       0x43,0x81,0x1B,0xE7,0x3B,0x03,0x12,0x6F,0x50,0x1B};
 
@@ -61,4 +64,4 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: 'id(virtual_uart_dev).start_scenario();'
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server.yaml

@@ -0,0 +1,124 @@
+esphome:
+  name: uart-mock-modbus-server-test
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_dev
+    baud_rate: 9600
+    rx_full_threshold: 120
+    rx_timeout: 2
+    auto_start: false
+    debug:
+    injections:
+      - delay: 100ms
+        inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] #  Read holding register 3 on device 1 (basic_read)
+      - delay: 100ms
+        #  Read holding register 7 on device 2
+        #  Reply from device 2
+        #  Read holding register 5 on device 1 (read_after_peer_response)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x02,
+            0x03,
+            0x02,
+            0x00,
+            0xF0,
+            0xFC,
+            0x00,
+            0x01,
+            0x03,
+            0x00,
+            0x05,
+            0x00,
+            0x01,
+            0x94,
+            0x0B,
+          ]
+      - delay: 100ms
+        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] #  Read holding register 7 on device 2, with no response
+      - delay: 100ms
+        #  Read holding register 7 on device 2, with no response
+        #  Read holding register A on device 1 (read_after_peer_timeout)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x01,
+            0x03,
+            0x00,
+            0x0A,
+            0x00,
+            0x01,
+            0xA4,
+            0x08,
+          ]
+
+modbus:
+  uart_id: virtual_uart_dev
+  role: server
+
+modbus_controller:
+  - address: 1
+    server_registers:
+      - address: 0x03
+        value_type: U_WORD
+        read_lambda: |-
+          id(basic_read).publish_state(1);
+          return 1;
+      - address: 0x05
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_response).publish_state(1);
+          return 1;
+      - address: 0x0A
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_timeout).publish_state(1);
+          return 1;
+
+sensor:
+  - platform: template
+    name: "basic_read"
+    id: basic_read
+  - platform: template
+    name: "read_after_peer_response"
+    id: read_after_peer_response
+  - platform: template
+    name: "read_after_peer_timeout"
+    id: read_after_peer_timeout
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server_controller.yaml

@@ -0,0 +1,180 @@
+esphome:
+  name: uart-mock-modbus-server-contro
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true  # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    update_interval: 1s
+    id: modbus_controller_1
+
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    server_registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 99;
+      - address: 0x03
+        value_type: S_WORD
+        read_lambda: return -99;
+      - address: 0x05
+        value_type: U_DWORD
+        read_lambda: return 16909060;
+      - address: 0x08
+        value_type: S_DWORD
+        read_lambda: return -16909060;
+      - address: 0x0B
+        value_type: U_DWORD_R
+        read_lambda: return 67305985;
+      - address: 0x0E
+        value_type: S_DWORD_R
+        read_lambda: return -67305985;
+      - address: 0x11
+        value_type: U_QWORD
+        read_lambda: return 72623859790382856;
+      - address: 0x16
+        value_type: S_QWORD
+        read_lambda: return -72623859790382856;
+      - address: 0x1B
+        value_type: U_QWORD_R
+        read_lambda: return 578437695752307201;
+      - address: 0x20
+        value_type: S_QWORD_R
+        read_lambda: return -578437695752307201;
+      - address: 0x25
+        value_type: FP32
+        read_lambda: return 3.14;
+      - address: 0x28
+        value_type: FP32_R
+        read_lambda: return 3.14;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_word"
+    address: 0x03
+    register_type: holding
+    value_type: S_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword"
+    address: 0x05
+    register_type: holding
+    value_type: U_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword"
+    address: 0x08
+    register_type: holding
+    value_type: S_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword_r"
+    address: 0x0B
+    register_type: holding
+    value_type: U_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword_r"
+    address: 0x0E
+    register_type: holding
+    value_type: S_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword"
+    address: 0x11
+    register_type: holding
+    value_type: U_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword"
+    address: 0x16
+    register_type: holding
+    value_type: S_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword_r"
+    address: 0x1B
+    register_type: holding
+    value_type: U_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword_r"
+    address: 0x20
+    register_type: holding
+    value_type: S_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32"
+    address: 0x25
+    register_type: holding
+    value_type: FP32
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32_r"
+    address: 0x28
+    register_type: holding
+    value_type: FP32_R
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_server).start_scenario();"
+      - lambda: "id(virtual_uart_controller).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -0,0 +1,118 @@
+esphome:
+  name: uart-mock-modbus-server-mult
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+  - id: virtual_uart_server_2
+    baud_rate: 9600
+    auto_start: true  # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true  # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_server_2
+    id: virtual_modbus_server_2
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_client
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_1
+  - address: 2
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_2
+
+  - address: 1
+    modbus_id: virtual_modbus_server
+    server_registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 919;
+  - address: 2
+    modbus_id: virtual_modbus_server_2
+    server_registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 929;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_2
+    name: "reg_u_word_2"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_server).start_scenario();"
+      - lambda: "id(virtual_uart_server_2).start_scenario();"
+      - lambda: "id(virtual_uart_controller).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server_controller_write.yaml

@@ -0,0 +1,330 @@
+esphome:
+  name: uart-mock-modbus-srv-write
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true  # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: stored_u_word
+    type: uint16_t
+    initial_value: "11"
+  - id: stored_s_word
+    type: int16_t
+    initial_value: "-11"
+  - id: stored_u_dword
+    type: uint32_t
+    initial_value: "1001"
+  - id: stored_s_dword
+    type: int32_t
+    initial_value: "-1001"
+  - id: stored_u_dword_r
+    type: uint32_t
+    initial_value: "3003"
+  - id: stored_s_dword_r
+    type: int32_t
+    initial_value: "-3003"
+  - id: stored_u_qword
+    type: uint64_t
+    initial_value: "5005"
+  - id: stored_s_qword
+    type: int64_t
+    initial_value: "-5005"
+  - id: stored_u_qword_r
+    type: uint64_t
+    initial_value: "7007"
+  - id: stored_s_qword_r
+    type: int64_t
+    initial_value: "-7007"
+  - id: stored_fp32
+    type: float
+    initial_value: "1.5"
+  - id: stored_fp32_r
+    type: float
+    initial_value: "2.5"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    update_interval: 2s
+    id: modbus_controller_1
+
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    server_registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return id(stored_u_word);
+        write_lambda: id(stored_u_word) = x; return true;
+      - address: 0x03
+        value_type: S_WORD
+        read_lambda: return id(stored_s_word);
+        write_lambda: id(stored_s_word) = x; return true;
+      - address: 0x05
+        value_type: U_DWORD
+        read_lambda: return id(stored_u_dword);
+        write_lambda: id(stored_u_dword) = x; return true;
+      - address: 0x08
+        value_type: S_DWORD
+        read_lambda: return id(stored_s_dword);
+        write_lambda: id(stored_s_dword) = x; return true;
+      - address: 0x0B
+        value_type: U_DWORD_R
+        read_lambda: return id(stored_u_dword_r);
+        write_lambda: id(stored_u_dword_r) = x; return true;
+      - address: 0x0E
+        value_type: S_DWORD_R
+        read_lambda: return id(stored_s_dword_r);
+        write_lambda: id(stored_s_dword_r) = x; return true;
+      - address: 0x11
+        value_type: U_QWORD
+        read_lambda: return id(stored_u_qword);
+        write_lambda: id(stored_u_qword) = x; return true;
+      - address: 0x16
+        value_type: S_QWORD
+        read_lambda: return id(stored_s_qword);
+        write_lambda: id(stored_s_qword) = x; return true;
+      - address: 0x1B
+        value_type: U_QWORD_R
+        read_lambda: return id(stored_u_qword_r);
+        write_lambda: id(stored_u_qword_r) = x; return true;
+      - address: 0x20
+        value_type: S_QWORD_R
+        read_lambda: return id(stored_s_qword_r);
+        write_lambda: id(stored_s_qword_r) = x; return true;
+      - address: 0x25
+        value_type: FP32
+        read_lambda: return id(stored_fp32);
+        write_lambda: id(stored_fp32) = x; return true;
+      - address: 0x28
+        value_type: FP32_R
+        read_lambda: return id(stored_fp32_r);
+        write_lambda: id(stored_fp32_r) = x; return true;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_word"
+    address: 0x03
+    register_type: holding
+    value_type: S_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword"
+    address: 0x05
+    register_type: holding
+    value_type: U_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword"
+    address: 0x08
+    register_type: holding
+    value_type: S_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword_r"
+    address: 0x0B
+    register_type: holding
+    value_type: U_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword_r"
+    address: 0x0E
+    register_type: holding
+    value_type: S_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword"
+    address: 0x11
+    register_type: holding
+    value_type: U_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword"
+    address: 0x16
+    register_type: holding
+    value_type: S_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword_r"
+    address: 0x1B
+    register_type: holding
+    value_type: U_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword_r"
+    address: 0x20
+    register_type: holding
+    value_type: S_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32"
+    address: 0x25
+    register_type: holding
+    value_type: FP32
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32_r"
+    address: 0x28
+    register_type: holding
+    value_type: FP32_R
+
+number:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+    min_value: 0
+    max_value: 65535
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_s_word"
+    address: 0x03
+    register_type: holding
+    value_type: S_WORD
+    min_value: -16777215
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_u_dword"
+    address: 0x05
+    register_type: holding
+    value_type: U_DWORD
+    min_value: 0
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_s_dword"
+    address: 0x08
+    register_type: holding
+    value_type: S_DWORD
+    min_value: -16777215
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_u_dword_r"
+    address: 0x0B
+    register_type: holding
+    value_type: U_DWORD_R
+    min_value: 0
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_s_dword_r"
+    address: 0x0E
+    register_type: holding
+    value_type: S_DWORD_R
+    min_value: -16777215
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_u_qword"
+    address: 0x11
+    register_type: holding
+    value_type: U_QWORD
+    min_value: 0
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_s_qword"
+    address: 0x16
+    register_type: holding
+    value_type: S_QWORD
+    min_value: -16777215
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_u_qword_r"
+    address: 0x1B
+    register_type: holding
+    value_type: U_QWORD_R
+    min_value: 0
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_s_qword_r"
+    address: 0x20
+    register_type: holding
+    value_type: S_QWORD_R
+    min_value: -16777215
+    max_value: 16777215
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_fp32"
+    address: 0x25
+    register_type: holding
+    value_type: FP32
+    min_value: -16777215
+    max_value: 16777215
+    step: 0.01
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_fp32_r"
+    address: 0x28
+    register_type: holding
+    value_type: FP32_R
+    min_value: -16777215
+    max_value: 16777215
+    step: 0.01
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_server).start_scenario();"
+      - lambda: "id(virtual_uart_controller).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_timing.yaml

@@ -22,6 +22,8 @@ uart_mock:
     baud_rate: 9600
     rx_full_threshold: 120
     rx_timeout: 2
+    # auto_start must be false to avoid races: the test presses the
+    # "Start Scenario" button only after subscribing to states.
     auto_start: false
     debug:
     on_tx:
@@ -61,4 +63,4 @@ button:
     name: "Start Scenario"
     id: start_scenario_btn
     on_press:
-      - lambda: 'id(virtual_uart_dev).start_scenario();'
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/state_utils.py

@@ -346,3 +346,109 @@ class SensorStateCollector:
         else:
             self._waiters.append((condition, future))
         return future
+
+
+class SensorTracker:
+    """Data-driven sensor state tracker with expected-value futures.
+
+    Tracks sensor state updates and resolves futures when sensors report
+    specific expected values. Eliminates per-sensor future boilerplate.
+
+    Usage::
+
+        tracker = SensorTracker(["reg_u_word", "reg_s_word"])
+        futures = tracker.expect_all({"reg_u_word": 99, "reg_s_word": -99})
+        # ... subscribe_states with tracker.on_state, start scenario ...
+        await tracker.await_all(futures)
+    """
+
+    def __init__(self, sensor_names: list[str]) -> None:
+        self.sensor_states: dict[str, list[float]] = {name: [] for name in sensor_names}
+        self.key_to_sensor: dict[int, str] = {}
+        self._expectations: dict[str, list[tuple[object, asyncio.Future]]] = {}
+
+    _ANY = object()  # Sentinel: match any value
+
+    def expect(self, name: str, value: object) -> asyncio.Future:
+        """Register an expected value for *name* and return a future for it."""
+        future: asyncio.Future = asyncio.get_running_loop().create_future()
+        self._expectations.setdefault(name, []).append((value, future))
+        return future
+
+    def expect_any(self, name: str) -> asyncio.Future:
+        """Register a future that resolves on *any* state update for *name*."""
+        return self.expect(name, self._ANY)
+
+    def expect_all(self, expected: dict[str, object]) -> dict[str, asyncio.Future]:
+        """Call ``expect`` for every entry and return a dict of futures."""
+        return {name: self.expect(name, value) for name, value in expected.items()}
+
+    def on_state(self, state: EntityState) -> None:
+        """State callback suitable for ``subscribe_states``."""
+        if not isinstance(state, SensorState) or state.missing_state:
+            return
+        sensor_name = self.key_to_sensor.get(state.key)
+        if not sensor_name or sensor_name not in self.sensor_states:
+            return
+        self.sensor_states[sensor_name].append(state.state)
+        for expected_value, future in self._expectations.get(sensor_name, []):
+            if not future.done() and (
+                expected_value is self._ANY or state.state == expected_value
+            ):
+                future.set_result(True)
+                break
+
+    async def await_change(
+        self, future: asyncio.Future, name: str, timeout: float = 2.0
+    ) -> None:
+        """Wait for a sensor future to resolve; fail the test on timeout."""
+        try:
+            await asyncio.wait_for(future, timeout=timeout)
+        except TimeoutError:
+            import pytest
+
+            pytest.fail(
+                f"Timeout waiting for {name} change. Received sensor states:\n"
+                f"  {name}: {self.sensor_states[name]}\n"
+            )
+
+    async def await_must_not_change(
+        self, future: asyncio.Future, name: str, timeout: float = 2.0
+    ) -> None:
+        """Assert a sensor future does NOT resolve within the timeout."""
+        try:
+            await asyncio.wait_for(future, timeout=timeout)
+        except TimeoutError:
+            return  # Expected
+        import pytest
+
+        pytest.fail(
+            f"{name} change should not have been triggered, but was. "
+            f"Received sensor states:\n  {name}: {self.sensor_states[name]}\n"
+        )
+
+    async def await_all(
+        self, futures: dict[str, asyncio.Future], timeout: float = 2.0
+    ) -> None:
+        """Await every future in *futures*, failing with per-sensor diagnostics."""
+        for name, future in futures.items():
+            await self.await_change(future, name, timeout=timeout)
+
+    async def setup_and_start_scenario(self, client) -> list:
+        """Wire up subscriptions, wait for initial states, press Start Scenario."""
+        entities, _ = await client.list_entities_services()
+        self.key_to_sensor.update(
+            build_key_to_entity_mapping(entities, list(self.sensor_states.keys()))
+        )
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(self.on_state))
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            import pytest
+
+            pytest.fail("Timeout waiting for initial states")
+        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
+        assert start_btn is not None, "Start Scenario button not found"
+        client.button_command(start_btn.key)
+        return entities

tests/integration/test_uart_mock_modbus.py

@@ -14,15 +14,67 @@ test_uart_mock_modbus_no_threshold :
 from __future__ import annotations
 
 import asyncio
-from pathlib import Path
+from collections.abc import Callable
+from dataclasses import dataclass
 
-from aioesphomeapi import ButtonInfo, EntityState, SensorState
+from aioesphomeapi import NumberInfo
 import pytest
 
-from .state_utils import InitialStateHelper, build_key_to_entity_mapping, find_entity
+from .state_utils import SensorTracker, find_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
+@dataclass
+class RegisterTestCase:
+    """Test parameters for a single modbus register write/read round-trip."""
+
+    initial_value: object
+    write_number_name: str
+    write_value: float
+    post_write_value: object
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_modbus_line_callback() -> tuple[Callable[[str], None], list[str], list[str]]:
+    """Return a (callback, error_lines, warning_lines) tuple for tracking modbus log output.
+
+    Only captures bus-level modbus messages ([modbus:]), not modbus_controller
+    scheduling noise (e.g. "Duplicate modbus command found").
+    """
+    error_log_lines: list[str] = []
+    warning_log_lines: list[str] = []
+
+    def line_callback(line: str) -> None:
+        if "[E][modbus:" in line:
+            error_log_lines.append(line)
+        if "[W][modbus:" in line:
+            warning_log_lines.append(line)
+
+    return line_callback, error_log_lines, warning_log_lines
+
+
+def _assert_no_modbus_errors(
+    error_log_lines: list[str], warning_log_lines: list[str]
+) -> None:
+    assert len(error_log_lines) == 0, (
+        "Expect no errors logged by the modbus mock, but got:\n"
+        + "\n".join(error_log_lines)
+    )
+    assert len(warning_log_lines) == 0, (
+        "Expect no warnings logged by the modbus mock, but got:\n"
+        + "\n".join(warning_log_lines)
+    )
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
 @pytest.mark.asyncio
 async def test_uart_mock_modbus(
     yaml_config: str,
@@ -30,127 +82,41 @@ async def test_uart_mock_modbus(
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
     """Test basic modbus data parsing."""
-    # Replace external component path placeholder
-    external_components_path = str(
-        Path(__file__).parent / "fixtures" / "external_components"
+
+    tracker = SensorTracker(
+        [
+            "basic_register",
+            "delayed_response",
+            "late_response",
+            "no_response",
+            "exception_response",
+        ]
     )
-    yaml_config = yaml_config.replace(
-        "EXTERNAL_COMPONENT_PATH", external_components_path
-    )
-
-    loop = asyncio.get_running_loop()
-
-    # Track sensor state updates (after initial state is swallowed)
-    sensor_states: dict[str, list[float]] = {
-        "basic_register": [],
-        "delayed_response": [],
-        "late_response": [],
-        "no_response": [],
-        "exception_response": [],
-    }
-
-    basic_register_changed = loop.create_future()
-    delayed_response_changed = loop.create_future()
-    late_response_changed = loop.create_future()
-    no_response_changed = loop.create_future()
-    exception_response_changed = loop.create_future()
-
-    def on_state(state: EntityState) -> None:
-        if isinstance(state, SensorState) and not state.missing_state:
-            sensor_name = key_to_sensor.get(state.key)
-            if sensor_name and sensor_name in sensor_states:
-                sensor_states[sensor_name].append(state.state)
-                if (
-                    sensor_name == "basic_register"
-                    and state.state == 259.0
-                    and not basic_register_changed.done()
-                ):
-                    basic_register_changed.set_result(True)
-                elif (
-                    sensor_name == "delayed_response"
-                    and state.state == 255.0
-                    and not delayed_response_changed.done()
-                ):
-                    delayed_response_changed.set_result(True)
-                elif (
-                    sensor_name == "late_response" and not late_response_changed.done()
-                ):
-                    late_response_changed.set_result(True)
-                elif sensor_name == "no_response" and not no_response_changed.done():
-                    no_response_changed.set_result(True)
-                elif (
-                    sensor_name == "exception_response"
-                    and not exception_response_changed.done()
-                ):
-                    exception_response_changed.set_result(True)
+    basic_register_changed = tracker.expect("basic_register", 259.0)
+    delayed_response_changed = tracker.expect("delayed_response", 255.0)
+    # late_response / no_response / exception_response: expect *any* value
+    # (these should never fire, so we use a permissive match via expect_any)
+    late_response_changed = tracker.expect_any("late_response")
+    no_response_changed = tracker.expect_any("no_response")
+    exception_response_changed = tracker.expect_any("exception_response")
 
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
     ):
-        entities, _ = await client.list_entities_services()
+        await tracker.setup_and_start_scenario(client)
 
-        # Build key mappings for all sensor types
-        all_names = list(sensor_states.keys())
-        key_to_sensor = build_key_to_entity_mapping(entities, all_names)
-
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
-
-        # Start the UART mock scenario now that we're subscribed
-        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
-        assert start_btn is not None, "Start Scenario button not found"
-        client.button_command(start_btn.key)
-
-        try:
-            await asyncio.wait_for(delayed_response_changed, timeout=2.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for delayed_response change. Received sensor states:\n"
-                f"  delayed_response: {sensor_states['delayed_response']}\n"
-            )
-
-        try:
-            await asyncio.wait_for(late_response_changed, timeout=2.0)
-            pytest.fail(
-                f"late_response change should not have been triggered, but was. Received sensor states:\n"
-                f"  late_response: {sensor_states['late_response']}\n"
-            )
-        except TimeoutError:
-            pass  # Expected timeout since we never inject a response for late_response
-
-        try:
-            await asyncio.wait_for(no_response_changed, timeout=2.0)
-            pytest.fail(
-                f"no_response change should not have been triggered, but was. Received sensor states:\n"
-                f"  no_response: {sensor_states['no_response']}\n"
-            )
-        except TimeoutError:
-            pass  # Expected timeout since we never inject a response for no_response
-
-        # Wait for basic register to be updated with successful parse
-        try:
-            await asyncio.wait_for(basic_register_changed, timeout=2.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for Basic Register change. Received sensor states:\n"
-                f"  basic_register: {sensor_states['basic_register']}\n"
-            )
-
-        try:
-            await asyncio.wait_for(exception_response_changed, timeout=2.0)
-            pytest.fail(
-                f"exception_response change should not have been triggered, but was. Received sensor states:\n"
-                f"  exception_response: {sensor_states['exception_response']}\n"
-            )
-        except TimeoutError:
-            pass
+        await tracker.await_change(delayed_response_changed, "delayed_response")
+        await tracker.await_change(basic_register_changed, "basic_register")
+        # Run all "must not change" checks concurrently — each waits the full
+        # timeout, so sequential execution would multiply the wall time.
+        await asyncio.gather(
+            tracker.await_must_not_change(late_response_changed, "late_response"),
+            tracker.await_must_not_change(no_response_changed, "no_response"),
+            tracker.await_must_not_change(
+                exception_response_changed, "exception_response"
+            ),
+        )
 
 
 @pytest.mark.asyncio
@@ -159,69 +125,17 @@ async def test_uart_mock_modbus_timing(
     run_compiled: RunCompiledFunction,
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    """Test basic modbus data parsing."""
-    # Replace external component path placeholder
-    external_components_path = str(
-        Path(__file__).parent / "fixtures" / "external_components"
-    )
-    yaml_config = yaml_config.replace(
-        "EXTERNAL_COMPONENT_PATH", external_components_path
-    )
+    """Test modbus timing with multi-register SDM meter response."""
 
-    loop = asyncio.get_running_loop()
-
-    # Track sensor state updates (after initial state is swallowed)
-    sensor_states: dict[str, list[float]] = {
-        "sdm_voltage": [],
-    }
-
-    voltage_changed = loop.create_future()
-
-    def on_state(state: EntityState) -> None:
-        if isinstance(state, SensorState) and not state.missing_state:
-            sensor_name = key_to_sensor.get(state.key)
-            if sensor_name and sensor_name in sensor_states:
-                sensor_states[sensor_name].append(state.state)
-                # Check if this is a good voltage reading (243V)
-                if (
-                    sensor_name == "sdm_voltage"
-                    and state.state > 200.0
-                    and not voltage_changed.done()
-                ):
-                    voltage_changed.set_result(True)
+    tracker = SensorTracker(["sdm_voltage"])
+    voltage_changed = tracker.expect_any("sdm_voltage")
 
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
     ):
-        entities, _ = await client.list_entities_services()
-
-        # Build key mappings for all sensor types
-        all_names = list(sensor_states.keys())
-        key_to_sensor = build_key_to_entity_mapping(entities, all_names)
-
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
-
-        # Start the UART mock scenario now that we're subscribed
-        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
-        assert start_btn is not None, "Start Scenario button not found"
-        client.button_command(start_btn.key)
-
-        # Wait for voltage to be updated with successful parse
-        try:
-            await asyncio.wait_for(voltage_changed, timeout=2.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for SDM voltage change. Received sensor states:\n"
-                f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
-            )
+        await tracker.setup_and_start_scenario(client)
+        await tracker.await_change(voltage_changed, "sdm_voltage")
 
 
 @pytest.mark.asyncio
@@ -234,66 +148,187 @@ async def test_uart_mock_modbus_no_threshold(
 
     Without the 50ms fallback timeout, the chunked response with a 40ms gap
     between USB packets would cause a false timeout and CRC failure cascade.
+    Bus-level warnings (CRC failures, buffer clears) are expected during
+    chunked reassembly — the test only verifies the final value arrives.
     """
-    # Replace external component path placeholder
-    external_components_path = str(
-        Path(__file__).parent / "fixtures" / "external_components"
-    )
-    yaml_config = yaml_config.replace(
-        "EXTERNAL_COMPONENT_PATH", external_components_path
-    )
 
-    loop = asyncio.get_running_loop()
-
-    # Track sensor state updates (after initial state is swallowed)
-    sensor_states: dict[str, list[float]] = {
-        "sdm_voltage": [],
-    }
-
-    voltage_changed = loop.create_future()
-
-    def on_state(state: EntityState) -> None:
-        if isinstance(state, SensorState) and not state.missing_state:
-            sensor_name = key_to_sensor.get(state.key)
-            if sensor_name and sensor_name in sensor_states:
-                sensor_states[sensor_name].append(state.state)
-                # Check if this is a good voltage reading (243V)
-                if (
-                    sensor_name == "sdm_voltage"
-                    and state.state > 200.0
-                    and not voltage_changed.done()
-                ):
-                    voltage_changed.set_result(True)
+    tracker = SensorTracker(["sdm_voltage"])
+    voltage_changed = tracker.expect_any("sdm_voltage")
 
     async with (
         run_compiled(yaml_config),
         api_client_connected() as client,
     ):
-        entities, _ = await client.list_entities_services()
+        await tracker.setup_and_start_scenario(client)
+        await tracker.await_change(voltage_changed, "sdm_voltage")
 
-        # Build key mappings for all sensor types
-        all_names = list(sensor_states.keys())
-        key_to_sensor = build_key_to_entity_mapping(entities, all_names)
 
-        # Set up initial state helper
-        initial_state_helper = InitialStateHelper(entities)
-        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+@pytest.mark.asyncio
+@pytest.mark.xfail(
+    reason="Modbus parser cannot handle server responses from other devices on the bus. Fix tracked in PR #11969.",
+    strict=True,
+)
+async def test_uart_mock_modbus_server(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test modbus server parsing with peer traffic on a shared bus."""
 
-        try:
-            await initial_state_helper.wait_for_initial_states()
-        except TimeoutError:
-            pytest.fail("Timeout waiting for initial states")
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-        # Start the UART mock scenario now that we're subscribed
-        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
-        assert start_btn is not None, "Start Scenario button not found"
-        client.button_command(start_btn.key)
+    tracker = SensorTracker(
+        ["basic_read", "read_after_peer_response", "read_after_peer_timeout"]
+    )
+    futures = tracker.expect_all(
+        {
+            "basic_read": 1,
+            "read_after_peer_response": 1,
+            "read_after_peer_timeout": 1,
+        }
+    )
 
-        # Wait for voltage to be updated with successful parse
-        try:
-            await asyncio.wait_for(voltage_changed, timeout=2.0)
-        except TimeoutError:
-            pytest.fail(
-                f"Timeout waiting for SDM voltage change. Received sensor states:\n"
-                f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        await tracker.setup_and_start_scenario(client)
+        await tracker.await_all(futures)
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_server_controller(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test server/controller functionality for all read register types."""
+
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
+
+    expected_values = {
+        "reg_u_word": 99,
+        "reg_s_word": -99,
+        "reg_u_dword": 16909060,
+        "reg_s_dword": -16909060,
+        "reg_u_dword_r": pytest.approx(67305985),
+        "reg_s_dword_r": pytest.approx(-67305985),
+        "reg_u_qword": pytest.approx(72623859790382856),
+        "reg_s_qword": pytest.approx(-72623859790382856),
+        "reg_u_qword_r": pytest.approx(578437695752307201),
+        "reg_s_qword_r": pytest.approx(-578437695752307201),
+        "reg_fp32": pytest.approx(3.14),
+        "reg_fp32_r": pytest.approx(3.14),
+    }
+    tracker = SensorTracker(list(expected_values.keys()))
+    futures = tracker.expect_all(expected_values)
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        await tracker.setup_and_start_scenario(client)
+        await tracker.await_all(futures)
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_server_controller_write(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test server/controller write functionality for all register value types.
+
+    Verifies that writing to modbus server registers via the controller updates
+    the server's stored values, which are then read back correctly on the next poll.
+    All 12 value types are tested: U/S_WORD, U/S_DWORD(_R), U/S_QWORD(_R), FP32(_R).
+    """
+
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
+
+    register_test_cases: dict[str, RegisterTestCase] = {
+        "reg_u_word": RegisterTestCase(11, "write_u_word", 42, 42),
+        "reg_s_word": RegisterTestCase(-11, "write_s_word", -42, -42),
+        "reg_u_dword": RegisterTestCase(1001, "write_u_dword", 2002, 2002),
+        "reg_s_dword": RegisterTestCase(-1001, "write_s_dword", -2002, -2002),
+        "reg_u_dword_r": RegisterTestCase(3003, "write_u_dword_r", 4004, 4004),
+        "reg_s_dword_r": RegisterTestCase(-3003, "write_s_dword_r", -4004, -4004),
+        "reg_u_qword": RegisterTestCase(5005, "write_u_qword", 6006, 6006),
+        "reg_s_qword": RegisterTestCase(-5005, "write_s_qword", -6006, -6006),
+        "reg_u_qword_r": RegisterTestCase(7007, "write_u_qword_r", 8008, 8008),
+        "reg_s_qword_r": RegisterTestCase(-7007, "write_s_qword_r", -8008, -8008),
+        "reg_fp32": RegisterTestCase(
+            pytest.approx(1.5, abs=0.01),
+            "write_fp32",
+            3.14,
+            pytest.approx(3.14, abs=0.01),
+        ),
+        "reg_fp32_r": RegisterTestCase(
+            pytest.approx(2.5, abs=0.01),
+            "write_fp32_r",
+            6.28,
+            pytest.approx(6.28, abs=0.01),
+        ),
+    }
+
+    tracker = SensorTracker(list(register_test_cases.keys()))
+
+    # Phase 1: expect initial baseline values
+    initial_futures = tracker.expect_all(
+        {name: case.initial_value for name, case in register_test_cases.items()}
+    )
+    # Phase 2: expect post-write values (registered now so on_state can match them)
+    written_futures = tracker.expect_all(
+        {name: case.post_write_value for name, case in register_test_cases.items()}
+    )
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        entities = await tracker.setup_and_start_scenario(client)
+
+        # Wait for initial baseline values to confirm the controller <-> server
+        # connection is working before issuing writes
+        await tracker.await_all(initial_futures, timeout=4.0)
+
+        # Issue write commands for all register types
+        for case in register_test_cases.values():
+            entity = find_entity(entities, case.write_number_name, NumberInfo)
+            assert entity is not None, (
+                f"{case.write_number_name} number entity not found"
             )
+            client.number_command(entity.key, case.write_value)
+
+        # Wait for sensors to reflect the written values (round-trip write+read)
+        await tracker.await_all(written_futures, timeout=4.0)
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)
+
+
+@pytest.mark.asyncio
+@pytest.mark.xfail(
+    reason="Modbus parser cannot handle server responses from other devices on the bus. Fix tracked in PR #11969.",
+    strict=True,
+)
+async def test_uart_mock_modbus_server_controller_multiple(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test server/controller functionality with multiple servers."""
+
+    line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
+
+    expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
+    tracker = SensorTracker(list(expected_values.keys()))
+    futures = tracker.expect_all(expected_values)
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        await tracker.setup_and_start_scenario(client)
+        await tracker.await_all(futures)
+        _assert_no_modbus_errors(error_log_lines, warning_log_lines)