Merge remote-tracking branch 'upstream/dev' into app-loop-optimize-speed

# Conflicts:
#	esphome/core/application.h

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/ota/test_esphome_ota.py

@@ -0,0 +1,105 @@
+"""Tests for the esphome OTA platform final_validate logic."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.esphome.ota import ota_esphome_final_validate
+from esphome.const import (
+    CONF_ESPHOME,
+    CONF_ID,
+    CONF_OTA,
+    CONF_PASSWORD,
+    CONF_PLATFORM,
+    CONF_PORT,
+    CONF_VERSION,
+)
+from esphome.core import ID
+import esphome.final_validate as fv
+
+
+def _make_ota_config(port: int = 3232, **kwargs: Any) -> dict[str, Any]:
+    config: dict[str, Any] = {
+        CONF_PLATFORM: CONF_ESPHOME,
+        CONF_ID: ID(f"ota_esphome_{port}", is_manual=False),
+        CONF_VERSION: 2,
+        CONF_PORT: port,
+    }
+    config.update(kwargs)
+    return config
+
+
+def test_single_esphome_ota_instance_accepted() -> None:
+    """A single ESPHome OTA config passes final_validate untouched."""
+    full_conf = {CONF_OTA: [_make_ota_config(port=3232)]}
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_PORT] == 3232
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_same_port_configs_merge(caplog: pytest.LogCaptureFixture) -> None:
+    """Two ESPHome OTA configs on the same port merge into one instance."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_PASSWORD: "pw"}),
+            _make_ota_config(port=3232),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_PORT] == 3232
+        assert any("Found and merged" in record.message for record in caplog.records), (
+            "Expected merge warning not found in log"
+        )
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_multiple_ports_rejected() -> None:
+    """Two ESPHome OTA configs on different ports raise cv.Invalid."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            _make_ota_config(port=3233),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(
+            cv.Invalid,
+            match=r"Only a single port is supported for 'ota' 'platform: esphome'",
+        ):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_non_esphome_ota_unaffected() -> None:
+    """Non-esphome OTA platforms are not subject to the single-instance rule."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+            {CONF_PLATFORM: "http_request", CONF_ID: ID("ota_hr", is_manual=False)},
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 3
+    finally:
+        fv.full_config.reset(token)

tests/components/core/test_helpers.cpp

@@ -0,0 +1,120 @@
+#include <gtest/gtest.h>
+#include <cstring>
+
+#include "esphome/core/helpers.h"
+
+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');
+}
+
+}  // 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/integration/fixtures/addressable_light_transition.yaml

@@ -0,0 +1,29 @@
+esphome:
+  name: addr-light-transition
+host:
+api:
+logger:
+  level: DEBUG
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+light:
+  - platform: mock_addressable_light
+    output_id: strip_output
+    id: strip
+    name: "Test Strip"
+    num_leds: 4
+    gamma_correct: 2.8
+    default_transition_length: 0s
+
+sensor:
+  - platform: template
+    name: "led0_red_raw"
+    id: led0_red_raw
+    update_interval: 10ms
+    accuracy_decimals: 0
+    lambda: |-
+      return (float) id(strip_output).get_raw_red(0);

tests/integration/fixtures/external_components/mock_addressable_light/__init__.py

@@ -0,0 +1 @@
+CODEOWNERS = ["@esphome/tests"]

tests/integration/fixtures/external_components/mock_addressable_light/light.py

@@ -0,0 +1,23 @@
+import esphome.codegen as cg
+from esphome.components import light
+import esphome.config_validation as cv
+from esphome.const import CONF_NUM_LEDS, CONF_OUTPUT_ID
+from esphome.types import ConfigType
+
+mock_addressable_light_ns = cg.esphome_ns.namespace("mock_addressable_light")
+MockAddressableLight = mock_addressable_light_ns.class_(
+    "MockAddressableLight", light.AddressableLight
+)
+
+CONFIG_SCHEMA = light.ADDRESSABLE_LIGHT_SCHEMA.extend(
+    {
+        cv.GenerateID(CONF_OUTPUT_ID): cv.declare_id(MockAddressableLight),
+        cv.Optional(CONF_NUM_LEDS, default=4): cv.positive_not_null_int,
+    }
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_OUTPUT_ID], config[CONF_NUM_LEDS])
+    await light.register_light(var, config)
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/mock_addressable_light/mock_addressable_light.h

@@ -0,0 +1,52 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+
+#include "esphome/components/light/addressable_light.h"
+#include "esphome/core/component.h"
+
+namespace esphome::mock_addressable_light {
+
+// In-memory addressable light for host-mode integration tests. Exposes the raw
+// per-LED byte buffer (post-gamma-correction, as the hardware would see it)
+// so tests can observe transition behavior without real hardware.
+class MockAddressableLight : public light::AddressableLight {
+ public:
+  explicit MockAddressableLight(uint16_t num_leds)
+      : num_leds_(num_leds), buf_(new uint8_t[num_leds * 4]()), effect_data_(new uint8_t[num_leds]()) {}
+
+  void setup() override {}
+  void write_state(light::LightState *state) override {}
+  int32_t size() const override { return this->num_leds_; }
+  void clear_effect_data() override {
+    for (uint16_t i = 0; i < this->num_leds_; i++)
+      this->effect_data_[i] = 0;
+  }
+  light::LightTraits get_traits() override {
+    auto traits = light::LightTraits();
+    traits.set_supported_color_modes({light::ColorMode::RGB});
+    return traits;
+  }
+
+  // Accessors for tests: return the raw stored byte (post gamma correction),
+  // which is what actual LED hardware would receive.
+  uint8_t get_raw_red(uint16_t index) const { return this->buf_[index * 4 + 0]; }
+  uint8_t get_raw_green(uint16_t index) const { return this->buf_[index * 4 + 1]; }
+  uint8_t get_raw_blue(uint16_t index) const { return this->buf_[index * 4 + 2]; }
+  uint8_t get_raw_white(uint16_t index) const { return this->buf_[index * 4 + 3]; }
+
+ protected:
+  light::ESPColorView get_view_internal(int32_t index) const override {
+    size_t pos = index * 4;
+    return {this->buf_.get() + pos + 0, this->buf_.get() + pos + 1,       this->buf_.get() + pos + 2,
+            this->buf_.get() + pos + 3, this->effect_data_.get() + index, &this->correction_};
+  }
+
+  uint16_t num_leds_;
+  std::unique_ptr<uint8_t[]> buf_;
+  std::unique_ptr<uint8_t[]> effect_data_;
+};
+
+}  // namespace esphome::mock_addressable_light

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_addressable_light_transition.py

