Merge branch 'platformio-pch-libretiny' into pch-strict-ci

tests/component_tests/light/test_color_name.py

@@ -0,0 +1,119 @@
+"""Tests for the light `color` option."""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.light.automation import (
+    LIGHT_CONTROL_ACTION_SCHEMA,
+    LIGHT_STATE_SCHEMA,
+)
+from esphome.const import (
+    CONF_BLUE,
+    CONF_COLOR_BRIGHTNESS,
+    CONF_GREEN,
+    CONF_ID,
+    CONF_RED,
+)
+
+LOGGER_NAME: str = "esphome.components.light.automation"
+
+
+def test_color_name_sets_rgb() -> None:
+    result = LIGHT_STATE_SCHEMA({"color": "Tomato"})
+    assert "color" not in result
+    assert result[CONF_RED] == 1.0
+    assert result[CONF_GREEN] == pytest.approx(0x63 / 0xFF)
+    assert result[CONF_BLUE] == pytest.approx(0x47 / 0xFF)
+    assert result[CONF_COLOR_BRIGHTNESS] == 1.0
+
+
+def test_color_name_in_control_action() -> None:
+    result = LIGHT_CONTROL_ACTION_SCHEMA({CONF_ID: "test_light", "color": "blue"})
+    assert (result[CONF_RED], result[CONF_GREEN], result[CONF_BLUE]) == (0, 0, 1.0)
+
+
+def test_unknown_color_name() -> None:
+    with pytest.raises(cv.Invalid, match="notacolor"):
+        LIGHT_STATE_SCHEMA({"color": "notacolor"})
+
+
+def test_color_name_conflicts_with_rgb() -> None:
+    with pytest.raises(cv.Invalid, match="cannot be used with"):
+        LIGHT_STATE_SCHEMA({"color": "red", CONF_GREEN: 0.5})
+
+
+def test_dark_color_sets_color_brightness() -> None:
+    result = LIGHT_STATE_SCHEMA({"color": "darkred"})
+    assert result[CONF_RED] == 1.0
+    assert result[CONF_GREEN] == 0.0
+    assert result[CONF_BLUE] == 0.0
+    assert result[CONF_COLOR_BRIGHTNESS] == pytest.approx(0x8B / 0xFF)
+
+
+def test_black_sets_zero_color_brightness() -> None:
+    result = LIGHT_STATE_SCHEMA({"color": "black"})
+    assert result[CONF_COLOR_BRIGHTNESS] == 0.0
+    assert (result[CONF_RED], result[CONF_GREEN], result[CONF_BLUE]) == (0, 0, 0)
+
+
+def test_explicit_color_brightness_wins(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.WARNING, logger=LOGGER_NAME):
+        result = LIGHT_STATE_SCHEMA({"color": "darkred", CONF_COLOR_BRIGHTNESS: 0.25})
+    assert result[CONF_COLOR_BRIGHTNESS] == 0.25
+    assert "overrides the brightness of color 'darkred'" in caplog.text
+
+
+def test_explicit_color_brightness_full_color(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger=LOGGER_NAME):
+        result = LIGHT_STATE_SCHEMA({"color": "red", CONF_COLOR_BRIGHTNESS: 0.25})
+    assert result[CONF_COLOR_BRIGHTNESS] == 0.25
+    assert not [r for r in caplog.records if r.name == LOGGER_NAME]
+
+
+@pytest.mark.parametrize("value", [0x223344, "0x223344", "0X223344"])
+def test_hex_color(value: int | str) -> None:
+    result = LIGHT_STATE_SCHEMA({"color": value})
+    assert "color" not in result
+    assert result[CONF_COLOR_BRIGHTNESS] == pytest.approx(0x44 / 0xFF)
+    assert result[CONF_BLUE] == 1.0
+    assert result[CONF_RED] == pytest.approx(0x22 / 0x44)
+    assert result[CONF_GREEN] == pytest.approx(0x33 / 0x44)
+
+
+def test_hex_color_full_level() -> None:
+    result = LIGHT_STATE_SCHEMA({"color": 0xFF8000})
+    assert result[CONF_COLOR_BRIGHTNESS] == 1.0
+    assert result[CONF_RED] == 1.0
+    assert result[CONF_GREEN] == pytest.approx(0x80 / 0xFF)
+    assert result[CONF_BLUE] == 0.0
+
+
+def test_hex_color_black() -> None:
+    result = LIGHT_STATE_SCHEMA({"color": 0})
+    assert result[CONF_COLOR_BRIGHTNESS] == 0.0
+    assert (result[CONF_RED], result[CONF_GREEN], result[CONF_BLUE]) == (0, 0, 0)
+
+
+@pytest.mark.parametrize("value", [0x1000000, -1, "0xGG", "0x1000000"])
+def test_hex_color_invalid(value: int | str) -> None:
+    with pytest.raises(cv.Invalid):
+        LIGHT_STATE_SCHEMA({"color": value})
+
+
+def test_hex_color_conflicts_with_rgb() -> None:
+    with pytest.raises(cv.Invalid, match="cannot be used with"):
+        LIGHT_STATE_SCHEMA({"color": 0x223344, CONF_RED: 0.5})
+
+
+def test_explicit_color_brightness_warning_names_hex_color(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger=LOGGER_NAME):
+        LIGHT_STATE_SCHEMA({"color": 0x223344, CONF_COLOR_BRIGHTNESS: 0.25})
+    assert "overrides the brightness of color '0x223344'" in caplog.text

tests/components/light/common.yaml

@@ -85,6 +85,12 @@ esphome:
           red: 100%
           green: 100%
           blue: 1.0
+      - light.turn_on:
+          id: test_rgb_light
+          color: Tomato
+      - light.turn_on:
+          id: test_rgb_light
+          color: 0x223344
       - light.control:
           id: test_monochromatic_light
           state: on
@@ -157,6 +163,17 @@ esphome:
           green: 100%
           blue: 0%
           transition_length: 1s
+      - light.addressable_set:
+          id: test_addressable_transition
+          range_from: 0
+          range_to: 4
+          color: darkred
+      - light.addressable_set:
+          id: test_addressable_transition
+          color: tomato
+      - light.addressable_set:
+          id: test_addressable_transition
+          color: 0x223344
 
 # Exercise light actions inside a trigger with non-empty Ts (number on_value
 # passes float).
@@ -336,6 +353,15 @@ light:
       red: 100%
       green: 50%
       blue: 50%
