Merge branch 'dev' into climate-control-action-bitmask

tests/component_tests/esp32/config/reproducible_build_arduino.yaml

@@ -0,0 +1,8 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+  variant: esp32
+  framework:
+    type: arduino

tests/component_tests/esp32/test_esp32.py

@@ -16,6 +16,7 @@ from esphome.const import (
     CONF_ESPHOME,
     CONF_IGNORE_PIN_VALIDATION_ERROR,
     CONF_NUMBER,
+    KEY_NATIVE_IDF,
     PlatformFramework,
 )
 from esphome.core import CORE
@@ -243,3 +244,30 @@ def test_platformio_idf_enables_reproducible_build(
 
     sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
     assert sdkconfig.get("CONFIG_APP_REPRODUCIBLE_BUILD") is True
+
+
+def test_platformio_arduino_enables_reproducible_build(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Test PlatformIO Arduino builds enable reproducible app metadata."""
+    generate_main(component_config_path("reproducible_build_arduino.yaml"))
+
+    sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
+    assert sdkconfig.get("CONFIG_APP_REPRODUCIBLE_BUILD") is True
+
+
+def test_native_idf_enables_reproducible_build(
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Test native ESP-IDF builds enable reproducible app metadata."""
+    from esphome.__main__ import generate_cpp_contents
+    from esphome.config import read_config
+
+    CORE.config_path = component_config_path("reproducible_build.yaml")
+    CORE.config = read_config({})
+    CORE.data[KEY_NATIVE_IDF] = True
+    generate_cpp_contents(CORE.config)
+
+    sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
+    assert sdkconfig.get("CONFIG_APP_REPRODUCIBLE_BUILD") is True

tests/component_tests/image/test_init.py

@@ -7,10 +7,12 @@ from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock, patch
 
+from PIL import Image as PILImage
 import pytest
 
 from esphome import config_validation as cv
 from esphome.components.image import (
+    CONF_ALPHA_CHANNEL,
     CONF_INVERT_ALPHA,
     CONF_OPAQUE,
     CONF_TRANSPARENCY,
@@ -411,3 +413,70 @@ async def test_svg_with_mm_dimensions_succeeds(
     assert 30 < height < 50, (
         f"Height should be around 39 pixels for 10mm at 100dpi, got {height}"
     )
+
+
+@pytest.mark.asyncio
+async def test_rgb565_alpha_animation_layout_per_frame(
+    tmp_path: Path,
+    mock_progmem_array: MagicMock,
+) -> None:
+    """RGB565+alpha animations must store each frame as a self-contained
+    [RGB plane | alpha plane] block. Animation::update_data_start_ steps frames
+    with a single per-frame stride, so any cross-frame layout (all RGB then all
+    alpha) makes the C++ alpha read land in the next frame's RGB bytes — that
+    was the regression behind issue #15999.
+    """
+    # Build a 2-frame APNG where each frame is a solid color with a known
+    # alpha. APNG preserves full RGBA per pixel (GIF only has 1-bit alpha so
+    # round-tripping mid-range alpha values does not work). Frame 0 is fully
+    # opaque red, frame 1 is fully transparent blue.
+    width = 4
+    height = 3
+    frame0 = PILImage.new("RGBA", (width, height), (255, 0, 0, 0xFF))
+    frame1 = PILImage.new("RGBA", (width, height), (0, 0, 255, 0x00))
+    apng_path = tmp_path / "anim.png"
+    frame0.save(
+        apng_path,
+        format="PNG",
+        save_all=True,
+        append_images=[frame1],
+        duration=100,
+        loop=0,
+    )
+
+    config = {
+        CONF_FILE: str(apng_path),
+        CONF_TYPE: "RGB565",
+        CONF_TRANSPARENCY: CONF_ALPHA_CHANNEL,
+        CONF_DITHER: "NONE",
+        CONF_INVERT_ALPHA: False,
+        CONF_RAW_DATA_ID: "test_raw_data_id",
+    }
+
+    _, _, _, _, _, frame_count = await write_image(config, all_frames=True)
+    assert frame_count == 2
+
+    # Recover the bytes handed to progmem_array. Signature is (id_, rhs).
+    _, raw_data = mock_progmem_array.call_args.args
+    data = [int(x) for x in raw_data]
+
+    rgb_size = width * height * 2
+    alpha_size = width * height
+    frame_size = rgb_size + alpha_size
+    assert len(data) == frame_size * frame_count, (
+        "RGB565+alpha animation buffer must be (RGB + alpha) per frame, not "
+        "all RGB followed by all alpha"
+    )
+
+    # Frame 0: RGB plane is red, alpha plane is 0xFF. Frame 1: alpha plane is
+    # 0x00. If the layout regresses to [all RGB | all alpha], the alpha bytes
+    # would all land at the tail of the buffer and the per-frame slices below
+    # would point at RGB565 noise instead.
+    frame0_alpha = data[rgb_size : rgb_size + alpha_size]
+    frame1_alpha = data[frame_size + rgb_size : frame_size + rgb_size + alpha_size]
+    assert all(a == 0xFF for a in frame0_alpha), (
+        f"Frame 0 alpha plane should be opaque, got {frame0_alpha}"
+    )
+    assert all(a == 0x00 for a in frame1_alpha), (
+        f"Frame 1 alpha plane should be transparent, got {frame1_alpha}"
+    )

tests/component_tests/lvgl/test_grid_layout.py

@@ -0,0 +1,239 @@
+"""Unit tests for the LVGL grid layout shorthand and rows/columns auto-sizing."""
+
+from __future__ import annotations
+
+import pytest
+from voluptuous import Invalid
+
+from esphome.components.lvgl.defines import (
+    CONF_GRID_COLUMNS,
+    CONF_GRID_ROWS,
+    CONF_LAYOUT,
+    CONF_WIDGETS,
+    TYPE_GRID,
+)
+from esphome.components.lvgl.layout import GridLayout, grid_dimension
+from esphome.const import CONF_TYPE
+
+FR1 = "LV_GRID_FR(1)"
+
+
+def _widgets(n: int) -> list[dict]:
+    """Build a list of `n` placeholder widgets for the validate() input."""
+    return [{"label": {}} for _ in range(n)]
+
+
+# ---------------------------------------------------------------------------
+# grid_dimension validator
+# ---------------------------------------------------------------------------
+
+
+def test_grid_dimension_int_expands_to_fr1_list() -> None:
+    """A positive integer should expand to a list of LV_GRID_FR(1) entries."""
+    assert grid_dimension(1) == [FR1]
+    assert grid_dimension(3) == [FR1, FR1, FR1]
+
+
+def test_grid_dimension_zero_or_negative_rejected() -> None:
+    """Non-positive integers must be rejected."""
+    with pytest.raises(Invalid):
+        grid_dimension(0)
+    with pytest.raises(Invalid):
+        grid_dimension(-2)
+
+
+def test_grid_dimension_list_passes_through() -> None:
+    """A list should be validated through the existing grid_spec list schema."""
+    result = grid_dimension(["100px", "content", "fr(2)"])
+    # `grid_spec` normalises each entry: pixel sizes become ints, the
+    # CONTENT keyword is uppercased and prefixed, and FR(n) is normalised.
+    assert result == [100, "LV_GRID_CONTENT", "LV_GRID_FR(2)"]
+
+
+def test_grid_dimension_invalid_string_rejected() -> None:
+    """A string is not a valid grid dimension and should be rejected."""
+    with pytest.raises(Invalid):
+        grid_dimension("not a list")
+
+
+def test_grid_dimension_empty_list_rejected() -> None:
+    """An empty list of grid specs must be rejected."""
+    with pytest.raises(Invalid, match="at least one entry"):
+        grid_dimension([])
+
+
+# ---------------------------------------------------------------------------
+# Shorthand string layouts
+# ---------------------------------------------------------------------------
+
+
+def test_shorthand_full_form_unchanged() -> None:
+    """`<rows>x<cols>` continues to work and yields the exact dimensions."""
+    config = {CONF_LAYOUT: "2x3", CONF_WIDGETS: _widgets(0)}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    assert layout[CONF_TYPE] == TYPE_GRID
+    assert layout[CONF_GRID_ROWS] == [FR1, FR1]
+    assert layout[CONF_GRID_COLUMNS] == [FR1, FR1, FR1]
+
+
+def test_shorthand_rows_only_calculates_columns_from_widgets() -> None:
+    """`<rows>x` derives the column count from the number of widgets."""
+    config = {CONF_LAYOUT: "3x", CONF_WIDGETS: _widgets(7)}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    # 7 widgets / 3 rows -> ceil = 3 columns.
+    assert len(layout[CONF_GRID_ROWS]) == 3
+    assert len(layout[CONF_GRID_COLUMNS]) == 3
+
+
+def test_shorthand_columns_only_calculates_rows_from_widgets() -> None:
+    """`x<cols>` derives the row count from the number of widgets."""
+    config = {CONF_LAYOUT: "x4", CONF_WIDGETS: _widgets(5)}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    # 5 widgets / 4 cols -> ceil = 2 rows.
+    assert len(layout[CONF_GRID_ROWS]) == 2
+    assert len(layout[CONF_GRID_COLUMNS]) == 4
+
+
+def test_shorthand_rows_only_no_widgets_defaults_columns_to_one() -> None:
+    """With no widgets and only rows specified, the column count defaults to 1."""
+    config = {CONF_LAYOUT: "3x", CONF_WIDGETS: []}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    assert len(layout[CONF_GRID_ROWS]) == 3
+    assert len(layout[CONF_GRID_COLUMNS]) == 1
+
+
+def test_shorthand_columns_only_no_widgets_defaults_rows_to_one() -> None:
+    """With no widgets and only columns specified, the row count defaults to 1."""
+    config = {CONF_LAYOUT: "x4", CONF_WIDGETS: []}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    assert len(layout[CONF_GRID_ROWS]) == 1
+    assert len(layout[CONF_GRID_COLUMNS]) == 4
+
+
+def test_shorthand_with_whitespace_accepted() -> None:
+    """The shorthand parser should tolerate whitespace around the components."""
+    config = {CONF_LAYOUT: "  3 x ", CONF_WIDGETS: _widgets(6)}
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    # 6 widgets / 3 rows -> 2 columns.
+    assert len(layout[CONF_GRID_ROWS]) == 3
+    assert len(layout[CONF_GRID_COLUMNS]) == 2
+
+
+def test_shorthand_bare_x_rejected() -> None:
+    """Pure `x` (no digits at all) is not a valid shorthand."""
+    config = {CONF_LAYOUT: "x", CONF_WIDGETS: _widgets(2)}
+    with pytest.raises(Invalid):
+        GridLayout().validate(config)
+
+
+@pytest.mark.parametrize(
+    "layout,bad_label",
+    [
+        ("0x3", "row"),
+        ("3x0", "column"),
+        ("0x", "row"),
+        ("x0", "column"),
+        ("0x0", "row"),
+    ],
+)
+def test_shorthand_zero_dimension_rejected(layout: str, bad_label: str) -> None:
+    """Shorthand row/column counts must be >= 1."""
+    config = {CONF_LAYOUT: layout, CONF_WIDGETS: _widgets(2)}
+    with pytest.raises(Invalid, match=f"{bad_label} count must be at least 1"):
+        GridLayout().validate(config)
+
+
+def test_shorthand_get_layout_schemas_recognizes_partial_forms() -> None:
+    """`<rows>x` and `x<cols>` should be picked up by GridLayout.get_layout_schemas."""
+    grid = GridLayout()
+    for layout in ("3x", "x4", "2x3"):
+        layout_schema, _ = grid.get_layout_schemas({CONF_LAYOUT: layout})
+        assert layout_schema is not None, f"{layout!r} should be recognised"
+    # Pure `x` and unrelated strings should not be picked up as a grid layout.
+    for layout in ("x", "horizontal"):
+        layout_schema, _ = grid.get_layout_schemas({CONF_LAYOUT: layout})
+        assert layout_schema is None, f"{layout!r} should not be recognised"
+
+
+# ---------------------------------------------------------------------------
+# Dict-form layouts with rows/columns auto-sizing
+# ---------------------------------------------------------------------------
+
+
+def test_dict_rows_only_calculates_columns_from_widgets() -> None:
+    """A dict layout with only rows fills in the column count from widget count."""
+    config = {
+        CONF_LAYOUT: {
+            CONF_TYPE: TYPE_GRID,
+            CONF_GRID_ROWS: [FR1, FR1],
+        },
+        CONF_WIDGETS: _widgets(5),
+    }
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    # 5 widgets / 2 rows -> ceil = 3 columns.
+    assert len(layout[CONF_GRID_ROWS]) == 2
+    assert layout[CONF_GRID_COLUMNS] == [FR1, FR1, FR1]
+
+
+def test_dict_columns_only_calculates_rows_from_widgets() -> None:
+    """A dict layout with only columns fills in the row count from widget count."""
+    config = {
+        CONF_LAYOUT: {
+            CONF_TYPE: TYPE_GRID,
+            CONF_GRID_COLUMNS: [FR1, FR1, FR1],
+        },
+        CONF_WIDGETS: _widgets(7),
+    }
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    # 7 widgets / 3 cols -> ceil = 3 rows.
+    assert layout[CONF_GRID_ROWS] == [FR1, FR1, FR1]
+    assert len(layout[CONF_GRID_COLUMNS]) == 3
+
+
+def test_dict_rows_only_no_widgets_defaults_columns_to_one() -> None:
+    """A dict layout with rows but no widgets defaults columns to 1."""
+    config = {
+        CONF_LAYOUT: {
+            CONF_TYPE: TYPE_GRID,
+            CONF_GRID_ROWS: [FR1, FR1, FR1],
+        },
+        CONF_WIDGETS: [],
+    }
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    assert len(layout[CONF_GRID_ROWS]) == 3
+    assert layout[CONF_GRID_COLUMNS] == [FR1]
+
+
+def test_dict_neither_rows_nor_columns_rejected() -> None:
+    """A grid layout dict without rows AND without columns must be rejected."""
+    config = {
+        CONF_LAYOUT: {CONF_TYPE: TYPE_GRID},
+        CONF_WIDGETS: _widgets(3),
+    }
+    with pytest.raises(Invalid):
+        GridLayout().validate(config)
+
+
+def test_dict_both_rows_and_columns_unchanged() -> None:
+    """When both dimensions are present they are preserved as-is."""
+    config = {
+        CONF_LAYOUT: {
+            CONF_TYPE: TYPE_GRID,
+            CONF_GRID_ROWS: [FR1, FR1],
+            CONF_GRID_COLUMNS: [FR1, FR1, FR1],
+        },
+        CONF_WIDGETS: _widgets(0),
+    }
+    result = GridLayout().validate(config)
+    layout = result[CONF_LAYOUT]
+    assert layout[CONF_GRID_ROWS] == [FR1, FR1]
+    assert layout[CONF_GRID_COLUMNS] == [FR1, FR1, FR1]

tests/components/api/test_proto_mac_varint.cpp

@@ -0,0 +1,123 @@
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <ios>
+#include <random>
+
+#include "esphome/components/api/api_buffer.h"
+#include "esphome/components/api/proto.h"
+
+namespace esphome::api::testing {
+
+// Generic varint decoder, used to verify the encoded bytes round-trip back to
+// the original 48-bit MAC value, independent of the specialized encoder under
+// test.
+static uint64_t decode_varint(const uint8_t *buf, size_t len, size_t *consumed) {
+  uint64_t value = 0;
+  int shift = 0;
+  for (size_t i = 0; i < len; i++) {
+    value |= static_cast<uint64_t>(buf[i] & 0x7F) << shift;
+    if ((buf[i] & 0x80) == 0) {
+      *consumed = i + 1;
+      return value;
+    }
+    shift += 7;
+  }
+  *consumed = 0;
+  return 0;
+}
+
+// Reference encoder mirroring ProtoEncode::encode_varint_raw_64.
+static size_t reference_encode(uint64_t value, uint8_t *out) {
+  uint8_t *p = out;
+  if (value < 128) {
+    *p++ = static_cast<uint8_t>(value);
+    return p - out;
+  }
+  do {
+    *p++ = static_cast<uint8_t>(value | 0x80);
+    value >>= 7;
+  } while (value > 0x7F);
+  *p++ = static_cast<uint8_t>(value);
+  return p - out;
+}
+
+// Encode `mac` via the 48-bit fast path and verify:
+//  - byte-identical output to the reference loop
+//  - encoded byte length matches `expected_bytes`
+//  - calc_uint64_48bit_force agrees on the size
+//  - the bytes round-trip through a generic varint decoder
+static void verify_mac(uint64_t mac, size_t expected_bytes) {
+  ASSERT_LT(mac, 1ULL << 48) << "test fixture mac exceeds 48 bits";
+
+  uint8_t ref_buf[16] = {0};
+  size_t ref_len = reference_encode(mac, ref_buf);
+
+  APIBuffer api_buf;
+  api_buf.resize(16);
+  uint8_t *pos = api_buf.data();
+#ifdef ESPHOME_DEBUG_API
+  uint8_t *proto_debug_end_ = api_buf.data() + api_buf.size();
+#endif
+  ProtoEncode::encode_varint_raw_48bit(pos PROTO_ENCODE_DEBUG_ARG, mac);
+  size_t new_len = pos - api_buf.data();
+
+  EXPECT_EQ(new_len, expected_bytes) << "mac=0x" << std::hex << mac << std::dec;
+  EXPECT_EQ(ref_len, expected_bytes) << "reference disagrees on length for mac=0x" << std::hex << mac << std::dec;
+
+  for (size_t i = 0; i < new_len; i++) {
+    EXPECT_EQ(api_buf.data()[i], ref_buf[i])
+        << "byte " << i << " differs for mac=0x" << std::hex << mac << " (got 0x" << static_cast<int>(api_buf.data()[i])
+        << ", expected 0x" << static_cast<int>(ref_buf[i]) << ")" << std::dec;
+  }
+
+  size_t consumed = 0;
+  uint64_t decoded = decode_varint(api_buf.data(), new_len, &consumed);
+  EXPECT_EQ(consumed, new_len) << "decoder did not consume all bytes for mac=0x" << std::hex << mac << std::dec;
+  EXPECT_EQ(decoded, mac) << "round-trip mismatch for mac=0x" << std::hex << mac << std::dec;
+
+  // Verify the size helper agrees. field_id_size = 1 (typical 1-byte tag).
+  uint32_t calc_size = ProtoSize::calc_uint64_48bit_force(1, mac);
+  EXPECT_EQ(calc_size, 1 + expected_bytes)
+      << "calc_uint64_48bit_force size mismatch for mac=0x" << std::hex << mac << std::dec;
+}
+
+// Compute the canonical varint byte length for a value < 1<<48.
+static size_t expected_varint_len(uint64_t v) {
+  if (v < (1ULL << 7))
+    return 1;
+  if (v < (1ULL << 14))
+    return 2;
+  if (v < (1ULL << 21))
+    return 3;
+  if (v < (1ULL << 28))
+    return 4;
+  if (v < (1ULL << 35))
+    return 5;
+  if (v < (1ULL << 42))
+    return 6;
+  return 7;
+}
+
+// --- Specific MACs requested for verification ---
+
+TEST(ProtoMacVarint, AllZeros) { verify_mac(0x000000000000ULL, 1); }        // 00:00:00:00:00:00
+TEST(ProtoMacVarint, FirstByteOnly) { verify_mac(0x110000000000ULL, 7); }   // 11:00:00:00:00:00
+TEST(ProtoMacVarint, SecondByteOnly) { verify_mac(0x00AA00000000ULL, 6); }  // 00:AA:00:00:00:00
+TEST(ProtoMacVarint, ThirdByteOnly) { verify_mac(0x0000BB000000ULL, 5); }   // 00:00:BB:00:00:00
+TEST(ProtoMacVarint, FourthByteOnly) { verify_mac(0x000000CC0000ULL, 4); }  // 00:00:00:CC:00:00
+TEST(ProtoMacVarint, FifthByteOnly) { verify_mac(0x00000000DD00ULL, 3); }   // 00:00:00:00:DD:00
+TEST(ProtoMacVarint, SixthByteOnly) { verify_mac(0x0000000000EEULL, 2); }   // 00:00:00:00:00:EE
+TEST(ProtoMacVarint, AllOnes) { verify_mac(0xFFFFFFFFFFFFULL, 7); }         // FF:FF:FF:FF:FF:FF
+
+// 100 deterministic-random 48-bit MACs to catch regressions across the space.
+TEST(ProtoMacVarint, RandomSample) {
+  // NOLINTNEXTLINE(cert-msc32-c,cert-msc51-cpp) -- intentional fixed seed for reproducibility.
+  std::mt19937_64 rng(0xC0FFEE);
+  for (int i = 0; i < 100; i++) {
+    uint64_t mac = rng() & 0xFFFFFFFFFFFFULL;
+    verify_mac(mac, expected_varint_len(mac));
+  }
+}
+
+}  // namespace esphome::api::testing

tests/components/climate/common.yaml

@@ -29,3 +29,59 @@ climate:
     heat_action:
       - switch.turn_on: climate_heater_switch
       - switch.turn_off: climate_cooler_switch
+  # Thermostat-based climate so climate.control: action variants get build
+  # coverage (bang_bang doesn't support fan modes, presets, etc.). Climate
+  # has no template platform, so thermostat is the right vehicle.
+  - platform: thermostat
+    id: climate_test_thermostat
+    name: Test Thermostat
+    sensor: climate_temperature_sensor
+    min_idle_time: 30s
+    min_heating_off_time: 300s
+    min_heating_run_time: 300s
+    min_cooling_off_time: 300s
+    min_cooling_run_time: 300s
+    heat_action:
+      - logger.log: heating
+    idle_action:
+      - logger.log: idle
+    cool_action:
+      - logger.log: cooling
+    auto_mode:
+      - logger.log: auto
+    heat_cool_mode:
+      - logger.log: heat_cool
+    preset:
+      - name: Default
+        default_target_temperature_low: 18°C
+        default_target_temperature_high: 22°C
+
+button:
+  # Exercise the climate.control: action so ControlAction templates get
+  # build coverage. Various field combinations are tested.
+  - platform: template
+    name: "Climate Control Mode"
+    on_press:
+      - climate.control:
+          id: climate_test_thermostat
+          mode: HEAT
+  - platform: template
+    name: "Climate Control Mode And Temps"
+    on_press:
+      - climate.control:
+          id: climate_test_thermostat
+          mode: HEAT_COOL
+          target_temperature_low: 19.0°C
+          target_temperature_high: 23.0°C
+  - platform: template
+    name: "Climate Control Lambda Temp"
+    on_press:
+      - climate.control:
+          id: climate_test_thermostat
+          target_temperature_high: !lambda "return 21.5;"
+  - platform: template
+    name: "Climate Control Off"
+    on_press:
+      - climate.control:
+          id: climate_test_thermostat
+          mode: "OFF"

tests/components/json/__init__.py

@@ -0,0 +1,9 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # json's to_code calls cg.add_library("bblanchon/ArduinoJson", ...). C++
+    # unit test builds that pull json in transitively (e.g. api) need that
+    # library registration to happen, otherwise json_util.cpp fails to find
+    # ArduinoJson.h.
+    manifest.enable_codegen()

tests/components/light/common.yaml

@@ -108,6 +108,10 @@ esphome:
           relative_brightness: 5%
           brightness_limits:
             max_brightness: 90%
+      - light.dim_relative:
+          id: test_monochromatic_light
+          relative_brightness: -5%
+          transition_length: 250ms
       - light.turn_on:
           id: test_addressable_transition
           brightness: 50%

tests/components/lvgl/lvgl-package.yaml

@@ -16,10 +16,19 @@ binary_sensor:
     platform: template
   - id: left_sensor
     platform: template
+  - platform: lvgl
+    name: Button A pressed
+    widget: button_a
+    state: pressed
+  - platform: lvgl
+    name: Button A checked
+    widget: button_a
+    state: checked
   - platform: lvgl
     id: button_checker
     name: LVGL button
     widget: button_button
+    state: checked
     on_state:
       then:
         - lvgl.checkbox.update:
@@ -29,6 +38,12 @@ binary_sensor:
                   auto y = x; // block inlining of one line return
                   return y;
 
+  - platform: lvgl
+    id: button_presser
+    name: Button pressed
+    widget: button_button
+    state: pressed
+
 lvgl:
   id: lvgl_id
   rotation: 90
@@ -1098,6 +1113,8 @@ lvgl:
                   pad_row: 6px
                   pad_column: 0
                   multiple_widgets_per_cell: true
+                  grid_cell_x_align: center
+                  grid_cell_y_align: center
                 widgets:
                   - image:
                       grid_cell_row_pos: 0
@@ -1290,6 +1307,87 @@ lvgl:
                 hidden: true
                 mode: text_lower
 
+        # Grid shorthand "<rows>x": 3 rows specified, columns derived
+        # from widget count (4 widgets / 3 rows -> 2 columns)
+        - obj:
+            id: grid_rows_only_shorthand
+            layout: 3x
+            widgets:
+              - label:
+                  text: "r1"
+              - label:
+                  text: "r2"
+              - label:
+                  text: "r3"
+              - label:
+                  text: "r4"
+
+        # Grid shorthand "x<cols>": 4 columns specified, rows derived
+        # from widget count (5 widgets / 4 cols -> 2 rows)
+        - obj:
+            id: grid_cols_only_shorthand
+            layout: x4
+            widgets:
+              - label:
+                  text: "a"
+              - label:
+                  text: "b"
+              - label:
+                  text: "c"
+              - label:
+                  text: "d"
+              - label:
+                  text: "e"
+
+        # Grid dict form with grid_rows as a plain integer; columns derived
+        - obj:
+            id: grid_rows_int
+            layout:
+              type: grid
+              grid_rows: 2
+            widgets:
+              - label:
+                  text: "1"
+              - label:
+                  text: "2"
+              - label:
+                  text: "3"
+
+        # Grid dict form with grid_columns as a plain integer; rows derived
+        - obj:
+            id: grid_cols_int
+            layout:
+              type: grid
+              grid_columns: 3
+            widgets:
+              - label:
+                  text: "x"
+              - label:
+                  text: "y"
+              - label:
+                  text: "z"
+              - label:
+                  text: "w"
+              - label:
+                  text: "v"
+
+        # Grid dict form with both grid_rows and grid_columns as plain integers
+        - obj:
+            id: grid_both_int
+            layout:
+              type: grid
+              grid_rows: 2
+              grid_columns: 2
+            widgets:
+              - label:
+                  text: "1,1"
+              - label:
+                  text: "1,2"
+              - label:
+                  text: "2,1"
+              - label:
+                  text: "2,2"
+
 font:
   - file: "gfonts://Roboto"
     id: space16

tests/components/mapping/common.yaml

@@ -21,6 +21,7 @@ mapping:
     entries:
       clear-night: image_1
       sunny: image_2
+    default_value: image_1
   - id: weather_map_2
     from: string
     to: image
@@ -35,6 +36,7 @@ mapping:
       2: "two"
       3: "three"
       77: "seventy-seven"
+    default_value: unknown
   - id: string_map
     from: string
     to: int

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

@@ -4,7 +4,7 @@ packages:
 
 display:
   spi_id: spi_bus
-  platform: ili9xxx
+  platform: mipi_spi
   id: main_lcd
   model: ili9342
   cs_pin: 12

tests/components/mapping/test.esp8266-ard.yaml

@@ -4,7 +4,7 @@ packages:
 
 display:
   spi_id: spi_bus
-  platform: ili9xxx
+  platform: mipi_spi
   id: main_lcd
   model: ili9342
   cs_pin: 5

tests/components/mapping/test.rp2040-ard.yaml

@@ -4,7 +4,7 @@ packages:
 
 display:
   spi_id: spi_bus
-  platform: ili9xxx
+  platform: mipi_spi
   id: main_lcd
   model: ili9342
   data_rate: 31.25MHz

tests/components/mdns/common-enabled-ethernet.yaml

@@ -0,0 +1,23 @@
+ethernet:
+  type: W5500
+  clk_pin: 18
+  mosi_pin: 19
+  miso_pin: 16
+  cs_pin: 17
+  interrupt_pin: 21
+  reset_pin: 20
+  manual_ip:
+    static_ip: 192.168.178.56
+    gateway: 192.168.178.1
+    subnet: 255.255.255.0
+  domain: .local
+  mac_address: "02:AA:BB:CC:DD:01"
+
+mdns:
+  disabled: false
+  services:
+    - service: _test_service
+      protocol: _tcp
+      port: 8888
+      txt:
+        static_string: Anything

tests/components/mdns/test-enabled-ethernet.rp2040-ard.yaml

@@ -0,0 +1 @@
+<<: !include common-enabled-ethernet.yaml

tests/components/modbus_controller/common.yaml

@@ -1,53 +1,11 @@
-modbus:
-  - id: mod_bus2
-    uart_id: uart_bus
-    role: server
-
 modbus_controller:
   - id: modbus_controller1
     address: 0x2
     modbus_id: modbus_bus
-    allow_duplicate_commands: false
     on_online:
       then:
         logger.log: "Module Online"
-  - id: modbus_controller2
-    address: 0x2
-    modbus_id: mod_bus2
-    server_registers:
-      - address: 0x0000
-        value_type: S_DWORD_R
-        read_lambda: |-
-          return 42.3;
-    max_cmd_retries: 0
-  - id: modbus_controller3
-    address: 0x3
-    modbus_id: mod_bus2
-    server_registers:
-      - address: 0x0009
-        value_type: S_DWORD
-        read_lambda: |-
-          return 31;
-        write_lambda: |-
-          printf("address=%d, value=%d", x);
-          return true;
-    max_cmd_retries: 0
-  - id: modbus_controller4
-    modbus_id: mod_bus2
-    address: 0x4
-    server_courtesy_response:
-      enabled: true
-      register_last_address: 100
-      register_value: 0
-    server_registers:
-      - address: 0x0001
-        value_type: U_WORD
-        read_lambda: |-
-          return 0x8;
-      - address: 0x0005
-        value_type: U_WORD
-        read_lambda: |-
-          return (random_uint32() % 100);
+
 binary_sensor:
   - platform: modbus_controller
     modbus_controller_id: modbus_controller1

tests/components/modbus_server/common.yaml

@@ -0,0 +1,41 @@
+modbus:
+  - id: mod_bus2
+    uart_id: uart_bus
+    role: server
+
+modbus_server:
+  - id: modbus_server2
+    address: 0x2
+    modbus_id: mod_bus2
+    registers:
+      - address: 0x0
+        value_type: S_DWORD_R
+        read_lambda: |-
+          return 42.3;
+  - id: modbus_server3
+    address: 0x3
+    modbus_id: mod_bus2
+    registers:
+      - address: 0x9
+        value_type: S_DWORD
+        read_lambda: |-
+          return 31;
+        write_lambda: |-
+          printf("address=%d, value=%d", x);
+          return true;
+  - id: modbus_server4
+    modbus_id: mod_bus2
+    address: 0x4
+    courtesy_response:
+      enabled: true
+      register_last_address: 100
+      register_value: 0
+    registers:
+      - address: 0x1
+        value_type: U_WORD
+        read_lambda: |-
+          return 0x8;
+      - address: 0x5
+        value_type: U_WORD
+        read_lambda: |-
+          return (random_uint32() % 100);

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

@@ -0,0 +1,4 @@
+packages:
+  modbus: !include ../../test_build_components/common/modbus/esp32-idf.yaml
+
+<<: !include common.yaml

tests/components/modbus_server/test.esp8266-ard.yaml

@@ -0,0 +1,4 @@
+packages:
+  modbus: !include ../../test_build_components/common/modbus/esp8266-ard.yaml
+
+<<: !include common.yaml

tests/components/modbus_server/test.rp2040-ard.yaml

@@ -0,0 +1,4 @@
+packages:
+  modbus: !include ../../test_build_components/common/modbus/rp2040-ard.yaml
+
+<<: !include common.yaml

tests/components/nrf52/test-dfu-pin.nrf52-xiao-ble.yaml

@@ -0,0 +1,9 @@
+nrf52:
+  dfu:
+    reset_pin:
+      number: 14
+      inverted: true
+      mode:
+        output: true
+  reg0:
+    voltage: 1.8V

tests/components/nrf52/test.nrf52-xiao-ble.yaml

@@ -1,9 +1,4 @@
 nrf52:
-  dfu:
-    reset_pin:
-      number: 14
-      inverted: true
-      mode:
-        output: true
+  dfu: true
   reg0:
     voltage: 1.8V

tests/components/template/common-base.yaml

@@ -293,6 +293,60 @@ cover:
             cover.is_closed: template_cover_with_triggers
           then:
             logger.log: Cover is closed
+  # Exercise cover.control / cover.template.publish action variants so they
+  # get build coverage in CI (and so memory-impact analysis on PRs that
+  # touch ControlAction / CoverPublishAction sees real instances).
+  - platform: template
+    name: "Template Cover Actions"
+    id: template_cover_actions
+    has_position: true
+    optimistic: true
+    open_action:
+      # CONF_STATE alias for the position bit
+      - cover.template.publish:
+          id: template_cover_actions
+          state: OPEN
+      - cover.template.publish:
+          id: template_cover_actions
+          position: 1.0
+      - cover.template.publish:
+          id: template_cover_actions
+          current_operation: IDLE
+    close_action:
+      - cover.template.publish:
+          id: template_cover_actions
+          position: 0.0
+          tilt: 0.0
+    stop_action:
+      - cover.template.publish:
+          id: template_cover_actions
+          current_operation: IDLE
+    tilt_action:
+      - lambda: |-
+          id(template_cover_actions).tilt = tilt;
+          id(template_cover_actions).publish_state();
+    on_idle:
+      # position only
+      - cover.control:
+          id: template_cover_actions
+          position: 50%
+      # tilt only
+      - cover.control:
+          id: template_cover_actions
+          tilt: 75%
+      # position + tilt
+      - cover.control:
+          id: template_cover_actions
+          position: 25%
+          tilt: 30%
+      # stop
+      - cover.control:
+          id: template_cover_actions
+          stop: true
+      # CONF_STATE alias for position
+      - cover.control:
+          id: template_cover_actions
+          state: OPEN
 
 number:
   - platform: template

tests/components/wifi/test.esp8266-ard.yaml

@@ -1,6 +1,7 @@
 wifi:
   min_auth_mode: WPA2
   post_connect_roaming: true
+  phy_mode: 11G
 
 packages:
   - !include common.yaml

tests/integration/fixtures/light_dim_relative_action.yaml

@@ -0,0 +1,60 @@
+esphome:
+  name: light-dim-relative-action-test
+host:
+api:
+logger:
+  level: DEBUG
+
+output:
+  - platform: template
+    id: test_out
+    type: float
+    write_action:
+      - lambda: ""
+
+light:
+  - platform: monochromatic
+    name: "Test Light"
+    id: test_light
+    output: test_out
+    default_transition_length: 0s
+
+button:
+  # Set up: turn on at 50% brightness
+  - platform: template
+    id: btn_setup
+    name: "Setup"
+    on_press:
+      - light.turn_on:
+          id: test_light
+          brightness: 50%
+
+  # Test 1: dim_relative without transition_length (HasTransitionLength=false)
+  - platform: template
+    id: btn_dim_up
+    name: "Dim Up"
+    on_press:
+      - light.dim_relative:
+          id: test_light
+          relative_brightness: 25%
+
+  # Test 2: dim_relative with transition_length (HasTransitionLength=true)
+  - platform: template
+    id: btn_dim_down
+    name: "Dim Down"
+    on_press:
+      - light.dim_relative:
+          id: test_light
+          relative_brightness: -10%
+          transition_length: 0s
+
+  # Test 3: dim_relative with brightness limits
+  - platform: template
+    id: btn_dim_clamp
+    name: "Dim Clamp"
+    on_press:
+      - light.dim_relative:
+          id: test_light
+          relative_brightness: 50%
+          brightness_limits:
+            max_brightness: 80%

tests/integration/fixtures/light_toggle_action.yaml

@@ -0,0 +1,37 @@
+esphome:
+  name: light-toggle-action-test
+host:
+api:
+logger:
+  level: DEBUG
+
+output:
+  - platform: template
+    id: test_out
+    type: float
+    write_action:
+      - lambda: ""
+
+light:
+  - platform: monochromatic
+    name: "Test Light"
+    id: test_light
+    output: test_out
+    default_transition_length: 0s
+
+button:
+  # Test 1: light.toggle without transition_length (HasTransitionLength=false)
+  - platform: template
+    id: btn_toggle
+    name: "Toggle"
+    on_press:
+      - light.toggle: test_light
+
+  # Test 2: light.toggle with transition_length (HasTransitionLength=true)
+  - platform: template
+    id: btn_toggle_with_trans
+    name: "Toggle With Trans"
+    on_press:
+      - light.toggle:
+          id: test_light
+          transition_length: 0s

tests/integration/fixtures/uart_mock_modbus_server.yaml

@@ -86,9 +86,9 @@ modbus:
   uart_id: virtual_uart_dev
   role: server
 
-modbus_controller:
+modbus_server:
   - address: 1
-    server_registers:
+    registers:
       - address: 0x03
         value_type: U_WORD
         read_lambda: |-

tests/integration/fixtures/uart_mock_modbus_server_controller.yaml

@@ -33,7 +33,7 @@ uart_mock:
               data: !lambda return data;
   - id: virtual_uart_controller
     baud_rate: 9600
-    auto_start: true  # See comment on virtual_uart_server above
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
@@ -56,10 +56,11 @@ modbus_controller:
     update_interval: 1s
     id: modbus_controller_1
 
+modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
     id: modbus_server_1
-    server_registers:
+    registers:
       - address: 0x01
         value_type: U_WORD
         read_lambda: return 99;

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -36,7 +36,7 @@ uart_mock:
               data: !lambda return data;
   - id: virtual_uart_server_2
     baud_rate: 9600
-    auto_start: true  # See comment on virtual_uart_server above
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
@@ -48,7 +48,7 @@ uart_mock:
               data: !lambda return data;
   - id: virtual_uart_controller
     baud_rate: 9600
-    auto_start: true  # See comment on virtual_uart_server above
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
@@ -81,15 +81,16 @@ modbus_controller:
     update_interval: 1s
     id: modbus_controller_2
 
+modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
-    server_registers:
+    registers:
       - address: 0x01
         value_type: U_WORD
         read_lambda: return 919;
   - address: 2
     modbus_id: virtual_modbus_server_2
-    server_registers:
+    registers:
       - address: 0x01
         value_type: U_WORD
         read_lambda: return 929;

tests/integration/fixtures/uart_mock_modbus_server_controller_write.yaml

@@ -33,7 +33,7 @@ uart_mock:
               data: !lambda return data;
   - id: virtual_uart_controller
     baud_rate: 9600
-    auto_start: true  # See comment on virtual_uart_server above
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
@@ -94,10 +94,11 @@ modbus_controller:
     update_interval: 2s
     id: modbus_controller_1
 
+modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
     id: modbus_server_1
-    server_registers:
+    registers:
       - address: 0x01
         value_type: U_WORD
         read_lambda: return id(stored_u_word);

tests/integration/test_light_dim_relative_action.py

@@ -0,0 +1,72 @@
+"""Integration test for light::DimRelativeAction.
+
+Tests both DimRelativeAction<HasTransitionLength=false> and
+DimRelativeAction<HasTransitionLength=true> instantiations.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_light_dim_relative_action(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test light.dim_relative with and without transition_length."""
+    loop = asyncio.get_running_loop()
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        light_state_future: asyncio.Future[LightState] | None = None
+
+        def on_state(state: EntityState) -> None:
+            if (
+                isinstance(state, LightState)
+                and light_state_future is not None
+                and not light_state_future.done()
+            ):
+                light_state_future.set_result(state)
+
+        async def wait_for_light_state(timeout: float = 5.0) -> LightState:
+            nonlocal light_state_future
+            light_state_future = loop.create_future()
+            try:
+                return await asyncio.wait_for(light_state_future, timeout)
+            finally:
+                light_state_future = None
+
+        entities, _ = await client.list_entities_services()
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        require_entity(entities, "test_light", LightInfo)
+
+        async def press_and_wait(name: str) -> LightState:
+            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
+            client.button_command(btn.key)
+            return await wait_for_light_state()
+
+        # Setup: turn on at 50%
+        state = await press_and_wait("Setup")
+        assert state.state is True
+        assert state.brightness == pytest.approx(0.5, abs=0.05)
+
+        # Test 1: dim_relative without transition_length: 50% + 25% = 75%
+        state = await press_and_wait("Dim Up")
+        assert state.brightness == pytest.approx(0.75, abs=0.05)
+
+        # Test 2: dim_relative with transition_length: 75% - 10% = 65%
+        state = await press_and_wait("Dim Down")
+        assert state.brightness == pytest.approx(0.65, abs=0.05)
+
+        # Test 3: dim_relative with max_brightness limit: 65% + 50% clamped to 80%
+        state = await press_and_wait("Dim Clamp")
+        assert state.brightness == pytest.approx(0.80, abs=0.05)

tests/integration/test_light_toggle_action.py

@@ -0,0 +1,67 @@
+"""Integration test for light::ToggleAction.
+
+Tests both ToggleAction<HasTransitionLength=false> and
+ToggleAction<HasTransitionLength=true> instantiations.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import ButtonInfo, EntityState, LightInfo, LightState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_light_toggle_action(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test light.toggle with and without transition_length."""
+    loop = asyncio.get_running_loop()
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        light_state_future: asyncio.Future[LightState] | None = None
+
+        def on_state(state: EntityState) -> None:
+            if (
+                isinstance(state, LightState)
+                and light_state_future is not None
+                and not light_state_future.done()
+            ):
+                light_state_future.set_result(state)
+
+        async def wait_for_light_state(timeout: float = 5.0) -> LightState:
+            nonlocal light_state_future
+            light_state_future = loop.create_future()
+            try:
+                return await asyncio.wait_for(light_state_future, timeout)
+            finally:
+                light_state_future = None
+
+        entities, _ = await client.list_entities_services()
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        require_entity(entities, "test_light", LightInfo)
+
+        async def press_and_wait(name: str) -> LightState:
+            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
+            client.button_command(btn.key)
+            return await wait_for_light_state()
+
+        # Test 1: toggle without transition_length flips off->on
+        state = await press_and_wait("Toggle")
+        assert state.state is True
+
+        # Test 2: toggle with transition_length flips on->off
+        state = await press_and_wait("Toggle With Trans")
+        assert state.state is False
+
+        # Test 3: toggle without transition_length flips off->on again
+        state = await press_and_wait("Toggle")
+        assert state.state is True

tests/script/test_check_import_time.py

@@ -0,0 +1,191 @@
+"""Unit tests for script/check_import_time.py."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Load the script-under-test as `check_import_time` (it's a hyphenated path
+# inside `script/` that mirrors the existing `determine_jobs` pattern).
+script_dir = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "..", "script")
+)
+sys.path.insert(0, script_dir)
+spec = importlib.util.spec_from_file_location(
+    "check_import_time", os.path.join(script_dir, "check_import_time.py")
+)
+check_import_time = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(check_import_time)
+
+
+def _entry(name: str, self_us: int, cumulative_us: int) -> dict:
+    """Build a minimal HAR entry matching `importtime_waterfall --har`."""
+    return {
+        "request": {"url": name},
+        "time": cumulative_us,
+        "timings": {"receive": self_us, "wait": cumulative_us - self_us},
+    }
+
+
+def _har(*entries: dict) -> dict:
+    return {"log": {"entries": list(entries)}}
+
+
+def test_root_cumulative_us_returns_time_for_root_module() -> None:
+    har = _har(
+        _entry("dep_a", 500, 500),
+        _entry("dep_b", 300, 300),
+        _entry("esphome.__main__", 100, 1000),
+    )
+    assert check_import_time.root_cumulative_us(har, "esphome.__main__") == 1000
+
+
+def test_root_cumulative_us_missing_module_raises() -> None:
+    har = _har(_entry("something.else", 100, 100))
+    with pytest.raises(RuntimeError, match="No HAR entry for 'esphome.__main__'"):
+        check_import_time.root_cumulative_us(har, "esphome.__main__")
+
+
+def test_top_offenders_ranks_by_self_time_descending() -> None:
+    har = _har(
+        _entry("small", 100, 100),
+        _entry("big", 5000, 5000),
+        _entry("medium", 2000, 2500),
+    )
+    result = check_import_time.top_offenders(har, n=10)
+    assert [name for name, _, _ in result] == ["big", "medium", "small"]
+    assert result[0] == ("big", 5000, 5000)
+
+
+def test_top_offenders_respects_n_limit() -> None:
+    har = _har(*[_entry(f"m{i}", i * 100, i * 100) for i in range(1, 20)])
+    assert len(check_import_time.top_offenders(har, n=5)) == 5
+
+
+def test_top_offenders_dedupes_repeat_names_keeping_first() -> None:
+    har = _har(
+        _entry("pkg", 5000, 5000),
+        _entry("pkg", 100, 100),  # reimport later in trace
+        _entry("other", 1000, 1000),
+    )
+    result = check_import_time.top_offenders(har, n=10)
+    assert [name for name, _, _ in result] == ["pkg", "other"]
+    # First occurrence wins
+    assert ("pkg", 5000, 5000) in result
+
+
+def test_format_us_switches_to_ms_at_threshold() -> None:
+    assert check_import_time._format_us(500) == "500us"
+    assert check_import_time._format_us(999) == "999us"
+    assert check_import_time._format_us(1000) == "1.0ms"
+    assert check_import_time._format_us(12345) == "12.3ms"
+
+
+def test_read_write_budget_roundtrip(tmp_path: Path) -> None:
+    budget_path = tmp_path / "budget.json"
+    with patch.object(check_import_time, "BUDGET_PATH", budget_path):
+        assert check_import_time.read_budget() == {}
+        check_import_time.write_budget(cumulative_us=12345, margin_pct=20)
+        loaded = check_import_time.read_budget()
+    assert loaded["cumulative_us"] == 12345
+    assert loaded["margin_pct"] == 20
+    assert loaded["target_module"] == check_import_time.TARGET_MODULE
+
+
+def test_cmd_check_passes_when_measured_within_ceiling(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,  # 100ms
+            }
+        )
+    )
+    # Measured 90ms: inside 100ms + 15% = 115ms ceiling
+    har = _har(_entry(check_import_time.TARGET_MODULE, 1000, 90000))
+    args = type("A", (), {"har": None})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "measure", return_value=har),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "measured   esphome.__main__:" in out
+    assert "budget 100.0ms" in out
+
+
+def test_cmd_check_fails_when_measured_exceeds_ceiling(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,
+            }
+        )
+    )
+    # Measured 120ms: over 100ms + 15% = 115ms ceiling
+    har = _har(
+        _entry("offender_a", 10000, 10000),
+        _entry(check_import_time.TARGET_MODULE, 1000, 120000),
+    )
+    args = type("A", (), {"har": None})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "measure", return_value=har),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 1
+    err = capsys.readouterr().err
+    assert "REGRESSION" in err
+    assert "120.0ms" in err
+    assert "offender_a" in err  # top offender table
+
+
+def test_cmd_check_returns_2_when_budget_missing(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "nonexistent.json"
+    args = type("A", (), {"har": None})()
+    with patch.object(check_import_time, "BUDGET_PATH", budget_path):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 2
+    assert "missing" in capsys.readouterr().err
+
+
+def test_cmd_check_writes_har_when_path_given(tmp_path: Path) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,
+            }
+        )
+    )
+    har_path = tmp_path / "out.har"
+    har_text = json.dumps(_har(_entry(check_import_time.TARGET_MODULE, 1000, 80000)))
+    args = type("A", (), {"har": str(har_path)})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "run_waterfall", return_value=har_text),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 0
+    assert har_path.exists()
+    assert json.loads(har_path.read_text()) == json.loads(har_text)

tests/script/test_determine_jobs.py

@@ -56,6 +56,13 @@ def mock_should_run_python_linters() -> Generator[Mock, None, None]:
         yield mock
 
 
+@pytest.fixture
+def mock_should_run_import_time() -> Generator[Mock, None, None]:
+    """Mock should_run_import_time from determine_jobs."""
+    with patch.object(determine_jobs, "should_run_import_time") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_determine_cpp_unit_tests() -> Generator[Mock, None, None]:
     """Mock determine_cpp_unit_tests from helpers."""
@@ -91,6 +98,7 @@ def test_main_all_tests_should_run(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -104,6 +112,7 @@ def test_main_all_tests_should_run(
     mock_should_run_clang_tidy.return_value = True
     mock_should_run_clang_format.return_value = True
     mock_should_run_python_linters.return_value = True
+    mock_should_run_import_time.return_value = True
     mock_determine_cpp_unit_tests.return_value = (False, ["wifi", "api", "sensor"])
 
     # Mock changed_files to return non-component files (to avoid memory impact)
@@ -158,6 +167,7 @@ def test_main_all_tests_should_run(
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is True
     assert output["python_linters"] is True
+    assert output["import_time"] is True
     assert output["changed_components"] == ["wifi", "api", "sensor"]
     # changed_components_with_tests will only include components that actually have test files
     assert "changed_components_with_tests" in output
@@ -189,6 +199,7 @@ def test_main_no_tests_should_run(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -202,6 +213,7 @@ def test_main_no_tests_should_run(
     mock_should_run_clang_tidy.return_value = False
     mock_should_run_clang_format.return_value = False
     mock_should_run_python_linters.return_value = False
+    mock_should_run_import_time.return_value = False
     mock_determine_cpp_unit_tests.return_value = (False, [])
 
     # Mock changed_files to return no component files
@@ -241,6 +253,7 @@ def test_main_no_tests_should_run(
     assert output["clang_tidy_mode"] == "disabled"
     assert output["clang_format"] is False
     assert output["python_linters"] is False
+    assert output["import_time"] is False
     assert output["changed_components"] == []
     assert output["changed_components_with_tests"] == []
     assert output["component_test_count"] == 0
@@ -261,6 +274,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -274,6 +288,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy.return_value = True
     mock_should_run_clang_format.return_value = False
     mock_should_run_python_linters.return_value = True
+    mock_should_run_import_time.return_value = True
     mock_determine_cpp_unit_tests.return_value = (False, ["mqtt"])
 
     # Mock changed_files to return non-component files (to avoid memory impact)
@@ -310,6 +325,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy.assert_called_once_with("main")
     mock_should_run_clang_format.assert_called_once_with("main")
     mock_should_run_python_linters.assert_called_once_with("main")
+    mock_should_run_import_time.assert_called_once_with("main")
 
     # Check output
     captured = capsys.readouterr()
@@ -322,6 +338,7 @@ def test_main_with_branch_argument(
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is False
     assert output["python_linters"] is True
+    assert output["import_time"] is True
     assert output["changed_components"] == ["mqtt"]
     # changed_components_with_tests will only include components that actually have test files
     assert "changed_components_with_tests" in output
@@ -597,6 +614,50 @@ def test_should_run_python_linters_with_branch() -> None:
         mock_changed.assert_called_once_with("release")
 
 
+@pytest.mark.parametrize(
+    ("changed_files", "expected_result"),
+    [
+        # esphome Python files trigger the check
+        (["esphome/__main__.py"], True),
+        (["esphome/components/wifi/__init__.py"], True),
+        (["esphome/core/config.py"], True),
+        (["esphome/types.pyi"], True),
+        # Dependency declarations and the check's own files trigger
+        (["requirements.txt"], True),
+        (["requirements_dev.txt"], True),
+        (["requirements_test.txt"], True),
+        (["pyproject.toml"], True),
+        (["script/check_import_time.py"], True),
+        (["script/import_time_budget.json"], True),
+        # Mixed: any triggering file is enough
+        (["docs/README.md", "esphome/config.py"], True),
+        # Python files outside esphome/ don't trigger
+        (["script/some_other_script.py"], False),
+        (["tests/script/test_determine_jobs.py"], False),
+        # Non-Python changes don't trigger
+        (["esphome/core/component.cpp"], False),
+        (["tests/components/wifi/test.esp32-idf.yaml"], False),
+        (["README.md"], False),
+        ([], False),
+    ],
+)
+def test_should_run_import_time(
+    changed_files: list[str], expected_result: bool
+) -> None:
+    """Test should_run_import_time function."""
+    with patch.object(determine_jobs, "changed_files", return_value=changed_files):
+        result = determine_jobs.should_run_import_time()
+        assert result == expected_result
+
+
+def test_should_run_import_time_with_branch() -> None:
+    """Test should_run_import_time with branch argument."""
+    with patch.object(determine_jobs, "changed_files") as mock_changed:
+        mock_changed.return_value = []
+        determine_jobs.should_run_import_time("release")
+        mock_changed.assert_called_once_with("release")
+
+
 @pytest.mark.parametrize(
     ("changed_files", "expected_result"),
     [

tests/script/test_test_helpers.py

@@ -258,3 +258,161 @@ def test_load_wraps_platform_component(tmp_path: Path) -> None:
     assert key == "bthome.sensor"
     assert isinstance(installed, ComponentManifestOverride)
     assert installed.to_code is None
+
+
+# ---------------------------------------------------------------------------
+# populate_dependency_config
+# ---------------------------------------------------------------------------
+
+
+def _make_component_stub(
+    *,
+    multi_conf: bool = False,
+    is_platform_component: bool = False,
+    config_schema=None,
+) -> MagicMock:
+    stub = MagicMock()
+    stub.multi_conf = multi_conf
+    stub.is_platform_component = is_platform_component
+    stub.config_schema = config_schema
+    return stub
+
+
+def test_populate_platform_component_listed_alone_uses_list() -> None:
+    """Regression: a platform component (sensor) with no `sensor.x` siblings
+    must land as `[]` in config. Previously it was populated as a dict via
+    `schema({})`, which then crashed the sibling `domain.platform` branch
+    when later dependencies tried `config.setdefault('sensor', []).append(...)`.
+    """
+    sensor = _make_component_stub(is_platform_component=True)
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["sensor"],
+        get_component_fn=lambda name: sensor if name == "sensor" else None,
+        register_platform_fn=lambda _: None,
+    )
+
+    assert config["sensor"] == []
+
+
+def test_populate_platform_component_then_platform_entry() -> None:
+    """When `sensor` is processed before `sensor.gpio` (sorted order),
+    the bare-component branch must leave `config['sensor']` as a list so
+    the platform-entry branch can append into it.
+    """
+    sensor = _make_component_stub(is_platform_component=True)
+    gpio = _make_component_stub()  # the bare `gpio` component
+    components: dict[str, object] = {"sensor": sensor, "gpio": gpio}
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["gpio", "sensor", "sensor.gpio"],
+        get_component_fn=components.get,
+        register_platform_fn=lambda _: None,
+    )
+
+    assert config["sensor"] == [{"platform": "gpio"}]
+
+
+def test_populate_multi_conf_component_uses_list() -> None:
+    multi = _make_component_stub(multi_conf=True)
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["multi"],
+        get_component_fn=lambda name: multi if name == "multi" else None,
+        register_platform_fn=lambda _: None,
+    )
+
+    assert config["multi"] == []
+
+
+def test_populate_plain_component_uses_schema_defaults() -> None:
+    schema = MagicMock(return_value={"default_key": 42})
+    plain = _make_component_stub(config_schema=schema)
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["plain"],
+        get_component_fn=lambda name: plain if name == "plain" else None,
+        register_platform_fn=lambda _: None,
+    )
+
+    schema.assert_called_once_with({})
+    assert config["plain"] == {"default_key": 42}
+
+
+def test_populate_plain_component_falls_back_when_schema_raises() -> None:
+    def picky_schema(_):
+        raise ValueError("required field missing")
+
+    plain = _make_component_stub(config_schema=picky_schema)
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["plain"],
+        get_component_fn=lambda name: plain if name == "plain" else None,
+        register_platform_fn=lambda _: None,
+    )
+
+    assert config["plain"] == {}
+
+
+def test_populate_skips_unresolvable_pseudo_components() -> None:
+    """`core` and other names that get_component returns None for are skipped
+    silently without inserting anything into the config.
+    """
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["core"],
+        get_component_fn=lambda _: None,
+        register_platform_fn=lambda _: None,
+    )
+
+    assert config == {}
+
+
+def test_populate_preserves_existing_plain_component_config() -> None:
+    """If a plain component already has a config entry (e.g. from the user's
+    YAML), the schema-defaults branch must not overwrite it.
+    """
+    schema = MagicMock()
+    plain = _make_component_stub(config_schema=schema)
+    config: dict = {"plain": {"user_key": "set_by_user"}}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["plain"],
+        get_component_fn=lambda name: plain if name == "plain" else None,
+        register_platform_fn=lambda _: None,
+    )
+
+    schema.assert_not_called()
+    assert config["plain"] == {"user_key": "set_by_user"}
+
+
+def test_populate_registers_platform_for_platform_entry() -> None:
+    """Each `domain.platform` entry triggers register_platform_fn(domain) so
+    USE_<DOMAIN> defines get emitted later in the build pipeline.
+    """
+    registered: list[str] = []
+    config: dict = {}
+
+    build_helpers.populate_dependency_config(
+        config,
+        ["sensor.gpio", "binary_sensor.gpio"],
+        get_component_fn=lambda _: None,
+        register_platform_fn=registered.append,
+    )
+
+    assert registered == ["sensor", "binary_sensor"]
+    assert config["sensor"] == [{"platform": "gpio"}]
+    assert config["binary_sensor"] == [{"platform": "gpio"}]

tests/unit_tests/analyze_memory/test_source_file_attribution.py

@@ -0,0 +1,43 @@
+"""Tests for source-file-to-component attribution in memory analyzer."""
+
+from unittest.mock import patch
+
+from esphome.analyze_memory import MemoryAnalyzer
+
+
+def _make_analyzer(external_components: set[str] | None = None) -> MemoryAnalyzer:
+    """Create a MemoryAnalyzer with mocked dependencies."""
+    with patch.object(MemoryAnalyzer, "__init__", lambda self, *a, **kw: None):
+        analyzer = MemoryAnalyzer.__new__(MemoryAnalyzer)
+        analyzer.external_components = external_components or set()
+        analyzer._lib_hash_to_name = {}
+    return analyzer
+
+
+def test_source_file_to_component_main_cpp_relative() -> None:
+    """ESPHome-generated src/main.cpp.o (nm path form) attributes to core."""
+    analyzer = _make_analyzer()
+    assert analyzer._source_file_to_component("src/main.cpp.o") == "[esphome]core"
+
+
+def test_source_file_to_component_main_cpp_pioenvs_path() -> None:
+    """Linker map paths like .pioenvs/<env>/src/main.cpp.o attribute to core."""
+    analyzer = _make_analyzer()
+    result = analyzer._source_file_to_component(".pioenvs/drivewaygate/src/main.cpp.o")
+    assert result == "[esphome]core"
+
+
+def test_source_file_to_component_esphome_core() -> None:
+    """Sources under src/esphome/core/ attribute to core."""
+    analyzer = _make_analyzer()
+    result = analyzer._source_file_to_component("src/esphome/core/application.cpp.o")
+    assert result == "[esphome]core"
+
+
+def test_source_file_to_component_known_component() -> None:
+    """Known ESPHome components attribute to their component name."""
+    analyzer = _make_analyzer()
+    result = analyzer._source_file_to_component(
+        "src/esphome/components/wifi/wifi_component.cpp.o"
+    )
+    assert result == "[esphome]wifi"

tests/unit_tests/test_external_files.py

@@ -1,5 +1,6 @@
 """Tests for external_files.py functions."""
 
+import os
 from pathlib import Path
 import time
 from unittest.mock import MagicMock, patch
@@ -8,8 +9,73 @@ import pytest
 import requests
 
 from esphome import external_files
-from esphome.config_validation import Invalid
-from esphome.core import CORE, TimePeriod
+from esphome.config_validation import Invalid, MultipleInvalid
+from esphome.core import CORE, EsphomeError, TimePeriod
+
+
+def _seed_etag(cache_file: Path, etag: str) -> Path:
+    """Write an ETag sidecar with its mtime synced to the cache file's mtime,
+    matching the invariant that `_write_etag` enforces in production.
+    """
+    sidecar = external_files._etag_sidecar_path(cache_file)
+    sidecar.write_text(etag)
+    file_mtime = int(cache_file.stat().st_mtime)
+    os.utime(sidecar, (file_mtime, file_mtime))
+    return sidecar
+
+
+@pytest.fixture
+def mock_requests_head() -> MagicMock:
+    """Patch `external_files.requests.head` so the conditional HEAD-request
+    validator can be tested without doing real HTTP.
+    """
+    with patch("esphome.external_files.requests.head") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_requests_get() -> MagicMock:
+    """Patch `external_files.requests.get` so the download path can be
+    tested without doing real HTTP.
+    """
+    with patch("esphome.external_files.requests.get") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_has_remote_file_changed() -> MagicMock:
+    """Patch `external_files.has_remote_file_changed` so download tests can
+    control the conditional check independently from the GET path.
+    """
+    with patch("esphome.external_files.has_remote_file_changed") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_write_file() -> MagicMock:
+    """Patch `external_files.write_file` so atomic-write failures can be
+    injected without involving the real filesystem helper.
+    """
+    with patch("esphome.external_files.write_file") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_download_content() -> MagicMock:
+    """Patch `external_files.download_content` for tests that exercise the
+    parallel batch helper without doing real I/O.
+    """
+    with patch("esphome.external_files.download_content") as m:
+        yield m
+
+
+@pytest.fixture
+def mock_download_content_many() -> MagicMock:
+    """Patch `external_files.download_content_many` for tests that exercise
+    the URL-collection helper without dispatching to the thread pool.
+    """
+    with patch("esphome.external_files.download_content_many") as m:
+        yield m
 
 
 def test_compute_local_file_dir(setup_core: Path) -> None:
@@ -88,9 +154,8 @@ def test_is_file_recent_with_zero_refresh(setup_core: Path) -> None:
         assert result is False
 
 
-@patch("esphome.external_files.requests.head")
 def test_has_remote_file_changed_not_modified(
-    mock_head: MagicMock, setup_core: Path
+    mock_requests_head: MagicMock, setup_core: Path
 ) -> None:
     """Test has_remote_file_changed returns False when file not modified."""
     test_file = setup_core / "cached.txt"
@@ -98,23 +163,23 @@ def test_has_remote_file_changed_not_modified(
 
     mock_response = MagicMock()
     mock_response.status_code = 304
-    mock_head.return_value = mock_response
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
 
     url = "https://example.com/file.txt"
     result = external_files.has_remote_file_changed(url, test_file)
 
     assert result is False
-    mock_head.assert_called_once()
+    mock_requests_head.assert_called_once()
 
-    call_args = mock_head.call_args
+    call_args = mock_requests_head.call_args
     headers = call_args[1]["headers"]
     assert external_files.IF_MODIFIED_SINCE in headers
     assert external_files.CACHE_CONTROL in headers
 
 
-@patch("esphome.external_files.requests.head")
 def test_has_remote_file_changed_modified(
-    mock_head: MagicMock, setup_core: Path
+    mock_requests_head: MagicMock, setup_core: Path
 ) -> None:
     """Test has_remote_file_changed returns True when file modified."""
     test_file = setup_core / "cached.txt"
@@ -122,7 +187,8 @@ def test_has_remote_file_changed_modified(
 
     mock_response = MagicMock()
     mock_response.status_code = 200
-    mock_head.return_value = mock_response
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
 
     url = "https://example.com/file.txt"
     result = external_files.has_remote_file_changed(url, test_file)
@@ -140,15 +206,16 @@ def test_has_remote_file_changed_no_local_file(setup_core: Path) -> None:
     assert result is True
 
 
-@patch("esphome.external_files.requests.head")
 def test_has_remote_file_changed_network_error(
-    mock_head: MagicMock, setup_core: Path
+    mock_requests_head: MagicMock, setup_core: Path
 ) -> None:
     """Test has_remote_file_changed returns False on network error when file is cached."""
     test_file = setup_core / "cached.txt"
     test_file.write_text("cached content")
 
-    mock_head.side_effect = requests.exceptions.RequestException("Network error")
+    mock_requests_head.side_effect = requests.exceptions.RequestException(
+        "Network error"
+    )
 
     url = "https://example.com/file.txt"
     result = external_files.has_remote_file_changed(url, test_file)
@@ -156,9 +223,8 @@ def test_has_remote_file_changed_network_error(
     assert result is False
 
 
-@patch("esphome.external_files.requests.head")
 def test_has_remote_file_changed_timeout(
-    mock_head: MagicMock, setup_core: Path
+    mock_requests_head: MagicMock, setup_core: Path
 ) -> None:
     """Test has_remote_file_changed respects timeout."""
     test_file = setup_core / "cached.txt"
@@ -166,15 +232,176 @@ def test_has_remote_file_changed_timeout(
 
     mock_response = MagicMock()
     mock_response.status_code = 304
-    mock_head.return_value = mock_response
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
 
     url = "https://example.com/file.txt"
     external_files.has_remote_file_changed(url, test_file)
 
-    call_args = mock_head.call_args
+    call_args = mock_requests_head.call_args
     assert call_args[1]["timeout"] == external_files.NETWORK_TIMEOUT
 
 
+def test_has_remote_file_changed_uses_etag(
+    mock_requests_head: MagicMock, setup_core: Path
+) -> None:
+    """Test has_remote_file_changed sends If-None-Match when ETag is cached."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_text("cached content")
+    _seed_etag(test_file, '"abc123"')
+
+    mock_response = MagicMock()
+    mock_response.status_code = 304
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    result = external_files.has_remote_file_changed(url, test_file)
+
+    assert result is False
+    headers = mock_requests_head.call_args[1]["headers"]
+    assert headers[external_files.IF_NONE_MATCH] == '"abc123"'
+
+
+def test_has_remote_file_changed_no_etag_no_if_none_match(
+    mock_requests_head: MagicMock, setup_core: Path
+) -> None:
+    """Test has_remote_file_changed omits If-None-Match when no ETag is cached."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_text("cached content")
+
+    mock_response = MagicMock()
+    mock_response.status_code = 304
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    external_files.has_remote_file_changed(url, test_file)
+
+    headers = mock_requests_head.call_args[1]["headers"]
+    assert external_files.IF_NONE_MATCH not in headers
+
+
+def test_has_remote_file_changed_refreshes_etag_on_304(
+    mock_requests_head: MagicMock, setup_core: Path
+) -> None:
+    """Test has_remote_file_changed updates the cached ETag when the 304 sends a new one."""
+    test_file = setup_core / "cached.txt"
+    test_file.write_text("cached content")
+    _seed_etag(test_file, '"old"')
+
+    mock_response = MagicMock()
+    mock_response.status_code = 304
+    mock_response.headers = {external_files.ETAG: '"new"'}
+    mock_requests_head.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    external_files.has_remote_file_changed(url, test_file)
+
+    assert external_files._etag_sidecar_path(test_file).read_text() == '"new"'
+
+
+def test_has_remote_file_changed_ignores_etag_when_mtime_diverges(
+    mock_requests_head: MagicMock, setup_core: Path
+) -> None:
+    """If the cache file was edited out-of-band (mtime no longer matches the
+    sidecar's), the cached ETag must not be used -- it no longer describes the
+    bytes on disk.
+    """
+    test_file = setup_core / "cached.txt"
+    test_file.write_text("cached content")
+    sidecar = _seed_etag(test_file, '"abc123"')
+
+    # Simulate an out-of-band edit to the cache file -- mtime advances by a
+    # full second (so it diverges at whole-second resolution) but the sidecar
+    # is left untouched, so the recorded ETag is now stale.
+    file_stat = test_file.stat()
+    os.utime(test_file, (file_stat.st_atime, file_stat.st_mtime + 1))
+
+    mock_response = MagicMock()
+    mock_response.status_code = 304
+    mock_response.headers = {}
+    mock_requests_head.return_value = mock_response
+
+    external_files.has_remote_file_changed("https://example.com/file.txt", test_file)
+
+    headers = mock_requests_head.call_args[1]["headers"]
+    assert external_files.IF_NONE_MATCH not in headers
+    # Stale sidecar should be removed so future calls don't keep paying the
+    # mtime-comparison cost on a known-bad sidecar.
+    assert not sidecar.exists()
+
+
+def test_download_content_pins_etag_mtime_to_file_mtime(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """After a successful download, the sidecar's mtime must equal the cache
+    file's mtime so `_read_etag` accepts it on the next call.
+    """
+    test_file = setup_core / "fresh.txt"
+    mock_has_remote_file_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = b"fresh content"
+    mock_response.headers = {external_files.ETAG: '"deadbeef"'}
+    mock_response.raise_for_status = MagicMock()
+    mock_requests_get.return_value = mock_response
+
+    external_files.download_content("https://example.com/file.txt", test_file)
+
+    sidecar = external_files._etag_sidecar_path(test_file)
+    assert int(sidecar.stat().st_mtime) == int(test_file.stat().st_mtime)
+
+
+def test_write_etag_swallows_write_file_failure(
+    mock_write_file: MagicMock, setup_core: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """If `write_file` raises, _write_etag must not propagate -- ETag
+    persistence is best-effort and a failure here must not abort the
+    surrounding download.
+    """
+    cache_file = setup_core / "cached.txt"
+    cache_file.write_text("cached content")
+    mock_write_file.side_effect = EsphomeError("disk full")
+
+    with caplog.at_level("DEBUG", logger="esphome.external_files"):
+        external_files._write_etag(cache_file, '"abc123"')
+
+    assert "Could not save ETag" in caplog.text
+    # Sidecar wasn't created, since write_file was mocked to fail before
+    # reaching the os.utime step.
+    assert not external_files._etag_sidecar_path(cache_file).exists()
+
+
+def test_write_etag_swallows_utime_failure(
+    setup_core: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """If `os.utime` raises while pinning the sidecar's mtime, _write_etag
+    must not propagate. The sidecar is still written; if its mtime later
+    fails to match the cache file, `_read_etag` will discard it on next
+    read.
+    """
+    cache_file = setup_core / "cached.txt"
+    cache_file.write_text("cached content")
+
+    with (
+        patch(
+            "esphome.external_files.os.utime",
+            side_effect=PermissionError("nope"),
+        ),
+        caplog.at_level("DEBUG", logger="esphome.external_files"),
+    ):
+        external_files._write_etag(cache_file, '"abc123"')
+
+    assert "Could not sync ETag sidecar mtime" in caplog.text
+    # write_file succeeded, so the sidecar exists with the new value even
+    # though we couldn't pin its mtime.
+    sidecar = external_files._etag_sidecar_path(cache_file)
+    assert sidecar.exists()
+    assert sidecar.read_text() == '"abc123"'
+
+
 def test_compute_local_file_dir_creates_parent_dirs(setup_core: Path) -> None:
     """Test compute_local_file_dir creates parent directories."""
     domain = "level1/level2/level3/level4"
@@ -200,10 +427,10 @@ def test_is_file_recent_handles_float_seconds(setup_core: Path) -> None:
     assert result is True
 
 
-@patch("esphome.external_files.requests.get")
-@patch("esphome.external_files.has_remote_file_changed")
 def test_download_content_with_network_error_uses_cache(
-    mock_has_changed: MagicMock, mock_get: MagicMock, setup_core: Path
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
 ) -> None:
     """Test download_content uses cached file when network fails."""
     test_file = setup_core / "cached.txt"
@@ -211,8 +438,10 @@ def test_download_content_with_network_error_uses_cache(
     test_file.write_bytes(cached_content)
 
     # Simulate file has changed, so it tries to download
-    mock_has_changed.return_value = True
-    mock_get.side_effect = requests.exceptions.RequestException("Network error")
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException(
+        "Network error"
+    )
 
     url = "https://example.com/file.txt"
     result = external_files.download_content(url, test_file)
@@ -220,17 +449,19 @@ def test_download_content_with_network_error_uses_cache(
     assert result == cached_content
 
 
-@patch("esphome.external_files.requests.get")
-@patch("esphome.external_files.has_remote_file_changed")
 def test_download_content_with_network_error_no_cache_fails(
-    mock_has_changed: MagicMock, mock_get: MagicMock, setup_core: Path
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
 ) -> None:
     """Test download_content raises error when network fails and no cache exists."""
     test_file = setup_core / "nonexistent.txt"
 
     # Simulate file has changed (doesn't exist), so it tries to download
-    mock_has_changed.return_value = True
-    mock_get.side_effect = requests.exceptions.RequestException("Network error")
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.side_effect = requests.exceptions.RequestException(
+        "Network error"
+    )
 
     url = "https://example.com/file.txt"
 
@@ -238,11 +469,9 @@ def test_download_content_with_network_error_no_cache_fails(
         external_files.download_content(url, test_file)
 
 
-@patch("esphome.external_files.requests.get")
-@patch("esphome.external_files.has_remote_file_changed")
 def test_download_content_skip_external_update_uses_cache(
-    mock_has_changed: MagicMock,
-    mock_get: MagicMock,
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
     setup_core: Path,
 ) -> None:
     """Test download_content skips network checks when CORE.skip_external_update is set."""
@@ -255,26 +484,25 @@ def test_download_content_skip_external_update_uses_cache(
     result = external_files.download_content(url, test_file)
 
     assert result == cached_content
-    mock_has_changed.assert_not_called()
-    mock_get.assert_not_called()
+    mock_has_remote_file_changed.assert_not_called()
+    mock_requests_get.assert_not_called()
 
 
-@patch("esphome.external_files.requests.get")
-@patch("esphome.external_files.has_remote_file_changed")
 def test_download_content_skip_external_update_downloads_when_missing(
-    mock_has_changed: MagicMock,
-    mock_get: MagicMock,
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
     setup_core: Path,
 ) -> None:
     """Test download_content still downloads when file is missing, even with skip_external_update."""
     test_file = setup_core / "missing.txt"
     new_content = b"fresh content"
 
-    mock_has_changed.return_value = True
+    mock_has_remote_file_changed.return_value = True
     mock_response = MagicMock()
     mock_response.content = new_content
+    mock_response.headers = {}
     mock_response.raise_for_status = MagicMock()
-    mock_get.return_value = mock_response
+    mock_requests_get.return_value = mock_response
 
     CORE.skip_external_update = True
     url = "https://example.com/file.txt"
@@ -282,3 +510,229 @@ def test_download_content_skip_external_update_downloads_when_missing(
 
     assert result == new_content
     assert test_file.read_bytes() == new_content
+
+
+def test_download_content_many_empty_is_noop(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """Empty input shouldn't spin up a thread pool or call download_content."""
+    external_files.download_content_many([])
+    mock_download_content.assert_not_called()
+
+
+def test_download_content_many_single_item_avoids_pool(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """A single item should be downloaded inline (no thread pool overhead)."""
+    item = ("https://example.com/file.txt", setup_core / "f.txt")
+    external_files.download_content_many([item])
+    mock_download_content.assert_called_once_with(
+        item[0], item[1], external_files.NETWORK_TIMEOUT
+    )
+
+
+def test_download_content_many_runs_in_parallel(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """Multiple items should run concurrently — total wall time ≈ max latency."""
+    import threading
+
+    barrier = threading.Barrier(3)
+
+    def slow_download(url: str, path: Path, timeout: int) -> bytes:
+        # If calls were serial this would deadlock (third caller never arrives
+        # while the first is blocked at the barrier).
+        barrier.wait(timeout=2.0)
+        return b""
+
+    mock_download_content.side_effect = slow_download
+    items = [
+        ("https://example.com/a", setup_core / "a"),
+        ("https://example.com/b", setup_core / "b"),
+        ("https://example.com/c", setup_core / "c"),
+    ]
+    external_files.download_content_many(items, max_workers=4)
+    assert mock_download_content.call_count == 3
+
+
+def test_download_content_many_propagates_single_error(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """A single failing worker should raise its `Invalid` directly, not wrap
+    it in a `MultipleInvalid` that the caller would have to unpack.
+    """
+
+    def fake_download(url: str, path: Path, timeout: int) -> bytes:
+        if url.endswith("bad"):
+            raise Invalid(f"could not download {url}")
+        return b""
+
+    mock_download_content.side_effect = fake_download
+    items = [
+        ("https://example.com/ok", setup_core / "ok"),
+        ("https://example.com/bad", setup_core / "bad"),
+    ]
+    with pytest.raises(Invalid, match="could not download") as exc_info:
+        external_files.download_content_many(items)
+    assert not isinstance(exc_info.value, MultipleInvalid)
+
+
+def test_download_content_many_aggregates_multiple_errors(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """Every failing worker should be reported in a single MultipleInvalid so
+    the user sees all broken URLs in one validation pass instead of fixing
+    them one network round-trip at a time.
+    """
+
+    def fake_download(url: str, path: Path, timeout: int) -> bytes:
+        if url.endswith("ok"):
+            return b""
+        raise Invalid(f"could not download {url}")
+
+    mock_download_content.side_effect = fake_download
+    items = [
+        ("https://example.com/ok", setup_core / "ok"),
+        ("https://example.com/bad1", setup_core / "bad1"),
+        ("https://example.com/bad2", setup_core / "bad2"),
+    ]
+    with pytest.raises(MultipleInvalid) as exc_info:
+        external_files.download_content_many(items)
+    messages = {str(e) for e in exc_info.value.errors}
+    assert messages == {
+        "could not download https://example.com/bad1",
+        "could not download https://example.com/bad2",
+    }
+
+
+def test_download_content_many_dedupes_by_path(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """Two items pointing at the same cache path must collapse to one
+    download -- otherwise concurrent writes race on the same file. Which
+    URL wins doesn't matter (in practice duplicate paths only arise when
+    the URL is duplicated), so we only assert the call count and path.
+    """
+    path = setup_core / "shared"
+    items = [
+        ("https://example.com/a", path),
+        ("https://example.com/b", path),
+        ("https://example.com/a", path),
+    ]
+    external_files.download_content_many(items)
+    assert mock_download_content.call_count == 1
+    args, _ = mock_download_content.call_args
+    assert args[1] == path
+
+
+def test_download_content_many_clamps_invalid_max_workers(
+    mock_download_content: MagicMock, setup_core: Path
+) -> None:
+    """`max_workers <= 0` must not raise from ThreadPoolExecutor; it should
+    be clamped up to at least 1 worker.
+    """
+    items = [
+        ("https://example.com/a", setup_core / "a"),
+        ("https://example.com/b", setup_core / "b"),
+    ]
+    external_files.download_content_many(items, max_workers=0)
+    assert mock_download_content.call_count == 2
+
+
+def test_download_web_files_in_config_filters_and_dispatches(
+    mock_download_content_many: MagicMock, setup_core: Path
+) -> None:
+    """Only `file.type == "web"` entries should be forwarded to
+    download_content_many, and the unmodified config should be returned so
+    the helper can sit in a `cv.All(...)` chain.
+    """
+
+    def path_for(file_dict: dict) -> Path:
+        return setup_core / file_dict["url"].rsplit("/", 1)[-1]
+
+    config = [
+        {"file": {"type": "web", "url": "https://example.com/a"}},
+        {"file": {"type": "local", "path": "/tmp/b"}},
+        {"file": {"type": "web", "url": "https://example.com/c"}},
+        {},  # no `file` key at all
+    ]
+    result = external_files.download_web_files_in_config(config, path_for)
+
+    assert result is config
+    mock_download_content_many.assert_called_once()
+    assert list(mock_download_content_many.call_args[0][0]) == [
+        ("https://example.com/a", setup_core / "a"),
+        ("https://example.com/c", setup_core / "c"),
+    ]
+
+
+def test_download_web_files_in_config_no_web_entries(
+    mock_download_content_many: MagicMock, setup_core: Path
+) -> None:
+    """A config with no web entries should still call through to
+    download_content_many (which is itself a no-op for empty input) so the
+    behavior stays consistent.
+    """
+    config = [{"file": {"type": "local", "path": "/tmp/a"}}]
+    external_files.download_web_files_in_config(config, lambda _: setup_core / "x")
+    mock_download_content_many.assert_called_once()
+    assert list(mock_download_content_many.call_args[0][0]) == []
+
+
+def test_download_content_saves_etag(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Test download_content writes the ETag sidecar after a successful download."""
+    test_file = setup_core / "fresh.txt"
+    new_content = b"fresh content"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = new_content
+    mock_response.headers = {external_files.ETAG: '"deadbeef"'}
+    mock_response.raise_for_status = MagicMock()
+    mock_requests_get.return_value = mock_response
+
+    url = "https://example.com/file.txt"
+    external_files.download_content(url, test_file)
+
+    assert external_files._etag_sidecar_path(test_file).read_text() == '"deadbeef"'
+
+
+def test_download_content_atomic_write_no_partial_on_failure(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    mock_write_file: MagicMock,
+    setup_core: Path,
+) -> None:
+    """If `write_file` (the atomic-write helper) fails, the existing cache
+    file must remain untouched and no temp files may be left behind. Patching
+    `write_file` directly exercises the atomic-rename path -- a failure inside
+    `write_file` is the only reason the rename wouldn't have happened.
+    """
+    from esphome.core import EsphomeError
+
+    test_file = setup_core / "cached.txt"
+    original_content = b"original content"
+    test_file.write_bytes(original_content)
+
+    mock_has_remote_file_changed.return_value = True
+    mock_response = MagicMock()
+    mock_response.content = b"new content"
+    mock_response.headers = {}
+    mock_response.raise_for_status = MagicMock()
+    mock_requests_get.return_value = mock_response
+
+    mock_write_file.side_effect = EsphomeError("disk full")
+
+    with pytest.raises(EsphomeError, match="disk full"):
+        external_files.download_content("https://example.com/file.txt", test_file)
+
+    # Original file is untouched -- write_file aborted before its rename step.
+    assert test_file.read_bytes() == original_content
+    # write_file is responsible for cleaning its own temp files; nothing leaks
+    # into the cache directory either way.
+    leftover_tmps = list(setup_core.glob("tmp*"))
+    assert leftover_tmps == []

tests/unit_tests/test_git.py

@@ -811,3 +811,193 @@ def test_clone_or_update_stale_clone_is_retried_after_cleanup(
     assert repo_dir.exists()
     assert call_count["clone"] == 2
     assert call_count["fetch"] == 2
+
+
+def test_clone_with_ref_uses_shallow_fetch(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Clone with a ref should use --depth=1 on both clone and fetch."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    ref = "pull/123/head"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, ref, domain)
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "clone":
+            repo_dir.mkdir(parents=True, exist_ok=True)
+            (repo_dir / ".git").mkdir(exist_ok=True)
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    git.clone_or_update(url=url, ref=ref, refresh=None, domain=domain)
+
+    call_list = mock_run_git_command.call_args_list
+
+    clone_calls = [c for c in call_list if "clone" in c[0][0]]
+    assert len(clone_calls) == 1
+    assert "--depth=1" in clone_calls[0][0][0]
+
+    fetch_calls = [c for c in call_list if "fetch" in c[0][0]]
+    assert len(fetch_calls) == 1
+    assert "--depth=1" in fetch_calls[0][0][0]
+    # Ref must still be passed so the requested commit/branch is fetched.
+    assert ref in fetch_calls[0][0][0]
+
+
+def test_clone_with_submodules_uses_shallow_submodule_update(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Submodule init on a fresh clone should use --depth=1."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "clone":
+            repo_dir.mkdir(parents=True, exist_ok=True)
+            (repo_dir / ".git").mkdir(exist_ok=True)
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    git.clone_or_update(
+        url=url,
+        ref=None,
+        refresh=None,
+        domain=domain,
+        submodules=["components/foo"],
+    )
+
+    submodule_calls = [
+        c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
+    ]
+    assert len(submodule_calls) == 1
+    cmd = submodule_calls[0][0][0]
+    assert "--depth=1" in cmd
+    assert "components/foo" in cmd
+    # The `--` terminator must precede the submodule paths so a path
+    # beginning with `-` cannot be parsed as an option.
+    assert cmd.index("--") < cmd.index("components/foo")
+
+
+def test_refresh_fetch_is_shallow(tmp_path: Path, mock_run_git_command: Mock) -> None:
+    """The refresh-path fetch should use --depth=1."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    ref = "main"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, ref, domain)
+
+    _setup_old_repo(repo_dir)
+    mock_run_git_command.return_value = "abc123"
+
+    git.clone_or_update(
+        url=url, ref=ref, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+
+    fetch_calls = [c for c in mock_run_git_command.call_args_list if "fetch" in c[0][0]]
+    assert len(fetch_calls) == 1
+    cmd = fetch_calls[0][0][0]
+    assert "--depth=1" in cmd
+    # Ref must still be in the refresh fetch so the right tip is updated.
+    assert cmd[-1] == ref
+
+
+def test_refresh_submodule_update_is_shallow(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """The refresh-path submodule update should use --depth=1."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    _setup_old_repo(repo_dir)
+    mock_run_git_command.return_value = "abc123"
+
+    git.clone_or_update(
+        url=url,
+        ref=None,
+        refresh=TimePeriodSeconds(days=1),
+        domain=domain,
+        submodules=["components/foo"],
+    )
+
+    submodule_calls = [
+        c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
+    ]
+    assert len(submodule_calls) == 1
+    cmd = submodule_calls[0][0][0]
+    assert "--depth=1" in cmd
+    assert "components/foo" in cmd
+    assert cmd.index("--") < cmd.index("components/foo")
+
+
+def test_refresh_picks_up_new_remote_commits(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Shallow fetch must still pull new commits when the remote tip moves.
+
+    Simulates a stale local repo at SHA "old" while the remote has advanced
+    to SHA "new". The refresh path must run fetch (with --depth=1) followed
+    by reset --hard FETCH_HEAD so the working tree advances to the new tip.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    ref = "main"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, ref, domain)
+
+    _setup_old_repo(repo_dir)
+
+    # rev-parse is called once before fetch to record the pre-update SHA.
+    rev_parse_calls = {"count": 0}
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        cmd_type = _get_git_command_type(cmd)
+        if cmd_type == "rev-parse":
+            rev_parse_calls["count"] += 1
+            return "old_sha"
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    _, revert = git.clone_or_update(
+        url=url, ref=ref, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+
+    # Verify the refresh sequence: rev-parse -> stash -> fetch (depth=1) -> reset
+    call_list = mock_run_git_command.call_args_list
+    cmd_sequence = [_get_git_command_type(c[0][0]) for c in call_list]
+    assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset"]
+
+    fetch_cmd = call_list[2][0][0]
+    assert "--depth=1" in fetch_cmd
+    assert fetch_cmd[-1] == ref
+
+    reset_cmd = call_list[3][0][0]
+    assert reset_cmd[-1] == "FETCH_HEAD"
+
+    # revert callback should reset back to the recorded pre-update SHA.
+    assert revert is not None
+    revert()
+    assert mock_run_git_command.call_args_list[-1][0][0] == [
+        "git",
+        "reset",
+        "--hard",
+        "old_sha",
+    ]

tests/unit_tests/test_loader.py

@@ -158,3 +158,167 @@ def test_component_manifest_resources_with_filter_source_files() -> None:
 
         # Verify the correct number of resources
         assert len(resources) == 3  # test.cpp, test.h, common.cpp
+
+
+# ---------------------------------------------------------------------------
+# recursive_sources — used only by the core "esphome" manifest so that files
+# in esphome/core/<subdir>/*.cpp (e.g. esphome/core/wake/wake_host.cpp) are
+# discovered without promoting <subdir>/ to a Python subpackage.
+# ---------------------------------------------------------------------------
+
+
+def _mock_file(filename: str) -> MagicMock:
+    m = MagicMock()
+    m.name = filename
+    m.is_file.return_value = True
+    m.is_dir.return_value = False
+    return m
+
+
+def _mock_dir(dirname: str, children: list, has_init: bool = False) -> MagicMock:
+    """Mock a directory entry with an iterdir() and joinpath('__init__.py')."""
+    d = MagicMock()
+    d.name = dirname
+    d.is_file.return_value = False
+    d.is_dir.return_value = True
+    d.iterdir.return_value = children
+    init_marker = MagicMock()
+    init_marker.is_file.return_value = has_init
+    d.joinpath.return_value = init_marker
+    return d
+
+
+def test_component_manifest_resources_non_recursive_skips_subdirs() -> None:
+    """Default (recursive_sources=False) does not descend into subdirectories."""
+    mock_module = MagicMock()
+    mock_module.__package__ = "esphome.components.test_component"
+    # No FILTER_SOURCE_FILES.
+    del mock_module.FILTER_SOURCE_FILES
+
+    manifest = ComponentManifest(mock_module)  # recursive_sources defaults to False
+
+    top_level = [
+        _mock_file("top.cpp"),
+        _mock_dir("subdir", [_mock_file("nested.cpp")]),
+    ]
+    with patch("importlib.resources.files") as mock_files_func:
+        pkg = MagicMock()
+        pkg.iterdir.return_value = top_level
+        mock_files_func.return_value = pkg
+
+        names = [r.resource for r in manifest.resources]
+
+    assert names == ["top.cpp"]
+
+
+def test_component_manifest_resources_recursive_walks_non_subpackage_subdirs() -> None:
+    """With recursive_sources=True, a subdir without __init__.py is walked."""
+    mock_module = MagicMock()
+    mock_module.__package__ = "esphome.core"
+    del mock_module.FILTER_SOURCE_FILES
+
+    manifest = ComponentManifest(mock_module, recursive_sources=True)
+
+    wake_dir = _mock_dir(
+        "wake",
+        [
+            _mock_file("wake_host.cpp"),
+            _mock_file("wake_host.h"),
+            _mock_file("README.md"),  # wrong suffix, excluded
+        ],
+        has_init=False,
+    )
+    top_level = [
+        _mock_file("wake.h"),
+        wake_dir,
+    ]
+    with patch("importlib.resources.files") as mock_files_func:
+        pkg = MagicMock()
+        pkg.iterdir.return_value = top_level
+        mock_files_func.return_value = pkg
+
+        names = sorted(r.resource for r in manifest.resources)
+
+    assert names == ["wake.h", "wake/wake_host.cpp", "wake/wake_host.h"]
+
+
+def test_component_manifest_resources_recursive_skips_subpackages() -> None:
+    """Subdirectories that ARE Python subpackages (contain __init__.py) are
+    skipped even with recursive_sources=True — those load as their own
+    ComponentManifest and would otherwise be double-counted."""
+    mock_module = MagicMock()
+    mock_module.__package__ = "esphome.components.haier"
+    del mock_module.FILTER_SOURCE_FILES
+
+    manifest = ComponentManifest(mock_module, recursive_sources=True)
+
+    button_pkg = _mock_dir(
+        "button",
+        [_mock_file("self_cleaning.cpp")],
+        has_init=True,  # Python subpackage — must be skipped.
+    )
+    top_level = [
+        _mock_file("haier.cpp"),
+        button_pkg,
+    ]
+    with patch("importlib.resources.files") as mock_files_func:
+        pkg = MagicMock()
+        pkg.iterdir.return_value = top_level
+        mock_files_func.return_value = pkg
+
+        names = [r.resource for r in manifest.resources]
+
+    assert names == ["haier.cpp"]
+
+
+def test_component_manifest_resources_recursive_skips_pycache() -> None:
+    """__pycache__ inside a recursive walk must never be descended into."""
+    mock_module = MagicMock()
+    mock_module.__package__ = "esphome.core"
+    del mock_module.FILTER_SOURCE_FILES
+
+    manifest = ComponentManifest(mock_module, recursive_sources=True)
+
+    # __pycache__ is_dir=True but must be skipped without checking __init__.py
+    # or calling iterdir (would yield compiled artifacts).
+    pycache = _mock_dir("__pycache__", [_mock_file("wake.cpython-314.pyc")])
+    top_level = [
+        _mock_file("wake.h"),
+        pycache,
+    ]
+    with patch("importlib.resources.files") as mock_files_func:
+        pkg = MagicMock()
+        pkg.iterdir.return_value = top_level
+        mock_files_func.return_value = pkg
+
+        names = [r.resource for r in manifest.resources]
+
+    assert names == ["wake.h"]
+
+
+def test_component_manifest_resources_recursive_filter_source_files_supports_subpaths() -> (
+    None
+):
+    """FILTER_SOURCE_FILES entries using '/'-joined subpaths exclude files
+    inside a recursively-walked subdir."""
+    mock_module = MagicMock()
+    mock_module.__package__ = "esphome.core"
+    mock_module.FILTER_SOURCE_FILES = lambda: ["wake/wake_host.cpp"]
+
+    manifest = ComponentManifest(mock_module, recursive_sources=True)
+
+    wake_dir = _mock_dir(
+        "wake",
+        [
+            _mock_file("wake_host.cpp"),  # excluded
+            _mock_file("wake_freertos.cpp"),  # kept
+        ],
+    )
+    with patch("importlib.resources.files") as mock_files_func:
+        pkg = MagicMock()
+        pkg.iterdir.return_value = [wake_dir]
+        mock_files_func.return_value = pkg
+
+        names = [r.resource for r in manifest.resources]
+
+    assert names == ["wake/wake_freertos.cpp"]