Merge remote-tracking branch 'upstream/dev' into optimize-value-accuracy

# Conflicts:
#	esphome/core/helpers.h
#	tests/components/core/test_helpers.cpp

tests/benchmarks/components/api/bench_log_response.cpp

@@ -0,0 +1,118 @@
+#include <benchmark/benchmark.h>
+
+#include "esphome/components/api/api_pb2.h"
+#include "esphome/components/api/api_buffer.h"
+
+namespace esphome::api::benchmarks {
+
+// Inner iteration count to amortize CodSpeed instrumentation overhead.
+static constexpr int kInnerIterations = 2000;
+
+// Typical log line: "[12:34:56][D][sensor:094]: 'Temperature': Sending state 23.50000 with 1 decimals of accuracy"
+static constexpr const char *kTypicalLogLine =
+    "[12:34:56][D][sensor:094]: 'Temperature': Sending state 23.50000 with 1 decimals of accuracy";
+
+// Short log line: "[12:34:56][I][app:029]: Running..."
+static constexpr const char *kShortLogLine = "[12:34:56][I][app:029]: Running...";
+
+// --- Encode ---
+
+static void Encode_LogResponse_Typical(benchmark::State &state) {
+  APIBuffer buffer;
+  SubscribeLogsResponse msg;
+  msg.level = enums::LOG_LEVEL_DEBUG;
+  msg.set_message(reinterpret_cast<const uint8_t *>(kTypicalLogLine), strlen(kTypicalLogLine));
+  uint32_t size = msg.calculate_size();
+  buffer.resize(size);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      ProtoWriteBuffer writer(&buffer, 0);
+      msg.encode(writer);
+    }
+    benchmark::DoNotOptimize(buffer.data());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Encode_LogResponse_Typical);
+
+static void Encode_LogResponse_Short(benchmark::State &state) {
+  APIBuffer buffer;
+  SubscribeLogsResponse msg;
+  msg.level = enums::LOG_LEVEL_INFO;
+  msg.set_message(reinterpret_cast<const uint8_t *>(kShortLogLine), strlen(kShortLogLine));
+  uint32_t size = msg.calculate_size();
+  buffer.resize(size);
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      ProtoWriteBuffer writer(&buffer, 0);
+      msg.encode(writer);
+    }
+    benchmark::DoNotOptimize(buffer.data());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Encode_LogResponse_Short);
+
+// --- Calculate Size ---
+
+static void CalculateSize_LogResponse_Typical(benchmark::State &state) {
+  SubscribeLogsResponse msg;
+  msg.level = enums::LOG_LEVEL_DEBUG;
+  msg.set_message(reinterpret_cast<const uint8_t *>(kTypicalLogLine), strlen(kTypicalLogLine));
+
+  for (auto _ : state) {
+    uint32_t result = 0;
+    for (int i = 0; i < kInnerIterations; i++) {
+      result += msg.calculate_size();
+    }
+    benchmark::DoNotOptimize(result);
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(CalculateSize_LogResponse_Typical);
+
+// --- Calc + Encode (steady state) ---
+
+static void CalcAndEncode_LogResponse_Typical(benchmark::State &state) {
+  APIBuffer buffer;
+  SubscribeLogsResponse msg;
+  msg.level = enums::LOG_LEVEL_DEBUG;
+  msg.set_message(reinterpret_cast<const uint8_t *>(kTypicalLogLine), strlen(kTypicalLogLine));
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      uint32_t size = msg.calculate_size();
+      buffer.resize(size);
+      ProtoWriteBuffer writer(&buffer, 0);
+      msg.encode(writer);
+    }
+    benchmark::DoNotOptimize(buffer.data());
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(CalcAndEncode_LogResponse_Typical);
+
+// --- Calc + Encode (fresh allocation each time) ---
+
+static void CalcAndEncode_LogResponse_Typical_Fresh(benchmark::State &state) {
+  SubscribeLogsResponse msg;
+  msg.level = enums::LOG_LEVEL_DEBUG;
+  msg.set_message(reinterpret_cast<const uint8_t *>(kTypicalLogLine), strlen(kTypicalLogLine));
+
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      APIBuffer buffer;
+      uint32_t size = msg.calculate_size();
+      buffer.resize(size);
+      ProtoWriteBuffer writer(&buffer, 0);
+      msg.encode(writer);
+      benchmark::DoNotOptimize(buffer.data());
+    }
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(CalcAndEncode_LogResponse_Typical_Fresh);
+
+}  // namespace esphome::api::benchmarks

tests/benchmarks/core/bench_helpers.cpp

@@ -1,4 +1,6 @@
 #include <benchmark/benchmark.h>
+#include <cinttypes>
+#include <cstdio>
 
 #include "esphome/core/helpers.h"
 
@@ -307,4 +309,58 @@ static void Base64Decode_32Bytes(benchmark::State &state) {
 }
 BENCHMARK(Base64Decode_32Bytes);
 
+// --- uint32_to_str() vs snprintf ---
+
+static void Uint32ToStr_Small(benchmark::State &state) {
+  char buf[UINT32_MAX_STR_SIZE];
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      uint32_to_str(buf, 12345);
+      benchmark::DoNotOptimize(buf);
+      benchmark::ClobberMemory();
+    }
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Uint32ToStr_Small);
+
+static void Snprintf_Uint32_Small(benchmark::State &state) {
+  char buf[UINT32_MAX_STR_SIZE];
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      snprintf(buf, sizeof(buf), "%" PRIu32, static_cast<uint32_t>(12345));
+      benchmark::DoNotOptimize(buf);
+      benchmark::ClobberMemory();
+    }
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Snprintf_Uint32_Small);
+
+static void Uint32ToStr_Large(benchmark::State &state) {
+  char buf[UINT32_MAX_STR_SIZE];
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      uint32_to_str(buf, 4294967295u);
+      benchmark::DoNotOptimize(buf);
+      benchmark::ClobberMemory();
+    }
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Uint32ToStr_Large);
+
+static void Snprintf_Uint32_Large(benchmark::State &state) {
+  char buf[UINT32_MAX_STR_SIZE];
+  for (auto _ : state) {
+    for (int i = 0; i < kInnerIterations; i++) {
+      snprintf(buf, sizeof(buf), "%" PRIu32, static_cast<uint32_t>(4294967295u));
+      benchmark::DoNotOptimize(buf);
+      benchmark::ClobberMemory();
+    }
+  }
+  state.SetItemsProcessed(state.iterations() * kInnerIterations);
+}
+BENCHMARK(Snprintf_Uint32_Large);
+
 }  // namespace esphome::benchmarks