+  # Test initial_state: with a color name (a dark color also sets color_brightness)
+  - platform: rgb
+    id: test_rgb_light_initial_state_color
+    name: RGB Light Initial State Color
+    red: test_ledc_1
+    green: test_ledc_2
+    blue: test_ledc_3
+    initial_state:
+      color: darkred
   # Test initial_state: with every templatable field set via !lambda (color_mode is
   # the only field that isn't templatable, so it's the only one left as a literal)
   - platform: rgbww

tests/components/mk2pvrouter/binary_sensor/mk2pvrouter_binary_sensor_test.cpp

@@ -0,0 +1,21 @@
+#include <gtest/gtest.h>
+
+#include "esphome/components/mk2pvrouter/binary_sensor/mk2pvrouter_binary_sensor.h"
+
+namespace esphome::mk2pvrouter::testing {
+
+TEST(Mk2PVRouterBinarySensorTest, ZeroPublishesOff) {
+  Mk2PVRouterBinarySensor sensor("R1");
+  sensor.publish_val("0");
+  ASSERT_TRUE(sensor.has_state());
+  EXPECT_FALSE(sensor.state);
+}
+
+TEST(Mk2PVRouterBinarySensorTest, OnePublishesOn) {
+  Mk2PVRouterBinarySensor sensor("R1");
+  sensor.publish_val("1");
+  ASSERT_TRUE(sensor.has_state());
+  EXPECT_TRUE(sensor.state);
+}
+
+}  // namespace esphome::mk2pvrouter::testing

tests/components/mk2pvrouter/common.yaml

@@ -59,3 +59,22 @@ sensor:
     name: Relay 10 State
     tag: R10
     mk2pvrouter_id: test_mk2pvrouter
+
+binary_sensor:
+  # Test relay as binary sensor
+  - platform: mk2pvrouter
+    name: Relay 2 Binary
+    tag: R2
+    mk2pvrouter_id: test_mk2pvrouter
+
+text_sensor:
+  - platform: mk2pvrouter
+    name: S per MC
+    tag: S_MC
+    mk2pvrouter_id: test_mk2pvrouter
+
+  # Test another text tag
+  - platform: mk2pvrouter
+    name: Status
+    tag: STATUS
+    mk2pvrouter_id: test_mk2pvrouter

tests/components/mk2pvrouter/text_sensor/mk2pvrouter_text_sensor_test.cpp

@@ -0,0 +1,21 @@
+#include <gtest/gtest.h>
+
+#include "esphome/components/mk2pvrouter/text_sensor/mk2pvrouter_text_sensor.h"
+
+namespace esphome::mk2pvrouter::testing {
+
+TEST(Mk2PVRouterTextSensorTest, PublishesValueVerbatim) {
+  Mk2PVRouterTextSensor sensor("S_MC");
+  sensor.publish_val("-1234");
+  ASSERT_TRUE(sensor.has_state());
+  EXPECT_EQ(sensor.get_state(), "-1234");
+}
+
+TEST(Mk2PVRouterTextSensorTest, ShorterValueReplacesPreviousOne) {
+  Mk2PVRouterTextSensor sensor("S_MC");
+  sensor.publish_val("-1234");
+  sensor.publish_val("5");
+  EXPECT_EQ(sensor.get_state(), "5");
+}
+
+}  // namespace esphome::mk2pvrouter::testing

tests/components/pn71xx/pn71xx_test.cpp