@@ -0,0 +1,119 @@
+"""Integration test for addressable light transitions with gamma correction.
+
+Regression test for a bug where a long turn-on transition on an addressable
+light with gamma correction (e.g. gamma_correct: 2.8) produced no visible
+output for ~90% of the transition duration, then jumped to the target in the
+final ~10%. Root cause: the transition algorithm read each LED's current value
+back through the 8-bit stored byte every step; at gamma 2.8 any pre-gamma value
+below ~27 rounds to stored byte 0, so the stored byte stalled at 0 until
+progress was high enough for a single step to produce a large-enough pre-gamma
+value to clear the gamma threshold.
+
+The fix interpolates against a cached start color when all LEDs started at the
+same value (the common case for plain turn_on/turn_off), avoiding the round-trip.
+
+This test uses a host-only mock addressable light that exposes the raw stored
+byte of each LED, so we can observe the transition directly.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import LightInfo, SensorInfo, SensorState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_addressable_light_transition(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """With gamma 2.8, the stored raw byte must rise visibly well before the end."""
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        entities, _ = await client.list_entities_services()
+        light = require_entity(entities, "test_strip", LightInfo)
+        sensor = require_entity(entities, "led0_red_raw", SensorInfo)
+
+        # Track the raw-byte sensor. It polls every 10ms in the fixture, and
+        # ESPHome sensors publish on every change, so we collect a time series.
+        # Samples are stored as absolute (loop_time, value); we rebase to the
+        # command-issue time after the run so pre-command samples are strictly
+        # negative and reliably excluded.
+        loop = asyncio.get_running_loop()
+        samples: list[tuple[float, float]] = []
+
+        def on_state(state: object) -> None:
+            if not isinstance(state, SensorState) or state.key != sensor.key:
+                return
+            samples.append((loop.time(), state.state))
+
+        # InitialStateHelper swallows the first state ESPHome sends per entity
+        # on subscribe, so on_state only sees real post-subscribe updates.
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        # Start transition: off -> full white over 1 second. This is the
+        # scenario from the bug report, compressed in time.
+        transition_s = 1.0
+        command_time = loop.time()
+        client.light_command(
+            key=light.key,
+            state=True,
+            rgb=(1.0, 1.0, 1.0),
+            brightness=1.0,
+            transition_length=transition_s,
+        )
+
+        # Let the full transition run, plus margin for the final sample.
+        await asyncio.sleep(transition_s + 0.2)
+
+        # Rebase to command-issue time. Pre-command samples have t < 0 and are
+        # excluded; everything else is in seconds since the command was issued.
+        post_command = [
+            (t - command_time, v) for (t, v) in samples if t >= command_time
+        ]
+        assert post_command, "no sensor samples received after command was issued"
+
+        # Assertion 1: the transition is not stalled. With the bug, the raw
+        # byte stays at 0 until ~90% of the transition duration. With the fix,
+        # it becomes nonzero in the first ~30% (for gamma 2.8, pre-gamma 76
+        # clears the gamma threshold at progress ~0.30). Require the first
+        # nonzero sample to land well before 50% of the transition duration,
+        # measured from the command-issue time. The 50% bound (rather than
+        # 70%) leaves headroom for assertion 2's mid-window check.
+        first_nonzero = next(((t, v) for (t, v) in post_command if v > 0), None)
+        assert first_nonzero is not None, (
+            "raw byte never rose above 0 during the transition — the fade stalled"
+        )
+        assert first_nonzero[0] < transition_s * 0.5, (
+            f"raw byte only rose above 0 at t={first_nonzero[0]:.3f}s "
+            f"(>{transition_s * 0.5:.3f}s after command) — transition is stalling"
+        )
+
+        # Assertion 2: by mid-late transition, the raw byte should have reached
+        # a substantial fraction of its final value. Bound the window to
+        # [50%, 90%] of the transition so the post-transition settled value
+        # (which always reaches 255) can't satisfy this assertion — that would
+        # let "stays at 0 then jumps at 99%" regressions slip through.
+        mid_window = [
+            v
+            for (t, v) in post_command
+            if transition_s * 0.5 <= t <= transition_s * 0.9
+        ]
+        assert mid_window, "no samples captured in mid-transition window"
+        assert max(mid_window) >= 100, (
+            f"raw byte peaked at only {max(mid_window)} between 50%–90% of "
+            "transition (expected >= 100 for white target at gamma 2.8)"
+        )
+
+        # Assertion 3: final value reaches target. Gamma 2.8 of 255 is 255.
+        final_samples = [v for (_, v) in post_command[-5:]]
+        assert max(final_samples) >= 250, (
+            f"final raw byte was {max(final_samples)}, expected >= 250"
+        )

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)