Merge remote-tracking branch 'origin/dev' into HEAD

tests/integration/fixtures/api_homeassistant_entities.yaml

@@ -0,0 +1,62 @@
+esphome:
+  name: ha-entities
+
+host:
+
+api:
+  actions:
+    - action: dump_select
+      variables:
+        tag: string
+      then:
+        - logger.log:
+            format: "dump %s: has_state=%d index=%d option='%s'"
+            args:
+              - tag.c_str()
+              - id(ha_select).has_state()
+              - (int) id(ha_select).active_index().value_or(-1)
+              - id(ha_select).current_option().c_str()
+
+logger:
+  level: DEBUG
+
+select:
+  - platform: homeassistant
+    name: HA Select
+    id: ha_select
+    entity_id: select.mode
+    internal: false
+    max_options: 4
+    options_buffer_size: 40
+    on_value:
+      - logger.log:
+          format: "select value '%s' at %u"
+          args: [x.c_str(), (unsigned) i]
+  - platform: homeassistant
+    name: HA Input Select
+    entity_id: input_select.colour
+    internal: false
+
+text:
+  - platform: homeassistant
+    name: HA Text
+    entity_id: text.note
+    internal: false
+    on_value:
+      - logger.log:
+          format: "text value '%s'"
+          args: [x.c_str()]
+  - platform: homeassistant
+    name: HA Input Text
+    entity_id: input_text.name
+    internal: false
+
+button:
+  - platform: homeassistant
+    name: HA Button
+    entity_id: button.doorbell
+    internal: false
+  - platform: homeassistant
+    name: HA Input Button
+    entity_id: input_button.reset
+    internal: false

tests/integration/fixtures/api_outgoing_connection.yaml

@@ -0,0 +1,14 @@
+esphome:
+  name: outgoing-conn-test
+
+host:
+
+logger:
+
+api:
+  encryption:
+    key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
+  outgoing_connection:
+    host: 127.0.0.1
+    port: OUTGOING_PORT
+    delay: 1s

tests/integration/fixtures/api_outgoing_connection_remembered.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: outgoing-conn-test
+
+host:
+
+logger:
+
+api:
+  encryption:
+    key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
+  outgoing_connection:
+    port: OUTGOING_PORT
+    delay: 1s

tests/integration/fixtures/api_string_lambda.yaml

@@ -37,7 +37,7 @@ api:
             args: [input_number]
 
         # Test that int lambdas still work correctly with to_string
-        # The TemplatableStringValue should automatically convert int to string
+        # Home Assistant action lambdas convert an int result to a string
         - homeassistant.event:
             event: esphome.test_int_lambda
             data:
@@ -54,7 +54,7 @@ api:
             args: [input_float]
 
         # Test that float lambdas still work correctly with to_string
-        # The TemplatableStringValue should automatically convert float to string
+        # Home Assistant action lambdas convert a float result to a string
         - homeassistant.event:
             event: esphome.test_float_lambda
             data:

tests/integration/fixtures/branch_resume_async.yaml

@@ -0,0 +1,261 @@
+esphome:
+  name: test-branch-resume-async
+
+host:
+
+logger:
+  level: DEBUG
+
+script:
+  - id: delay_in_if
+    then:
+      - if:
+          condition:
+            lambda: 'return true;'
+          then:
+            - delay: 50ms
+            - logger.log: "if-branch-after-delay"
+          else:
+            - logger.log: "if-else-should-not-run"
+      - logger.log: "after-if"
+
+  - id: delay_in_else
+    then:
+      - if:
+          condition:
+            lambda: 'return false;'
+          then:
+            - logger.log: "if-then-should-not-run"
+          else:
+            - delay: 50ms
+            - logger.log: "else-branch-after-delay"
+      - logger.log: "after-else"
+
+  - id: delay_in_while
+    then:
+      - lambda: 'id(counter) = 0;'
+      - while:
+          condition:
+            lambda: 'return id(counter) < 3;'
+          then:
+            - delay: 20ms
+            - lambda: 'id(counter) += 1;'
+            - logger.log:
+                format: "while-body-%d"
+                args: ['id(counter)']
+      - logger.log: "after-while"
+
+  - id: stop_in_branch
+    then:
+      - if:
+          condition:
+            lambda: 'return true;'
+          then:
+            - delay: 500ms
+            - logger.log: "stopped-branch-should-not-run"
+      - logger.log: "stopped-after-if-should-not-run"
+
+  - id: parallel_branch
+    mode: parallel
+    parameters:
+      run: int
+    then:
+      - if:
+          condition:
+            lambda: 'return run >= 0;'
+          then:
+            - delay: 100ms
+            - logger.log:
+                format: "parallel-branch-done-%d"
+                args: ['run']
+      - logger.log:
+          format: "parallel-after-if-%d"
+          args: ['run']
+
+  - id: nested_if_last_in_while
+    then:
+      - lambda: 'id(counter) = 0;'
+      - while:
+          condition:
+            lambda: 'return id(counter) < 2;'
+          then:
+            - lambda: 'id(counter) += 1;'
+            - if:
+                condition:
+                  lambda: 'return true;'
+                then:
+                  - delay: 20ms
+                  - logger.log:
+                      format: "nested-if-in-while-%d"
+                      args: ['id(counter)']
+      - logger.log: "after-nested-while"
+
+  - id: stop_mid_while
+    then:
+      - lambda: 'id(counter) = 0;'
+      - while:
+          condition:
+            lambda: 'return id(counter) < 5;'
+          then:
+            - delay: 200ms
+            - lambda: 'id(counter) += 1;'
+            - logger.log:
+                format: "stop-while-body-%d"
+                args: ['id(counter)']
+      - logger.log: "stop-while-done"
+
+  - id: queued_branch
+    mode: queued
+    parameters:
+      run: int
+    then:
+      - if:
+          condition:
+            lambda: 'return true;'
+          then:
+            - delay: 30ms
+            - logger.log:
+                format: "queued-branch-%d"
+                args: ['run']
+      - logger.log:
+          format: "queued-after-if-%d"
+          args: ['run']
+
+  - id: restart_branch
+    mode: restart
+    parameters:
+      run: int
+    then:
+      - if:
+          condition:
+            lambda: 'return true;'
+          then:
+            - delay: 100ms
+            - logger.log:
+                format: "restart-branch-%d"
+                args: ['run']
+      - logger.log:
+          format: "restart-after-if-%d"
+          args: ['run']
+
+  - id: delay_last_in_branch
+    then:
+      - if:
+          condition:
+            lambda: 'return true;'
+          then:
+            - logger.log: "delay-last-start"
+            - delay: 30ms
+      - logger.log: "after-delay-last"
+
+  - id: stop_in_if_condition
+    then:
+      - if:
+          condition:
+            lambda: |-
+              id(stop_in_if_condition).stop();
+              return true;
+          then:
+            - logger.log: "if-stopped-by-condition-should-not-run"
+          else:
+            - logger.log: "if-else-stopped-by-condition-should-not-run"
+      - logger.log: "after-if-stopped-by-condition-should-not-run"
+  - id: stop_in_if_else_condition
+    then:
+      - if:
+          condition:
+            lambda: |-
+              id(stop_in_if_else_condition).stop();
+              return false;
+          else:
+            - logger.log: "else-stopped-by-condition-should-not-run"
+      - logger.log: "after-else-stopped-by-condition-should-not-run"
+  - id: stop_in_while_condition
+    then:
+      - while:
+          condition:
+            lambda: |-
+              id(stop_in_while_condition).stop();
+              return true;
+          then:
+            - logger.log: "while-stopped-by-condition-should-not-run"
+      - logger.log: "after-while-stopped-by-condition-should-not-run"
+
+globals:
+  - id: counter
+    type: int
+    initial_value: '0'
+
+api:
+  actions:
+    - action: run_delay_in_if
+      then:
+        - script.execute: delay_in_if
+    - action: run_delay_in_else
+      then:
+        - script.execute: delay_in_else
+    - action: run_delay_in_while
+      then:
+        - script.execute: delay_in_while
+    - action: run_stop_in_branch
+      then:
+        - script.execute: stop_in_branch
+        - delay: 100ms
+        - script.stop: stop_in_branch
+        - lambda: |-
+            ESP_LOGI("test", "stop-running=%d", id(stop_in_branch).is_running() ? 1 : 0);
+        - delay: 600ms
+        - logger.log: "stop-check-done"
+    - action: run_nested_if_last_in_while
+      then:
+        - script.execute: nested_if_last_in_while
+    - action: run_stop_mid_while
+      then:
+        - script.execute: stop_mid_while
+        - delay: 300ms
+        - script.stop: stop_mid_while
+        - lambda: |-
+            ESP_LOGI("test", "stop-while-running=%d", id(stop_mid_while).is_running() ? 1 : 0);
+        - delay: 400ms
+        - script.execute: stop_mid_while
+    - action: run_queued_branch
+      then:
+        - script.execute:
+            id: queued_branch
+            run: 1
+        - script.execute:
+            id: queued_branch
+            run: 2
+        - script.execute:
+            id: queued_branch
+            run: 3
+    - action: run_restart_branch
+      then:
+        - script.execute:
+            id: restart_branch
+            run: 1
+        - delay: 30ms
+        - script.execute:
+            id: restart_branch
+            run: 2
+    - action: run_delay_last_in_branch
+      then:
+        - script.execute: delay_last_in_branch
+    - action: run_stop_in_conditions
+      then:
+        - script.execute: stop_in_if_condition
+        - script.execute: stop_in_if_else_condition
+        - script.execute: stop_in_while_condition
+        - delay: 200ms
+        - logger.log: "stop-in-conditions-done"
+    - action: run_parallel_branch
+      then:
+        - script.execute:
+            id: parallel_branch
+            run: 1
+        - script.execute:
+            id: parallel_branch
+            run: 2
+        - script.execute:
+            id: parallel_branch
+            run: 3

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