@@ -0,0 +1,194 @@
+#include <gtest/gtest.h>
+
+#include <deque>
+
+#include "esphome/components/pn71xx/pn71xx.h"
+
+namespace esphome::pn71xx {
+
+namespace {
+
+// Stands in for the bus: records every frame written and replays queued frames on read.
+class FakePN71xx : public PN71xx {
+ public:
+  using PN71xx::card_emu_t4t_get_response_;
+  using PN71xx::transceive_;
+
+  std::deque<std::vector<uint8_t>> to_read;
+  std::vector<std::vector<uint8_t>> written;
+  uint8_t write_failures{0};
+
+ protected:
+  uint8_t verify_reset(nfc::NciMessage &rx, bool reset_config) override { return nfc::STATUS_OK; }
+  uint8_t process_init_response(nfc::NciMessage &rx) override { return nfc::STATUS_OK; }
+  std::span<const uint8_t> pmu_config() const override { return {}; }
+  std::span<const uint8_t> listen_mode_routing_config() const override { return {}; }
+
+  uint8_t read_nfcc(nfc::NciMessage &rx, uint16_t timeout) override {
+    if (this->to_read.empty())
+      return nfc::STATUS_FAILED;
+    rx = nfc::NciMessage(this->to_read.front());
+    this->to_read.pop_front();
+    return nfc::STATUS_OK;
+  }
+  uint8_t write_nfcc(nfc::NciMessage &tx) override {
+    if (this->write_failures > 0) {
+      this->write_failures--;
+      return nfc::STATUS_FAILED;
+    }
+    this->written.push_back(tx.encode());
+    return nfc::STATUS_OK;
+  }
+};
+
+std::vector<uint8_t> apdu(std::initializer_list<uint8_t> bytes) {
+  std::vector<uint8_t> msg = {nfc::NCI_PKT_MT_DATA, 0x00, static_cast<uint8_t>(bytes.size())};
+  msg.insert(msg.end(), bytes);
+  return msg;
+}
+
+std::vector<uint8_t> respond(FakePN71xx &nfcc, std::initializer_list<uint8_t> bytes) {
+  std::vector<uint8_t> response;
+  nfcc.card_emu_t4t_get_response_(apdu(bytes), response);
+  return response;
+}
+
+void select_ndef_file(FakePN71xx &nfcc) {
+  respond(nfcc, {0x00, 0xA4, 0x04, 0x00, 0x07, 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00});
+  respond(nfcc, {0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x04});
+}
+
+const std::vector<uint8_t> SW_OK = {0x90, 0x00};
+const std::vector<uint8_t> SW_NOT_FOUND = {0x6A, 0x82};
+
+}  // namespace
+
+// A timed-out read must not cause the command to be sent again (NCI forbids a second command before the response).
+TEST(PN71xxTransceive, ReadTimeoutDoesNotResend) {
+  FakePN71xx nfcc;
+  nfc::NciMessage tx(nfc::NCI_PKT_MT_CTRL_COMMAND, nfc::RF_GID, nfc::RF_DEACTIVATE_OID, {0x00});
+  nfc::NciMessage rx;
+  EXPECT_NE(nfcc.transceive_(tx, rx), nfc::STATUS_OK);
+  EXPECT_EQ(nfcc.written.size(), 1u);
+}
+
+// A notification with the same GID/OID as the response (RF_DEACTIVATE_NTF) is not mistaken for it.
+TEST(PN71xxTransceive, SkipsNotificationAheadOfResponse) {
+  FakePN71xx nfcc;
+  nfcc.to_read.push_back({0x61, 0x06, 0x02, 0x00, 0x00});
+  nfcc.to_read.push_back({0x41, 0x06, 0x01, 0x00});
+  nfc::NciMessage tx(nfc::NCI_PKT_MT_CTRL_COMMAND, nfc::RF_GID, nfc::RF_DEACTIVATE_OID, {0x00});
+  nfc::NciMessage rx;
+  EXPECT_EQ(nfcc.transceive_(tx, rx), nfc::STATUS_OK);
+  EXPECT_EQ(rx.get_message(), (std::vector<uint8_t>{0x41, 0x06, 0x01, 0x00}));
+  EXPECT_EQ(nfcc.written.size(), 1u);
+}
+
+TEST(PN71xxTransceive, NotificationAloneIsNotAResponse) {
+  FakePN71xx nfcc;
+  nfcc.to_read.push_back({0x61, 0x06, 0x02, 0x00, 0x00});
+  nfc::NciMessage tx(nfc::NCI_PKT_MT_CTRL_COMMAND, nfc::RF_GID, nfc::RF_DEACTIVATE_OID, {0x00});
+  nfc::NciMessage rx;
+  EXPECT_NE(nfcc.transceive_(tx, rx), nfc::STATUS_OK);
+}
+
+// A late response to an earlier, timed-out command must not be taken as the response to this one.
+TEST(PN71xxTransceive, SkipsStaleResponseFromEarlierCommand) {
+  FakePN71xx nfcc;
+  nfcc.to_read.push_back({0x41, 0x06, 0x01, 0x00});  // RF_DEACTIVATE_RSP, arriving late
+  nfcc.to_read.push_back({0x41, 0x03, 0x01, 0x00});  // RF_DISCOVER_RSP
+  nfc::NciMessage tx(nfc::NCI_PKT_MT_CTRL_COMMAND, nfc::RF_GID, nfc::RF_DISCOVER_OID, {0x00});
+  nfc::NciMessage rx;
+  EXPECT_EQ(nfcc.transceive_(tx, rx), nfc::STATUS_OK);
+  EXPECT_EQ(rx.get_message(), (std::vector<uint8_t>{0x41, 0x03, 0x01, 0x00}));
+  EXPECT_EQ(nfcc.written.size(), 1u);
+}
+
+// A refused write (e.g. NFCC in standby) is sent again.
+TEST(PN71xxTransceive, RefusedWriteIsRetried) {
+  FakePN71xx nfcc;
+  nfcc.write_failures = 1;
+  nfcc.to_read.push_back({0x41, 0x06, 0x01, 0x00});
+  nfc::NciMessage tx(nfc::NCI_PKT_MT_CTRL_COMMAND, nfc::RF_GID, nfc::RF_DEACTIVATE_OID, {0x00});
+  nfc::NciMessage rx;
+  EXPECT_EQ(nfcc.transceive_(tx, rx), nfc::STATUS_OK);
+  EXPECT_EQ(nfcc.written.size(), 1u);
+}
+
+TEST(PN71xxCardEmulation, CcReadOutOfRangeIsRejected) {
+  FakePN71xx nfcc;
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/test", false);
+  respond(nfcc, {0x00, 0xA4, 0x04, 0x00, 0x07, 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00});
+  respond(nfcc, {0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x03});
+  // the CC file is 15 bytes; reading 17 or reading far past its end must not return memory beyond it
+  EXPECT_EQ(respond(nfcc, {0x00, 0xB0, 0x00, 0x00, 0x11}), SW_NOT_FOUND);
+  respond(nfcc, {0x00, 0xA4, 0x04, 0x00, 0x07, 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00});
+  respond(nfcc, {0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x03});
+  EXPECT_EQ(respond(nfcc, {0x00, 0xB0, 0x01, 0x00, 0x0F}), SW_NOT_FOUND);
+}
+
+TEST(PN71xxCardEmulation, CcReadInRange) {
+  FakePN71xx nfcc;
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/test", false);
+  respond(nfcc, {0x00, 0xA4, 0x04, 0x00, 0x07, 0xD2, 0x76, 0x00, 0x00, 0x85, 0x01, 0x01, 0x00});
+  respond(nfcc, {0x00, 0xA4, 0x00, 0x0C, 0x02, 0xE1, 0x03});
+  auto response = respond(nfcc, {0x00, 0xB0, 0x00, 0x00, 0x0F});
+  ASSERT_EQ(response.size(), sizeof(CARD_EMU_T4T_CC) + 2);
+  EXPECT_TRUE(std::equal(std::begin(CARD_EMU_T4T_CC), std::end(CARD_EMU_T4T_CC), response.begin()));
+}
+
+// Reading the NDEF file in small chunks returns NLEN followed by the message, in order.
+TEST(PN71xxCardEmulation, ChunkedNdefReadMatchesFile) {
+  FakePN71xx nfcc;
+  auto message = std::make_shared<nfc::NdefMessage>();
+  message->add_uri_record("https://www.home-assistant.io/tag/0123456789abcdef");
+  const auto encoded = message->encode();
+  nfcc.set_tag_emulation_message(message);
+  select_ndef_file(nfcc);
+
+  std::vector<uint8_t> expected = {static_cast<uint8_t>(encoded.size() >> 8),
+                                   static_cast<uint8_t>(encoded.size() & 0xFF)};
+  expected.insert(expected.end(), encoded.begin(), encoded.end());
+
+  std::vector<uint8_t> file;
+  for (size_t offset = 0; offset < expected.size(); offset += 5) {
+    const uint8_t length = std::min<size_t>(5, expected.size() - offset);
+    auto response =
+        respond(nfcc, {0x00, 0xB0, static_cast<uint8_t>(offset >> 8), static_cast<uint8_t>(offset), length});
+    ASSERT_EQ(response.size(), length + 2u);
+    EXPECT_EQ(std::vector<uint8_t>(response.end() - 2, response.end()), SW_OK);
+    file.insert(file.end(), response.begin(), response.end() - 2);
+  }
+  EXPECT_EQ(file, expected);
+}
+
+TEST(PN71xxCardEmulation, NdefReadPastEndIsRejected) {
+  FakePN71xx nfcc;
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/test", false);
+  select_ndef_file(nfcc);
+  EXPECT_EQ(respond(nfcc, {0x00, 0xB0, 0x00, 0x02, 0xFD}), SW_NOT_FOUND);
+}
+
+TEST(PN71xxCardEmulation, TruncatedApdusAreRejected) {
+  FakePN71xx nfcc;
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/test", false);
+  select_ndef_file(nfcc);
+  EXPECT_EQ(respond(nfcc, {0x00, 0xB0, 0x00}), SW_NOT_FOUND);
+  select_ndef_file(nfcc);
+  // UPDATE BINARY claiming 16 bytes of data but carrying only 2
+  EXPECT_EQ(respond(nfcc, {0x00, 0xD6, 0x00, 0x00, 0x10, 0x00, 0x00}), SW_NOT_FOUND);
+}
+
+// A message too large for the emulated NDEF file is refused when it is set, keeping the previous one.
+TEST(PN71xxCardEmulation, OversizedMessageRejectedWhenSet) {
+  FakePN71xx nfcc;
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/test", false);
+  nfcc.set_tag_emulation_message("https://www.home-assistant.io/tag/" + std::string(300, 'x'), false);
+  select_ndef_file(nfcc);
+  auto response = respond(nfcc, {0x00, 0xB0, 0x00, 0x00, 0x02});
+  ASSERT_EQ(response.size(), 4u);
+  EXPECT_LT((response[0] << 8) | response[1], 0xFF - 2);
+  EXPECT_EQ(std::vector<uint8_t>(response.end() - 2, response.end()), SW_OK);
+}
+
+}  // namespace esphome::pn71xx