tests/component_tests/packages/test_packages.py

@@ -1,11 +1,18 @@
 """Tests for the packages component."""
 
+import logging
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from esphome.components.packages import CONFIG_SCHEMA, do_packages_pass, merge_packages
+from esphome.components.packages import (
+    CONFIG_SCHEMA,
+    _walk_packages,
+    do_packages_pass,
+    is_package_definition,
+    merge_packages,
+)
 from esphome.components.substitutions import do_substitution_pass
 import esphome.config as config_module
 from esphome.config import resolve_extend_remove
@@ -37,7 +44,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.util import OrderedDict
-from esphome.yaml_util import add_context
+from esphome.yaml_util import IncludeFile, add_context
 
 # Test strings
 TEST_DEVICE_NAME = "test_device_name"
@@ -79,6 +86,44 @@ def packages_pass(config):
     return config
 
 
+_INCLUDE_FILE = "INCLUDE_FILE"
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        # IncludeFile objects are package definitions
+        (_INCLUDE_FILE, True),
+        # Git URL shorthand strings are package definitions
+        ("github://esphome/firmware/base.yaml@main", True),
+        # Remote package dicts (with url key) are package definitions
+        ({"url": "https://github.com/esphome/firmware", "file": "base.yaml"}, True),
+        # Plain config dicts are NOT package definitions (they are config fragments)
+        ({"wifi": {"ssid": "test"}}, False),
+        # None is not a package definition
+        (None, False),
+        # Lists are not package definitions
+        ([{"wifi": {"ssid": "test"}}], False),
+        # Empty dicts are not package definitions
+        ({}, False),
+    ],
+    ids=[
+        "include_file",
+        "git_shorthand",
+        "remote_package",
+        "config_fragment",
+        "none",
+        "list",
+        "empty_dict",
+    ],
+)
+def test_is_package_definition(value: object, expected: bool) -> None:
+    """Test that is_package_definition correctly identifies package definitions."""
+    if value is _INCLUDE_FILE:
+        value = MagicMock(spec=IncludeFile)
+    assert is_package_definition(value) is expected
+
+
 def test_package_unused(basic_esphome, basic_wifi) -> None:
     """
     Ensures do_package_pass does not change a config if packages aren't used.
@@ -1061,6 +1106,51 @@ def test_packages_invalid_type_raises() -> None:
         do_packages_pass(config)
 
 
+@patch("esphome.components.packages.resolve_include")
+def test_packages_include_file_resolves_to_list(mock_resolve_include) -> None:
+    """When packages: is an IncludeFile that resolves to a list, it is processed correctly."""
+    include_file = MagicMock(spec=IncludeFile)
+    package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
+    mock_resolve_include.return_value = ([package_content], None)
+
+    config = {CONF_PACKAGES: include_file}
+    result = do_packages_pass(config)
+    result = merge_packages(result)
+
+    assert result == {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
+
+
+@patch("esphome.components.packages.resolve_include")
+def test_packages_include_file_resolves_to_dict(mock_resolve_include) -> None:
+    """When packages: is an IncludeFile that resolves to a dict, it is processed correctly."""
+    include_file = MagicMock(spec=IncludeFile)
+    package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
+    mock_resolve_include.return_value = ({"network": package_content}, None)
+
+    config = {CONF_PACKAGES: include_file}
+    result = do_packages_pass(config)
+    result = merge_packages(result)
+
+    assert result == {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
+
+
+@patch("esphome.components.packages.resolve_include")
+def test_packages_include_file_resolves_to_invalid_type_raises(
+    mock_resolve_include,
+) -> None:
+    """When packages: is an IncludeFile that resolves to an invalid type, cv.Invalid is raised."""
+    include_file = MagicMock(spec=IncludeFile)
+    mock_resolve_include.return_value = ("not_a_dict_or_list", None)
+
+    config = {CONF_PACKAGES: include_file}
+    with pytest.raises(
+        cv.Invalid, match="Packages must be a key to value mapping or list"
+    ) as exc_info:
+        do_packages_pass(config)
+
+    assert exc_info.value.path == [CONF_PACKAGES]
+
+
 @pytest.mark.parametrize(
     "invalid_package",
     [
@@ -1107,6 +1197,134 @@ def test_invalid_package_contents_masked_by_deprecation(
         do_packages_pass(config)
 
 
+def test_named_dict_with_include_files_no_false_deprecation_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Package errors in named dicts must not trigger the deprecated fallback."""
+    good_include = MagicMock(spec=IncludeFile)
+    bad_include = MagicMock(spec=IncludeFile)
+
+    config = {
+        CONF_PACKAGES: {
+            "good_pkg": good_include,
+            "bad_pkg": bad_include,
+        },
+    }
+
+    call_count = 0
+
+    def failing_callback(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            # First package processes fine
+            return {CONF_WIFI: {CONF_SSID: "test"}}
+        # Second package has an error (e.g. jinja syntax error)
+        raise cv.Invalid("simulated jinja error in bad_pkg")
+
+    with (
+        caplog.at_level(logging.WARNING),
+        pytest.raises(cv.Invalid, match="simulated jinja error"),
+    ):
+        _walk_packages(config, failing_callback)
+
+    # Must NOT emit the deprecated single-package warning
+    assert "deprecated" not in caplog.text.lower()
+
+
+def test_validate_deprecated_false_raises_directly(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """With validate_deprecated=False, errors raise directly without fallback.
+
+    This is the codepath used for remote packages where _process_remote_package
+    returns already-resolved dicts that is_package_definition cannot detect.
+    """
+    config = {
+        CONF_PACKAGES: {
+            "pkg_a": {CONF_WIFI: {CONF_SSID: "test"}},
+            "pkg_b": {CONF_WIFI: {CONF_SSID: "test2"}},
+        },
+    }
+
+    call_count = 0
+
+    def failing_callback(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return package_config
+        raise cv.Invalid("nested error")
+
+    with (
+        caplog.at_level(logging.WARNING),
+        pytest.raises(cv.Invalid, match="nested error"),
+    ):
+        _walk_packages(config, failing_callback, validate_deprecated=False)
+
+    assert "deprecated" not in caplog.text.lower()
+
+
+def test_error_on_first_declared_package_still_detected() -> None:
+    """When the first declared package errors, it's the last processed in reverse.
+
+    All other entries are already resolved to dicts, but the failing entry
+    retains its original IncludeFile value since assignment was skipped.
+    """
+    config = {
+        CONF_PACKAGES: {
+            "first_pkg": MagicMock(spec=IncludeFile),
+            "second_pkg": MagicMock(spec=IncludeFile),
+            "third_pkg": MagicMock(spec=IncludeFile),
+        },
+    }
+
+    call_count = 0
+
+    def fail_on_last(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        # Reverse iteration: third_pkg (1), second_pkg (2), first_pkg (3)
+        if call_count < 3:
+            return {CONF_WIFI: {CONF_SSID: "test"}}
+        raise cv.Invalid("error in first_pkg")
+
+    with pytest.raises(cv.Invalid, match="error in first_pkg"):
+        _walk_packages(config, fail_on_last)
+
+
+def test_deprecated_single_package_fallback_still_works(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The deprecated single-package form still falls back at the top level.
+
+    When a dict's values are plain config fragments (not package definitions)
+    and the callback fails, the deprecated fallback wraps the dict in a list
+    and retries with a deprecation warning.
+    """
+    config = {
+        CONF_PACKAGES: {
+            CONF_WIFI: {CONF_SSID: "test", CONF_PASSWORD: "secret"},
+        },
+    }
+
+    attempt = 0
+
+    def fail_then_succeed(package_config: dict, context: object) -> dict:
+        nonlocal attempt
+        attempt += 1
+        if attempt == 1:
+            # First attempt: treating as named dict fails
+            raise cv.Invalid("not a valid package")
+        # Second attempt: after fallback wraps as list, succeeds
+        return package_config
+
+    with caplog.at_level(logging.WARNING):
+        _walk_packages(config, fail_then_succeed)
+
+    assert "deprecated" in caplog.text.lower()
+
+
 def test_merge_packages_invalid_nested_type_raises() -> None:
     """Invalid nested packages type during merge raises cv.Invalid."""
     config = {

tests/components/canbus/common.yaml

@@ -50,6 +50,13 @@ button:
   - platform: template
     name: Canbus Actions
     on_press:
+      - canbus.send:
+          can_id: 0x601
+          data: [0, 1, 2]
+      - canbus.send:
+          can_id: 0x1FFFFFFF
+          use_extended_id: true
+          data: [0, 1, 2]
       - canbus.send: "abc"
       - canbus.send: [0, 1, 2]
       - canbus.send: !lambda return {0, 1, 2};

tests/components/core/test_helpers.cpp

@@ -3,7 +3,119 @@
 
 #include "esphome/core/helpers.h"
 
-namespace esphome::testing {
+namespace esphome::core::testing {
+
+// --- format_hex_to() ---
+
+TEST(FormatHexTo, Basic) {
+  const uint8_t data[] = {0xAB, 0xCD, 0xEF};
+  char buffer[7];  // 3 * 2 + 1
+  format_hex_to(buffer, data, 3);
+  EXPECT_STREQ(buffer, "abcdef");
+}
+
+TEST(FormatHexTo, SingleByte) {
+  const uint8_t data[] = {0x0F};
+  char buffer[3];
+  format_hex_to(buffer, data, 1);
+  EXPECT_STREQ(buffer, "0f");
+}
+
+TEST(FormatHexTo, ZeroLength) {
+  char buffer[4] = "xxx";
+  format_hex_to(buffer, static_cast<size_t>(sizeof(buffer)), static_cast<const uint8_t *>(nullptr), 0);
+  EXPECT_STREQ(buffer, "");
+}
+
+TEST(FormatHexTo, ZeroBufferSize) {
+  char buffer[4] = "xxx";
+  const uint8_t data[] = {0xAB};
+  format_hex_to(buffer, static_cast<size_t>(0), data, 1);
+  // Should not crash, buffer unchanged
+  EXPECT_EQ(buffer[0], 'x');
+}
+
+TEST(FormatHexTo, BufferTooSmall) {
+  const uint8_t data[] = {0xAB, 0xCD, 0xEF};
+  char buffer[5];  // only room for 2 bytes
+  format_hex_to(buffer, data, 3);
+  EXPECT_STREQ(buffer, "abcd");
+}
+
+TEST(FormatHexTo, MacAddress) {
+  const uint8_t mac[] = {0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF};
+  char buffer[13];
+  format_hex_to(buffer, mac, 6);
+  EXPECT_STREQ(buffer, "aabbccddeeff");
+}
+
+// --- format_hex_pretty_to() ---
+
+TEST(FormatHexPrettyTo, BasicColon) {
+  const uint8_t data[] = {0xAB, 0xCD, 0xEF};
+  char buffer[9];  // 3 * 3
+  format_hex_pretty_to(buffer, data, 3);
+  EXPECT_STREQ(buffer, "AB:CD:EF");
+}
+
+TEST(FormatHexPrettyTo, SingleByte) {
+  const uint8_t data[] = {0x0F};
+  char buffer[3];
+  format_hex_pretty_to(buffer, data, 1);
+  EXPECT_STREQ(buffer, "0F");
+}
+
+TEST(FormatHexPrettyTo, ZeroLength) {
+  char buffer[4] = "xxx";
+  format_hex_pretty_to(buffer, static_cast<size_t>(sizeof(buffer)), static_cast<const uint8_t *>(nullptr), 0);
+  EXPECT_STREQ(buffer, "");
+}
+
+TEST(FormatHexPrettyTo, ZeroBufferSize) {
+  char buffer[4] = "xxx";
+  const uint8_t data[] = {0xAB};
+  format_hex_pretty_to(buffer, static_cast<size_t>(0), data, 1);
+  EXPECT_EQ(buffer[0], 'x');
+}
+
+TEST(FormatHexPrettyTo, CustomSeparator) {
+  const uint8_t data[] = {0xAA, 0xBB, 0xCC};
+  char buffer[9];
+  format_hex_pretty_to(buffer, data, 3, '-');
+  EXPECT_STREQ(buffer, "AA-BB-CC");
+}
+
+// --- format_mac_addr_upper() ---
+
+TEST(FormatMacAddrUpper, Basic) {
+  const uint8_t mac[] = {0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF};
+  char buffer[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
+  format_mac_addr_upper(mac, buffer);
+  EXPECT_STREQ(buffer, "AA:BB:CC:DD:EE:FF");
+}
+
+TEST(FormatMacAddrUpper, AllZeros) {
+  const uint8_t mac[] = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
+  char buffer[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
+  format_mac_addr_upper(mac, buffer);
+  EXPECT_STREQ(buffer, "00:00:00:00:00:00");
+}
+
+// --- format_hex_char() ---
+
+TEST(FormatHexChar, LowercaseDigits) {
+  EXPECT_EQ(format_hex_char(0), '0');
+  EXPECT_EQ(format_hex_char(9), '9');
+  EXPECT_EQ(format_hex_char(10), 'a');
+  EXPECT_EQ(format_hex_char(15), 'f');
+}
+
+TEST(FormatHexChar, UppercaseDigits) {
+  EXPECT_EQ(format_hex_pretty_char(0), '0');
+  EXPECT_EQ(format_hex_pretty_char(9), '9');
+  EXPECT_EQ(format_hex_pretty_char(10), 'A');
+  EXPECT_EQ(format_hex_pretty_char(15), 'F');
+}
 
 // --- small_pow10() ---
 
@@ -12,68 +124,6 @@ TEST(SmallPow10, One) { EXPECT_EQ(small_pow10(1), 10u); }
 TEST(SmallPow10, Two) { EXPECT_EQ(small_pow10(2), 100u); }
 TEST(SmallPow10, Three) { EXPECT_EQ(small_pow10(3), 1000u); }
 
-// --- uint32_to_str() ---
-
-TEST(Uint32ToStr, Zero) {
-  char buf[12];
-  char *end = uint32_to_str_(buf, 0);
-  *end = '\0';
-  EXPECT_STREQ(buf, "0");
-  EXPECT_EQ(end - buf, 1);
-}
-
-TEST(Uint32ToStr, SingleDigit) {
-  char buf[12];
-  char *end = uint32_to_str_(buf, 7);
-  *end = '\0';
-  EXPECT_STREQ(buf, "7");
-}
-
-TEST(Uint32ToStr, MultiDigit) {
-  char buf[12];
-  char *end = uint32_to_str_(buf, 12345);
-  *end = '\0';
-  EXPECT_STREQ(buf, "12345");
-  EXPECT_EQ(end - buf, 5);
-}
-
-TEST(Uint32ToStr, Large) {
-  char buf[12];
-  char *end = uint32_to_str_(buf, 4294967295u);
-  *end = '\0';
-  EXPECT_STREQ(buf, "4294967295");
-  EXPECT_EQ(end - buf, 10);
-}
-
-TEST(Uint32ToStr, PowersOfTen) {
-  char buf[12];
-  char *end;
-
-  end = uint32_to_str_(buf, 10);
-  *end = '\0';
-  EXPECT_STREQ(buf, "10");
-
-  end = uint32_to_str_(buf, 100);
-  *end = '\0';
-  EXPECT_STREQ(buf, "100");
-
-  end = uint32_to_str_(buf, 1000);
-  *end = '\0';
-  EXPECT_STREQ(buf, "1000");
-}
-
-// --- uint32_to_str() (public, template with size check) ---
-
-TEST(Uint32ToStr, PublicApi) {
-  char buf[UINT32_MAX_STR_SIZE];
-  EXPECT_EQ(uint32_to_str(buf, 0), 1u);
-  EXPECT_STREQ(buf, "0");
-  EXPECT_EQ(uint32_to_str(buf, 12345), 5u);
-  EXPECT_STREQ(buf, "12345");
-  EXPECT_EQ(uint32_to_str(buf, 4294967295u), 10u);
-  EXPECT_STREQ(buf, "4294967295");
-}
-
 // --- frac_to_str_() ---
 
 TEST(FracToStr, OneDigit) {
@@ -163,4 +213,4 @@ TEST(BufAppendSepStr, Truncation) {
   EXPECT_EQ(end - buf, 7);
 }
 
-}  // namespace esphome::testing
+}  // namespace esphome::core::testing

tests/components/core/test_uint32_to_str.cpp

@@ -0,0 +1,77 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/helpers.h"
+
+namespace esphome::core::testing {
+
+// --- uint32_to_str_unchecked() (internal, raw pointer) ---
+
+TEST(Uint32ToStr, InternalZero) {
+  char buf[UINT32_MAX_STR_SIZE];
+  char *end = uint32_to_str_unchecked(buf, 0);
+  *end = '\0';
+  EXPECT_STREQ(buf, "0");
+  EXPECT_EQ(end - buf, 1);
+}
+
+TEST(Uint32ToStr, InternalSingleDigit) {
+  char buf[UINT32_MAX_STR_SIZE];
+  char *end = uint32_to_str_unchecked(buf, 7);
+  *end = '\0';
+  EXPECT_STREQ(buf, "7");
+}
+
+TEST(Uint32ToStr, InternalMultiDigit) {
+  char buf[UINT32_MAX_STR_SIZE];
+  char *end = uint32_to_str_unchecked(buf, 12345);
+  *end = '\0';
+  EXPECT_STREQ(buf, "12345");
+  EXPECT_EQ(end - buf, 5);
+}
+
+TEST(Uint32ToStr, InternalMaxValue) {
+  char buf[UINT32_MAX_STR_SIZE];
+  char *end = uint32_to_str_unchecked(buf, 4294967295u);
+  *end = '\0';
+  EXPECT_STREQ(buf, "4294967295");
+  EXPECT_EQ(end - buf, 10);
+}
+
+TEST(Uint32ToStr, InternalPowersOfTen) {
+  char buf[UINT32_MAX_STR_SIZE];
+  char *end;
+
+  end = uint32_to_str_unchecked(buf, 10);
+  *end = '\0';
+  EXPECT_STREQ(buf, "10");
+
+  end = uint32_to_str_unchecked(buf, 100);
+  *end = '\0';
+  EXPECT_STREQ(buf, "100");
+
+  end = uint32_to_str_unchecked(buf, 1000000);
+  *end = '\0';
+  EXPECT_STREQ(buf, "1000000");
+}
+
+// --- uint32_to_str() (public, span API) ---
+
+TEST(Uint32ToStr, SpanZero) {
+  char buf[UINT32_MAX_STR_SIZE];
+  EXPECT_EQ(uint32_to_str(buf, 0), 1u);
+  EXPECT_STREQ(buf, "0");
+}
+
+TEST(Uint32ToStr, SpanMultiDigit) {
+  char buf[UINT32_MAX_STR_SIZE];
+  EXPECT_EQ(uint32_to_str(buf, 12345), 5u);
+  EXPECT_STREQ(buf, "12345");
+}
+
+TEST(Uint32ToStr, SpanMaxValue) {
+  char buf[UINT32_MAX_STR_SIZE];
+  EXPECT_EQ(uint32_to_str(buf, 4294967295u), 10u);
+  EXPECT_STREQ(buf, "4294967295");
+}
+
+}  // namespace esphome::core::testing

tests/components/globals/common.yaml

@@ -4,6 +4,14 @@ esphome:
       - globals.set:
           id: glob_int
           value: "10"
+      # Set a float global with an integer literal - must emit the correct
+      # return type so TemplatableFn stores a direct function pointer.
+      - globals.set:
+          id: glob_float
+          value: "102"
+      - globals.set:
+          id: glob_float
+          value: !lambda "return 42;"
 
 globals:
   - id: glob_int

tests/components/lvgl/test.host.yaml

@@ -20,6 +20,7 @@ lvgl:
   - id: lvgl_0
     default_font: space16
     displays: sdl0
+    rotation: 180
     top_layer:
 
   - id: lvgl_1

tests/components/mcp23016/common.yaml

@@ -1,6 +1,10 @@
 mcp23016:
-  i2c_id: i2c_bus
-  id: mcp23016_hub
+  - i2c_id: i2c_bus
+    id: mcp23016_hub
+  - i2c_id: i2c_bus
+    id: mcp23016_hub_int
+    address: 0x21
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/components/pca6416a/common.yaml

@@ -2,6 +2,10 @@ pca6416a:
   - id: pca6416a_hub
     i2c_id: i2c_bus
     address: 0x21
+  - id: pca6416a_hub_int
+    i2c_id: i2c_bus
+    address: 0x22
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/components/tca9555/common.yaml

@@ -2,6 +2,10 @@ tca9555:
   - id: tca9555_hub
     i2c_id: i2c_bus
     address: 0x21
+  - id: tca9555_hub_int
+    i2c_id: i2c_bus
+    address: 0x22
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

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

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/integration/fixtures/status_flags.yaml

@@ -0,0 +1,141 @@
+esphome:
+  name: status-flags-test
+
+host:
+api:
+  actions:
+    # Warning flag services for sensor_a
+    - action: set_warning_a
+      then:
+        - lambda: "id(sensor_a)->status_set_warning();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+    - action: clear_warning_a
+      then:
+        - lambda: "id(sensor_a)->status_clear_warning();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+
+    # Warning flag services for sensor_b
+    - action: set_warning_b
+      then:
+        - lambda: "id(sensor_b)->status_set_warning();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+    - action: clear_warning_b
+      then:
+        - lambda: "id(sensor_b)->status_clear_warning();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+
+    # Error flag services for sensor_a
+    - action: set_error_a
+      then:
+        - lambda: "id(sensor_a)->status_set_error();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+    - action: clear_error_a
+      then:
+        - lambda: "id(sensor_a)->status_clear_error();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+
+    # Error flag services for sensor_b
+    - action: set_error_b
+      then:
+        - lambda: "id(sensor_b)->status_set_error();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+    - action: clear_error_b
+      then:
+        - lambda: "id(sensor_b)->status_clear_error();"
+        - component.update: app_warning_bit
+        - component.update: app_error_bit
+
+    # Snapshot of the status_led_light's output state for observation.
+    - action: snapshot_led
+      then:
+        - component.update: status_led_writes
+        - component.update: status_led_last_state
+
+logger:
+
+# Tracks each write to the fake status_led output.
+globals:
+  - id: status_led_write_count
+    type: uint32_t
+    restore_value: no
+    initial_value: "0"
+  - id: status_led_last_write
+    type: bool
+    restore_value: no
+    initial_value: "false"
+
+# Fake binary output — status_led_light writes to this instead of a pin.
+# Every write bumps a counter and records the last value, both of which
+# are exposed below so the test can verify status_led_light's loop is
+# actually reading App.get_app_state() and responding.
+output:
+  - platform: template
+    id: fake_status_led
+    type: binary
+    write_action:
+      - globals.set:
+          id: status_led_write_count
+          value: !lambda "return id(status_led_write_count) + 1;"
+      - globals.set:
+          id: status_led_last_write
+          value: !lambda "return state;"
+
+# Actual status_led_light component under test.
+light:
+  - platform: status_led
+    name: Status LED
+    id: status_led_light_id
+    output: fake_status_led
+
+sensor:
+  # Two components that the test will toggle warning/error flags on.
+  - platform: template
+    name: Sensor A
+    id: sensor_a
+    update_interval: 24h
+    lambda: return 1.0;
+  - platform: template
+    name: Sensor B
+    id: sensor_b
+    update_interval: 24h
+    lambda: return 2.0;
+
+  # Expose App.app_state_'s STATUS_LED_WARNING / STATUS_LED_ERROR bits
+  # as 0.0 / 1.0. force_update ensures every manual component.update
+  # publishes even if the value is unchanged.
+  - platform: template
+    name: App Warning Bit
+    id: app_warning_bit
+    update_interval: 24h
+    force_update: true
+    lambda: |-
+      return (App.get_app_state() & STATUS_LED_WARNING) != 0 ? 1.0 : 0.0;
+  - platform: template
+    name: App Error Bit
+    id: app_error_bit
+    update_interval: 24h
+    force_update: true
+    lambda: |-
+      return (App.get_app_state() & STATUS_LED_ERROR) != 0 ? 1.0 : 0.0;
+
+  # Observables for the fake status_led output.
+  - platform: template
+    name: Status LED Writes
+    id: status_led_writes
+    update_interval: 24h
+    force_update: true
+    lambda: return id(status_led_write_count);
+  - platform: template
+    name: Status LED Last State
+    id: status_led_last_state
+    update_interval: 24h
+    force_update: true
+    lambda: |-
+      return id(status_led_last_write) ? 1.0 : 0.0;

tests/integration/test_status_flags.py

@@ -0,0 +1,209 @@
+"""Integration tests for Component::status_set/clear_warning/error propagation.
+
+Verifies that toggling STATUS_LED_WARNING / STATUS_LED_ERROR on individual
+components correctly updates the app-wide bits on Application::app_state_,
+AND that the status_led_light component actually responds to those bits
+by writing to its output (the full chain from component.status_set_warning
+→ App.app_state_ → status_led_light.loop() reading get_app_state()).
+
+Exercises the multi-component OR semantics (the app bit stays set while
+any component still has the flag, and only clears when the last component
+clears its bit), the independence of warning and error, and the actual
+status_led_light read of the bits via a fake template output that counts
+writes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .state_utils import InitialStateHelper, SensorTracker, build_key_to_entity_mapping
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+# Time to let the host-mode main loop run so status_led_light.loop() can
+# execute enough iterations to produce measurable write-count changes on
+# the fake template output. 300 ms is well above the minimum needed.
+STATUS_LED_SETTLE_S = 0.3
+
+
+@pytest.mark.asyncio
+async def test_status_flags(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        entities, services = await client.list_entities_services()
+
+        # Map every custom API service by name for the test to execute.
+        svc = {s.name: s for s in services}
+        for name in (
+            "set_warning_a",
+            "clear_warning_a",
+            "set_warning_b",
+            "clear_warning_b",
+            "set_error_a",
+            "clear_error_a",
+            "set_error_b",
+            "clear_error_b",
+            "snapshot_led",
+        ):
+            assert name in svc, f"service {name} not registered"
+
+        # Track every sensor we care about. SensorTracker gives us
+        # expect(value) / expect_any() futures that resolve when a
+        # matching state arrives; much simpler than manual bookkeeping.
+        tracker = SensorTracker(
+            [
+                "app_warning_bit",
+                "app_error_bit",
+                "status_led_writes",
+                "status_led_last_state",
+            ]
+        )
+        tracker.key_to_sensor.update(
+            build_key_to_entity_mapping(entities, list(tracker.sensor_states.keys()))
+        )
+
+        # Swallow initial state broadcasts so the test only reacts to
+        # state changes triggered by our service calls.
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(tracker.on_state))
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        async def call(name: str) -> None:
+            await client.execute_service(svc[name], {})
+
+        async def call_and_expect_bits(
+            service_name: str, *, warning: float, error: float
+        ) -> None:
+            """Execute a service and wait for both app bit sensors to match.
+
+            Each bit-toggling service calls component.update on both
+            app_warning_bit and app_error_bit, so both sensors publish.
+            """
+            futures = tracker.expect_all(
+                {"app_warning_bit": warning, "app_error_bit": error}
+            )
+            await call(service_name)
+            await tracker.await_all(futures)
+
+        async def snapshot_led_writes() -> int:
+            """Trigger a publish of the fake status_led output counter and return it."""
+            future = tracker.expect_any("status_led_writes")
+            await call("snapshot_led")
+            await tracker.await_change(future, "status_led_writes")
+            return int(tracker.sensor_states["status_led_writes"][-1])
+
+        # ---- Baseline: everything clean ----
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
+
+        # ================================================================
+        # Part 1 — STATUS_LED_WARNING propagation to App.app_state_
+        # ================================================================
+
+        # Single component set/clear
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
+
+        # Multi-component OR: both set, clear A, bit stays (B still has it), clear B, gone
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        await call_and_expect_bits("set_warning_b", warning=1.0, error=0.0)
+        await call_and_expect_bits("clear_warning_a", warning=1.0, error=0.0)
+        await call_and_expect_bits("clear_warning_b", warning=0.0, error=0.0)
+
+        # Opposite clear order
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        await call_and_expect_bits("set_warning_b", warning=1.0, error=0.0)
+        await call_and_expect_bits("clear_warning_b", warning=1.0, error=0.0)
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
+
+        # ================================================================
+        # Part 2 — STATUS_LED_ERROR propagation (same scenarios)
+        # ================================================================
+
+        await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
+        await call_and_expect_bits("clear_error_a", warning=0.0, error=0.0)
+
+        await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
+        await call_and_expect_bits("set_error_b", warning=0.0, error=1.0)
+        await call_and_expect_bits("clear_error_a", warning=0.0, error=1.0)
+        await call_and_expect_bits("clear_error_b", warning=0.0, error=0.0)
+
+        # ================================================================
+        # Part 3 — warning and error are independent
+        # ================================================================
+
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        await call_and_expect_bits("set_error_b", warning=1.0, error=1.0)
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=1.0)
+        await call_and_expect_bits("clear_error_b", warning=0.0, error=0.0)
+
+        # ================================================================
+        # Part 4 — status_led_light actually reads App.app_state_
+        # ================================================================
+        # The fake status_led_light output increments status_led_write_count
+        # on every write. status_led_light::loop() writes its output on every
+        # iteration while an error/warning bit is set, so after holding a
+        # warning for ~300 ms we should see the counter move significantly.
+        # This is the end-to-end proof that the bits we set above actually
+        # reach status_led_light and drive its behavior.
+
+        count_before_warning = await snapshot_led_writes()
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        # Let status_led_light's loop run long enough to toggle the pin
+        # several times (it reads get_app_state() every main loop iteration).
+        await asyncio.sleep(STATUS_LED_SETTLE_S)
+        count_after_warning = await snapshot_led_writes()
+        assert count_after_warning > count_before_warning, (
+            "status_led_light did not respond to STATUS_LED_WARNING being set: "
+            f"write count stayed at {count_before_warning} → {count_after_warning}. "
+            "The full chain Component::status_set_warning → App.app_state_ → "
+            "status_led_light::loop reading get_app_state() is broken."
+        )
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)
+
+        # Same check for ERROR
+        count_before_error = await snapshot_led_writes()
+        await call_and_expect_bits("set_error_a", warning=0.0, error=1.0)
+        await asyncio.sleep(STATUS_LED_SETTLE_S)
+        count_after_error = await snapshot_led_writes()
+        assert count_after_error > count_before_error, (
+            "status_led_light did not respond to STATUS_LED_ERROR being set: "
+            f"write count stayed at {count_before_error} → {count_after_error}. "
+        )
+        await call_and_expect_bits("clear_error_a", warning=0.0, error=0.0)
+
+        # ---- Set → clear → re-set round-trip ----
+        # After clearing, status_led_light stops writing (steady state).
+        # Re-setting the flag must make it resume. This guards against a
+        # future idle optimization (e.g. #15642) where status_led disables
+        # its own loop when idle: if the re-enable path were broken, the
+        # second set would not produce writes.
+        #
+        # Snapshot AFTER the clear to avoid counting writes that were still
+        # in-flight from the error-set phase.
+        count_after_clear = await snapshot_led_writes()
+        await asyncio.sleep(STATUS_LED_SETTLE_S)
+        count_after_idle = await snapshot_led_writes()
+        assert count_after_idle - count_after_clear <= 5, (
+            "status_led_light kept writing after warning/error was cleared: "
+            f"count grew from {count_after_clear} to {count_after_idle}. "
+            "Expected it to stop writing once all status bits were clear."
+        )
+        # Re-set warning — writes must resume.
+        await call_and_expect_bits("set_warning_a", warning=1.0, error=0.0)
+        await asyncio.sleep(STATUS_LED_SETTLE_S)
+        count_after_reset = await snapshot_led_writes()
+        assert count_after_reset > count_after_idle + 5, (
+            "status_led_light did not resume writing after re-setting "
+            f"STATUS_LED_WARNING: count went from {count_after_idle} to "
+            f"{count_after_reset}. If an idle optimization disabled the "
+            "loop, the re-enable path may be broken."
+        )
+        await call_and_expect_bits("clear_warning_a", warning=0.0, error=0.0)

tests/unit_tests/conftest.py

@@ -84,9 +84,9 @@ def mock_decode_pc() -> Generator[Mock, None, None]:
 
 
 @pytest.fixture
-def mock_run_external_command() -> Generator[Mock, None, None]:
-    """Mock run_external_command for platformio_api."""
-    with patch("esphome.platformio_api.run_external_command") as mock:
+def mock_run_external_process() -> Generator[Mock, None, None]:
+    """Mock run_external_process for platformio_api."""
+    with patch("esphome.platformio_api.run_external_process") as mock:
         yield mock
 
 

tests/unit_tests/fixtures/substitutions/13-packages_as_included_list.approved.yaml

@@ -0,0 +1,3 @@
+wifi:
+  password: pkg_password
+  ssid: main_ssid

tests/unit_tests/fixtures/substitutions/13-packages_as_included_list.input.yaml

@@ -0,0 +1,4 @@
+packages: !include 13-packages_list.yaml
+
+wifi:
+  ssid: main_ssid

tests/unit_tests/fixtures/substitutions/13-packages_list.yaml

@@ -0,0 +1,2 @@
+- wifi:
+    password: pkg_password

tests/unit_tests/fixtures/substitutions/14-packages_as_included_dict.approved.yaml

@@ -0,0 +1,3 @@
+wifi:
+  password: pkg_password
+  ssid: main_ssid

tests/unit_tests/fixtures/substitutions/14-packages_as_included_dict.input.yaml

@@ -0,0 +1,4 @@
+packages: !include 14-packages_dict.yaml
+
+wifi:
+  ssid: main_ssid

tests/unit_tests/fixtures/substitutions/14-packages_dict.yaml

@@ -0,0 +1,3 @@
+network:
+  wifi:
+    password: pkg_password

tests/unit_tests/test_main.py

@@ -1231,6 +1231,48 @@ def test_upload_using_esptool_path_conversion(
     assert partitions_path.endswith("partitions.bin")
 
 
+def test_upload_using_esptool_skips_missing_extra_flash_images(
+    tmp_path: Path,
+    mock_run_external_command_main: Mock,
+    mock_get_idedata: Mock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A non-existent path in extra_flash_images must be filtered out with a
+    warning, and must not appear in the esptool command line. Only the valid
+    images are flashed. Regression test for
+    https://github.com/esphome/esphome/issues/15634.
+    """
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path, name="test")
+    CORE.data[KEY_ESP32] = {KEY_VARIANT: VARIANT_ESP32}
+
+    missing_path = tmp_path / "variants" / "tasmota" / "tinyuf2.bin"
+
+    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
+    mock_idedata.extra_flash_images = [
+        platformio_api.FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
+        platformio_api.FlashImage(path=missing_path, offset="0x2d0000"),
+    ]
+    mock_get_idedata.return_value = mock_idedata
+
+    (tmp_path / "firmware.bin").touch()
+    (tmp_path / "bootloader.bin").touch()
+    # Intentionally do NOT create missing_path
+
+    config = {CONF_ESPHOME: {"platformio_options": {}}}
+
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        result = upload_using_esptool(config, "/dev/ttyUSB0", None, None)
+
+    assert result == 0
+    assert "Skipping missing flash image" in caplog.text
+    assert str(missing_path) in caplog.text
+
+    cmd_list = list(mock_run_external_command_main.call_args[0][1:])
+    assert str(missing_path) not in cmd_list
+    assert "0x2d0000" not in cmd_list
+
+
 def test_upload_using_esptool_with_file_path(
     tmp_path: Path,
     mock_run_external_command_main: Mock,

tests/unit_tests/test_platformio_api.py

@@ -1,7 +1,8 @@
 """Tests for platformio_api.py path functions."""
 
+# pylint: disable=protected-access
+
 import json
-import logging
 import os
 from pathlib import Path
 import shutil
@@ -10,7 +11,7 @@ from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
 
-from esphome import platformio_api
+from esphome import platformio_api, platformio_runner
 from esphome.core import CORE, EsphomeError
 
 
@@ -281,13 +282,13 @@ def test_run_idedata_raises_on_invalid_json(
 
 
 def test_run_platformio_cli_sets_environment_variables(
-    setup_core: Path, mock_run_external_command: Mock
+    setup_core: Path, mock_run_external_process: Mock
 ) -> None:
     """Test run_platformio_cli sets correct environment variables."""
     CORE.build_path = str(setup_core / "build" / "test")
 
     with patch.dict(os.environ, {}, clear=False):
-        mock_run_external_command.return_value = 0
+        mock_run_external_process.return_value = 0
         platformio_api.run_platformio_cli("test", "arg")
 
         # Check environment variables were set
@@ -300,10 +301,12 @@ def test_run_platformio_cli_sets_environment_variables(
         assert "PLATFORMIO_LIBDEPS_DIR" in os.environ
         assert "PYTHONWARNINGS" in os.environ
 
-        # Check command was called correctly
-        mock_run_external_command.assert_called_once()
-        args = mock_run_external_command.call_args[0]
-        assert "platformio" in args
+        # Check command was called correctly — runs PlatformIO as a subprocess
+        # via the esphome.platformio_runner entry point.
+        mock_run_external_process.assert_called_once()
+        args = mock_run_external_process.call_args[0]
+        assert "-m" in args
+        assert "esphome.platformio_runner" in args
         assert "test" in args
         assert "arg" in args
 
@@ -444,7 +447,7 @@ def test_patch_structhash(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Verify both modules had clean_build_dir patched
         # Check that clean_build_dir was set on both modules
@@ -496,7 +499,7 @@ def test_patched_clean_build_dir_removes_outdated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -546,7 +549,7 @@ def test_patched_clean_build_dir_keeps_updated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -594,7 +597,7 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -719,7 +722,7 @@ def test_patch_file_downloader_succeeds_first_try() -> None:
             ),
         },
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -758,7 +761,7 @@ def test_patch_file_downloader_retries_on_failure() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -799,7 +802,7 @@ def test_patch_file_downloader_raises_after_max_retries() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -847,7 +850,7 @@ def test_patch_file_downloader_closes_session_and_response_between_retries() ->
         ),
         patch("time.sleep"),
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -882,9 +885,9 @@ def test_patch_file_downloader_idempotent() -> None:
         },
     ):
         # Patch multiple times
-        platformio_api.patch_file_downloader()
-        platformio_api.patch_file_downloader()
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -895,19 +898,18 @@ def test_patch_file_downloader_idempotent() -> None:
         assert call_count == 1
 
 
-def test_platformio_log_filter_allows_non_platformio_messages() -> None:
-    """Test that non-platformio logger messages are allowed through."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="esphome.core",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg="Some esphome message",
-        args=(),
-        exc_info=None,
+def _filter_through_redirect(line: str) -> str:
+    """Write a line through RedirectText with FILTER_PLATFORMIO_LINES and return what passes."""
+    import io
+
+    from esphome.util import RedirectText
+
+    captured = io.StringIO()
+    redirect = RedirectText(
+        captured, filter_lines=platformio_runner.FILTER_PLATFORMIO_LINES
     )
-    assert log_filter.filter(record) is True
+    redirect.write(line + "\n")
+    return captured.getvalue()
 
 
 @pytest.mark.parametrize(
@@ -930,19 +932,9 @@ def test_platformio_log_filter_allows_non_platformio_messages() -> None:
         "Memory Usage -> https://bit.ly/pio-memory-usage",
     ],
 )
-def test_platformio_log_filter_blocks_noisy_messages(msg: str) -> None:
-    """Test that noisy platformio messages are filtered out."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="platformio.builder",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg=msg,
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is False
+def test_filter_platformio_lines_blocks_noisy_messages(msg: str) -> None:
+    """Test that noisy platformio output lines are filtered out by RedirectText."""
+    assert _filter_through_redirect(msg) == ""
 
 
 @pytest.mark.parametrize(
@@ -954,39 +946,6 @@ def test_platformio_log_filter_blocks_noisy_messages(msg: str) -> None:
         "warning: unused variable",
     ],
 )
-def test_platformio_log_filter_allows_other_platformio_messages(msg: str) -> None:
-    """Test that non-noisy platformio messages are allowed through."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="platformio.builder",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg=msg,
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is True
-
-
-@pytest.mark.parametrize(
-    "logger_name",
-    [
-        "PLATFORMIO.builder",
-        "PlatformIO.core",
-        "platformio.run",
-    ],
-)
-def test_platformio_log_filter_case_insensitive_logger_name(logger_name: str) -> None:
-    """Test that platformio logger name matching is case insensitive."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name=logger_name,
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg="Found 5 compatible libraries",
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is False
+def test_filter_platformio_lines_allows_other_messages(msg: str) -> None:
+    """Test that non-noisy platformio output lines pass through RedirectText."""
+    assert _filter_through_redirect(msg) == msg + "\n"