@@ -0,0 +1,25 @@
+"""Host-only stand-in for remote_transmitter used by the IR/RF integration tests."""
+
+import esphome.codegen as cg
+from esphome.components import remote_base
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@esphome/tests"]
+MULTI_CONF = True
+AUTO_LOAD = ["remote_base"]
+
+remote_transmitter_mock_ns = cg.esphome_ns.namespace("remote_transmitter_mock")
+MockRemoteTransmitter = remote_transmitter_mock_ns.class_(
+    "MockRemoteTransmitter", remote_base.RemoteTransmitterBase, cg.Component
+)
+
+CONFIG_SCHEMA = cv.Schema(
+    {cv.GenerateID(): cv.declare_id(MockRemoteTransmitter)}
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/remote_transmitter_mock/remote_transmitter_mock.cpp

@@ -0,0 +1,40 @@
+#include "remote_transmitter_mock.h"
+#include "esphome/core/log.h"
+
+#include <cinttypes>
+
+namespace esphome::remote_transmitter_mock {
+
+static const char *const TAG = "remote_transmitter_mock";
+
+void MockRemoteTransmitter::dump_config() { ESP_LOGCONFIG(TAG, "Mock Remote Transmitter"); }
+
+void MockRemoteTransmitter::flush_pending_completion() {
+  if (!this->busy_)
+    return;
+  // The RMT backend blocks here until the hardware is idle; the mock only reports the overlap
+  ESP_LOGW(TAG, "Overlap: a frame was requested while seq=%" PRIu16 " was in flight", this->current_seq_);
+  this->cancel_timeout("complete");
+  this->finish_();
+}
+
+void MockRemoteTransmitter::send_internal(uint32_t send_times, uint32_t send_wait) {
+  uint64_t total_us = static_cast<uint64_t>(send_wait) * (send_times - 1);
+  for (int32_t value : this->temp_.get_data()) {
+    total_us += static_cast<uint64_t>(value < 0 ? -value : value) * send_times;
+  }
+  const auto duration_ms = static_cast<uint32_t>(total_us / 1000);
+
+  this->busy_ = true;
+  ESP_LOGI(TAG, "TX seq=%" PRIu16 " timings=%zu repeat=%" PRIu32 " duration=%" PRIu32 "ms", this->current_seq_,
+           this->temp_.get_data().size(), send_times, duration_ms);
+  this->set_timeout("complete", duration_ms, [this]() { this->finish_(); });
+}
+
+void MockRemoteTransmitter::finish_() {
+  this->busy_ = false;
+  ESP_LOGI(TAG, "Complete seq=%" PRIu16, this->current_seq_);
+  this->notify_complete_(true);
+}
+
+}  // namespace esphome::remote_transmitter_mock

tests/integration/fixtures/external_components/remote_transmitter_mock/remote_transmitter_mock.h

@@ -0,0 +1,31 @@
+#pragma once
+
+// ============================================================================
+// HOST-ONLY TEST COMPONENT — DO NOT COPY TO PRODUCTION CODE
+//
+// Emulates a non-blocking remote transmitter: a frame "occupies" the
+// transmitter for its real duration and the completion callback fires from a
+// scheduler timeout, the same way the ESP32 RMT backend reports completion.
+// A frame submitted while another is in flight is logged as an overlap; the
+// real backends block the main loop in that case.
+// ============================================================================
+
+#include "esphome/core/component.h"
+#include "esphome/components/remote_base/remote_base.h"
+
+namespace esphome::remote_transmitter_mock {
+
+class MockRemoteTransmitter : public remote_base::RemoteTransmitterBase, public Component {
+ public:
+  MockRemoteTransmitter() : remote_base::RemoteTransmitterBase(nullptr) {}
+  void dump_config() override;
+
+ protected:
+  void flush_pending_completion() override;
+  void send_internal(uint32_t send_times, uint32_t send_wait) override;
+  void finish_();
+
+  bool busy_{false};
+};
+
+}  // namespace esphome::remote_transmitter_mock

tests/integration/fixtures/external_components/wake_test_component/wake_test_component.cpp

@@ -16,4 +16,16 @@ void WakeTestComponent::start_async_wake() {
   }).detach();
 }
 