tests/components/snapshot/common.yaml

@@ -26,6 +26,11 @@ esphome:
     - snapshot.take:
         id: snapshot_rotated
         filename: !lambda 'return "rotated.bmp";'
+    - snapshot.take:
+        id: snapshot_display
+        filename: animation.gif
+        frames: 5
+        frame_rate: 10fps
 
 interval:
   # A generated name has the time in it, so this one can repeat.

tests/components/snapshot/snapshot_test.cpp

@@ -0,0 +1,125 @@
+#ifdef USE_HOST
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <chrono>
+#include <cstdlib>
+#include <filesystem>
+#include <fstream>
+#include <iterator>
+#include <new>
+#include <string>
+#include <thread>
+#include <vector>
+#include "esphome/components/snapshot/snapshot.h"
+#include "esphome/core/application.h"
+#include "esphome/core/hal.h"
+
+namespace esphome::snapshot::testing {
+namespace fs = std::filesystem;
+
+/// A display of a fixed size that can be told to fail a capture.
+class FakeSnapshot : public Snapshot {
+ public:
+  /// The capture with this number (counting from 1) fails. Zero means none do.
+  int fail_on{0};
+  int captures{0};
+
+ protected:
+  int snapshot_width() override { return 8; }
+  int snapshot_height() override { return 4; }
+  bool capture_bgr(uint8_t *dest, size_t row_stride) override {
+    this->captures++;
+    if (this->captures == this->fail_on)
+      return false;
+    for (int y = 0; y != 4; y++) {
+      for (int x = 0; x != 8 * 3; x++)
+        dest[y * row_stride + x] = static_cast<uint8_t>(x * 7 + y * 31 + this->captures);
+    }
+    return true;
+  }
+};
+
+class SnapshotAnimationTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    this->dir_ = fs::temp_directory_path() / "esphome_snapshot_test";
+    fs::remove_all(this->dir_);
+    fs::create_directories(this->dir_);
+    setenv("ESPHOME_SNAPSHOT_DIR", this->dir_.c_str(), 1);
+    // The test main does not construct App as generated code does, and recording needs its scheduler.
+    static const bool app_constructed = (new (&App) Application(), true);
+    (void) app_constructed;
+    App.pre_setup("test_snapshot", 10, "", 0);
+  }
+  void TearDown() override {
+    unsetenv("ESPHOME_SNAPSHOT_DIR");
+    fs::remove_all(this->dir_);
+  }
+
+  /// Names of the files in the snapshot directory, in order.
+  std::vector<std::string> files() const {
+    std::vector<std::string> names;
+    for (const auto &entry : fs::directory_iterator(this->dir_))
+      names.push_back(entry.path().filename().string());
+    std::sort(names.begin(), names.end());
+    return names;
+  }
+
+  /// Let the scheduler run for long enough that a fast recording has finished.
+  static void run_scheduler() {
+    for (int i = 0; i != 50; i++) {
+      App.scheduler.call(millis());
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+  }
+
+  fs::path dir_;
+};
+
+TEST_F(SnapshotAnimationTest, FailedFirstCaptureLeavesNoFile) {
+  FakeSnapshot display;
+  display.fail_on = 1;
+  EXPECT_FALSE(display.take_animation("first.gif", 3, 50.0f));
+  EXPECT_TRUE(this->files().empty());
+}
+
+TEST_F(SnapshotAnimationTest, FailedCaptureMidRecordingRemovesPartialFile) {
+  FakeSnapshot display;
+  display.fail_on = 2;
+  ASSERT_TRUE(display.take_animation("partial.gif", 3, 50.0f));
+  // The first frame is already in the file.
+  EXPECT_EQ(this->files(), std::vector<std::string>{"partial.gif"});
+
+  run_scheduler();
+  EXPECT_TRUE(this->files().empty());
+  // The failure ends the recording, so the display can record again.
+  display.fail_on = 0;
+  EXPECT_TRUE(display.take_animation("again.gif", 1, 50.0f));
+  EXPECT_EQ(this->files(), std::vector<std::string>{"again.gif"});
+}
+
+TEST_F(SnapshotAnimationTest, SecondRecordingIsRefusedUntilTheFirstEnds) {
+  FakeSnapshot display;
+  ASSERT_TRUE(display.take_animation("one.gif", 3, 50.0f));
+  EXPECT_FALSE(display.take_animation("two.gif", 3, 50.0f));
+  EXPECT_EQ(this->files(), std::vector<std::string>{"one.gif"});
+
+  run_scheduler();
+  std::ifstream in(this->dir_ / "one.gif", std::ios::binary);
+  std::string data((std::istreambuf_iterator<char>(in)), std::istreambuf_iterator<char>());
+  ASSERT_FALSE(data.empty());
+  EXPECT_EQ(data.back(), ';');  // the GIF trailer: the recording finished
+
+  EXPECT_TRUE(display.take_animation("two.gif", 1, 50.0f));
+  EXPECT_EQ(this->files(), (std::vector<std::string>{"one.gif", "two.gif"}));
+}
+
+TEST_F(SnapshotAnimationTest, RejectsNoFramesAndNoFrameRate) {
+  FakeSnapshot display;
+  EXPECT_FALSE(display.take_animation("none.gif", 0, 10.0f));
+  EXPECT_FALSE(display.take_animation("none.gif", 3, 0.0f));
+  EXPECT_TRUE(this->files().empty());
+}
+
+}  // namespace esphome::snapshot::testing
+#endif

tests/components/socket/test-ipv6.rp2040-ard.yaml

@@ -0,0 +1,4 @@
+substitutions:
+  network_enable_ipv6: "true"
+
+<<: !include common.yaml

tests/components/xiaomi_body_scale/common-ln.yaml

@@ -0,0 +1,16 @@
+sensor:
+  - platform: xiaomi_body_scale
+    mac_address: "AA:BB:CC:DD:EE:FF"
+    bindkey: "00112233445566778899aabbccddeeff"
+    weight:
+      name: "Scale Weight"
+    impedance_low:
+      name: "Scale Impedance Low"
+    impedance_high:
+      name: "Scale Impedance High"
+    heart_rate:
+      name: "Scale Heart Rate"
+    profile_id:
+      name: "Scale Profile ID"
+    stabilized:
+      name: "Scale Stabilized"

tests/components/xiaomi_body_scale/common.h

@@ -0,0 +1,78 @@
+#pragma once
+
+#include "esphome/components/xiaomi_body_scale/xiaomi_body_scale.h"
+#include "esphome/core/application.h"
+
+#include <array>
+#include <vector>
+
+namespace esphome::xiaomi_body_scale::testing {
+
+using Frame = std::array<uint8_t, 24>;
+
+// Real captures and keys from the xiaomi-ble test suite (Bluetooth-Devices/xiaomi-ble)
+static constexpr uint64_t SCALE_A = 0x8CD0B2F6BEEFULL;
+static constexpr const char *KEY_A = "0728974d657a4b60964c1b1677f35f7c";
+// Weight, heart rate and the 50 kHz impedance
+static constexpr Frame PACKET_1 = {0x48, 0x59, 0xd5, 0x3b, 0x0a, 0xbc, 0x07, 0x8f, 0xf2, 0x34, 0x8c, 0x84,
+                                   0x41, 0x38, 0xe9, 0x30, 0x22, 0x00, 0x00, 0x00, 0x9e, 0x53, 0x85, 0x99};
+// Only the 250 kHz impedance, the end of a bare feet measurement
+static constexpr Frame PACKET_2 = {0x48, 0x59, 0xd5, 0x3b, 0x0b, 0xd6, 0xef, 0x0b, 0x25, 0xdb, 0x72, 0x78,
+                                   0x5e, 0x7e, 0x2f, 0x46, 0xd6, 0x00, 0x00, 0x00, 0xd8, 0x64, 0x2d, 0xf6};
+// PACKET_1 re-encrypted with KEY_A and frame count 0xFF
+static constexpr Frame PACKET_1_COUNT_FF = {0x48, 0x59, 0xd5, 0x3b, 0xff, 0x6c, 0x2f, 0xd8, 0xdb, 0xaa, 0x70, 0xc7,
+                                            0xbb, 0x1f, 0x0a, 0x08, 0x9c, 0x00, 0x00, 0x00, 0x31, 0xce, 0xf1, 0x95};
+
+static constexpr uint64_t SCALE_B = 0x04AE4767C67CULL;
+static constexpr const char *KEY_B = "02d2900363ef629c736a4549677acbee";
+// Weight without impedance, the end of a measurement with socks
+static constexpr Frame SOCKS = {0x48, 0x59, 0xd5, 0x3b, 0x71, 0x53, 0x04, 0x38, 0xb5, 0x89, 0x4b, 0x24,
+                                0x2c, 0x20, 0x99, 0x08, 0xda, 0x00, 0x00, 0x00, 0x47, 0x9e, 0xcd, 0xa3};
+// All metrics zero, stepped off the scale
+static constexpr Frame STEP_OFF = {0x48, 0x59, 0xd5, 0x3b, 0x72, 0x03, 0x6c, 0x67, 0x94, 0x35, 0x5a, 0x19,
+                                   0xdb, 0xc8, 0x64, 0xbf, 0xb3, 0x00, 0x00, 0x00, 0xe4, 0x15, 0x1d, 0xc8};
+
+// S200: weight only, 62.25 kg for profile 1
+static constexpr uint64_t SCALE_S200 = 0xD07B6F27D729ULL;
+static constexpr const char *KEY_S200 = "653b1b10e1cb35e4ac5e60fa45f3bf29";
+static constexpr Frame S200_WEIGHT = {0x48, 0x59, 0x04, 0x4c, 0x01, 0x9a, 0x80, 0xa2, 0x75, 0x93, 0x90, 0x10,
+                                      0xf0, 0xab, 0xc4, 0xfa, 0xdc, 0x06, 0x00, 0x00, 0x3d, 0x29, 0xc0, 0x44};
+
+inline ble_device_base::ESPBTDevice advert(uint64_t address, const Frame &frame) {
+  // Service data AD structure for UUID 0xFE95
+  std::vector<uint8_t> adv = {static_cast<uint8_t>(frame.size() + 3), 0x16, 0x95, 0xFE};
+  adv.insert(adv.end(), frame.begin(), frame.end());
+  uint8_t mac[6];
+  for (size_t i = 0; i < 6; i++)
+    mac[i] = static_cast<uint8_t>(address >> (i * 8));
+  ble_device_base::ESPBTDevice device;
+  device.from_scan_result(mac, -60, 0, adv.data(), static_cast<uint16_t>(adv.size()));
+  return device;
+}
+
+struct Harness {
+  Harness(uint64_t address, const char *key) : scale(address, key) {
+    // The test main does not construct App as generated code does; the stabilized reset needs its scheduler
+    static const bool app_constructed = (new (&App) Application(), true);
+    (void) app_constructed;
+    App.pre_setup("test_scale", 10, "", 0);
+    this->scale.set_weight(&this->weight);
+    this->scale.set_impedance_low(&this->impedance_low);
+    this->scale.set_impedance_high(&this->impedance_high);
+    this->scale.set_heart_rate(&this->heart_rate);
+    this->scale.set_profile_id(&this->profile_id);
+    this->scale.set_stabilized(&this->stabilized);
+  }
+
+  ~Harness() {
+    // Drop the pending stabilized reset (id 0) while the scale is alive; App outlives each test
+    App.scheduler.cancel_timeout(&this->scale, 0u);
+    App.scheduler.call(millis());
+  }
+
+  XiaomiBodyScale scale;
+  sensor::Sensor weight, impedance_low, impedance_high, heart_rate, profile_id;
+  binary_sensor::BinarySensor stabilized;
+};
+
+}  // namespace esphome::xiaomi_body_scale::testing