+void WakeTestComponent::start_async_timeout(uint32_t delay_ms) {
+  const uint32_t start_time = millis();
+  std::thread([this, start_time, delay_ms] {
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    const int start_loop_count = this->get_loop_count();
+    this->set_timeout(delay_ms, [this, start_time, start_loop_count, delay_ms] {
+      ESP_LOGI(TAG, "SCHEDULER_WAKE_RESULT delay=%u elapsed=%u loop_delta=%d", static_cast<unsigned int>(delay_ms),
+               static_cast<unsigned int>(millis() - start_time), this->get_loop_count() - start_loop_count);
+    });
+  }).detach();
+}
+
 }  // namespace esphome::wake_test_component

tests/integration/fixtures/external_components/wake_test_component/wake_test_component.h

@@ -18,6 +18,10 @@ class WakeTestComponent : public Component {
   // loop_interval_ has been raised high enough to gate it off otherwise.
   void start_async_wake();
 
+  // Spawn a detached thread that inserts a timeout (or a defer when delay_ms
+  // is 0) while the main loop sleeps, and report how soon it ran.
+  void start_async_timeout(uint32_t delay_ms);
+
   float get_setup_priority() const override { return setup_priority::DATA; }
 
  protected:

tests/integration/fixtures/ir_rf_transmit_complete.yaml

@@ -0,0 +1,30 @@
+esphome:
+  name: ir-rf-transmit-complete-test
+
+host:
+api:
+logger:
+  level: DEBUG
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+remote_transmitter_mock:
+  - id: rf_tx
+  - id: ir_tx
+
+# two entities on one transmitter: each must get only its own completion
+radio_frequency:
+  - platform: ir_rf_proxy
+    name: RF Transmitter A
+    remote_transmitter_id: rf_tx
+  - platform: ir_rf_proxy
+    name: RF Transmitter B
+    remote_transmitter_id: rf_tx
+
+infrared:
+  - platform: ir_rf_proxy
+    name: IR Transmitter
+    remote_transmitter_id: ir_tx

tests/integration/fixtures/scheduler_background_wake.yaml

@@ -0,0 +1,31 @@
+esphome:
+  name: scheduler-background-wake
+  on_boot:
+    priority: -100
+    then:
+      - lambda: |-
+          App.set_loop_interval(5000);
+
+host:
+api:
+logger:
+  level: INFO
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [wake_test_component]
+
+wake_test_component:
+  id: wake_test
+
+button:
+  - platform: template
+    name: Start Scheduler Timeout
+    on_press:
+      - lambda: id(wake_test)->start_async_timeout(100);
+  - platform: template
+    name: Start Scheduler Defer
+    on_press:
+      - lambda: id(wake_test)->start_async_timeout(0);

tests/integration/fixtures/sensor_timeout_throttle_filter.yaml

@@ -0,0 +1,67 @@
+esphome:
+  name: test-timeout-throttle
+
+host:
+api:
+  batch_delay: 0ms # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+sensor:
+  - platform: template
+    name: "Source"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # The default radar pair; codegen merges it into one TimeoutThrottleFilter
+  - platform: copy
+    source_id: source_sensor
+    name: "Merged"
+    filters:
+      - timeout:
+          timeout: 200ms
+          value: last
+      - throttle_with_priority: 200ms
+
+  # Same chain with an identity filter in between, which keeps the two filter objects
+  - platform: copy
+    source_id: source_sensor
+    name: "Reference"
+    filters:
+      - timeout:
+          timeout: 200ms
+          value: last
+      - multiply: 1.0
+      - throttle_with_priority: 200ms
+
+script:
+  - id: drive
+    then:
+      # Burst faster than the period: first passes, the rest are dropped, the timeout republishes the last
+      - sensor.template.publish: {id: source_sensor, state: 1.0}
+      - delay: 30ms
+      - sensor.template.publish: {id: source_sensor, state: 2.0}
+      - delay: 30ms
+      - sensor.template.publish: {id: source_sensor, state: 3.0}
+      - delay: 450ms
+      # NaN inside a burst always passes
+      - sensor.template.publish: {id: source_sensor, state: 4.0}
+      - delay: 30ms
+      - sensor.template.publish: {id: source_sensor, state: !lambda "return NAN;"}
+      - delay: 30ms
+      - sensor.template.publish: {id: source_sensor, state: 5.0}
+      - delay: 450ms
+      # Values right around the period
+      - sensor.template.publish: {id: source_sensor, state: 6.0}
+      - delay: 250ms
+      - sensor.template.publish: {id: source_sensor, state: 7.0}
+      - delay: 190ms
+      - sensor.template.publish: {id: source_sensor, state: 8.0}
+      - delay: 600ms
+      - sensor.template.publish: {id: source_sensor, state: 99.0}
+
+button:
+  - platform: template
+    name: "Drive"
+    on_press:
+      - script.execute: drive

tests/integration/fixtures/status_momentary.yaml

@@ -0,0 +1,40 @@
+esphome:
+  name: status-momentary-test
+
+host:
+api:
+  actions:
+    - action: momentary_warning
+      variables:
+        length: int
+      then:
+        - lambda: "id(probe)->status_momentary_warning(length);"
+    - action: momentary_error
+      variables:
+        length: int
+      then:
+        - lambda: "id(probe)->status_momentary_error(length);"
+    - action: momentary_warning_message
+      then:
+        - lambda: 'id(probe)->status_momentary_warning(LOG_STR("probe warning reason"), 300);'
+    - action: momentary_error_message
+      then:
+        - lambda: 'id(probe)->status_momentary_error(LOG_STR("probe error reason"), 300);'
+
+logger:
+
+sensor:
+  - platform: template
+    id: probe
+    update_interval: 24h
+    lambda: return 1.0;
+
+binary_sensor:
+  - platform: template
+    name: Probe Warning
+    id: probe_warning
+    lambda: return id(probe)->status_has_warning();
+  - platform: template
+    name: Probe Error
+    id: probe_error
+    lambda: return id(probe)->status_has_error();

tests/integration/fixtures/uart_tcp_bridge.yaml

@@ -12,6 +12,10 @@ uart:
   - id: uart_bus
     baud_rate: 115200
     port: PTY_PATH
+  # uart_tcp needs its own UART.
+  - id: denied_bus
+    baud_rate: 115200
+    port: DENIED_PTY_PATH
 
 uart_tcp:
   - id: bridge
@@ -23,7 +27,7 @@ uart_tcp:
     connected:
       name: Bridge Connected
   - id: denied_bridge
-    uart_id: uart_bus
+    uart_id: denied_bus
     role: server
     port: 18127
     allowed_ips:

tests/integration/test_api_homeassistant_entities.py

@@ -0,0 +1,213 @@
+"""Test the homeassistant select, text and button platforms against a fake Home Assistant."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+
+from aioesphomeapi import APIClient, HomeassistantServiceCall, UserService
+import pytest
+
+from .log_utils import LineWaiter
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+SUBSCRIPTIONS = {
+    ("select.mode", "options"),
+    ("select.mode", None),
+    ("input_select.colour", "options"),
+    ("input_select.colour", None),
+    ("text.note", None),
+    ("text.note", "min"),
+    ("text.note", "max"),
+    ("text.note", "mode"),
+    ("input_text.name", None),
+    ("input_text.name", "min"),
+    ("input_text.name", "max"),
+    ("input_text.name", "mode"),
+}
+
+# Each is rejected by the options parser, leaving the previous options in place
+MALFORMED_OPTIONS = (
+    "None",
+    "",
+    "[",
+    "['a'",
+    "['a',",
+    "['a' 'b']",
+    "['a'] x",
+    "[1]",
+    "['a\\q']",
+    "['a\\",
+    "['\\x4']",
+    "['\\x0']",
+    "['\\x00']",
+    "['\\U00110000']",
+)
+
+
+@pytest.mark.asyncio
+async def test_api_homeassistant_entities(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Options, state and actions of the homeassistant select, text and button platforms."""
+    loop = asyncio.get_running_loop()
+    waiter = LineWaiter()
+    subscribed: set[tuple[str, str | None]] = set()
+    all_subscribed = loop.create_future()
+    calls: asyncio.Queue[HomeassistantServiceCall] = asyncio.Queue()
+
+    def on_state_sub(entity_id: str, attribute: str | None) -> None:
+        subscribed.add((entity_id, attribute or None))
+        if not all_subscribed.done() and subscribed >= SUBSCRIPTIONS:
+            all_subscribed.set_result(None)
+
+    async def next_call() -> HomeassistantServiceCall:
+        return await asyncio.wait_for(calls.get(), timeout=5.0)
+
+    async with (
+        run_compiled(yaml_config, line_callback=waiter.callback),
+        api_client_connected() as client,
+    ):
+        entities, services = await client.list_entities_services()
+        keys = {entity.name: entity.key for entity in entities}
+        dump_select = next(s for s in services if s.name == "dump_select")
+        client.subscribe_service_calls(calls.put_nowait)
+        client.subscribe_home_assistant_states(on_state_sub)
+        try:
+            await asyncio.wait_for(all_subscribed, timeout=5.0)
+        except TimeoutError:
+            pytest.fail(f"never subscribed: {SUBSCRIPTIONS - subscribed}")
+
+        await _check_select(client, waiter, keys, dump_select, next_call)
+        await _check_text(client, waiter, keys, next_call)
+
+        client.button_command(keys["HA Button"])
+        call = await next_call()
+        assert call.service == "button.press"
+        assert call.data == {"entity_id": "button.doorbell"}
+        client.button_command(keys["HA Input Button"])
+        call = await next_call()
+        assert call.service == "input_button.press"
+        assert call.data == {"entity_id": "input_button.reset"}
+
+        assert calls.empty()
+
+
+async def _check_select(
+    client: APIClient,
+    waiter: LineWaiter,
+    keys: dict[str, int],
+    dump_select: UserService,
+    next_call: Callable[[], Awaitable[HomeassistantServiceCall]],
+) -> None:
+    async def dump(tag: str) -> str:
+        await client.execute_service(dump_select, {"tag": tag})
+        return await waiter.wait_for(f"dump {tag}:", timeout=5.0)
+
+    # Built with str() exactly as Home Assistant does, so every escape it emits is covered
+    options = ["Low", "It's", "café\t\x7f\\", "\u200b\U000e0001\r\n'\""]
+    client.send_home_assistant_state("select.mode", "options", str(options))
+    await waiter.wait_for("'select.mode': Got 4 options", timeout=5.0)
+
+    client.send_home_assistant_state("select.mode", "", "It's")
+    await waiter.wait_for("select value 'It's' at 1", timeout=5.0)
+    assert "has_state=1 index=1 option='It's'" in await dump("first")
+
+    # Decoded options round trip through select.select_option unchanged
+    for option in options[2:]:
+        client.select_command(keys["HA Select"], option)
+        call = await next_call()
+        assert call.service == "select.select_option"
+        assert call.data == {"entity_id": "select.mode", "option": option}
+
+    # The active option keeps its state when it moves; double quoted options parse too
+    client.send_home_assistant_state("select.mode", "options", "[\"It's\", 'L\\\"ow']")
+    await waiter.wait_for("'select.mode': Got 2 options", timeout=5.0)
+    assert "has_state=1 index=0 option='It's'" in await dump("moved")
+    client.select_command(keys["HA Select"], 'L"ow')
+    assert (await next_call()).data["option"] == 'L"ow'
+
+    # Dropping the active option clears the state
+    client.send_home_assistant_state(
+        "select.mode", "options", "[ 'Low' , 'Max' , 'High' ]"
+    )
+    await waiter.wait_for("'select.mode': Got 3 options", timeout=5.0)
+    assert "has_state=0" in await dump("dropped")
+
+    client.send_home_assistant_state("select.mode", "", "unavailable")
+    await waiter.wait_for("State 'unavailable' is not one of the options", timeout=5.0)
+
+    # Rejected lists keep the previous options
+    client.send_home_assistant_state("select.mode", "options", str(list("abcde")))
+    await waiter.wait_for("5 options exceed max_options (4)", timeout=5.0)
+    client.send_home_assistant_state("select.mode", "options", str(["x" * 50]))
+    await waiter.wait_for(
+        "Options need 51 bytes, more than options_buffer_size (40)", timeout=5.0
+    )
+    for malformed in MALFORMED_OPTIONS:
+        client.send_home_assistant_state("select.mode", "options", malformed)
+    client.send_home_assistant_state("select.mode", "", "Max")
+    await waiter.wait_for("select value 'Max' at 1", timeout=5.0)
+    assert sum("Can't parse options" in line for line in waiter.lines) == len(
+        MALFORMED_OPTIONS
+    )
+
+    # Repeating the active state does not publish again
+    client.send_home_assistant_state("select.mode", "", "Max")
+    client.send_home_assistant_state("select.mode", "options", "[]")
+    await waiter.wait_for("'select.mode': Got 0 options", timeout=5.0)
+    assert sum("select value 'Max'" in line for line in waiter.lines) == 1
+    assert "has_state=0 index=-1" in await dump("empty")
+
+    client.send_home_assistant_state(
+        "input_select.colour", "options", "['Red', 'Green']"
+    )
+    await waiter.wait_for("'input_select.colour': Got 2 options", timeout=5.0)
+    client.select_command(keys["HA Input Select"], "Green")
+    call = await next_call()
+    assert call.service == "input_select.select_option"
+    assert call.data == {"entity_id": "input_select.colour", "option": "Green"}
+
+
+async def _check_text(
+    client: APIClient,
+    waiter: LineWaiter,
+    keys: dict[str, int],
+    next_call: Callable[[], Awaitable[HomeassistantServiceCall]],
+) -> None:
+    client.send_home_assistant_state("text.note", "min", "2")
+    client.send_home_assistant_state("text.note", "max", "10")
+    client.send_home_assistant_state("text.note", "mode", "password")
+    await waiter.wait_for("'text.note': Mode retrieved: password", timeout=5.0)
+
+    client.send_home_assistant_state("input_text.name", "min", "abc")
+    client.send_home_assistant_state("input_text.name", "max", "x")
+    client.send_home_assistant_state("input_text.name", "mode", "fancy")
+    await waiter.wait_for(
+        "'input_text.name': Unknown 'mode' value 'fancy'", timeout=5.0
+    )
+    await waiter.wait_for("Can't convert 'min' value 'abc'", timeout=5.0)
+    await waiter.wait_for("Can't convert 'max' value 'x'", timeout=5.0)
+    client.send_home_assistant_state("input_text.name", "mode", "text")
+    await waiter.wait_for("'input_text.name': Mode retrieved: text", timeout=5.0)
+
+    client.send_home_assistant_state("text.note", "", "hello")
+    client.send_home_assistant_state("text.note", "", "hello")
+    client.send_home_assistant_state("text.note", "", "there")
+    await waiter.wait_for("text value 'there'", timeout=5.0)
+    assert sum("text value 'hello'" in line for line in waiter.lines) == 1
+
+    # The limits from Home Assistant apply: too short and too long are refused
+    client.text_command(keys["HA Text"], "a")
+    client.text_command(keys["HA Text"], "abcdefghijk")
+    client.text_command(keys["HA Text"], "fine")
+    call = await next_call()
+    assert call.service == "text.set_value"
+    assert call.data == {"entity_id": "text.note", "value": "fine"}
+
+    client.text_command(keys["HA Input Text"], "Bob")
+    call = await next_call()
+    assert call.service == "input_text.set_value"
+    assert call.data == {"entity_id": "input_text.name", "value": "Bob"}

tests/integration/test_api_outgoing_connection.py

@@ -0,0 +1,175 @@
+"""Integration tests for the api outgoing_connection option.
+
+The device dials out to the test's listener when no dial-back target client is
+connected. The listener plays the Home Assistant side over the accepted socket
+using aioesphomeapi's sans-IO Noise handshake: the device sends its server
+hello first so the listener could pick the right key, and the NNpsk0 handshake
+then verifies both sides. Protocol roles stay unchanged, so the client speaks
+exactly the same frames as over a normal connection. A client becomes the
+remembered dial-back target by setting the outgoing_connection_target flag in
+its hello.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import socket
+from typing import Any
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from .raw_api_client import MESSAGE_TYPE_OF
+from .types import RunCompiledFunction
+
+KEY = "bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU="
+DEVICE_NAME = "outgoing-conn-test"
+HA_CLIENT_INFO = "Home Assistant 2026.8.0"
+# HelloRequest field 4 (outgoing_connection_target) as raw protobuf bytes; the
+# installed aioesphomeapi's api_pb2 predates the field, so append it manually.
+HELLO_TARGET_FLAG = b"\x20\x01"
+
+
+def _frame(payload: bytes) -> bytes:
+    return bytes((0x01, len(payload) >> 8, len(payload) & 0xFF)) + payload
+
+
+async def _read_frame(reader: asyncio.StreamReader, timeout: float = 10.0) -> bytes:
+    header = await asyncio.wait_for(reader.readexactly(3), timeout)
+    assert header[0] == 0x01, f"Bad frame indicator: {header[0]}"
+    return await asyncio.wait_for(
+        reader.readexactly((header[1] << 8) | header[2]), timeout
+    )
+
+
+def _check_server_hello(server_hello: bytes) -> None:
+    assert server_hello[0] == 0x01, "Bad chosen proto in server hello"
+    name, mac, _rest = server_hello[1:].split(b"\x00", 2)
+    assert name.decode() == DEVICE_NAME
+    assert len(mac) == 12, f"Expected bare MAC, got {mac!r}"
+
+
+async def _run_ha_session(
+    reader: asyncio.StreamReader,
+    writer: asyncio.StreamWriter,
+    *,
+    device_dialed_out: bool,
+) -> None:
+    """Handshake and exchange the usual first messages as Home Assistant would."""
+    # Lazy import per the module's own contract (pulls in the noise stack)
+    from aioesphomeapi.noise import NoiseHandshake
+
+    if device_dialed_out:
+        # On an outgoing connection the device announces itself first so the
+        # peer can pick the matching key before its PSK-mixed first message.
+        _check_server_hello(await _read_frame(reader))
+
+    handshake = NoiseHandshake(KEY, b"NoiseAPIInit\x00\x00")
+    writer.write(b"\x01\x00\x00" + _frame(b"\x00" + handshake.write_message()))
+    await writer.drain()
+
+    if not device_dialed_out:
+        _check_server_hello(await _read_frame(reader))
+
+    reply = await _read_frame(reader)
+    assert reply[0] == 0, f"Handshake rejected: {reply[1:].decode(errors='replace')}"
+    handshake.read_message(reply[1:])
+    encrypt_cipher, decrypt_cipher = handshake.get_ciphers()
+
+    async def transact(msg: Any, response_cls: Any, extra_payload: bytes = b"") -> Any:
+        msg_type = MESSAGE_TYPE_OF[type(msg)]
+        payload = msg.SerializeToString() + extra_payload
+        plaintext = (
+            bytes(
+                (msg_type >> 8, msg_type & 0xFF, len(payload) >> 8, len(payload) & 0xFF)
+            )
+            + payload
+        )
+        writer.write(_frame(encrypt_cipher.encrypt(plaintext)))
+        await writer.drain()
+        want = MESSAGE_TYPE_OF[response_cls]
+        while True:
+            plain = decrypt_cipher.decrypt(await _read_frame(reader))
+            if ((plain[0] << 8) | plain[1]) == want:
+                response = response_cls()
+                response.ParseFromString(bytes(plain[4:]))
+                return response
+
+    # Declare this client a dial-back target in the hello
+    await transact(
+        api_pb2.HelloRequest(client_info=HA_CLIENT_INFO),
+        api_pb2.HelloResponse,
+        extra_payload=HELLO_TARGET_FLAG,
+    )
+    device_info = await transact(
+        api_pb2.DeviceInfoRequest(), api_pb2.DeviceInfoResponse
+    )
+    assert device_info.name == DEVICE_NAME
+
+
+async def _serve_home_assistant(listener: socket.socket) -> None:
+    """Accept one dial-in from the device and run the client side over it."""
+    loop = asyncio.get_running_loop()
+    conn, _ = await asyncio.wait_for(loop.sock_accept(listener), timeout=30)
+    reader, writer = await asyncio.open_connection(sock=conn)
+    try:
+        await _run_ha_session(reader, writer, device_dialed_out=True)
+    finally:
+        writer.close()
+
+
+@pytest.mark.asyncio
+async def test_api_outgoing_connection(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+) -> None:
+    """With a configured host the device dials out and speaks the normal API."""
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(2)
+    listener.setblocking(False)
+    port = listener.getsockname()[1]
+
+    try:
+        yaml = yaml_config.replace("OUTGOING_PORT", str(port))
+        async with run_compiled(yaml):
+            await _serve_home_assistant(listener)
+    finally:
+        listener.close()
+
+
+@pytest.mark.asyncio
+async def test_api_outgoing_connection_remembered(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+) -> None:
+    """No host configured: the device remembers the client whose hello carried
+    the dial-back flag and dials that address after a restart."""
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    # Bound but not yet listening so first-phase dials cannot queue stale
+    # connections; whether the device attempts any dial before the restart
+    # is timing dependent and not asserted here.
+    listener.bind(("127.0.0.1", 0))
+    port = listener.getsockname()[1]
+
+    try:
+        yaml = yaml_config.replace("OUTGOING_PORT", str(port))
+
+        async with run_compiled(yaml):
+            # Connect inbound with the dial-back flag; the device persists the
+            # peer address during the hello.
+            reader, writer = await asyncio.open_connection("127.0.0.1", unused_tcp_port)
+            try:
+                await _run_ha_session(reader, writer, device_dialed_out=False)
+            finally:
+                writer.close()
+
+        # Restart with the same preferences: the device now dials the
+        # remembered address on its own.
+        listener.listen(2)
+        listener.setblocking(False)
+        async with run_compiled(yaml):
+            await _serve_home_assistant(listener)
+    finally:
+        listener.close()

tests/integration/test_api_string_lambda.py

@@ -1,4 +1,4 @@
-"""Integration test for TemplatableStringValue with string lambdas."""
+"""Integration test for Home Assistant action lambdas returning strings and numbers."""
 
 from __future__ import annotations
 
@@ -16,7 +16,7 @@ async def test_api_string_lambda(
     run_compiled: RunCompiledFunction,
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    """Test TemplatableStringValue works with lambdas that return different types."""
+    """Test Home Assistant action lambdas that return different types are sent as strings."""
     loop = asyncio.get_running_loop()
 
     # Track log messages for all four service calls

tests/integration/test_branch_resume_async.py

@@ -0,0 +1,140 @@
+"""Test that if/else and while branches resume correctly after async actions, stop and parallel runs."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_branch_resume_async(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """A branch that ends after a delay must resume its owner exactly once."""
+    lines: list[str] = []
+    waiters: list[tuple[str, asyncio.Future[None]]] = []
+
+    def has_marker(text: str, line: str) -> bool:
+        # Whole markers only: "after-if" must not match "queued-after-if-1"
+        return re.search(rf"(?<![\w-]){re.escape(text)}(?![\w-])", line) is not None
+
+    def check_output(line: str) -> None:
+        lines.append(line)
+        for text, future in waiters:
+            if has_marker(text, line) and not future.done():
+                future.set_result(None)
+
+    def wait_for(text: str) -> asyncio.Future[None]:
+        future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+        waiters.append((text, future))
+        return future
+
+    def count(text: str) -> int:
+        return sum(has_marker(text, line) for line in lines)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        _, services = await client.list_entities_services()
+        service = {s.name: s for s in services}
+
+        done = wait_for("after-if")
+        await client.execute_service(service["run_delay_in_if"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert count("if-branch-after-delay") == 1
+        assert count("if-else-should-not-run") == 0
+        assert count("after-if") == 1
+
+        done = wait_for("after-else")
+        await client.execute_service(service["run_delay_in_else"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert count("else-branch-after-delay") == 1
+        assert count("if-then-should-not-run") == 0
+        assert count("after-else") == 1
+
+        done = wait_for("after-while")
+        await client.execute_service(service["run_delay_in_while"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert [count(f"while-body-{i}") for i in (1, 2, 3)] == [1, 1, 1]
+        assert count("while-body-4") == 0
+        assert count("after-while") == 1
+
+        done = wait_for("stop-check-done")
+        await client.execute_service(service["run_stop_in_branch"], {})
+        await asyncio.wait_for(done, timeout=3.0)
+        assert count("stop-running=0") == 1
+        assert count("stopped-branch-should-not-run") == 0
+        assert count("stopped-after-if-should-not-run") == 0
+
+        done = wait_for("after-nested-while")
+        await client.execute_service(service["run_nested_if_last_in_while"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert [count(f"nested-if-in-while-{i}") for i in (1, 2)] == [1, 1]
+        assert count("nested-if-in-while-3") == 0
+        assert count("after-nested-while") == 1
+
+        # Stop during the second iteration's delay: that iteration and the line after the loop never run;
+        # a fresh run then completes
+        done = wait_for("stop-while-done")
+        await client.execute_service(service["run_stop_mid_while"], {})
+        await asyncio.wait_for(done, timeout=3.0)
+        assert count("stop-while-running=0") == 1
+        assert count("stop-while-body-1") == 2
+        assert [count(f"stop-while-body-{i}") for i in (2, 3, 4, 5)] == [1, 1, 1, 1]
+        assert count("stop-while-done") == 1
+
+        done = wait_for("queued-after-if-3")
+        await client.execute_service(service["run_queued_branch"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        queued = [line for line in lines if "queued-" in line]
+        order = [
+            "queued-branch-1",
+            "queued-after-if-1",
+            "queued-branch-2",
+            "queued-after-if-2",
+            "queued-branch-3",
+            "queued-after-if-3",
+        ]
+        assert [next(o for o in order if o in line) for line in queued] == order
+
+        done = wait_for("restart-after-if-2")
+        await client.execute_service(service["run_restart_branch"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        await asyncio.sleep(0.2)
+        assert count("restart-branch-1") == 0
+        assert count("restart-after-if-1") == 0
+        assert count("restart-branch-2") == 1
+        assert count("restart-after-if-2") == 1
+
+        done = wait_for("after-delay-last")
+        await client.execute_service(service["run_delay_last_in_branch"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert count("delay-last-start") == 1
+        assert count("after-delay-last") == 1
+
+        done = wait_for("parallel-after-if-3")
+        await client.execute_service(service["run_parallel_branch"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        await asyncio.sleep(0.2)
+        for run in (1, 2, 3):
+            assert count(f"parallel-branch-done-{run}") == 1
+            assert count(f"parallel-after-if-{run}") == 1
+
+        # A condition that stops its own script must not start a branch or continue the chain
+        done = wait_for("stop-in-conditions-done")
+        await client.execute_service(service["run_stop_in_conditions"], {})
+        await asyncio.wait_for(done, timeout=2.0)
+        assert count("if-stopped-by-condition-should-not-run") == 0
+        assert count("if-else-stopped-by-condition-should-not-run") == 0
+        assert count("after-if-stopped-by-condition-should-not-run") == 0
+        assert count("else-stopped-by-condition-should-not-run") == 0
+        assert count("after-else-stopped-by-condition-should-not-run") == 0
+        assert count("while-stopped-by-condition-should-not-run") == 0
+        assert count("after-while-stopped-by-condition-should-not-run") == 0

tests/integration/test_ir_rf_transmit_complete.py

@@ -0,0 +1,131 @@
+"""IR/RF transmit completion replies (API 1.18) and the client pacing built on them.
+
+The transmitter is a host-only mock that takes a frame's real duration to
+"send" it and reports completion from a scheduler timeout, like the ESP32 RMT
+backend. The client is expected to hold the next frame until the device
+replies, so the mock never sees an overlapping frame.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+from aioesphomeapi import InfraredInfo, RadioFrequencyInfo
+from aioesphomeapi.api_pb2 import InfraredRFTransmitCompleteResponse
+import pytest
+
+from .state_utils import find_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+FRAME_COUNT = 5
+# 10 marks and 10 spaces of 500 us, sent twice: 20 ms per frame
+TIMINGS = [500, -500] * 10
+REPEAT = 2
+MOCK_EVENT = re.compile(
+    r"remote_transmitter_mock[^\]]*\]: (TX|Complete|Overlap)\b.*?seq=(\d+)"
+)
+
+
+@pytest.mark.shared_yaml("ir_rf_transmit_complete")
+@pytest.mark.asyncio
+async def test_ir_rf_transmit_complete_boot(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """The host build with the mock transmitters boots and lists all entities on any client."""
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        entities, _ = await client.list_entities_services()
+    assert find_entity(entities, "rf_transmitter_a", RadioFrequencyInfo) is not None
+    assert find_entity(entities, "rf_transmitter_b", RadioFrequencyInfo) is not None
+    assert find_entity(entities, "ir_transmitter", InfraredInfo) is not None
+
+
+@pytest.mark.shared_yaml("ir_rf_transmit_complete")
+@pytest.mark.asyncio
+async def test_ir_rf_transmit_complete(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Frames are answered once they leave the transmitter, never overlap, and a refused
+    request is answered at once. Two RF entities share a transmitter and each gets its own
+    reply; the infrared entity on its own transmitter is answered the same way."""
+    loop = asyncio.get_running_loop()
+    events: list[tuple[str, int]] = []
+    all_sent = loop.create_future()
+    all_logged = loop.create_future()
+
+    def line_callback(line: str) -> None:
+        if (match := MOCK_EVENT.search(line)) is None:
+            return
+        events.append((match.group(1), int(match.group(2))))
+        if (
+            match.group(1) == "Complete"
+            and not all_sent.done()
+            and sum(kind == "Complete" for kind, _ in events) == FRAME_COUNT
+        ):
+            all_sent.set_result(None)
+        # the log reader stops with the device, so wait for the last line before asserting on it
+        if len(events) == 2 * (FRAME_COUNT + 1) and not all_logged.done():
+            all_logged.set_result(None)
+
+    completions: list[InfraredRFTransmitCompleteResponse] = []
+    all_replied = loop.create_future()
+    ir_replied = loop.create_future()
+    refused = loop.create_future()
+
+    def on_complete(msg: InfraredRFTransmitCompleteResponse) -> None:
+        if not msg.success:
+            if not refused.done():
+                refused.set_result(msg)
+            return
+        completions.append(msg)
+        if len(completions) == FRAME_COUNT and not all_replied.done():
+            all_replied.set_result(None)
+        if len(completions) == FRAME_COUNT + 1 and not ir_replied.done():
+            ir_replied.set_result(None)
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+        rf = find_entity(entities, "rf_transmitter_a", RadioFrequencyInfo)
+        rf_b = find_entity(entities, "rf_transmitter_b", RadioFrequencyInfo)
+        ir = find_entity(entities, "ir_transmitter", InfraredInfo)
+        assert rf is not None and rf_b is not None, "RF transmitter entities not found"
+        assert ir is not None, "IR transmitter entity not found"
+
+        client._connection.add_message_callback(
+            on_complete, (InfraredRFTransmitCompleteResponse,)
+        )
+        # alternate between the two entities sharing the transmitter
+        keys = [rf.key if i % 2 == 0 else rf_b.key for i in range(FRAME_COUNT)]
+        for key in keys:
+            client.radio_frequency_transmit_raw_timings(
+                key, 433920000, TIMINGS, repeat_count=REPEAT
+            )
+
+        await asyncio.wait_for(all_replied, timeout=10)
+        await asyncio.wait_for(all_sent, timeout=10)
+
+        # the infrared entity goes through the same path on its own transmitter
+        client.infrared_rf_transmit_raw_timings(ir.key, 38000, TIMINGS)
+        await asyncio.wait_for(ir_replied, timeout=10)
+        await asyncio.wait_for(all_logged, timeout=10)
+
+        # A request the entity refuses is answered right away with success false;
+        # no timings, so the proxy rejects it before it reaches the transmitter
+        client.radio_frequency_transmit_raw_timings(rf.key, 433920000, [])
+        refused_msg = await asyncio.wait_for(refused, timeout=10)
+
+    assert [msg.key for msg in completions] == [*keys, ir.key]
+    assert refused_msg.key == rf.key
+
+    # The mocks saw one frame at a time: every transmit follows the previous completion
+    kinds = [kind for kind, _ in events]
+    assert kinds == ["TX", "Complete"] * (FRAME_COUNT + 1), events
+    seqs = [seq for _, seq in events]
+    assert seqs[::2] == seqs[1::2], events

tests/integration/test_scheduler_background_wake.py

@@ -0,0 +1,84 @@
+"""A background scheduler insertion must interrupt the loop's current sleep.
+
+Covers both set_timeout and defer: a zero-delay insert from another thread
+takes the separate defer queue on multi-threaded builds.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+import re
+
+from aioesphomeapi import ButtonInfo
+import pytest
+
+from .state_utils import require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_scheduler_background_wake(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+    results: dict[int, asyncio.Future[tuple[int, int]]] = {
+        100: loop.create_future(),
+        0: loop.create_future(),
+    }
+
+    def on_log_line(line: str) -> None:
+        match = re.search(
+            r"SCHEDULER_WAKE_RESULT delay=(\d+) elapsed=(\d+) loop_delta=(-?\d+)",
+            line,
+        )
+        if match is None:
+            return
+        result = results[int(match.group(1))]
+        if not result.done():
+            result.set_result((int(match.group(2)), int(match.group(3))))
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "scheduler-background-wake"
+        entities, _ = await client.list_entities_services()
+        for object_id, delay in (
+            ("start_scheduler_timeout", 100),
+            ("start_scheduler_defer", 0),
+        ):
+            start_button = require_entity(
+                entities,
+                object_id,
+                ButtonInfo,
+                description=f"{object_id} button",
+            )
+            client.button_command(start_button.key)
+
+            try:
+                elapsed, loop_delta = await asyncio.wait_for(
+                    results[delay], timeout=10.0
+                )
+            except TimeoutError:
+                pytest.fail(f"background insert with delay={delay} did not fire")
+
+            assert elapsed < 1000, (
+                f"background insert with delay={delay} should interrupt the "
+                f"five-second sleep; it fired after {elapsed}ms"
+            )
+            assert loop_delta == 0, (
+                f"scheduler-only wake must not run component loops; observed "
+                f"loop_delta={loop_delta} for delay={delay}"
+            )

tests/integration/test_sensor_timeout_throttle_filter.py

@@ -0,0 +1,61 @@
+"""The merged timeout + throttle_with_priority filter publishes exactly what the two filter chain does."""
+
+from __future__ import annotations
+
+import asyncio
+import math
+
+from aioesphomeapi import ButtonInfo, EntityState, SensorState
+import pytest
+
+from .state_utils import InitialStateHelper, build_key_to_entity_mapping
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_sensor_timeout_throttle_filter(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    loop = asyncio.get_running_loop()
+    values: dict[str, list[float]] = {"merged": [], "reference": []}
+    done = {name: loop.create_future() for name in values}
+
+    def on_state(state: EntityState) -> None:
+        if not isinstance(state, SensorState):
+            return
+        name = key_to_sensor.get(state.key)
+        if name not in values:
+            return
+        values[name].append(math.nan if state.missing_state else state.state)
+        # 99 is published last and republished once by the timeout
+        if values[name].count(99.0) == 2 and not done[name].done():
+            done[name].set_result(True)
+
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        entities, _ = await client.list_entities_services()
+        key_to_sensor = build_key_to_entity_mapping(
+            entities, {"merged": "Merged", "reference": "Reference"}
+        )
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        button = next(
+            e.key for e in entities if isinstance(e, ButtonInfo) and e.name == "Drive"
+        )
+        client.button_command(button)
+        try:
+            await asyncio.wait_for(asyncio.gather(*done.values()), timeout=5.0)
+        except TimeoutError:
+            pytest.fail(f"Timed out: {values}")
+
+        merged, reference = values["merged"], values["reference"]
+        assert len(merged) == len(reference), (merged, reference)
+        for got, want in zip(merged, reference, strict=True):
+            assert (math.isnan(got) and math.isnan(want)) or got == want, (
+                merged,
+                reference,
+            )
+        assert any(math.isnan(v) for v in merged)

tests/integration/test_status_momentary.py

@@ -0,0 +1,87 @@
+"""Integration test for status_momentary_warning/error without a name."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import BinarySensorState, EntityState
+import pytest
+
+from .state_utils import InitialStateHelper
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_status_momentary(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    loop = asyncio.get_running_loop()
+    warning_logged = loop.create_future()
+    error_logged = loop.create_future()
+
+    def on_log_line(line: str) -> None:
+        if (
+            "set Warning flag: probe warning reason" in line
+            and not warning_logged.done()
+        ):
+            warning_logged.set_result(True)
+        if "set Error flag: probe error reason" in line and not error_logged.done():
+            error_logged.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        entities, services = await client.list_entities_services()
+        svc = {s.name: s for s in services}
+        keys = {e.object_id: e.key for e in entities}
+        warning_key = keys["probe_warning"]
+        error_key = keys["probe_error"]
+
+        changes: asyncio.Queue[tuple[int, bool, float]] = asyncio.Queue()
+
+        def on_state(state: EntityState) -> None:
+            if isinstance(state, BinarySensorState):
+                changes.put_nowait((state.key, state.state, loop.time()))
+
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+        assert initial_state_helper.initial_states[warning_key].state is False
+        assert initial_state_helper.initial_states[error_key].state is False
+
+        async def next_change(key: int) -> tuple[bool, float]:
+            while True:
+                k, value, when = await asyncio.wait_for(changes.get(), timeout=5)
+                if k == key:
+                    return value, when
+
+        # The flag sets at once and clears after the length
+        for service, key in (
+            ("momentary_warning", warning_key),
+            ("momentary_error", error_key),
+        ):
+            start = loop.time()
+            await client.execute_service(svc[service], {"length": 300})
+            assert (await next_change(key))[0] is True
+            value, cleared = await next_change(key)
+            assert value is False
+            assert cleared - start >= 0.25
+
+        # A second call restarts the timeout instead of adding a second one
+        start = loop.time()
+        await client.execute_service(svc["momentary_warning"], {"length": 1000})
+        assert (await next_change(warning_key))[0] is True
+        await asyncio.sleep(0.6)
+        await client.execute_service(svc["momentary_warning"], {"length": 1000})
+        value, cleared = await next_change(warning_key)
+        assert value is False
+        assert cleared - start >= 1.4
+
+        # A message replaces "unspecified" in the log line
+        await client.execute_service(svc["momentary_warning_message"], {})
+        await asyncio.wait_for(warning_logged, timeout=5)
+        await client.execute_service(svc["momentary_error_message"], {})
+        await asyncio.wait_for(error_logged, timeout=5)

tests/integration/test_uart_tcp_bridge.py

@@ -27,13 +27,17 @@ async def test_uart_tcp_bridge(
     server_port = unused_tcp_port_factory()
     denied_port = unused_tcp_port_factory()
     controller_fd, device_fd = os.openpty()
+    denied_controller_fd, denied_device_fd = os.openpty()
     os.set_blocking(controller_fd, False)
     # uart's validate_port wants a two segment device path; Linux ptys live at
     # /dev/pts/N, so hand the config a /tmp symlink instead.
     pty_link = f"/tmp/uart-tcp-pty-{os.getpid()}"
     pathlib.Path(pty_link).symlink_to(os.ttyname(device_fd))
+    denied_pty_link = f"/tmp/uart-tcp-denied-pty-{os.getpid()}"
+    pathlib.Path(denied_pty_link).symlink_to(os.ttyname(denied_device_fd))
     yaml_config = yaml_config.replace("port: 18126", f"port: {server_port}")
     yaml_config = yaml_config.replace("port: 18127", f"port: {denied_port}")
+    yaml_config = yaml_config.replace("DENIED_PTY_PATH", denied_pty_link)
     yaml_config = yaml_config.replace("PTY_PATH", pty_link)
 
     lines = LineWaiter()
@@ -117,4 +121,7 @@ async def test_uart_tcp_bridge(
         loop.remove_reader(controller_fd)
         os.close(controller_fd)
         os.close(device_fd)
+        os.close(denied_controller_fd)
+        os.close(denied_device_fd)
         pathlib.Path(pty_link).unlink()
+        pathlib.Path(denied_pty_link).unlink()