tests/components/xiaomi_body_scale/common.yaml

@@ -0,0 +1,27 @@
+esp32_ble_tracker:
+  id: ble_tracker_hub
+
+sensor:
+  # Explicit ble_hub_id: pins the neutral binding as a declared key.
+  - platform: xiaomi_body_scale
+    ble_hub_id: ble_tracker_hub
+    mac_address: "AA:BB:CC:DD:EE:FF"
+    bindkey: "00112233445566778899aabbccddeeff"
+    weight:
+      name: "Scale Weight"
+    impedance_low:
+      name: "Scale Impedance Low"
+    impedance_high:
+      name: "Scale Impedance High"
+    heart_rate:
+      name: "Scale Heart Rate"
+    profile_id:
+      name: "Scale Profile ID"
+    stabilized:
+      name: "Scale Stabilized"
+  # No ble_hub_id: exercises the generated binding real configs use.
+  - platform: xiaomi_body_scale
+    mac_address: "AA:BB:CC:DD:EE:00"
+    bindkey: "ffeeddccbbaa99887766554433221100"
+    weight:
+      name: "Second Scale Weight"

tests/components/xiaomi_body_scale/sensor/xiaomi_body_scale_test.cpp

@@ -0,0 +1,112 @@
+#include "../common.h"
+
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <thread>
+
+namespace esphome::xiaomi_body_scale::testing {
+
+TEST(XiaomiBodyScale, DecodesWeightHeartRateAndLowImpedance) {
+  Harness h(SCALE_A, KEY_A);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_1)));
+  EXPECT_FLOAT_EQ(h.weight.state, 69.9f);
+  EXPECT_FLOAT_EQ(h.impedance_low.state, 543.2f);
+  EXPECT_FLOAT_EQ(h.heart_rate.state, 92.0f);
+  EXPECT_FLOAT_EQ(h.profile_id.state, 1.0f);
+  EXPECT_FALSE(h.impedance_high.has_state());
+  EXPECT_FALSE(h.stabilized.state);
+}
+
+TEST(XiaomiBodyScale, HighImpedancePacketCompletesTheMeasurement) {
+  Harness h(SCALE_A, KEY_A);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_2)));
+  EXPECT_FLOAT_EQ(h.impedance_high.state, 497.6f);
+  EXPECT_FALSE(h.impedance_low.has_state());
+  EXPECT_FALSE(h.weight.has_state());
+  EXPECT_TRUE(h.stabilized.state);
+}
+
+TEST(XiaomiBodyScale, StabilizedClearsAfterOneSecond) {
+  Harness h(SCALE_A, KEY_A);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_2)));
+  ASSERT_TRUE(h.stabilized.state);
+  const uint32_t start = millis();
+  while (h.stabilized.state && millis() - start < 2000) {
+    App.scheduler.call(millis());
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+  EXPECT_FALSE(h.stabilized.state);
+  EXPECT_GE(millis() - start, 900u);
+}
+
+TEST(XiaomiBodyScale, WeightWithoutImpedanceCompletesTheMeasurement) {
+  Harness h(SCALE_B, KEY_B);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_B, SOCKS)));
+  EXPECT_FLOAT_EQ(h.weight.state, 74.7f);
+  EXPECT_FALSE(h.impedance_low.has_state());
+  EXPECT_FALSE(h.impedance_high.has_state());
+  EXPECT_TRUE(h.stabilized.state);
+}
+
+TEST(XiaomiBodyScale, SteppingOffClearsStabilized) {
+  Harness h(SCALE_B, KEY_B);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_B, SOCKS)));
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_B, STEP_OFF)));
+  EXPECT_FALSE(h.stabilized.state);
+  EXPECT_FLOAT_EQ(h.weight.state, 74.7f);  // a zero weight is not published
+}
+
+TEST(XiaomiBodyScale, DecodesS200Weight) {
+  Harness h(SCALE_S200, KEY_S200);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_S200, S200_WEIGHT)));
+  EXPECT_FLOAT_EQ(h.weight.state, 62.25f);
+  EXPECT_FLOAT_EQ(h.profile_id.state, 1.0f);
+  // The S200 has no impedance or heart rate, and does not drive stabilized
+  EXPECT_FALSE(h.impedance_low.has_state());
+  EXPECT_FALSE(h.impedance_high.has_state());
+  EXPECT_FALSE(h.heart_rate.has_state());
+  EXPECT_FALSE(h.stabilized.has_state());
+}
+
+TEST(XiaomiBodyScale, IgnoresOtherAddresses) {
+  Harness h(SCALE_A, KEY_A);
+  EXPECT_FALSE(h.scale.parse_device(advert(SCALE_B, PACKET_1)));
+  EXPECT_FALSE(h.weight.has_state());
+}
+
+TEST(XiaomiBodyScale, RejectsAWrongBindkey) {
+  Harness h(SCALE_A, KEY_B);
+  EXPECT_FALSE(h.scale.parse_device(advert(SCALE_A, PACKET_1)));
+  EXPECT_FALSE(h.weight.has_state());
+}
+
+TEST(XiaomiBodyScale, RejectsAPlaintextFrame) {
+  Harness h(SCALE_A, KEY_A);
+  Frame plain = PACKET_1;
+  plain[0] &= ~0x08;
+  EXPECT_FALSE(h.scale.parse_device(advert(SCALE_A, plain)));
+  EXPECT_FALSE(h.weight.has_state());
+}
+
+TEST(XiaomiBodyScale, IgnoresARepeatedFrame) {
+  Harness h(SCALE_A, KEY_A);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_1)));
+  EXPECT_FALSE(h.scale.parse_device(advert(SCALE_A, PACKET_1)));
+}
+
+TEST(XiaomiBodyScale, AcceptsFrameCountFFAsTheFirstFrame) {
+  Harness h(SCALE_A, KEY_A);
+  ASSERT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_1_COUNT_FF)));
+  EXPECT_FLOAT_EQ(h.weight.state, 69.9f);
+}
+
+TEST(XiaomiBodyScale, AFailedFrameDoesNotBlockTheRealOne) {
+  Harness h(SCALE_A, KEY_A);
+  Frame forged = PACKET_1;
+  forged[23] ^= 0xFF;  // corrupt the tag, same frame count
+  EXPECT_FALSE(h.scale.parse_device(advert(SCALE_A, forged)));
+  EXPECT_TRUE(h.scale.parse_device(advert(SCALE_A, PACKET_1)));
+}
+
+}  // namespace esphome::xiaomi_body_scale::testing

tests/components/xiaomi_body_scale/test.esp32-idf.yaml

@@ -0,0 +1,3 @@
+packages:
+  ble: !include ../../test_build_components/common/ble/esp32-idf.yaml
+  xiaomi_body_scale: !include common.yaml

tests/components/xiaomi_body_scale/test.ln882x-ard.yaml

@@ -0,0 +1,3 @@
+packages:
+  ln882h_ble_tracker: !include ../ln882h_ble_tracker/common.yaml
+  xiaomi_body_scale: !include common-ln.yaml

tests/components/xiaomi_body_scale/validate.bk72xx-ard.yaml

@@ -0,0 +1,28 @@
+# Config-only: the CI bk72xx board cannot compile the BLE 5.x tracker
+bk72xx_ble_tracker:
+  id: ble_tracker_hub
+
+sensor:
+  # Explicit ble_hub_id: pins the neutral binding as a declared key.
+  - platform: xiaomi_body_scale
+    ble_hub_id: ble_tracker_hub
+    mac_address: "AA:BB:CC:DD:EE:FF"
+    bindkey: "00112233445566778899aabbccddeeff"
+    weight:
+      name: "Scale Weight"
+    impedance_low:
+      name: "Scale Impedance Low"
+    impedance_high:
+      name: "Scale Impedance High"
+    heart_rate:
+      name: "Scale Heart Rate"
+    profile_id:
+      name: "Scale Profile ID"
+    stabilized:
+      name: "Scale Stabilized"
+  # No ble_hub_id: exercises the generated binding real configs use.
+  - platform: xiaomi_body_scale
+    mac_address: "AA:BB:CC:DD:EE:00"
+    bindkey: "ffeeddccbbaa99887766554433221100"
+    weight:
+      name: "Second Scale Weight"

tests/integration/fixtures/snapshot_display.yaml

@@ -14,6 +14,47 @@ api:
             id: snapshot_display
             filename: !lambda return name;
 
+    - action: take_animation
+      variables:
+        name: string
+      then:
+        - snapshot.take:
+            id: snapshot_display
+            filename: !lambda return name;
+            frames: 5
+            frame_rate: 20fps
+
+    - action: take_noise_snapshot
+      variables:
+        name: string
+      then:
+        - snapshot.take:
+            id: noise_display
+            filename: !lambda return name;
+
+    # A busy picture needs more compression codes than a GIF dictionary holds, so recording it makes
+    # the encoder widen its codes and start the dictionary over.
+    - action: take_noise_animation
+      variables:
+        name: string
+      then:
+        - snapshot.take:
+            id: noise_display
+            filename: !lambda return name;
+            frames: 3
+            frame_rate: 20fps
+
+    # Slow enough that a second request arrives while this one is still recording.
+    - action: take_slow_animation
+      variables:
+        name: string
+      then:
+        - snapshot.take:
+            id: snapshot_display
+            filename: !lambda return name;
+            frames: 3
+            frame_rate: 2fps
+
 logger:
   level: DEBUG
 
@@ -26,3 +67,23 @@ display:
     dimensions:
       width: 101
       height: 64
+
+  # Every pixel is one of 256 colours picked by a hash of its position, so the picture is
+  # effectively random but fits a GIF palette exactly.
+  - platform: snapshot
+    id: noise_display
+    update_interval: 1s
+    dimensions:
+      width: 200
+      height: 150
+    lambda: |-
+      for (int y = 0; y != it.get_height(); y++) {
+        for (int x = 0; x != it.get_width(); x++) {
+          uint32_t h = static_cast<uint32_t>(x) * 73856093u ^ static_cast<uint32_t>(y) * 19349663u;
+          h ^= h >> 13;
+          h *= 0x5bd1e995u;
+          h ^= h >> 15;
+          uint8_t c = h & 0xFF;
+          it.draw_pixel_at(x, y, Color(c & 0xE0, (c << 3) & 0xE0, (c << 6) & 0xC0));
+        }
+      }

tests/integration/test_snapshot_display.py

@@ -3,19 +3,66 @@
 from __future__ import annotations
 
 import asyncio
+import io
 from pathlib import Path
 
 from aioesphomeapi import LogLevel
+from PIL import Image, UnidentifiedImageError
 import pytest
 
-from .bmp_utils import capture_when_drawn, wait_for_bmp
+from .bmp_utils import Bmp, capture_when_drawn, wait_for_bmp
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 WIDTH = 101
 HEIGHT = 64
 
+ANIMATION_FRAMES = 5
+# The fixture asks for 20 frames a second, and a GIF counts time in milliseconds here.
+ANIMATION_FRAME_MS = 50
+
+NOISE_WIDTH = 200
+NOISE_HEIGHT = 150
+NOISE_FRAMES = 3
+
 # Part of the message the writer logs when it will not write over a file that is already there.
 REFUSAL_MESSAGE = b"not overwriting"
+# Part of the message logged when an animation is asked for while another is still being recorded.
+BUSY_MESSAGE = b"Already recording"
+
+
+async def wait_for_gif(path: Path, frames: int, timeout: float = 5.0) -> Image.Image:
+    """Wait for a complete animated GIF with the given number of frames and return it.
+
+    The file exists from the moment the recording starts and grows a frame at a time, so keep
+    reading until it holds all of them and ends with the GIF trailer.
+    """
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        try:
+            data = path.read_bytes()
+            # Open the bytes just read, so the trailer check and the decode see the same file.
+            image = Image.open(io.BytesIO(data))
+            if data.endswith(b";") and image.n_frames == frames:
+                # Decoding every frame proves the compressed data is all there and valid.
+                for frame in range(frames):
+                    image.seek(frame)
+                    image.load()
+                image.seek(0)
+                return image
+        except (FileNotFoundError, UnidentifiedImageError, OSError, EOFError):
+            pass
+        await asyncio.sleep(0.05)
+    raise AssertionError(
+        f"no complete {frames} frame GIF appeared at {path} within {timeout}s"
+    )
+
+
+def bmp_as_image(bmp: Bmp) -> Image.Image:
+    """The picture in a decoded BMP, as an RGB image."""
+    return Image.frombytes(
+        "RGB", (bmp.width, bmp.height), bmp.pixels, "raw", "BGR", 0, -1
+    )
 
 
 @pytest.mark.asyncio
@@ -39,12 +86,50 @@ async def test_snapshot_display(
         async def take(name: str) -> None:
             await client.execute_service(service, {"name": name})
 
+        animation_service = next(s for s in services if s.name == "take_animation")
+
         # The test card is drawn in several colours, so once it is on the screen the picture is
         # not one flat shade. Capturing until that is true waits out the first update rather than
         # racing it.
         image, capture = await capture_when_drawn(take, snapshot_dir)
         assert (image.width, image.height, image.bits) == (WIDTH, HEIGHT, 24)
 
+        # Asking for frames records a GIF: every frame is the size of the display and lasts as long
+        # as the frame rate says. The test card does not move, so every frame is the picture
+        # captured above.
+        await client.execute_service(animation_service, {"name": "movie"})
+        movie = await wait_for_gif(snapshot_dir / "movie.gif", ANIMATION_FRAMES)
+        assert movie.size == (WIDTH, HEIGHT)
+        expected = bmp_as_image(image)
+        for frame in range(ANIMATION_FRAMES):
+            movie.seek(frame)
+            assert movie.info["duration"] == ANIMATION_FRAME_MS
+            assert movie.convert("RGB").tobytes() == expected.tobytes()
+
+        # A picture of 256 colours in no pattern makes a long run of compression codes, so recording
+        # it passes the point where the code width grows and where the dictionary is started over.
+        # The picture uses no more than 256 colours, so it is stored exactly and every frame must
+        # come back identical to the one captured as a BMP.
+        noise_service = next(s for s in services if s.name == "take_noise_snapshot")
+        noise_animation = next(s for s in services if s.name == "take_noise_animation")
+
+        async def take_noise(name: str) -> None:
+            await client.execute_service(noise_service, {"name": name})
+
+        noise, _ = await capture_when_drawn(take_noise, snapshot_dir, "noisedrawn")
+        assert (noise.width, noise.height) == (NOISE_WIDTH, NOISE_HEIGHT)
+        noise_expected = bmp_as_image(noise)
+        assert (
+            len(noise_expected.getcolors(NOISE_WIDTH * NOISE_HEIGHT)) > 200
+        )  # a busy picture
+        await client.execute_service(noise_animation, {"name": "noise"})
+        noise_movie = await wait_for_gif(
+            snapshot_dir / "noise.gif", NOISE_FRAMES, timeout=15.0
+        )
+        for frame in range(NOISE_FRAMES):
+            noise_movie.seek(frame)
+            assert noise_movie.convert("RGB").tobytes() == noise_expected.tobytes()
+
         # An extension is only added when there is not one already, whatever its case.
         await take("UPPER.BMP")
         await wait_for_bmp(snapshot_dir / "UPPER.BMP")
@@ -54,18 +139,30 @@ async def test_snapshot_display(
         await take("../escape")
         await wait_for_bmp(snapshot_dir / ".._escape.bmp")
 
-        # A second capture under a name already used must fail rather than write over the first.
-        # Wait for the device to report the refusal: on its own, an unchanged file cannot tell a
-        # refusal apart from a request the device has not got to yet, so a regression that wrote
-        # over the file could still pass on a busy machine.
+        # The device says so when it refuses a request. Waiting for that message is the only way to
+        # tell a refusal from a request the device has not got to yet: an unchanged file, or a file
+        # that never appears, would look the same on a busy machine.
         refused = asyncio.Event()
+        busy = asyncio.Event()
 
         def on_log(msg) -> None:
             if REFUSAL_MESSAGE in msg.message:
                 refused.set()
+            if BUSY_MESSAGE in msg.message:
+                busy.set()
 
         client.subscribe_logs(on_log, log_level=LogLevel.LOG_LEVEL_DEBUG)
 
+        # A display records one animation at a time. A second request made while the first is still
+        # running is refused and leaves no file, and the first one carries on to the end.
+        slow_animation = next(s for s in services if s.name == "take_slow_animation")
+        await client.execute_service(slow_animation, {"name": "slow"})
+        await client.execute_service(slow_animation, {"name": "second"})
+        await asyncio.wait_for(busy.wait(), timeout=10.0)
+        assert not (snapshot_dir / "second.gif").exists()
+        await wait_for_gif(snapshot_dir / "slow.gif", 3)
+
+        # A second capture under a name already used must fail rather than write over the first.
         before = capture.read_bytes()
         await take(capture.name)
         await asyncio.wait_for(refused.wait(), timeout=10.0)
@@ -73,6 +170,12 @@ async def test_snapshot_display(
         # Nothing beyond what was asked for, leaving out however many captures it took to wait
         # for the first frame.
         written = sorted(
-            p.name for p in snapshot_dir.iterdir() if not p.name.startswith("drawn-")
+            p.name for p in snapshot_dir.iterdir() if "drawn-" not in p.name
         )
-        assert written == [".._escape.bmp", "UPPER.BMP"]
+        assert written == [
+            ".._escape.bmp",
+            "UPPER.BMP",
+            "movie.gif",
+            "noise.gif",
+            "slow.gif",
+        ]

tests/script/test_determine_jobs.py

@@ -1017,6 +1017,9 @@ _ESP32_PLATFORMIO_FULL_LIST_FILES = [
     # Workflow / harness files
     ["script/test_build_components.py"],
     [".github/workflows/ci.yml"],
+    # The base config and bus packages every generated build includes
+    ["tests/test_build_components/build_components_base.esp32-idf.yaml"],
+    ["tests/test_build_components/common/uart/esp32-idf.yaml"],
 ]
 
 
@@ -3187,6 +3190,9 @@ def test_memory_impact_elf_layouts_are_found(tmp_path: Path) -> None:
         "esphome/platformio/toolchain.py",
         # The composite cache action must not ship unexercised
         ".github/actions/cache-arduino8266/action.yml",
+        # The base config and bus packages every generated build includes
+        "tests/test_build_components/build_components_base.esp8266-ard.yaml",
+        "tests/test_build_components/common/uart/esp8266-ard.yaml",
     ],
 )
 def test_esp8266_native_components_full_list_on_infra_change(changed: str) -> None: