Merge remote-tracking branch 'origin/dev' into web-server-offline-hint

# Conflicts:
#	esphome/components/web_server/__init__.py
#	esphome/components/web_server_base/__init__.py

esphome/components/captive_portal/__init__.py

@@ -3,7 +3,6 @@ import logging
 import esphome.codegen as cg
 from esphome.components import web_server_base, wifi
 from esphome.components.web_server_base import CONF_WEB_SERVER_BASE_ID
-from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_AP,
@@ -15,7 +14,6 @@ from esphome.const import (
     PLATFORM_LN882X,
     PLATFORM_RP2,
     PLATFORM_RTL87XX,
-    PlatformFramework,
 )
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
@@ -76,17 +74,7 @@ def _final_validate(config: ConfigType) -> None:
             "Add 'ap:' to your WiFi configuration to enable the captive portal."
         )
 
-    # Register socket needs for DNS server and additional HTTP connections
-    # - 1 UDP socket for DNS server
-    # - 3 TCP sockets for captive portal detection probes + configuration requests
-    #   OS captive portal detection makes multiple probe requests that stay in TIME_WAIT.
-    #   Need headroom for actual user configuration requests.
-    #   LRU purging will reclaim idle sockets to prevent exhaustion from repeated attempts.
-    # The listening socket is registered by web_server_base (shared HTTP server).
-    from esphome.components import socket
-
-    socket.consume_sockets(3, "captive_portal")(config)
-    socket.consume_sockets(1, "captive_portal", socket.SocketType.UDP)(config)
+    web_server_base.consume_captive_dns_sockets(config, "captive_portal")
 
 
 FINAL_VALIDATE_SCHEMA = _final_validate
@@ -106,16 +94,4 @@ async def to_code(config: ConfigType) -> None:
     if config[CONF_COMPRESSION] == "gzip":
         cg.add_define("USE_CAPTIVE_PORTAL_GZIP")
 
-    if CORE.using_arduino and (CORE.is_esp8266 or CORE.is_libretiny or CORE.is_rp2):
-        cg.add_library("DNSServer", None)
-
-
-# Only compile the ESP-IDF DNS server when using ESP-IDF framework
-FILTER_SOURCE_FILES = filter_source_files_from_platform(
-    {
-        "dns_server_esp32_idf.cpp": {
-            PlatformFramework.ESP32_ARDUINO,
-            PlatformFramework.ESP32_IDF,
-        },
-    }
-)
+    web_server_base.add_captive_dns_library()

esphome/components/captive_portal/captive_portal.cpp

@@ -102,17 +102,7 @@ void CaptivePortal::start() {
     this->base_->add_handler_without_auth(this);
   }
 
-  network::IPAddress ip = wifi::global_wifi_component->wifi_soft_ap_ip();
-
-#if defined(USE_ESP32)
-  // Create DNS server instance for ESP-IDF
-  this->dns_server_ = make_unique<DNSServer>();
-  this->dns_server_->start(ip);
-#elif defined(USE_ARDUINO)
-  this->dns_server_ = make_unique<DNSServer>();
-  this->dns_server_->setErrorReplyCode(DNSReplyCode::NoError);
-  this->dns_server_->start(53, ESPHOME_F("*"), ip);
-#endif
+  this->dns_.start(wifi::global_wifi_component->wifi_soft_ap_ip());
 
   this->initialized_ = true;
   this->active_ = true;

esphome/components/captive_portal/captive_portal.h

@@ -1,16 +1,11 @@
 #pragma once
 #include "esphome/core/defines.h"
 #ifdef USE_CAPTIVE_PORTAL
-#include <memory>
-#if defined(USE_ESP32)
-#include "dns_server_esp32_idf.h"
-#elif defined(USE_ARDUINO)
-#include <DNSServer.h>
-#endif
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/preferences.h"
 #include "esphome/components/web_server_base/web_server_base.h"
+#include "esphome/components/web_server_base/captive_dns.h"
 
 namespace esphome::captive_portal {
 
@@ -19,17 +14,7 @@ class CaptivePortal final : public AsyncWebHandler, public Component {
   CaptivePortal(web_server_base::WebServerBase *base);
   void setup() override;
   void dump_config() override;
-  void loop() override {
-#if defined(USE_ESP32)
-    if (this->dns_server_ != nullptr) {
-      this->dns_server_->process_next_request();
-    }
-#elif defined(USE_ARDUINO)
-    if (this->dns_server_ != nullptr) {
-      this->dns_server_->processNextRequest();
-    }
-#endif
-  }
+  void loop() override { this->dns_.loop(); }
   float get_setup_priority() const override;
   void start();
   bool is_active() const { return this->active_; }
@@ -37,10 +22,7 @@ class CaptivePortal final : public AsyncWebHandler, public Component {
     this->active_ = false;
     this->disable_loop();  // Stop processing DNS requests
     this->base_->deinit();
-    if (this->dns_server_ != nullptr) {
-      this->dns_server_->stop();
-      this->dns_server_ = nullptr;
-    }
+    this->dns_.stop();
   }
 
   bool canHandle(AsyncWebServerRequest *request) const override {
@@ -60,9 +42,7 @@ class CaptivePortal final : public AsyncWebHandler, public Component {
   web_server_base::WebServerBase *base_;
   bool initialized_{false};
   bool active_{false};
-#if defined(USE_ARDUINO) || defined(USE_ESP32)
-  std::unique_ptr<DNSServer> dns_server_{nullptr};
-#endif
+  web_server_base::CaptiveDNS dns_;
 };
 
 extern CaptivePortal *global_captive_portal;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)

esphome/components/web_server/__init__.py

@@ -12,6 +12,7 @@ from esphome.components.logger import request_log_listener
 from esphome.components.web_server_base import CONF_WEB_SERVER_BASE_ID
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_AP,
     CONF_AUTH,
     CONF_COMPRESSION,
     CONF_CSS_INCLUDE,
@@ -23,15 +24,19 @@ from esphome.const import (
     CONF_JS_URL,
     CONF_LOCAL,
     CONF_LOG,
+    CONF_MANUAL_IP,
     CONF_NAME,
+    CONF_NETWORKS,
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PORT,
+    CONF_STATIC_IP,
     CONF_TYPE,
     CONF_USERNAME,
     CONF_VERSION,
     CONF_WEB_SERVER,
     CONF_WEB_SERVER_ID,
+    CONF_WIFI,
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
@@ -46,7 +51,23 @@ from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
 
-AUTO_LOAD = ["json", "web_server_base"]
+
+def AUTO_LOAD() -> list[str]:
+    # No config parameter on purpose: that would make this a late (dynamic) auto-load and
+    # ota.web_server's dependency on web_server_base would not be satisfied in time.
+    auto_load = ["json", "web_server_base"]
+    # The AP mode DNS server (web_server_base/dns_server_esp32_idf) uses socket; only
+    # configs with a WiFi access point can end up in AP mode. CORE.raw_config is set
+    # after package merging, so a wifi block from a package is visible here.
+    wifi = CORE.raw_config.get(CONF_WIFI) if CORE.raw_config else None
+    if (
+        CORE.is_esp32
+        and wifi is not None
+        and (not isinstance(wifi, dict) or CONF_AP in wifi)
+    ):
+        auto_load.append("socket")
+    return auto_load
+
 
 AUTH_TYPE_BASIC = "basic"
 AUTH_TYPE_DIGEST = "digest"
@@ -205,9 +226,6 @@ def _final_validate_sorting(config: ConfigType) -> None:
         )
 
 
-FINAL_VALIDATE_SCHEMA = _final_validate_sorting
-
-
 def _consume_web_server_sockets(config: ConfigType) -> ConfigType:
     """Register socket needs for web_server component."""
     from esphome.components import socket
@@ -334,6 +352,95 @@ async def add_entity_config(entity: MockObj, config: ConfigType) -> None:
     )
 
 
+def wifi_is_ap_only(wifi_config: ConfigType | None) -> bool:
+    """AP only: an access point and no network to join, so the device is only ever reached
+    through its own AP."""
+    return (
+        wifi_config is not None
+        and CONF_AP in wifi_config
+        and not wifi_config.get(CONF_NETWORKS)
+    )
+
+
+def serve_local(config: ConfigType, wifi_config: ConfigType | None) -> bool:
+    """Embed the interface unless ``local:`` says otherwise; AP only WiFi has no internet
+    for the hosted page. Version 1 has no local mode."""
+    if (local := config.get(CONF_LOCAL)) is not None:
+        return local
+    return config[CONF_VERSION] != 1 and wifi_is_ap_only(wifi_config)
+
+
+def serve_captive(config: ConfigType, full_config: ConfigType) -> bool:
+    """web_server runs its own captive portal while the AP is up: embedded interface plus
+    an access point, unless captive_portal (which owns that role) is configured. Only on
+    port 80: the OS captive portal probes and the DHCP portal URI always use port 80, so
+    a portal on another port could never be discovered."""
+    wifi_config = full_config.get(CONF_WIFI)
+    return (
+        "captive_portal" not in full_config
+        and config[CONF_PORT] == 80
+        and wifi_config is not None
+        and CONF_AP in wifi_config
+        and serve_local(config, wifi_config)
+    )
+
+
+def _final_validate_ap_mode(config: ConfigType) -> None:
+    full_config = fv.full_config.get()
+    wifi_config = full_config.get(CONF_WIFI)
+    captive = serve_captive(config, full_config)
+    local = serve_local(config, wifi_config)
+    if captive:
+        web_server_base.consume_captive_dns_sockets(config, "web_server")
+    # Surface behavior that the config does not spell out.
+    if local and CONF_LOCAL not in config:
+        _LOGGER.info(
+            "WiFi is AP only: embedding the web interface in the firmware "
+            "(local: true, roughly 13 KB of flash for version 2, 78 KB for version 3)%s. "
+            "Set 'local: false' to load it from the internet instead.",
+            " and serving it as a captive portal on the access point"
+            if captive
+            else "",
+        )
+    elif captive:
+        _LOGGER.info(
+            "web_server will act as a captive portal while the %saccess point is active.",
+            "" if wifi_is_ap_only(wifi_config) else "fallback ",
+        )
+    if not wifi_is_ap_only(wifi_config):
+        return
+    if not local:
+        _LOGGER.warning(
+            "WiFi is AP only and the web_server interface is loaded from the internet, "
+            "which browsers on the access point usually cannot reach; the page stays "
+            "blank. %s so the interface is embedded in the firmware.",
+            "Remove 'local: false'"
+            if config.get(CONF_LOCAL) is False
+            else "Migrate to version 2 or 3",
+        )
+    elif config[CONF_PORT] != 80:
+        ap_ip = "192.168.4.1"
+        if (manual_ip := wifi_config[CONF_AP].get(CONF_MANUAL_IP)) is not None:
+            ap_ip = str(manual_ip[CONF_STATIC_IP])
+        _LOGGER.warning(
+            "WiFi is AP only and web_server uses port %d. The interface cannot open "
+            "automatically on the access point (captive portal detection only works on "
+            "port 80); open http://%s:%d/ manually, or remove 'port:' to use 80.",
+            config[CONF_PORT],
+            ap_ip,
+            config[CONF_PORT],
+        )
+
+
+def _final_validate(config: ConfigType) -> None:
+    # Called one after the other rather than via cv.All: these return None.
+    _final_validate_sorting(config)
+    _final_validate_ap_mode(config)
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
 def build_index_html(config: ConfigType) -> str:
     html = "<!DOCTYPE html><html><head><meta charset=UTF-8><link rel=icon href=data:>"
     css_include = config.get(CONF_CSS_INCLUDE)
@@ -434,8 +541,12 @@ async def to_code(config: ConfigType) -> None:
         with path.open(encoding="utf-8") as js_file:
             add_resource_as_progmem("JS_INCLUDE", js_file.read())
     cg.add(var.set_include_internal(config[CONF_INCLUDE_INTERNAL]))
-    if CONF_LOCAL in config and config[CONF_LOCAL]:
+    if serve_local(config, CORE.config.get(CONF_WIFI)):
         cg.add_define("USE_WEBSERVER_LOCAL")
+    if serve_captive(config, CORE.config):
+        # AP mode: DNS server plus redirect of unknown URLs so phones open the interface
+        cg.add_define("USE_WEBSERVER_CAPTIVE")
+        web_server_base.add_captive_dns_library()
     if config[CONF_COMPRESSION] == "gzip":
         cg.add_define("USE_WEBSERVER_GZIP")
 

esphome/components/web_server/web_server.cpp

@@ -44,6 +44,10 @@
 #include "esphome/components/radio_frequency/radio_frequency.h"
 #endif
 
+#ifdef USE_WEBSERVER_CAPTIVE
+#include "esphome/components/wifi/wifi_component.h"
+#endif
+
 #ifdef USE_WEBSERVER_LOCAL
 #if USE_WEBSERVER_VERSION == 2
 #include "server_index_v2.h"
@@ -334,7 +338,15 @@ void DeferredUpdateEventSourceList::on_client_disconnect_(DeferredUpdateEventSou
 }
 #endif
 
-WebServer::WebServer(web_server_base::WebServerBase *base) : base_(base) {}
+#ifdef USE_WEBSERVER_CAPTIVE
+WebServer *global_web_server = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+#endif
+
+WebServer::WebServer(web_server_base::WebServerBase *base) : base_(base) {
+#ifdef USE_WEBSERVER_CAPTIVE
+  global_web_server = this;
+#endif
+}
 
 #ifdef USE_WEBSERVER_CSS_INCLUDE
 void WebServer::set_css_include(const char *css_include) { this->css_include_ = css_include; }
@@ -380,6 +392,11 @@ void WebServer::setup() {
   this->base_->add_handler(&this->events_);
 #endif
   this->base_->add_handler(this);
+#ifdef USE_WEBSERVER_CAPTIVE
+  // Not-found fallback (outside the auth middleware): the OS captive portal probes hit
+  // arbitrary URLs and must get the redirect without credentials.
+  this->base_->get_server()->onNotFound([this](AsyncWebServerRequest *request) { this->handle_not_found_(request); });
+#endif
 
   // OTA is now handled by the web_server OTA platform
 
@@ -395,16 +412,52 @@ void WebServer::setup() {
   });
 }
 void WebServer::loop() {
-  // No SSE clients connected; stop looping until a new client connects via
+  bool keep_looping = this->events_.loop();
+#ifdef USE_WEBSERVER_CAPTIVE
+  this->dns_.loop();
+  keep_looping |= this->dns_.is_running();
+#endif
+  // No SSE clients connected (and no captive DNS to serve); stop looping until a new client connects via
   // enable_loop_soon_any_context(). This is safe because:
   // - set_interval/set_timeout/defer run via the Scheduler, independent of loop()
   // - deferrable_send_state early-outs when no clients are connected
   // - try_send_nodefer (log, ping) iterates sessions which are empty
   // - REST API handlers use defer() which runs via the Scheduler
-  if (!this->events_.loop())
+  if (!keep_looping)
     this->disable_loop();
 }
 
+#ifdef USE_WEBSERVER_CAPTIVE
+void WebServer::start_captive() {
+  // CaptiveDNS::start() no-ops too; this guard just avoids repeating the log and enable_loop
+  if (this->dns_.is_running())
+    return;
+  network::IPAddress ip = wifi::global_wifi_component->wifi_soft_ap_ip();
+  this->dns_.start(ip);
+  this->enable_loop();
+  char ip_buf[network::IP_ADDRESS_BUFFER_SIZE];
+  ESP_LOGI(TAG, "AP mode: serving the web interface as captive portal at http://%s/", ip.str_to(ip_buf));
+}
+
+void WebServer::end_captive() { this->dns_.stop(); }
+
+void WebServer::handle_not_found_(AsyncWebServerRequest *request) {
+  // OS captive portal probe (or any other unknown page) while the AP is up: send the browser
+  // to the real page. A redirect rather than the page itself, because the interface resolves
+  // its /events and REST paths relative to the page URL.
+  if (this->dns_.is_running() && request->method() == HTTP_GET) {
+    // Captive mode requires port 80 (enforced at validation), so no port suffix is needed.
+    char location[7 + network::IP_ADDRESS_BUFFER_SIZE + 1];
+    size_t pos = buf_append_str(location, sizeof(location), 0, "http://");
+    wifi::global_wifi_component->wifi_soft_ap_ip().str_to(location + pos);
+    buf_append_str(location, sizeof(location), strlen(location), "/");
+    request->redirect(location);
+    return;
+  }
+  request->send(404);
+}
+#endif
+
 #ifdef USE_LOGGER
 void WebServer::on_log(uint8_t level, const char *tag, const char *message, size_t message_len) {
   (void) level;

esphome/components/web_server/web_server.h

@@ -4,6 +4,9 @@
 
 #include "esphome/components/json/json_util.h"
 #include "esphome/components/web_server_base/web_server_base.h"
+#ifdef USE_WEBSERVER_CAPTIVE
+#include "esphome/components/web_server_base/captive_dns.h"
+#endif
 #ifdef USE_WEBSERVER
 #include "esphome/core/component.h"
 #include "esphome/core/controller.h"
@@ -276,6 +279,18 @@ class WebServer final : public Controller, public Component, public AsyncWebHand
   /// Handle an index request under '/'.
   void handle_index_request(AsyncWebServerRequest *request);
 
+#ifdef USE_WEBSERVER_CAPTIVE
+  /** AP mode: run a DNS server that answers every name with the AP address and redirect any
+   * unknown URL to the interface, so a phone joining the AP opens it through the OS captive
+   * portal check. Started and ended by the wifi component with the access point. start may run
+   * before setup() (wifi sets up first): safe because enable_loop() is a no-op before setup;
+   * nothing but the DNS server may be touched, in particular not base_ or the handlers.
+   */
+  void start_captive();
+  void end_captive();
+  bool is_captive() const { return this->dns_.is_running(); }
+#endif
+
   /// Return the webserver configuration as JSON.
   json::SerializationBuffer<> get_config_json();
 
@@ -597,6 +612,10 @@ class WebServer final : public Controller, public Component, public AsyncWebHand
 #elif USE_ARDUINO
   DeferredUpdateEventSourceList events_;
 #endif
+#ifdef USE_WEBSERVER_CAPTIVE
+  void handle_not_found_(AsyncWebServerRequest *request);
+  web_server_base::CaptiveDNS dns_;
+#endif
 
 #if USE_WEBSERVER_VERSION == 1
   const char *css_url_{nullptr};
@@ -696,5 +715,9 @@ class WebServer final : public Controller, public Component, public AsyncWebHand
 #endif
 };
 
+#ifdef USE_WEBSERVER_CAPTIVE
+extern WebServer *global_web_server;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+#endif
+
 }  // namespace esphome::web_server
 #endif

esphome/components/web_server_base/__init__.py

@@ -1,8 +1,9 @@
 from pathlib import Path
 
 import esphome.codegen as cg
+from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
-from esphome.const import CONF_ID
+from esphome.const import CONF_ID, PlatformFramework
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
 from esphome.helpers import copy_file_if_changed
@@ -26,6 +27,22 @@ WebServerBase = web_server_base_ns.class_("WebServerBase")
 CONF_WEB_SERVER_BASE_ID = "web_server_base_id"
 
 
+def consume_captive_dns_sockets(config: ConfigType, name: str) -> None:
+    """Register the sockets a captive portal needs on top of the shared HTTP server:
+    1 UDP socket for the DNS server and 3 TCP sockets for the OS captive portal probes,
+    which make several requests that linger in TIME_WAIT."""
+    from esphome.components import socket
+
+    socket.consume_sockets(3, name)(config)
+    socket.consume_sockets(1, name, socket.SocketType.UDP)(config)
+
+
+def add_captive_dns_library() -> None:
+    """Pull in the Arduino DNSServer library used by CaptiveDNS off ESP32."""
+    if CORE.using_arduino and (CORE.is_esp8266 or CORE.is_libretiny or CORE.is_rp2):
+        cg.add_library("DNSServer", None)
+
+
 def _consume_web_server_base_sockets(config: ConfigType) -> ConfigType:
     """Register the shared listening socket for the HTTP server.
 
@@ -81,3 +98,15 @@ async def to_code(config: ConfigType) -> None:
             cg.add_platformio_option("extra_scripts", ["pre:fix_rp2040_hash.py"])
         # https://github.com/ESP32Async/ESPAsyncWebServer/blob/main/library.json
         cg.add_library("ESP32Async/ESPAsyncWebServer", "3.9.6")
+
+
+# The DNS server used for captive portals on ESP32; other platforms use the Arduino
+# DNSServer library. Its source is also guarded by USE_CAPTIVE_PORTAL / USE_WEBSERVER_CAPTIVE.
+FILTER_SOURCE_FILES = filter_source_files_from_platform(
+    {
+        "dns_server_esp32_idf.cpp": {
+            PlatformFramework.ESP32_ARDUINO,
+            PlatformFramework.ESP32_IDF,
+        },
+    }
+)

esphome/components/web_server_base/captive_dns.h

@@ -0,0 +1,58 @@
+#pragma once
+#include "esphome/core/defines.h"
+// DNS server that answers every name with the access point address, so a phone joining the
+// AP runs its captive portal check against the device. Shared by captive_portal and the
+// web_server AP mode; hides the ESP32 (own implementation) vs Arduino (DNSServer library) split.
+#if defined(USE_CAPTIVE_PORTAL) || defined(USE_WEBSERVER_CAPTIVE)
+#include <memory>
+
+#include "esphome/components/network/ip_address.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/progmem.h"
+#if defined(USE_ESP32)
+#include "dns_server_esp32_idf.h"
+#elif defined(USE_ARDUINO)
+#include <DNSServer.h>
+#endif
+
+namespace esphome::web_server_base {
+
+// The server object only exists while running, so an idle owner (AP not up) pays one pointer.
+class CaptiveDNS {
+ public:
+  void start(const network::IPAddress &ip) {
+    if (this->dns_server_ != nullptr)
+      return;
+    this->dns_server_ = make_unique<DNSServer>();
+#if defined(USE_ESP32)
+    this->dns_server_->start(ip);
+#elif defined(USE_ARDUINO)
+    this->dns_server_->setErrorReplyCode(DNSReplyCode::NoError);
+    this->dns_server_->start(53, ESPHOME_F("*"), ip);
+#endif
+  }
+  void stop() {
+    if (this->dns_server_ == nullptr)
+      return;
+    this->dns_server_->stop();
+    this->dns_server_ = nullptr;
+  }
+  /// Answer one pending query; call from the owner's loop() while running.
+  void loop() {
+    if (this->dns_server_ == nullptr)
+      return;
+#if defined(USE_ESP32)
+    this->dns_server_->process_next_request();
+#elif defined(USE_ARDUINO)
+    this->dns_server_->processNextRequest();
+#endif
+  }
+  bool is_running() const { return this->dns_server_ != nullptr; }
+
+ protected:
+  // ESP32: web_server_base::DNSServer from dns_server_esp32_idf.h; Arduino: the library class.
+  std::unique_ptr<DNSServer> dns_server_;
+};
+
+}  // namespace esphome::web_server_base
+#endif  // USE_CAPTIVE_PORTAL || USE_WEBSERVER_CAPTIVE

esphome/components/web_server_base/dns_server_esp32_idf.cpp

@@ -1,5 +1,5 @@
 #include "dns_server_esp32_idf.h"
-#ifdef USE_ESP32
+#if defined(USE_ESP32) && (defined(USE_CAPTIVE_PORTAL) || defined(USE_WEBSERVER_CAPTIVE))
 
 #include "esphome/core/log.h"
 #include "esphome/core/hal.h"
@@ -7,9 +7,9 @@
 #include <lwip/sockets.h>
 #include <lwip/inet.h>
 
-namespace esphome::captive_portal {
+namespace esphome::web_server_base {
 
-static const char *const TAG = "captive_portal.dns";
+static const char *const TAG = "web_server_base.dns";
 
 // DNS constants
 static constexpr uint16_t DNS_PORT = 53;
@@ -202,6 +202,6 @@ void DNSServer::process_next_request() {
   }
 }
 
-}  // namespace esphome::captive_portal
+}  // namespace esphome::web_server_base
 
-#endif  // USE_ESP32
+#endif  // USE_ESP32 && (USE_CAPTIVE_PORTAL || USE_WEBSERVER_CAPTIVE)

esphome/components/web_server_base/dns_server_esp32_idf.h

@@ -1,11 +1,15 @@
 #pragma once
-#ifdef USE_ESP32
+#include "esphome/core/defines.h"
+// Small DNS server that answers every query with the access point address, so a
+// phone joining the AP opens the captive portal or web_server page on its own.
+// Shared by captive_portal and the web_server AP mode.
+#if defined(USE_ESP32) && (defined(USE_CAPTIVE_PORTAL) || defined(USE_WEBSERVER_CAPTIVE))
 
 #include "esphome/core/helpers.h"
 #include "esphome/components/network/ip_address.h"
 #include "esphome/components/socket/socket.h"
 
-namespace esphome::captive_portal {
+namespace esphome::web_server_base {
 
 class DNSServer {
  public:
@@ -27,6 +31,6 @@ class DNSServer {
   uint8_t buffer_[DNS_BUFFER_SIZE];
 };
 
-}  // namespace esphome::captive_portal
+}  // namespace esphome::web_server_base
 
-#endif  // USE_ESP32
+#endif  // USE_ESP32 && (USE_CAPTIVE_PORTAL || USE_WEBSERVER_CAPTIVE)

esphome/components/web_server_idf/web_server_idf.cpp

@@ -325,9 +325,9 @@ StringRef AsyncWebServerRequest::url_to(std::span<char, URL_BUF_SIZE> buffer) co
   return StringRef(buffer.data(), decoded_len);
 }
 
-void AsyncWebServerRequest::redirect(const std::string &url) {
+void AsyncWebServerRequest::redirect(const char *url) {
   httpd_resp_set_status(*this, "302 Found");
-  httpd_resp_set_hdr(*this, "Location", url.c_str());
+  httpd_resp_set_hdr(*this, "Location", url);
   httpd_resp_set_hdr(*this, "Connection", "close");
   httpd_resp_send(*this, nullptr, 0);
 }

esphome/components/web_server_idf/web_server_idf.h

@@ -126,7 +126,8 @@ class AsyncWebServerRequest {
   void requestAuthentication() const;
 #endif
 
-  void redirect(const std::string &url);
+  void redirect(const char *url);
+  void redirect(const std::string &url) { this->redirect(url.c_str()); }
 
   inline void ESPHOME_ALWAYS_INLINE send(AsyncWebServerResponse *response) {
     httpd_resp_send(*this, response->get_content_data(), response->get_content_size());

esphome/components/wifi/wifi_component.cpp

@@ -36,6 +36,9 @@
 #ifdef USE_CAPTIVE_PORTAL
 #include "esphome/components/captive_portal/captive_portal.h"
 #endif
+#ifdef USE_WEBSERVER_CAPTIVE
+#include "esphome/components/web_server/web_server.h"
+#endif
 
 #ifdef USE_IMPROV
 #include "esphome/components/esp32_improv/esp32_improv_component.h"
@@ -741,9 +744,9 @@ void WiFiComponent::start() {
     if (captive_portal::global_captive_portal != nullptr) {
       this->wifi_sta_pre_setup_();
       this->start_scanning();
-      captive_portal::global_captive_portal->start();
     }
 #endif
+    this->start_ap_portal_();
 #endif  // USE_WIFI_AP
   }
 #ifdef USE_IMPROV
@@ -804,8 +807,8 @@ void WiFiComponent::loop() {
           this->check_connecting_finished(now);
           break;
         }
-        // Use longer cooldown when captive portal/improv is active to avoid disrupting user config
-        bool portal_active = this->is_captive_portal_active_() || this->is_esp32_improv_active_();
+        // Use longer cooldown when a portal/improv is active to avoid disrupting a user on the AP
+        bool portal_active = this->is_ap_portal_active_() || this->is_esp32_improv_active_();
         uint32_t cooldown_duration = portal_active ? WIFI_COOLDOWN_WITH_AP_ACTIVE_MS : WIFI_COOLDOWN_DURATION_MS;
         if (now - this->action_started_ > cooldown_duration) {
           // After cooldown we either restarted the adapter because of
@@ -882,13 +885,11 @@ void WiFiComponent::loop() {
         ESP_LOGI(TAG, "Starting fallback AP");
         this->setup_ap_config_();
 #ifdef USE_CAPTIVE_PORTAL
-        if (captive_portal::global_captive_portal != nullptr) {
-          // Reset so we force one full scan after captive portal starts
-          // (previous scans were filtered because captive portal wasn't active yet)
-          this->has_completed_scan_after_captive_portal_start_ = false;
-          captive_portal::global_captive_portal->start();
-        }
+        // Reset so we force one full scan after captive portal starts
+        // (previous scans were filtered because captive portal wasn't active yet)
+        this->has_completed_scan_after_captive_portal_start_ = false;
 #endif
+        this->start_ap_portal_();
       }
     }
 #endif  // USE_WIFI_AP
@@ -1635,10 +1636,8 @@ void WiFiComponent::check_connecting_finished(uint32_t now) {
     this->retry_phase_ = WiFiRetryPhase::INITIAL_CONNECT;
     this->num_retried_ = 0;
     if (this->has_ap()) {
-#ifdef USE_CAPTIVE_PORTAL
-      if (this->is_captive_portal_active_()) {
-        captive_portal::global_captive_portal->end();
-      }
+#ifdef USE_WIFI_AP
+      this->end_ap_portal_();
 #endif
       ESP_LOGD(TAG, "Disabling AP");
       this->wifi_mode_({}, false);
@@ -1965,10 +1964,10 @@ bool WiFiComponent::transition_to_phase_(WiFiRetryPhase new_phase) {
       break;
 
     case WiFiRetryPhase::RESTARTING_ADAPTER:
-      // Skip actual adapter restart if captive portal/improv is active
+      // Skip actual adapter restart if a portal/improv is active
       // This allows state machine to reset num_retried_ and trigger fresh scan
-      // without disrupting the captive portal/improv connection
-      if (!this->is_captive_portal_active_() && !this->is_esp32_improv_active_()) {
+      // without disrupting the portal/improv connection
+      if (!this->is_ap_portal_active_() && !this->is_esp32_improv_active_()) {
         this->restart_adapter();
       } else {
         // Even when skipping full restart, disconnect to clear driver state
@@ -2227,6 +2226,38 @@ bool WiFiComponent::is_captive_portal_active_() {
   return false;
 #endif
 }
+
+bool WiFiComponent::is_ap_portal_active_() {
+#ifdef USE_WEBSERVER_CAPTIVE
+  if (web_server::global_web_server->is_captive())
+    return true;
+#endif
+  return this->is_captive_portal_active_();
+}
+
+#ifdef USE_WIFI_AP
+// global_web_server needs no null check: codegen always instantiates WebServer when
+// USE_WEBSERVER_CAPTIVE is defined, and the constructor assigns the global.
+void WiFiComponent::start_ap_portal_() {
+#ifdef USE_CAPTIVE_PORTAL
+  if (captive_portal::global_captive_portal != nullptr)
+    captive_portal::global_captive_portal->start();
+#endif
+#ifdef USE_WEBSERVER_CAPTIVE
+  web_server::global_web_server->start_captive();
+#endif
+}
+
+void WiFiComponent::end_ap_portal_() {
+#ifdef USE_CAPTIVE_PORTAL
+  if (this->is_captive_portal_active_())
+    captive_portal::global_captive_portal->end();
+#endif
+#ifdef USE_WEBSERVER_CAPTIVE
+  web_server::global_web_server->end_captive();
+#endif
+}
+#endif  // USE_WIFI_AP
 bool WiFiComponent::is_esp32_improv_active_() {
 #ifdef USE_IMPROV
   return esp32_improv::global_improv_component != nullptr && esp32_improv::global_improv_component->is_active();

esphome/components/wifi/wifi_component.h

@@ -796,6 +796,12 @@ class WiFiComponent final : public Component {
   network::IPAddress wifi_dns_ip_(int num);
 
   bool is_captive_portal_active_();
+  /// captive_portal or the web_server AP mode is serving a user on the access point
+  bool is_ap_portal_active_();
+#ifdef USE_WIFI_AP
+  void start_ap_portal_();
+  void end_ap_portal_();
+#endif
   bool is_esp32_improv_active_();
 
 #ifdef USE_WIFI_FAST_CONNECT

esphome/components/wifi/wifi_component_esp_idf.cpp

@@ -1130,10 +1130,16 @@ bool WiFiComponent::wifi_ap_ip_config_(const optional<ManualIP> &manual_ip) {
     return false;
   }
 
-#if defined(USE_CAPTIVE_PORTAL) && ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 4, 0)
-  // Configure DHCP Option 114 (Captive Portal URI) if captive portal is enabled
-  // This provides a standards-compliant way for clients to discover the captive portal
-  if (captive_portal::global_captive_portal != nullptr) {
+#if (defined(USE_CAPTIVE_PORTAL) || defined(USE_WEBSERVER_CAPTIVE)) && ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 4, 0)
+  // Configure DHCP Option 114 (Captive Portal URI) if captive portal or the web_server AP
+  // mode is enabled. This provides a standards-compliant way for clients to discover the portal
+#ifdef USE_WEBSERVER_CAPTIVE
+  // web_server AP mode always serves the portal when compiled in
+  const bool has_portal = true;
+#else
+  const bool has_portal = captive_portal::global_captive_portal != nullptr;
+#endif
+  if (has_portal) {
     // Buffer must be static - dhcps_set_option_info stores pointer, doesn't copy
     static char captive_portal_uri[24];        // "http://" (7) + IPv4 max (15) + null
     memcpy(captive_portal_uri, "http://", 7);  // NOLINT(bugprone-not-null-terminated-result) - str_to null-terminates

esphome/core/defines.h

@@ -366,6 +366,7 @@
 #define USE_WEBSERVER
 #define USE_WEBSERVER_AUTH
 #define USE_WEBSERVER_AUTH_DIGEST
+#define USE_WEBSERVER_CAPTIVE
 #define USE_WEBSERVER_OTA
 #define USE_WEBSERVER_PORT 80  // NOLINT
 #define USE_WEBSERVER_GZIP
@@ -479,6 +480,7 @@
 #define USE_WEBSERVER
 #define USE_WEBSERVER_AUTH
 #define USE_WEBSERVER_AUTH_DIGEST
+#define USE_WEBSERVER_CAPTIVE
 #define USE_WEBSERVER_PORT 80  // NOLINT
 #endif
 
@@ -536,6 +538,7 @@
 #define USE_WEBSERVER
 #define USE_WEBSERVER_AUTH
 #define USE_WEBSERVER_AUTH_DIGEST
+#define USE_WEBSERVER_CAPTIVE
 #define USE_WEBSERVER_PORT 80  // NOLINT
 #define USE_ESPHOME_TASK_LOG_BUFFER
 #define ESPHOME_TASK_LOG_BUFFER_SIZE 768

script/helpers.py

@@ -1104,10 +1104,6 @@ def get_components_per_integration_fixture() -> dict[str, set[str]]:
 
 
 _TEST_FUNC_RE = re.compile(r"async def (test_\w+)")
-# Any usage form (decorator, pytestmark assignment or list element); only
-# test_*.py files are scanned, so the marker docs elsewhere cannot false-hit
-_SHARED_YAML_USE_RE = re.compile(r"\bmark\.shared_yaml")
-_SHARED_YAML_ARG_RE = re.compile(r"\(\s*[\"'](\w+)[\"']\s*\)")
 
 
 @cache
@@ -1127,19 +1123,6 @@ def get_fixture_to_test_files() -> dict[str, frozenset[str]]:
         for func in _TEST_FUNC_RE.findall(content):
             base_name = func.replace("test_", "").partition("[")[0]
             result.setdefault(base_name, set()).add(rel_path)
-        # Shared fixtures are named by marker, not by a test function; each
-        # decorator must carry a string literal or its fixture would silently
-        # map to no tests
-        for use in _SHARED_YAML_USE_RE.finditer(content):
-            arg = _SHARED_YAML_ARG_RE.match(content, use.end())
-            if arg is None:
-                line = content.count("\n", 0, use.start()) + 1
-                raise ValueError(
-                    f"{rel_path}:{line}: shared_yaml marker must take a "
-                    "single-line string literal so CI test selection can map "
-                    "its fixture"
-                )
-            result.setdefault(arg.group(1), set()).add(rel_path)
 
     return {k: frozenset(v) for k, v in result.items()}
 

tests/components/web_server/common-ap-fallback.yaml

@@ -0,0 +1,11 @@
+# STA with AP fallback plus local: true opts the fallback into captive AP mode; exercises
+# the runtime start (fallback branch in wifi loop) and end (on STA connect) paths.
+wifi:
+  ssid: MySSID
+  password: password1
+  ap:
+    ssid: "ESPHome-Test"
+    password: "Test1234!"
+
+web_server:
+  local: true

tests/components/web_server/common-ap-mode.yaml

@@ -0,0 +1,9 @@
+# AP mode: with a WiFi access point and the interface embedded in the firmware, web_server
+# runs its own captive portal (DNS server, unknown URLs redirect to the page).
+wifi:
+  ap:
+    ssid: "ESPHome-Test"
+    password: "Test1234!"
+
+web_server:
+  local: true

tests/components/web_server/test-ap-fallback.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  web_server: !include common-ap-fallback.yaml

tests/components/web_server/test-ap-mode.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  web_server: !include common-ap-mode.yaml

tests/components/web_server/test-ap-mode.esp8266-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  web_server: !include common-ap-mode.yaml

tests/integration/README.md

@@ -21,13 +21,6 @@ The `yaml_config` fixture automatically loads YAML configurations based on the t
 - The fixture file must exist or the test will fail with a clear error message
 - The fixture automatically injects a dynamic port number into the API configuration
 
-Tests marked `@pytest.mark.shared_yaml("name")` load `fixtures/name.yaml` instead
-of the test-named file and compile it in a shared, hash-keyed build directory, so
-the whole group pays one full compile and each test only a relink. The marker
-argument must be a single-line string literal (CI test selection maps fixtures to
-test files by scanning for it), and marked tests must hand the `yaml_config`
-content to `run_compiled` unmodified.
-
 ### Key Fixtures
 
 - `run_compiled` - Combines write, compile, and run operations into a single context manager

tests/integration/conftest.py

@@ -4,22 +4,17 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import AsyncGenerator, Callable, Generator
-from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
 import fcntl
-from functools import cache
-import hashlib
 import logging
 import os
 from pathlib import Path
 import platform
-import re
-import shutil
 import signal
 import socket
 import subprocess
 import sys
 import tempfile
-import time
 from typing import TextIO
 
 from aioesphomeapi import APIClient, APIConnectionError, LogParser, ReconnectLogic
@@ -28,13 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
-from esphome.helpers import (
-    get_usable_cpu_count,
-    read_file,
-    rmtree,
-    write_file,
-    write_file_if_changed,
-)
+from esphome.helpers import get_usable_cpu_count
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -67,21 +56,6 @@ import pty  # not available on Windows
 pytest.register_assert_rewrite("tests.integration.entity_utils")
 
 
-def pytest_configure(config: pytest.Config) -> None:
-    config.addinivalue_line(
-        "markers",
-        "shared_yaml(name): load fixtures/<name>.yaml and compile it in a shared, "
-        "hash-keyed incremental build directory",
-    )
-
-
-FIXTURES_DIR = Path(__file__).parent / "fixtures"
-REPO_ROOT = Path(__file__).resolve().parent.parent.parent
-
-# CI caches parts of this path; keep in sync with ci.yml integration-tests.
-INTEGRATION_TESTS_ROOT = Path.home() / ".esphome-integration-tests"
-
-
 def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     """Get environment variables for PlatformIO with shared cache."""
     env = os.environ.copy()
@@ -104,7 +78,7 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
         )
     # Compile with THIS tree's esphome sources, not wherever the venv's editable
     # install points (which may be a different git worktree or checkout).
-    repo_root = str(REPO_ROOT)
+    repo_root = str(Path(__file__).resolve().parent.parent.parent)
     existing = env.get("PYTHONPATH")
     env["PYTHONPATH"] = f"{repo_root}{os.pathsep}{existing}" if existing else repo_root
     return env
@@ -114,7 +88,8 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
 def shared_platformio_cache() -> Generator[Path]:
     """Initialize a shared PlatformIO cache for all integration tests."""
     # Use a dedicated directory for integration tests to avoid conflicts.
-    test_cache_dir = INTEGRATION_TESTS_ROOT
+    # CI caches parts of this path; keep in sync with ci.yml integration-tests.
+    test_cache_dir = Path.home() / ".esphome-integration-tests"
     cache_dir = test_cache_dir / "platformio"
 
     # Use a lock file in the home directory to ensure only one process initializes the cache
@@ -137,9 +112,7 @@ def shared_platformio_cache() -> Generator[Path]:
                 init_dir = Path(tmpdir)
                 fixture_path = Path(__file__).parent / "fixtures" / "cache_init.yaml"
                 config_path = init_dir / "cache_init.yaml"
-                config_path.write_text(
-                    fixture_path.read_text(encoding="utf-8"), encoding="utf-8"
-                )
+                config_path.write_text(fixture_path.read_text())
 
                 # Run compilation to populate the cache
                 # We must succeed here to avoid race conditions where multiple
@@ -208,29 +181,21 @@ def unused_tcp_port(reserved_tcp_port: tuple[int, socket.socket]) -> int:
     return reserved_tcp_port[0]
 
 
-@pytest.fixture(autouse=True)
-def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
-    """Give every test its own host prefs dir; prefs are keyed only by device
-    name, which tests sharing a fixture also share."""
-    prefdir = tmp_path / "prefs"
-    monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
-    return prefdir
-
-
 @pytest_asyncio.fixture
 async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> str:
     """Load YAML configuration based on test name."""
-    shared_name = _shared_yaml_name(request)
-    # Base test name: test_ prefix and any parametrization stripped
-    base_name = shared_name or request.node.name.replace("test_", "").partition("[")[0]
+    # Get the test function name
+    test_name: str = request.node.name
+    # Extract the base test name (remove test_ prefix and any parametrization)
+    base_name = test_name.replace("test_", "").partition("[")[0]
 
     # Load the fixture file
-    fixture_path = FIXTURES_DIR / f"{base_name}.yaml"
+    fixture_path = Path(__file__).parent / "fixtures" / f"{base_name}.yaml"
     if not fixture_path.exists():
         raise FileNotFoundError(f"Fixture file not found: {fixture_path}")
 
     loop = asyncio.get_running_loop()
-    content = await loop.run_in_executor(None, read_file, fixture_path)
+    content = await loop.run_in_executor(None, fixture_path.read_text)
 
     # Replace the port in the config if it contains api section
     if "api:" in content:
@@ -254,13 +219,11 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
 
     # Replace external component path placeholder if present
     if "EXTERNAL_COMPONENT_PATH" in content:
-        external_components_path = str(FIXTURES_DIR / "external_components")
+        external_components_path = str(
+            Path(__file__).parent / "fixtures" / "external_components"
+        )
         content = content.replace("EXTERNAL_COMPONENT_PATH", external_components_path)
 
-    if shared_name is not None:
-        # _compile verifies the marked test compiles this content unmodified
-        request.node._shared_yaml_content = content
-
     return content
 
 
@@ -270,218 +233,24 @@ async def write_yaml_config(
 ) -> AsyncGenerator[ConfigWriter]:
     """Write YAML configuration to a file."""
     # Get the test name for default filename
-    base_name = request.node.name.replace("test_", "").partition("[")[0]
+    test_name = request.node.name
+    base_name = test_name.replace("test_", "").split("[")[0]
 
     async def _write_config(content: str, filename: str | None = None) -> Path:
         if filename is None:
             filename = f"{base_name}.yaml"
         config_path = integration_test_dir / filename
         loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, write_file, config_path, content)
+        await loop.run_in_executor(None, config_path.write_text, content)
         return config_path
 
     yield _write_config
 
 
-# Deliberately not CI-cached (ci.yml caches only platformio/ subpaths); stale
-# dirs for a fixture are pruned when its content hash changes.
-SHARED_BUILDS_ROOT = INTEGRATION_TESTS_ROOT / "builds"
-
-# In the dir name (not just the hash) so pruning stays inside this checkout
-_REPO_KEY = hashlib.sha256(str(REPO_ROOT).encode()).hexdigest()[:8]
-
-# Give a contended shared build lock time for a full cold compile ahead of us
-_SHARED_LOCK_TIMEOUT_S = 900
-_SHARED_LOCK_POLL_S = 0.1
-_SHARED_LOCK_REPORT_S = 30
-
-# Reclaims dirs orphaned by fixture renames or deleted checkouts
-_STALE_BUILD_MAX_AGE_S = 30 * 24 * 3600
-
-# ELF path per shared build dir; constant once compiled, so resolve it only once
-_shared_elf_paths: dict[Path, Path] = {}
-
-# Dirs this process already swept; pruning is session-scoped work
-_pruned_dirs: set[Path] = set()
-
-
-def _shared_yaml_name(request: pytest.FixtureRequest) -> str | None:
-    """Name passed to the shared_yaml marker, or None when unmarked."""
-    marker = request.node.get_closest_marker("shared_yaml")
-    if marker is None:
-        return None
-    # Exactly one \w+ positional arg: the name doubles as a build dir
-    # component, and CI test selection (script/helpers.py) parses the same shape
-    if (
-        len(marker.args) != 1
-        or marker.kwargs
-        or not re.fullmatch(r"\w+", str(marker.args[0]))
-    ):
-        raise ValueError(
-            "shared_yaml marker requires exactly one \\w+ fixture name literal"
-        )
-    return marker.args[0]
-
-
-def _shared_build_prefix(name: str) -> str:
-    return f"{name}-{_REPO_KEY}-"
-
-
-@cache
-def _shared_build_dir(name: str) -> Path:
-    """Dir keyed by checkout and fixture source, before per-test injections."""
-    key = hashlib.sha256((FIXTURES_DIR / f"{name}.yaml").read_bytes()).hexdigest()[:16]
-    return SHARED_BUILDS_ROOT / (_shared_build_prefix(name) + key)
-
-
-def _read_stamp(stamp: Path, shared_dir: Path) -> Path | None:
-    """ELF path recorded by the last completed compile, or None."""
-    try:
-        text = stamp.read_text(encoding="utf-8").strip()
-    except FileNotFoundError:
-        return None
-    except OSError as err:
-        print(f"Cannot read {stamp}: {err}")
-        return None
-    if not text:
-        print(f"Ignoring empty stamp {stamp}")
-        return None
-    built = Path(text)
-    # Never trust a stamp pointing outside its own build dir as an unlink target
-    if shared_dir.resolve() in built.resolve().parents:
-        return built
-    print(f"Ignoring stamp {stamp} pointing outside {shared_dir}")
-    return None
-
-
-def _unused_since(stale: Path, cutoff: float) -> bool:
-    """Whether a build dir looks untouched since cutoff; unknown counts as used."""
-    # Newest of the .built stamp (rewritten by every completed compile) and the
-    # dir itself (freshened by a worker claiming the dir before locking)
-    newest: float | None = None
-    for probe in (stale / ".built", stale):
-        try:
-            mtime = probe.stat().st_mtime
-        except FileNotFoundError:
-            continue
-        except NotADirectoryError:
-            return True  # a stray file where a dir should be; reclaimable
-        except OSError as err:
-            print(f"Cannot age-probe {stale}: {err}")
-            return False  # unknown never authorizes deletion
-        newest = mtime if newest is None else max(newest, mtime)
-    return newest is not None and newest < cutoff
-
-
-def _prune_stale_builds(name: str, keep: Path) -> None:
-    """Remove outdated build dirs (blocking, run in executor): this checkout's
-    other dirs for the fixture, plus anything untouched for 30 days. Tolerates
-    other workers pruning the same dirs concurrently."""
-    cutoff = time.time() - _STALE_BUILD_MAX_AGE_S
-    prefix = _shared_build_prefix(name)
-    for stale in SHARED_BUILDS_ROOT.iterdir():
-        if stale == keep:
-            continue
-        same_fixture = stale.name.startswith(prefix)
-        if not same_fixture and not _unused_since(stale, cutoff):
-            continue
-        # Creating .lock bumps the dir mtime, so remember whether the re-probe
-        # under the lock can trust it
-        lock_preexisting = (stale / ".lock").exists()
-        try:
-            lock_file = (stale / ".lock").open("w")
-        except FileNotFoundError:
-            continue  # pruned by another worker meanwhile
-        except NotADirectoryError:
-            print(f"Removing stray file {stale}")
-            stale.unlink(missing_ok=True)
-            continue
-        except OSError as err:
-            print(f"Cannot prune {stale}: {err}")
-            continue
-        with lock_file:
-            try:
-                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-            except BlockingIOError:
-                continue  # still in use by another run
-            # Re-probe under the lock: a worker freshens its dir before
-            # locking, so a just-claimed dir no longer looks unused. A dir
-            # whose .lock we just created cannot be held by anyone, and our
-            # own open bumped its mtime, so its pre-open probe stands
-            if (
-                lock_preexisting
-                and not same_fixture
-                and not _unused_since(stale, cutoff)
-            ):
-                continue
-            # rmtree tolerates races; a leftover partial tree only costs a
-            # rebuild, since the ELF is deleted before every compile
-            try:
-                rmtree(stale)
-            except OSError as err:
-                print(f"Failed to prune {stale}: {err}")
-
-
-async def _run_esphome_compile(
-    config_path: Path, cwd: Path, env: dict[str, str]
-) -> None:
-    """Run `esphome compile`, retrying up to 3 times on a segfault."""
-    max_retries = 3
-    for attempt in range(max_retries):
-        # Compile using subprocess, inheriting stdout/stderr to show progress
-        proc = await asyncio.create_subprocess_exec(
-            sys.executable,
-            "-m",
-            "esphome",
-            "compile",
-            str(config_path),
-            cwd=cwd,
-            stdout=None,  # Inherit stdout
-            stderr=None,  # Inherit stderr
-            stdin=asyncio.subprocess.DEVNULL,
-            # Start in a new process group to isolate signal handling
-            start_new_session=True,
-            env=env,
-            close_fds=False,
-        )
-        await proc.wait()
-
-        if proc.returncode == 0:
-            break
-        if proc.returncode == -11 and attempt < max_retries - 1:
-            # Segfault (-11 = SIGSEGV), retry
-            print(
-                f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
-            )
-            await asyncio.sleep(1)  # Brief pause before retry
-            continue
-        raise RuntimeError(
-            f"Failed to compile {config_path}, return code: {proc.returncode}. "
-            f"Run with 'pytest -s' to see compilation output."
-        )
-
-
-def _resolve_compiled_binary(config_path: Path) -> Path:
-    """Load the config to learn the compiled ELF path (blocking, run in executor)."""
-    CORE.reset()  # Reset CORE state between test runs
-    CORE.config_path = config_path
-    config = esphome.config.read_config(
-        {"command": "compile", "config": str(config_path)}
-    )
-    if config is None:
-        raise RuntimeError(f"Failed to read config from {config_path}")
-    idedata = get_idedata(config)
-    binary_path = Path(idedata.firmware_elf_path)
-    if not binary_path.exists():
-        raise RuntimeError(f"Compiled binary not found at {binary_path}")
-    return binary_path
-
-
 @pytest_asyncio.fixture
 async def compile_esphome(
     integration_test_dir: Path,
     shared_platformio_cache: Path,
-    request: pytest.FixtureRequest,
 ) -> AsyncGenerator[CompileFunction]:
     """Compile an ESPHome configuration and return the binary path."""
 
@@ -489,96 +258,66 @@ async def compile_esphome(
         # Use the shared PlatformIO cache for faster compilation
         # This avoids re-downloading dependencies for each test
         env = _get_platformio_env(shared_platformio_cache)
+
+        # Retry compilation up to 3 times if we get a segfault
+        max_retries = 3
+        for attempt in range(max_retries):
+            # Compile using subprocess, inheriting stdout/stderr to show progress
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable,
+                "-m",
+                "esphome",
+                "compile",
+                str(config_path),
+                cwd=integration_test_dir,
+                stdout=None,  # Inherit stdout
+                stderr=None,  # Inherit stderr
+                stdin=asyncio.subprocess.DEVNULL,
+                # Start in a new process group to isolate signal handling
+                start_new_session=True,
+                env=env,
+                close_fds=False,
+            )
+            await proc.wait()
+
+            if proc.returncode == 0:
+                # Success!
+                break
+            if proc.returncode == -11 and attempt < max_retries - 1:
+                # Segfault (-11 = SIGSEGV), retry
+                print(
+                    f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
+                )
+                await asyncio.sleep(1)  # Brief pause before retry
+                continue
+            # Other error or final retry
+            raise RuntimeError(
+                f"Failed to compile {config_path}, return code: {proc.returncode}. "
+                f"Run with 'pytest -s' to see compilation output."
+            )
+
+        # Load the config to get idedata (blocking call, must use executor)
         loop = asyncio.get_running_loop()
 
-        name = _shared_yaml_name(request)
-        if name is None:
-            await _run_esphome_compile(config_path, integration_test_dir, env)
-            return await loop.run_in_executor(
-                None, _resolve_compiled_binary, config_path
+        def _read_config_and_get_binary():
+            CORE.reset()  # Reset CORE state between test runs
+            CORE.config_path = config_path
+            config = esphome.config.read_config(
+                {"command": "compile", "config": str(config_path)}
             )
+            if config is None:
+                raise RuntimeError(f"Failed to read config from {config_path}")
 
-        # Shared fixture: build in a hash-keyed dir so tests sharing a config
-        # pay one full compile and later only a main.cpp (port) rebuild + relink
-        shared_dir = _shared_build_dir(name)
-        shared_dir.mkdir(parents=True, exist_ok=True)
-        # Freshen the dir before locking so a concurrent age sweep, which
-        # re-probes under the lock, never reaps a dir a worker just claimed;
-        # if a peer reaped it already, the guarded lock open recreates it
-        with suppress(FileNotFoundError):
-            os.utime(shared_dir)
-        if shared_dir not in _pruned_dirs:
-            _pruned_dirs.add(shared_dir)
-            await loop.run_in_executor(None, _prune_stale_builds, name, shared_dir)
-        shared_config = shared_dir / f"{name}.yaml"
-        private_binary = integration_test_dir / f"{name}.elf"
-        content = await loop.run_in_executor(None, read_file, config_path)
-        if content != getattr(request.node, "_shared_yaml_content", None):
-            # The dir is keyed by the fixture source; a mutated config would be
-            # cached under a hash that does not describe it
-            raise RuntimeError(
-                "shared_yaml tests must compile the yaml_config content unmodified"
-            )
-        # flock serializes concurrent xdist workers; closing the fd releases it.
-        # Hand-rolled rather than filelock.FileLock: non-blocking retries keep
-        # the wait cancellable, while a blocking acquire in an executor thread
-        # would survive test cancellation holding the fd
-        try:
-            lock_file = (shared_dir / ".lock").open("w")
-        except FileNotFoundError:
-            # A peer run pruning divergent hashes reaped the dir between our
-            # mkdir and this open; recreate it and pay a full rebuild
-            shared_dir.mkdir(parents=True, exist_ok=True)
-            lock_file = (shared_dir / ".lock").open("w")
-        with lock_file:
-            start = time.monotonic()
-            last_report = start
-            while True:
-                try:
-                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-                    break
-                except BlockingIOError:
-                    now = time.monotonic()
-                    if now - start > _SHARED_LOCK_TIMEOUT_S:
-                        raise RuntimeError(
-                            f"Timed out waiting for the {shared_dir} lock"
-                        ) from None
-                    if now - last_report >= _SHARED_LOCK_REPORT_S:
-                        last_report = now
-                        print(
-                            f"Waited {now - start:.0f}s for another worker's "
-                            f"build of {shared_dir.name}"
-                        )
-                    await asyncio.sleep(_SHARED_LOCK_POLL_S)
-            # .built carries the ELF path of the last completed compile, so
-            # later workers skip the config re-read in _resolve_compiled_binary
-            stamp = shared_dir / ".built"
-            if (built := _shared_elf_paths.get(shared_dir)) is None:
-                built = await loop.run_in_executor(None, _read_stamp, stamp, shared_dir)
-            # Delete the ELF before compiling: whatever exists afterwards is
-            # this compile's output, so no staleness check is ever needed.
-            # With no usable stamp, sweep any leftover at the known layout
-            if built is not None:
-                built.unlink(missing_ok=True)
-            else:
-                # Layout-agnostic: ESPHOME_BUILD_PATH can move the build tree
-                for leftover in shared_dir.rglob("program"):
-                    if leftover.is_file():
-                        leftover.unlink()
-            await loop.run_in_executor(
-                None, write_file_if_changed, shared_config, content
-            )
-            await _run_esphome_compile(shared_config, shared_dir, env)
-            if built is None or not built.exists():
-                built = await loop.run_in_executor(
-                    None, _resolve_compiled_binary, shared_config
-                )
-            _shared_elf_paths[shared_dir] = built
-            await loop.run_in_executor(None, write_file, stamp, str(built))
-            # Copy out before unlocking: another worker may relink firmware.elf
-            # while this test is still running its private copy
-            await loop.run_in_executor(None, shutil.copy2, built, private_binary)
-        return private_binary
+            # Get the compiled binary path
+            idedata = get_idedata(config)
+            return Path(idedata.firmware_elf_path)
+
+        binary_path = await loop.run_in_executor(None, _read_config_and_get_binary)
+
+        if not binary_path.exists():
+            raise RuntimeError(f"Compiled binary not found at {binary_path}")
+
+        return binary_path
 
     yield _compile
 

tests/integration/fixtures/sensor_filters_batch_window.yaml

@@ -0,0 +1,58 @@
+esphome:
+  name: test-batch-window-filters
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+# Template sensor that we'll use to publish values
+sensor:
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # Batch window filters (window_size == send_every) - use streaming filters
+  - platform: copy
+    source_id: source_sensor
+    name: "Min Sensor"
+    id: min_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Max Sensor"
+    id: max_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Moving Avg Sensor"
+    id: moving_avg_sensor
+    filters:
+      - sliding_window_moving_average:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+# Button to trigger publishing test values
+button:
+  - platform: template
+    name: "Publish Values Button"
+    id: publish_button
+    on_press:
+      - lambda: |-
+          // Publish 10 values: 1.0, 2.0, ..., 10.0
+          for (int i = 1; i <= 10; i++) {
+            id(source_sensor).publish_state(float(i));
+          }

tests/integration/fixtures/uart_mock_modbus_client_read_write.yaml

@@ -0,0 +1,111 @@
+esphome:
+  name: uart-mock-modbus-cli-rw
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+# Two virtual buses looped back to each other: the client's transmissions reach the server and the
+# server's replies reach the client. auto_start so forwarding is active before the button fires.
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_client
+              data: !lambda return data;
+  - id: virtual_uart_client
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: stored_1
+    type: uint16_t
+    initial_value: "0"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_client
+    id: virtual_modbus_client
+    role: client
+    turnaround_time: 10ms
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    registers:
+      # Writable + readable register: the read publishes what it returns, so the test can confirm the
+      # write half of the 0x17 ran before the read half (Modbus 6.17).
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: |-
+          id(srv_read_1).publish_state(id(stored_1));
+          return id(stored_1);
+        write_lambda: |-
+          id(stored_1) = x;
+          id(srv_write_1).publish_state(x);
+          return true;
+      # Read-only register, returned together with 0x01 by the 2-register read half.
+      - address: 0x02
+        value_type: U_WORD
+        read_lambda: return 0x00AA;
+
+sensor:
+  # Server-side observations.
+  - platform: template
+    name: "srv_write_1"
+    id: srv_write_1
+  - platform: template
+    name: "srv_read_1"
+    id: srv_read_1
+  # Client-side read-back: the values the client's on_response received.
+  - platform: template
+    name: "client_read_0"
+    id: client_read_0
+  - platform: template
+    name: "client_read_1"
+    id: client_read_1
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      # FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
+      - modbus_client.read_write_multiple_registers:
+          address: 0x01
+          read_address: 0x0001
+          read_count: 2
+          write_address: 0x0001
+          values: [0x1234]
+          on_response:
+            then:
+              - lambda: |-
+                  // values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
+                  if (values.size() >= 2) {
+                    id(client_read_0).publish_state(values[0]);
+                    id(client_read_1).publish_state(values[1]);
+                  }

tests/integration/fixtures/uart_mock_modbus_custom_pdu.yaml

@@ -0,0 +1,88 @@
+esphome:
+  name: uart-mock-modbus-custom-pdu
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 259;
+
+sensor:
+  # Plain read to confirm the controller <-> server link is up.
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "plain_read"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  # Custom PDU: read holding register 0x0001, count 1. The PDU is
+  # {function code, address hi, address lo, count hi, count lo}; the device
+  # address and CRC are added by the hub. The lambda parses the response payload
+  # (the register value, big-endian).
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "custom_read"
+    custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
+    lambda: |-
+      if (data.size() < 2) return {};
+      return (float) ((data[0] << 8) | data[1]);
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_deprecated_write_buffer.yaml

@@ -0,0 +1,106 @@
+esphome:
+  name: uart-mock-modbus-dep-buffer
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: reg10
+    type: uint16_t
+    initial_value: "0"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: |-
+          id(reg10) = x;
+          return true;
+
+# A number whose write_lambda uses the DEPRECATED buffer parameter (fills `payload` with a legacy raw
+# frame as words: device address + function code + data) instead of the new item->write_* API. The write
+# must still land with its legacy semantics, and the one-time deprecation warning must fire only once per
+# entity no matter how many writes happen.
+number:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "buf_number"
+    id: buf_number
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+    min_value: 0
+    max_value: 1000
+    step: 1
+    write_lambda: |-
+      // Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0010, value.
+      payload.push_back(0x0106);
+      payload.push_back(0x0010);
+      payload.push_back((uint16_t) x);
+      return {};
+
+# Reports the server-side register so the test can observe that the deprecated buffer write landed.
+sensor:
+  - platform: template
+    name: "written_value"
+    id: written_value
+    update_interval: 0.5s
+    lambda: "return id(reg10);"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # The test drives the writes via number_command; the mock is autostart.

tests/integration/fixtures/uart_mock_modbus_lambda_invert.yaml

@@ -0,0 +1,95 @@
+esphome:
+  name: uart-mock-modbus-lambda-invert
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: reg40
+    type: uint16_t
+    initial_value: "5"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x40
+        value_type: U_WORD
+        read_lambda: return id(reg40);
+        write_lambda: id(reg40) = x; return true;
+
+# An active-low holding switch: the write_lambda inverts the wire value, but the entity must still
+# report the REQUESTED state. assumed_state keeps the register unpolled, so the published state comes
+# only from write_state() - turning ON writes 0x0000 yet the switch shows ON.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "invert_switch"
+    register_type: holding
+    address: 0x40
+    assumed_state: true
+    write_lambda: |-
+      return !x;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_40"
+    address: 0x40
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_lambda_write.yaml

@@ -0,0 +1,97 @@
+esphome:
+  name: uart-mock-modbus-lambda-write
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: reg30
+    type: uint16_t
+    initial_value: "0"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x30
+        value_type: U_WORD
+        read_lambda: return id(reg30);
+        write_lambda: id(reg30) = x; return true;
+
+# A COIL-type switch (assumed_state, write-only) whose write_lambda ignores its own coil type and instead
+# drives a HOLDING-REGISTER write on the mock server through the entity itself: `item` IS the command, so
+# item->write_single_register() sends a register write from a coil entity (cross-type). Returning nothing
+# (an empty optional) tells the write path the lambda already dispatched the frame - no default coil write.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "cross_switch"
+    register_type: coil
+    address: 0x00
+    assumed_state: true
+    write_lambda: |-
+      item->write_single_register(0x30, x ? 1234 : 0);
+      return {};
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_30"
+    address: 0x30
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_loopback.yaml

@@ -1,233 +0,0 @@
-esphome:
-  name: uart-mock-modbus-loopback
-
-host:
-api:
-logger:
-  level: VERBOSE
-
-external_components:
-  - source:
-      type: local
-      path: EXTERNAL_COMPONENT_PATH
-
-# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
-# The actual UART bus used is the uart_mock component below
-uart:
-  baud_rate: 115200
-  port: /dev/null
-
-# Shared loopback fixture (see the shared_yaml markers in the test file);
-# register spaces are disjoint so each test only observes its own entities.
-uart_mock:
-  - id: virtual_uart_server
-    baud_rate: 9600
-    auto_start: true
-    debug:
-    on_tx:
-      - then:
-          - uart_mock.inject_rx:
-              id: virtual_uart_controller
-              data: !lambda return data;
-  - id: virtual_uart_controller
-    baud_rate: 9600
-    auto_start: true
-    debug:
-    on_tx:
-      - then:
-          - uart_mock.inject_rx:
-              id: virtual_uart_server
-              data: !lambda return data;
-
-globals:
-  - id: reg10
-    type: uint16_t
-    initial_value: "100"
-  - id: reg11
-    type: uint16_t
-    initial_value: "200"
-  - id: reg12
-    type: uint16_t
-    initial_value: "300"
-  - id: reg13
-    type: uint16_t
-    initial_value: "0xABCD"
-  - id: reg30
-    type: uint16_t
-    initial_value: "0"
-  - id: reg40
-    type: uint16_t
-    initial_value: "5"
-  - id: reg50
-    type: uint16_t
-    initial_value: "0"
-
-modbus:
-  - uart_id: virtual_uart_server
-    id: virtual_modbus_server
-    role: server
-  - uart_id: virtual_uart_controller
-    id: virtual_modbus_controller
-    role: client
-    turnaround_time: 10ms
-
-modbus_controller:
-  - address: 1
-    modbus_id: virtual_modbus_controller
-    id: modbus_controller_1
-    update_interval: 1s
-
-modbus_server:
-  - address: 1
-    modbus_id: virtual_modbus_server
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 259;
-      - address: 0x10
-        value_type: U_WORD
-        read_lambda: return id(reg10);
-        write_lambda: id(reg10) = x; return true;
-      - address: 0x11
-        value_type: U_WORD
-        read_lambda: return id(reg11);
-        write_lambda: id(reg11) = x; return true;
-      - address: 0x12
-        value_type: U_WORD
-        read_lambda: return id(reg12);
-        write_lambda: id(reg12) = x; return true;
-      - address: 0x13
-        value_type: U_WORD
-        read_lambda: return id(reg13);
-      - address: 0x30
-        value_type: U_WORD
-        read_lambda: return id(reg30);
-        write_lambda: id(reg30) = x; return true;
-      - address: 0x40
-        value_type: U_WORD
-        read_lambda: return id(reg40);
-        write_lambda: id(reg40) = x; return true;
-      - address: 0x50
-        value_type: U_WORD
-        read_lambda: return id(reg50);
-        write_lambda: id(reg50) = x; return true;
-
-# Byte-based offset: 2 bytes -> register 0x11 (the old code folded it in as a
-# register count, hitting 0x12). assumed_state keeps the switch write-only.
-switch:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "offset_switch"
-    register_type: holding
-    address: 0x10
-    offset: 2
-    assumed_state: true
-  # Reading switch, byte offset 6 -> register 0x13; the pre-fix resolution (0x16)
-  # would draw ILLEGAL_DATA_ADDRESS and never publish.
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "read_offset_switch"
-    register_type: holding
-    address: 0x10
-    offset: 6
-    bitmask: 0x1
-  # Coil switch whose write_lambda dispatches a holding-register write via `item`;
-  # returning an empty optional suppresses the default coil write.
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "cross_switch"
-    register_type: coil
-    address: 0x00
-    assumed_state: true
-    write_lambda: |-
-      item->write_single_register(0x30, x ? 1234 : 0);
-      return {};
-  # Active-low: the write_lambda inverts the wire value but the entity must still
-  # report the requested state (assumed_state keeps the register unpolled).
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "invert_switch"
-    register_type: holding
-    address: 0x40
-    assumed_state: true
-    write_lambda: |-
-      return !x;
-
-# Uses the deprecated buffer parameter (legacy raw frame as words); the write
-# must land and the deprecation warning must fire only once per entity.
-number:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "buf_number"
-    id: buf_number
-    address: 0x50
-    register_type: holding
-    value_type: U_WORD
-    min_value: 0
-    max_value: 1000
-    step: 1
-    write_lambda: |-
-      // Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0050, value.
-      payload.push_back(0x0106);
-      payload.push_back(0x0050);
-      payload.push_back((uint16_t) x);
-      return {};
-
-sensor:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "plain_read"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  # Custom PDU: read holding register 0x0001; device address and CRC are added
-  # by the hub. The lambda parses the big-endian register value.
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "custom_read"
-    custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
-    lambda: |-
-      if (data.size() < 2) return {};
-      return (float) ((data[0] << 8) | data[1]);
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_10"
-    address: 0x10
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_11"
-    address: 0x11
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_12"
-    address: 0x12
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_30"
-    address: 0x30
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_40"
-    address: 0x40
-    register_type: holding
-    value_type: U_WORD
-  # Reports the server-side register so the test can observe that the deprecated buffer write landed.
-  - platform: template
-    name: "written_value"
-    id: written_value
-    update_interval: 0.5s
-    lambda: "return id(reg50);"
-
-button:
-  - platform: template
-    name: "Start Scenario"
-    id: start_scenario_btn
-    # Nothing to start (mock is autostart); tests drive entities directly

tests/integration/fixtures/uart_mock_modbus_register_offset.yaml

@@ -0,0 +1,138 @@
+esphome:
+  name: uart-mock-modbus-reg-offset
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: reg10
+    type: uint16_t
+    initial_value: "100"
+  - id: reg11
+    type: uint16_t
+    initial_value: "200"
+  - id: reg12
+    type: uint16_t
+    initial_value: "300"
+  - id: reg13
+    type: uint16_t
+    initial_value: "0xABCD"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: id(reg10) = x; return true;
+      - address: 0x11
+        value_type: U_WORD
+        read_lambda: return id(reg11);
+        write_lambda: id(reg11) = x; return true;
+      - address: 0x12
+        value_type: U_WORD
+        read_lambda: return id(reg12);
+        write_lambda: id(reg12) = x; return true;
+      - address: 0x13
+        value_type: U_WORD
+        read_lambda: return id(reg13);
+        write_lambda: id(reg13) = x; return true;
+
+# A holding-register switch at 0x10 with a 2-BYTE offset. offset is byte-based, so the write must target
+# register 0x10 + 2/2 = 0x11. The old (pre-fix) behavior folded offset into the address as a register
+# count, hitting 0x12 instead. assumed_state keeps the switch write-only so it does not read any register.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "offset_switch"
+    register_type: holding
+    address: 0x10
+    offset: 2
+    assumed_state: true
+  # A holding-register switch that READS its state. Byte offset 6 -> register 0x10 + 6/2 = 0x13. Post-fix
+  # the switch itself resolves to 0x13 (the even byte offset folds into the address as whole registers) and
+  # joins the 0x10..0x13 range, so no separate 0x13 sensor is needed. Pre-fix the whole byte offset folds
+  # into the address (0x16), where the server answers ILLEGAL_DATA_ADDRESS and the switch never publishes.
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "read_offset_switch"
+    register_type: holding
+    address: 0x10
+    offset: 6
+    bitmask: 0x1
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_10"
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_11"
+    address: 0x11
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_12"
+    address: 0x12
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server.yaml

@@ -0,0 +1,124 @@
+esphome:
+  name: uart-mock-modbus-server-test
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_dev
+    baud_rate: 9600
+    rx_full_threshold: 120
+    rx_timeout: 2
+    auto_start: false
+    debug:
+    injections:
+      - delay: 100ms
+        inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] #  Read holding register 3 on device 1 (basic_read)
+      - delay: 100ms
+        #  Read holding register 7 on device 2
+        #  Reply from device 2
+        #  Read holding register 5 on device 1 (read_after_peer_response)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x02,
+            0x03,
+            0x02,
+            0x00,
+            0xF0,
+            0xFC,
+            0x00,
+            0x01,
+            0x03,
+            0x00,
+            0x05,
+            0x00,
+            0x01,
+            0x94,
+            0x0B,
+          ]
+      - delay: 100ms
+        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] #  Read holding register 7 on device 2, with no response
+      - delay: 100ms
+        #  Read holding register 7 on device 2, with no response
+        #  Read holding register A on device 1 (read_after_peer_timeout)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x01,
+            0x03,
+            0x00,
+            0x0A,
+            0x00,
+            0x01,
+            0xA4,
+            0x08,
+          ]
+
+modbus:
+  uart_id: virtual_uart_dev
+  role: server
+
+modbus_server:
+  - address: 1
+    registers:
+      - address: 0x03
+        value_type: U_WORD
+        read_lambda: |-
+          id(basic_read).publish_state(1);
+          return 1;
+      - address: 0x05
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_response).publish_state(1);
+          return 1;
+      - address: 0x0A
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_timeout).publish_state(1);
+          return 1;
+
+sensor:
+  - platform: template
+    name: "basic_read"
+    id: basic_read
+  - platform: template
+    name: "read_after_peer_response"
+    id: read_after_peer_response
+  - platform: template
+    name: "read_after_peer_timeout"
+    id: read_after_peer_timeout
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server_controller.yaml

@@ -0,0 +1,203 @@
+esphome:
+  name: uart-mock-modbus-server-contro
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 99;
+      - address: 0x02
+        value_type: U_WORD_S
+        read_lambda: return 4660;
+      - address: 0x03
+        value_type: S_WORD
+        read_lambda: return -99;
+      - address: 0x04
+        value_type: S_WORD_S
+        read_lambda: return -2;
+      - address: 0x05
+        value_type: U_DWORD
+        read_lambda: return 16909060;
+      - address: 0x08
+        value_type: S_DWORD
+        read_lambda: return -16909060;
+      - address: 0x0B
+        value_type: U_DWORD_R
+        read_lambda: return 67305985;
+      - address: 0x0E
+        value_type: S_DWORD_R
+        read_lambda: return -67305985;
+      - address: 0x11
+        value_type: U_QWORD
+        read_lambda: return 72623859790382856;
+      - address: 0x16
+        value_type: S_QWORD
+        read_lambda: return -72623859790382856;
+      - address: 0x1B
+        value_type: U_QWORD_R
+        read_lambda: return 578437695752307201;
+      - address: 0x20
+        value_type: S_QWORD_R
+        read_lambda: return -578437695752307201;
+      - address: 0x25
+        value_type: FP32
+        read_lambda: return 3.14;
+      - address: 0x28
+        value_type: FP32_R
+        read_lambda: return 3.14;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word_s"
+    address: 0x02
+    register_type: holding
+    value_type: U_WORD_S
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word_s_raw"
+    address: 0x02
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_word"
+    address: 0x03
+    register_type: holding
+    value_type: S_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_word_s"
+    address: 0x04
+    register_type: holding
+    value_type: S_WORD_S
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword"
+    address: 0x05
+    register_type: holding
+    value_type: U_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword"
+    address: 0x08
+    register_type: holding
+    value_type: S_DWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_dword_r"
+    address: 0x0B
+    register_type: holding
+    value_type: U_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_dword_r"
+    address: 0x0E
+    register_type: holding
+    value_type: S_DWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword"
+    address: 0x11
+    register_type: holding
+    value_type: U_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword"
+    address: 0x16
+    register_type: holding
+    value_type: S_QWORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_qword_r"
+    address: 0x1B
+    register_type: holding
+    value_type: U_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_s_qword_r"
+    address: 0x20
+    register_type: holding
+    value_type: S_QWORD_R
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32"
+    address: 0x25
+    register_type: holding
+    value_type: FP32
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_fp32_r"
+    address: 0x28
+    register_type: holding
+    value_type: FP32_R
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_bits.yaml

@@ -0,0 +1,147 @@
+esphome:
+  name: uart-mock-modbus-srv-bits
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: stored_bit_2
+    type: bool
+    initial_value: "false"
+  - id: stored_bit_3
+    type: bool
+    initial_value: "true"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    update_interval: 1s
+    id: modbus_controller_1
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    bits:
+      - address: 0x00
+        read_lambda: return true;
+      - address: 0x01
+        read_lambda: return false;
+      - address: 0x02
+        read_lambda: return id(stored_bit_2);
+        write_lambda: id(stored_bit_2) = x; return true;
+      - address: 0x03
+        read_lambda: return id(stored_bit_3);
+        write_lambda: id(stored_bit_3) = x; return true;
+
+# The same four bits are read both as coils (FC 0x01) and as discrete inputs
+# (FC 0x02): the server serves both from one shared bit table, so the two
+# views must always agree.
+binary_sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_coil_0"
+    address: 0x00
+    register_type: coil
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_coil_1"
+    address: 0x01
+    register_type: coil
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_coil_2"
+    address: 0x02
+    register_type: coil
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_coil_3"
+    address: 0x03
+    register_type: coil
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_di_0"
+    address: 0x00
+    register_type: discrete_input
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_di_1"
+    address: 0x01
+    register_type: discrete_input
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_di_2"
+    address: 0x02
+    register_type: discrete_input
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "bit_di_3"
+    address: 0x03
+    register_type: discrete_input
+
+# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
+# multiple-coils write (FC 0x0F) so both server write paths are exercised.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_bit_2"
+    address: 0x02
+    register_type: coil
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "write_bit_3"
+    address: 0x03
+    register_type: coil
+    use_write_multiple: true
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -0,0 +1,116 @@
+esphome:
+  name: uart-mock-modbus-server-mult
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+  - id: virtual_uart_server_2
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_server_2
+    id: virtual_modbus_server_2
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_client
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_1
+  - address: 2
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_2
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 919;
+  - address: 2
+    modbus_id: virtual_modbus_server_2
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 929;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_2
+    name: "reg_u_word_2"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_write.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-mesh
+  name: uart-mock-modbus-srv-write
 
 host:
 api:
@@ -17,14 +17,13 @@ uart:
   baud_rate: 115200
   port: /dev/null
 
-# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed registers
-# backed by writable globals, addr 5 = the read/write 0x17 target, addr 2/3/6
-# on the second server hub. auto_start everywhere: the controller polls at
-# boot, so the forwarding must already be live or early requests generate warnings.
-# Every test presses Start Scenario, so all merged actions fire in every test.
 uart_mock:
   - id: virtual_uart_server
     baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
     auto_start: true
     debug:
     on_tx:
@@ -32,120 +31,79 @@ uart_mock:
           - uart_mock.inject_rx:
               id: virtual_uart_controller
               data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_server_2
-              data: !lambda return data;
-  - id: virtual_uart_server_2
-    baud_rate: 9600
-    auto_start: true
-    debug:
-    on_tx:
-      - then:
-          - uart_mock.inject_rx:
-              id: virtual_uart_server
-              data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_controller
-              data: !lambda return data;
   - id: virtual_uart_controller
     baud_rate: 9600
-    auto_start: true
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
           - uart_mock.inject_rx:
               id: virtual_uart_server
               data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_server_2
-              data: !lambda return data;
 
 globals:
-  - id: stored_1
-    type: uint16_t
-    initial_value: "0"
   - id: stored_u_word
     type: uint16_t
-    initial_value: "99"
+    initial_value: "11"
   - id: stored_u_word_s
     type: uint16_t
     initial_value: "4660"
   - id: stored_s_word
     type: int16_t
-    initial_value: "-99"
+    initial_value: "-11"
   - id: stored_s_word_s
     type: int16_t
     initial_value: "-2"
   - id: stored_u_dword
     type: uint32_t
-    initial_value: "16909060"
+    initial_value: "1001"
   - id: stored_s_dword
     type: int32_t
-    initial_value: "-16909060"
+    initial_value: "-1001"
   - id: stored_u_dword_r
     type: uint32_t
-    initial_value: "67305985"
+    initial_value: "3003"
   - id: stored_s_dword_r
     type: int32_t
-    initial_value: "-67305985"
+    initial_value: "-3003"
   - id: stored_u_qword
     type: uint64_t
-    initial_value: "72623859790382856"
+    initial_value: "5005"
   - id: stored_s_qword
     type: int64_t
-    initial_value: "-72623859790382856"
+    initial_value: "-5005"
   - id: stored_u_qword_r
     type: uint64_t
-    initial_value: "578437695752307201"
+    initial_value: "7007"
   - id: stored_s_qword_r
     type: int64_t
-    initial_value: "-578437695752307201"
+    initial_value: "-7007"
   - id: stored_fp32
     type: float
-    initial_value: "3.14"
+    initial_value: "1.5"
   - id: stored_fp32_r
     type: float
     initial_value: "2.5"
-  - id: stored_bit_2
-    type: bool
-    initial_value: "false"
-  - id: stored_bit_3
-    type: bool
-    initial_value: "true"
 
 modbus:
   - uart_id: virtual_uart_server
     id: virtual_modbus_server
     role: server
-  - uart_id: virtual_uart_server_2
-    id: virtual_modbus_server_2
-    role: server
   - uart_id: virtual_uart_controller
-    id: virtual_modbus_client
+    id: virtual_modbus_controller
     role: client
     turnaround_time: 10ms
 
 modbus_controller:
   - address: 1
-    modbus_id: virtual_modbus_client
+    modbus_id: virtual_modbus_controller
+    update_interval: 2s
     id: modbus_controller_1
-    update_interval: 1s
-  - address: 2
-    modbus_id: virtual_modbus_client
-    id: modbus_controller_2
-    update_interval: 1s
-  - address: 3
-    modbus_id: virtual_modbus_client
-    id: modbus_controller_3
-    update_interval: 1s
-  - address: 6
-    modbus_id: virtual_modbus_client
-    id: modbus_controller_6
-    update_interval: 1s
 
 modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
+    id: modbus_server_1
     registers:
       - address: 0x01
         value_type: U_WORD
@@ -203,47 +161,6 @@ modbus_server:
         value_type: FP32_R
         read_lambda: return id(stored_fp32_r);
         write_lambda: id(stored_fp32_r) = x; return true;
-  - address: 5
-    modbus_id: virtual_modbus_server
-    registers:
-      # Writable + readable register: srv_write_1 plus the client's read-back
-      # confirm the write half of the 0x17 ran before the read half (Modbus 6.17).
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return id(stored_1);
-        write_lambda: |-
-          id(stored_1) = x;
-          id(srv_write_1).publish_state(x);
-          return true;
-      # Read-only register, returned together with 0x01 by the 2-register read half.
-      - address: 0x02
-        value_type: U_WORD
-        read_lambda: return 0x00AA;
-  - address: 2
-    modbus_id: virtual_modbus_server_2
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 919;
-  - address: 3
-    modbus_id: virtual_modbus_server_2
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 929;
-  - address: 6
-    modbus_id: virtual_modbus_server_2
-    bits:
-      - address: 0x00
-        read_lambda: return true;
-      - address: 0x01
-        read_lambda: return false;
-      - address: 0x02
-        read_lambda: return id(stored_bit_2);
-        write_lambda: id(stored_bit_2) = x; return true;
-      - address: 0x03
-        read_lambda: return id(stored_bit_3);
-        write_lambda: id(stored_bit_3) = x; return true;
 
 sensor:
   - platform: modbus_controller
@@ -258,12 +175,6 @@ sensor:
     address: 0x02
     register_type: holding
     value_type: U_WORD_S
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_u_word_s_raw"
-    address: 0x02
-    register_type: holding
-    value_type: U_WORD
   - platform: modbus_controller
     modbus_controller_id: modbus_controller_1
     name: "reg_s_word"
@@ -336,31 +247,7 @@ sensor:
     address: 0x28
     register_type: holding
     value_type: FP32_R
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_2
-    name: "multi_reg_a"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_3
-    name: "multi_reg_b"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  # client_read_write observations, server- and client-side.
-  - platform: template
-    name: "srv_write_1"
-    id: srv_write_1
-  - platform: template
-    name: "client_read_0"
-    id: client_read_0
-  - platform: template
-    name: "client_read_1"
-    id: client_read_1
 
-# The number schema caps min/max at 16777215 (float32 integer precision), so
-# the large dword/qword baselines cannot be written back through these numbers.
 number:
   - platform: modbus_controller
     modbus_controller_id: modbus_controller_1
@@ -477,82 +364,8 @@ number:
     max_value: 16777215
     step: 0.01
 
-# The four bits are read both as coils (FC 0x01) and discrete inputs (FC 0x02);
-# the server serves both from one shared table, so the two views must agree.
-binary_sensor:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_coil_0"
-    address: 0x00
-    register_type: coil
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_coil_1"
-    address: 0x01
-    register_type: coil
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_coil_2"
-    address: 0x02
-    register_type: coil
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_coil_3"
-    address: 0x03
-    register_type: coil
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_di_0"
-    address: 0x00
-    register_type: discrete_input
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_di_1"
-    address: 0x01
-    register_type: discrete_input
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_di_2"
-    address: 0x02
-    register_type: discrete_input
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "bit_di_3"
-    address: 0x03
-    register_type: discrete_input
-
-# write_bit_2 uses the single-coil write (FC 0x05); write_bit_3 opts into the
-# multiple-coils write (FC 0x0F) so both server write paths are exercised.
-switch:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "write_bit_2"
-    address: 0x02
-    register_type: coil
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_6
-    name: "write_bit_3"
-    address: 0x03
-    register_type: coil
-    use_write_multiple: true
-
 button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    on_press:
-      # FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
-      - modbus_client.read_write_multiple_registers:
-          address: 5
-          read_address: 0x0001
-          read_count: 2
-          write_address: 0x0001
-          values: [0x1234]
-          on_response:
-            then:
-              - lambda: |-
-                  // values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
-                  if (values.size() >= 2) {
-                    id(client_read_0).publish_state(values[0]);
-                    id(client_read_1).publish_state(values[1]);
-                  }
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_read_write.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-srv-injected
+  name: uart-mock-modbus-srv-rw
 
 host:
 api:
@@ -17,8 +17,6 @@ uart:
   baud_rate: 115200
   port: /dev/null
 
-# Shared server-role fixture (see the shared_yaml markers in the test file);
-# the injections concatenate and each test waits only on its own sensors.
 uart_mock:
   - id: virtual_uart_dev
     baud_rate: 9600
@@ -27,31 +25,18 @@ uart_mock:
     auto_start: false
     debug:
     injections:
-      - delay: 100ms
-        inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] #  Read holding register 3 on device 1 (basic_read)
-      - delay: 100ms
-        #  Read holding register 7 on device 2, its reply, then read holding
-        #  register 5 on device 1 (read_after_peer_response)
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
-                    0x02, 0x03, 0x02, 0x00, 0xF0, 0xFC,
-                    0x00, 0x01, 0x03, 0x00, 0x05, 0x00, 0x01, 0x94, 0x0B]
-      - delay: 100ms
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] #  Read holding register 7 on device 2, with no response
-      - delay: 100ms
-        #  Read holding register 7 on device 2 with no response, then read
-        #  holding register A on device 1 (read_after_peer_timeout)
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
-                    0x01, 0x03, 0x00, 0x0A, 0x00, 0x01, 0xA4, 0x08]
-      # FC 0x17 on device 1: write reg 0x0001 = 0x1234 then read 0x0001..0x0002;
-      # per Modbus 6.17 the write runs first, so 0x0001 must read back 0x1234.
+      # FC 0x17 Read/Write Multiple Registers on device 1:
+      #   write reg 0x0001 = 0x1234 (qty 1), then read regs 0x0001..0x0002 (qty 2).
+      # Per Modbus 6.17 the write is performed before the read, so reg 0x0001 must
+      # read back the just-written 0x1234 in the same request.
       - delay: 100ms
         inject_rx:
           [0x01, 0x17, 0x00, 0x01, 0x00, 0x02, 0x00, 0x01, 0x00, 0x01, 0x02, 0x12, 0x34, 0x49, 0xD8]
-      # FC 0x17: write reg 0x0006 = 0x5678 (qty 1), then read reg 0x0006 (qty 1) -
+      # FC 0x17: write reg 0x0003 = 0x5678 (qty 1), then read reg 0x0003 (qty 1) -
       # a write and read targeting a different register block.
       - delay: 100ms
         inject_rx:
-          [0x01, 0x17, 0x00, 0x06, 0x00, 0x01, 0x00, 0x06, 0x00, 0x01, 0x02, 0x56, 0x78, 0x8B, 0x55]
+          [0x01, 0x17, 0x00, 0x03, 0x00, 0x01, 0x00, 0x03, 0x00, 0x01, 0x02, 0x56, 0x78, 0x9B, 0x10]
 
 globals:
   - id: stored_1
@@ -85,18 +70,8 @@ modbus_server:
         read_lambda: |-
           id(rw_read_2).publish_state(0x00AA);
           return 0x00AA;
+      # Second writable + readable register, targeted by the second request.
       - address: 0x03
-        value_type: U_WORD
-        read_lambda: |-
-          id(basic_read).publish_state(1);
-          return 1;
-      - address: 0x05
-        value_type: U_WORD
-        read_lambda: |-
-          id(read_after_peer_response).publish_state(1);
-          return 1;
-      # Second writable + readable register, targeted by the second FC 0x17 request.
-      - address: 0x06
         value_type: U_WORD
         read_lambda: |-
           id(rw_read_3).publish_state(id(stored_3));
@@ -105,22 +80,8 @@ modbus_server:
           id(stored_3) = x;
           id(rw_write_3).publish_state(x);
           return true;
-      - address: 0x0A
-        value_type: U_WORD
-        read_lambda: |-
-          id(read_after_peer_timeout).publish_state(1);
-          return 1;
 
 sensor:
-  - platform: template
-    name: "basic_read"
-    id: basic_read
-  - platform: template
-    name: "read_after_peer_response"
-    id: read_after_peer_response
-  - platform: template
-    name: "read_after_peer_timeout"
-    id: read_after_peer_timeout
   - platform: template
     name: "rw_write_1"
     id: rw_write_1

tests/integration/host_prefs.py

@@ -1,7 +1,7 @@
 """Helpers for manipulating the host platform's preferences file.
 
 ESPHome's host platform stores preferences in
-``$ESPHOME_PREFDIR/<app_name>.prefs`` using a simple binary layout that
+``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
 mirrors ``HostPreferences::sync()``:
 ``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
 
@@ -11,21 +11,13 @@ boot (e.g. forcing safe mode) or to clear stale state between runs.
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 import struct
 
 
 def host_prefs_path(device_name: str) -> Path:
-    """Return the on-disk prefs file path for a host-platform device.
-
-    Requires ESPHOME_PREFDIR, which the autouse isolated_preferences fixture
-    sets; refusing the ~/.esphome/prefs fallback keeps tests off real user
-    data if the fixture is ever bypassed."""
-    prefdir = os.environ.get("ESPHOME_PREFDIR")
-    if not prefdir:
-        raise RuntimeError("ESPHOME_PREFDIR is not set; refusing the real prefs dir")
-    return Path(prefdir) / f"{device_name}.prefs"
+    """Return the on-disk prefs file path for a host-platform device."""
+    return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
 
 
 def clear_host_prefs(device_name: str) -> None:

tests/integration/test_api_zero_psk_provisioning.py

@@ -24,6 +24,12 @@ NEW_KEY = base64.b64encode(b"n" * 32)
 KEY_ACTIVATION_DELAY = 0.5
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    """Keep host preferences per-test so every run starts unprovisioned."""
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+
+
 @pytest.mark.asyncio
 async def test_api_zero_psk_provisioning(
     yaml_config: str,

tests/integration/test_host_preferences_suspend_resume.py

@@ -41,6 +41,15 @@ async def _poll_until_exists(path: Path) -> None:
         await asyncio.sleep(0.05)
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Path:
+    """Keep host preferences per-test so this test never touches the real
+    ~/.esphome/prefs and never races other tests over ESPHOME_PREFDIR."""
+    prefdir = tmp_path / "prefs"
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
+    return prefdir / f"{DEVICE_NAME}.prefs"
+
+
 @pytest.mark.asyncio
 async def test_host_preferences_suspend_resume(
     yaml_config: str,
@@ -49,7 +58,7 @@ async def test_host_preferences_suspend_resume(
     isolated_preferences: Path,
 ) -> None:
     """Test that a running syncer flushes, a suspended one doesn't, and resume restores flushing."""
-    pref_file = isolated_preferences / f"{DEVICE_NAME}.prefs"
+    pref_file = isolated_preferences
 
     loop = asyncio.get_running_loop()
     saved_in_memory = loop.create_future()

tests/integration/test_light_initial_state.py

@@ -11,6 +11,14 @@ from .state_utils import InitialStateHelper, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    """Keep host preferences per-test so RESTORE_AND_ON never loads a stale value left
+    behind by a previous run (host preferences otherwise persist to ~/.esphome/prefs,
+    keyed only by device name)."""
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+
+
 @pytest.mark.asyncio
 async def test_light_initial_state(
     yaml_config: str,

tests/integration/test_uart_mock_modbus.py

@@ -19,40 +19,23 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Callable
+from dataclasses import dataclass
 
 from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
 import pytest
 
-from .state_utils import SensorTracker, find_entity, require_entity, wait_for_state
+from .state_utils import SensorTracker, find_entity, wait_for_state
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
-def _swap16(value: int) -> int:
-    """Byte-swapped view of a 16-bit register as the raw U_WORD wire value."""
-    return ((value & 0xFF) << 8) | (value >> 8)
+@dataclass
+class RegisterTestCase:
+    """Test parameters for a single modbus register write/read round-trip."""
 
-
-# Raw U_WORD view of reg_u_word_s's initial 0x1234
-MESH_RAW_U_WORD_S = _swap16(4660)
-
-# Initial values of the mesh fixture's address 1 registers; the
-# server_controller test reads them and the write test uses them as baseline.
-MESH_INITIAL_VALUES: dict[str, object] = {
-    "reg_u_word": 99,
-    "reg_u_word_s": 4660,
-    "reg_s_word": -99,
-    "reg_s_word_s": -2,
-    "reg_u_dword": 16909060,
-    "reg_s_dword": -16909060,
-    "reg_u_dword_r": pytest.approx(67305985),
-    "reg_s_dword_r": pytest.approx(-67305985),
-    "reg_u_qword": pytest.approx(72623859790382856),
-    "reg_s_qword": pytest.approx(-72623859790382856),
-    "reg_u_qword_r": pytest.approx(578437695752307201),
-    "reg_s_qword_r": pytest.approx(-578437695752307201),
-    "reg_fp32": pytest.approx(3.14),
-    "reg_fp32_r": pytest.approx(2.5),
-}
+    initial_value: object
+    write_number_name: str
+    write_value: float
+    post_write_value: object
 
 
 # ---------------------------------------------------------------------------
@@ -190,7 +173,6 @@ async def test_uart_mock_modbus_no_threshold(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server(
     yaml_config: str,
@@ -221,7 +203,6 @@ async def test_uart_mock_modbus_server(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_read_write(
     yaml_config: str,
@@ -250,8 +231,8 @@ async def test_uart_mock_modbus_server_read_write(
             "rw_write_1": 4660,  # 0x1234 written to reg 0x0001
             "rw_read_1": 4660,  # reg 0x0001 reads back the just-written value
             "rw_read_2": 170,  # 0x00AA read from reg 0x0002 in the same request
-            "rw_write_3": 22136,  # 0x5678 written to reg 0x0006
-            "rw_read_3": 22136,  # reg 0x0006 reads back the just-written value
+            "rw_write_3": 22136,  # 0x5678 written to reg 0x0003
+            "rw_read_3": 22136,  # reg 0x0003 reads back the just-written value
         }
     )
 
@@ -260,8 +241,7 @@ async def test_uart_mock_modbus_server_read_write(
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
-        # The FC 0x17 injections fire last, behind four earlier 100ms delays
-        await tracker.await_all(futures, timeout=4.0)
+        await tracker.await_all(futures)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
@@ -316,7 +296,6 @@ async def test_uart_mock_modbus_server_read_write_invalid(
     )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller(
     yaml_config: str,
@@ -327,7 +306,23 @@ async def test_uart_mock_modbus_server_controller(
 
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    expected_values = MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
+    expected_values = {
+        "reg_u_word": 99,
+        "reg_u_word_s": 4660,
+        "reg_u_word_s_raw": 13330,
+        "reg_s_word": -99,
+        "reg_s_word_s": -2,
+        "reg_u_dword": 16909060,
+        "reg_s_dword": -16909060,
+        "reg_u_dword_r": pytest.approx(67305985),
+        "reg_s_dword_r": pytest.approx(-67305985),
+        "reg_u_qword": pytest.approx(72623859790382856),
+        "reg_s_qword": pytest.approx(-72623859790382856),
+        "reg_u_qword_r": pytest.approx(578437695752307201),
+        "reg_s_qword_r": pytest.approx(-578437695752307201),
+        "reg_fp32": pytest.approx(3.14),
+        "reg_fp32_r": pytest.approx(3.14),
+    }
     tracker = SensorTracker(list(expected_values.keys()))
     futures = tracker.expect_all(expected_values)
 
@@ -335,12 +330,14 @@ async def test_uart_mock_modbus_server_controller(
         run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
+        # The controller polls from boot, so the first values can already be in
+        # the states the device sends on connect; matching them there saves
+        # waiting for the next poll
         await tracker.setup_and_start_scenario(client, match_initial_states=True)
         await tracker.await_all(futures)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller_write(
     yaml_config: str,
@@ -356,47 +353,51 @@ async def test_uart_mock_modbus_server_controller_write(
 
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    # Per read-back sensor: the number entity to write through and the value;
-    # floats read back within tolerance, everything else exactly
-    register_writes: dict[str, tuple[str, int | float]] = {
-        "reg_u_word": ("write_u_word", 42),
-        "reg_u_word_s": ("write_u_word_s", 17185),
-        "reg_s_word": ("write_s_word", -42),
-        "reg_s_word_s": ("write_s_word_s", -257),
-        "reg_u_dword": ("write_u_dword", 2002),
-        "reg_s_dword": ("write_s_dword", -2002),
-        "reg_u_dword_r": ("write_u_dword_r", 4004),
-        "reg_s_dword_r": ("write_s_dword_r", -4004),
-        "reg_u_qword": ("write_u_qword", 6006),
-        "reg_s_qword": ("write_s_qword", -6006),
-        "reg_u_qword_r": ("write_u_qword_r", 8008),
-        "reg_s_qword_r": ("write_s_qword_r", -8008),
-        "reg_fp32": ("write_fp32", 6.28),
-        "reg_fp32_r": ("write_fp32_r", 9.42),
+    register_test_cases: dict[str, RegisterTestCase] = {
+        "reg_u_word": RegisterTestCase(11, "write_u_word", 42, 42),
+        "reg_u_word_s": RegisterTestCase(4660, "write_u_word_s", 17185, 17185),
+        "reg_s_word": RegisterTestCase(-11, "write_s_word", -42, -42),
+        "reg_s_word_s": RegisterTestCase(-2, "write_s_word_s", -257, -257),
+        "reg_u_dword": RegisterTestCase(1001, "write_u_dword", 2002, 2002),
+        "reg_s_dword": RegisterTestCase(-1001, "write_s_dword", -2002, -2002),
+        "reg_u_dword_r": RegisterTestCase(3003, "write_u_dword_r", 4004, 4004),
+        "reg_s_dword_r": RegisterTestCase(-3003, "write_s_dword_r", -4004, -4004),
+        "reg_u_qword": RegisterTestCase(5005, "write_u_qword", 6006, 6006),
+        "reg_s_qword": RegisterTestCase(-5005, "write_s_qword", -6006, -6006),
+        "reg_u_qword_r": RegisterTestCase(7007, "write_u_qword_r", 8008, 8008),
+        "reg_s_qword_r": RegisterTestCase(-7007, "write_s_qword_r", -8008, -8008),
+        "reg_fp32": RegisterTestCase(
+            pytest.approx(1.5, abs=0.01),
+            "write_fp32",
+            3.14,
+            pytest.approx(3.14, abs=0.01),
+        ),
+        "reg_fp32_r": RegisterTestCase(
+            pytest.approx(2.5, abs=0.01),
+            "write_fp32_r",
+            6.28,
+            pytest.approx(6.28, abs=0.01),
+        ),
     }
 
-    tracker = SensorTracker([*register_writes, "reg_u_word_s_raw"])
+    tracker = SensorTracker(list(register_test_cases.keys()))
 
-    # The raw U_WORD view of 0x02 pins the byte swap on the write path: the
-    # round trip through write_u_word_s applies the swap an even number of
-    # times, so only the raw sensor can catch a symmetrically dropped swap.
     # Phase 1: expect initial baseline values
     initial_futures = tracker.expect_all(
-        MESH_INITIAL_VALUES | {"reg_u_word_s_raw": MESH_RAW_U_WORD_S}
+        {name: case.initial_value for name, case in register_test_cases.items()}
     )
     # Phase 2: expect post-write values (registered now so on_state can match them)
     written_futures = tracker.expect_all(
-        {
-            name: pytest.approx(value, abs=0.01) if isinstance(value, float) else value
-            for name, (_, value) in register_writes.items()
-        }
-        | {"reg_u_word_s_raw": _swap16(register_writes["reg_u_word_s"][1])}
+        {name: case.post_write_value for name, case in register_test_cases.items()}
     )
 
     async with (
         run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
+        # The controller polls from boot, so the baseline can already be in the
+        # states the device sends on connect; matching it there saves waiting for
+        # the next poll
         entities = await tracker.setup_and_start_scenario(
             client, match_initial_states=True
         )
@@ -405,22 +406,19 @@ async def test_uart_mock_modbus_server_controller_write(
         # connection is working before issuing writes
         await tracker.await_all(initial_futures, timeout=4.0)
 
-        # Issue write commands for all register types; exact object_id match,
-        # since several write_* names are prefixes of a sibling
-        numbers = {
-            e.object_id.lower(): e for e in entities if isinstance(e, NumberInfo)
-        }
-        for number_name, value in register_writes.values():
-            entity = numbers.get(number_name)
-            assert entity is not None, f"{number_name} number entity not found"
-            client.number_command(entity.key, value)
+        # Issue write commands for all register types
+        for case in register_test_cases.values():
+            entity = find_entity(entities, case.write_number_name, NumberInfo)
+            assert entity is not None, (
+                f"{case.write_number_name} number entity not found"
+            )
+            client.number_command(entity.key, case.write_value)
 
         # Wait for sensors to reflect the written values (round-trip write+read)
         await tracker.await_all(written_futures, timeout=4.0)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller_bits(
     yaml_config: str,
@@ -466,6 +464,8 @@ async def test_uart_mock_modbus_server_controller_bits(
         run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
+        # The controller polls from boot and binary sensors drop repeats, so the
+        # baseline can arrive only in the states the device sends on connect
         entities = await tracker.setup_and_start_scenario(
             client, match_initial_states=True
         )
@@ -476,7 +476,8 @@ async def test_uart_mock_modbus_server_controller_bits(
 
         # Flip both writable bits: 0x02 false -> true, 0x03 true -> false
         for switch_name, value in (("write_bit_2", True), ("write_bit_3", False)):
-            entity = require_entity(entities, switch_name, SwitchInfo)
+            entity = find_entity(entities, switch_name, SwitchInfo)
+            assert entity is not None, f"{switch_name} switch entity not found"
             client.switch_command(entity.key, value)
 
         # Wait for both read views to reflect the written values
@@ -484,7 +485,6 @@ async def test_uart_mock_modbus_server_controller_bits(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller_multiple(
     yaml_config: str,
@@ -495,7 +495,7 @@ async def test_uart_mock_modbus_server_controller_multiple(
 
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    expected_values = {"multi_reg_a": 919, "multi_reg_b": 929}
+    expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
     tracker = SensorTracker(list(expected_values.keys()))
     futures = tracker.expect_all(expected_values)
 
@@ -503,6 +503,9 @@ async def test_uart_mock_modbus_server_controller_multiple(
         run_compiled(yaml_config, line_callback=line_callback),
         api_client_connected() as client,
     ):
+        # The controller polls from boot, so the first values can already be in
+        # the states the device sends on connect; matching them there saves
+        # waiting for the next poll
         await tracker.setup_and_start_scenario(client, match_initial_states=True)
         await tracker.await_all(futures)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
@@ -703,7 +706,6 @@ async def test_uart_mock_modbus_shared_address(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_custom_pdu(
     yaml_config: str,
@@ -930,7 +932,6 @@ async def test_uart_mock_modbus_broadcast_write(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_client_read_write(
     yaml_config: str,
@@ -946,7 +947,9 @@ async def test_uart_mock_modbus_client_read_write(
     """
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    tracker = SensorTracker(["srv_write_1", "client_read_0", "client_read_1"])
+    tracker = SensorTracker(
+        ["srv_write_1", "srv_read_1", "client_read_0", "client_read_1"]
+    )
     futures = tracker.expect_all(
         {
             "srv_write_1": 4660,  # server wrote 0x1234 to reg 0x0001
@@ -964,7 +967,6 @@ async def test_uart_mock_modbus_client_read_write(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_register_offset(
     yaml_config: str,
@@ -1020,7 +1022,6 @@ async def test_uart_mock_modbus_register_offset(
         )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_lambda_write(
     yaml_config: str,
@@ -1057,7 +1058,6 @@ async def test_uart_mock_modbus_lambda_write(
         await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_lambda_invert(
     yaml_config: str,
@@ -1113,7 +1113,6 @@ async def test_uart_mock_modbus_lambda_invert(
         )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_deprecated_write_buffer(
     yaml_config: str,

tests/script/test_helpers.py

@@ -2122,34 +2122,6 @@ def test_get_cpp_changed_components_independent_of_cwd(
     ) == ["time"]
 
 
-def test_fixture_map_includes_shared_yaml_markers() -> None:
-    """Fixtures named only by shared_yaml markers must map to their test file."""
-    helpers.get_fixture_to_test_files.cache_clear()
-    mapping = helpers.get_fixture_to_test_files()
-    for fixture in (
-        "uart_mock_modbus_loopback",
-        "uart_mock_modbus_mesh",
-        "uart_mock_modbus_server_injected",
-    ):
-        assert mapping[fixture] == frozenset(
-            {"tests/integration/test_uart_mock_modbus.py"}
-        )
-
-
-def test_no_orphan_integration_fixtures() -> None:
-    """Every fixture must reach CI test selection; an orphan selects nothing."""
-    helpers.get_fixture_to_test_files.cache_clear()
-    mapping = helpers.get_fixture_to_test_files()
-    fixtures_dir = (Path(__file__).parent.parent / "integration" / "fixtures").resolve()
-    fixtures = list(fixtures_dir.glob("*.yaml"))
-    assert fixtures, f"no fixtures found under {fixtures_dir}"
-    # cache_init is covered via INTEGRATION_TESTS_TRIGGER_FILES instead
-    orphans = [
-        f.stem for f in fixtures if f.stem != "cache_init" and f.stem not in mapping
-    ]
-    assert not orphans, f"fixtures invisible to CI test selection: {orphans}"
-
-
 def test_lpt_partition_balances_skewed_weights() -> None:
     """Heavy items spread across groups instead of clustering."""
     items = [f"i{n}" for n in range(6)]

tests/unit_tests/components/test_web_server.py

@@ -0,0 +1,150 @@
+"""Tests for the web_server AP mode helpers."""
+
+import logging
+
+import pytest
+
+from esphome.components.web_server import (
+    _final_validate_ap_mode,
+    serve_captive,
+    serve_local,
+)
+from esphome.const import (
+    CONF_AP,
+    CONF_LOCAL,
+    CONF_NETWORKS,
+    CONF_PORT,
+    CONF_SSID,
+    CONF_VERSION,
+    CONF_WIFI,
+)
+import esphome.final_validate as fv
+
+AP_ONLY = {CONF_AP: {}}
+AP_FALLBACK = {CONF_AP: {}, CONF_NETWORKS: [{CONF_SSID: "x"}]}
+STA_ONLY = {CONF_NETWORKS: [{CONF_SSID: "x"}]}
+
+
+@pytest.mark.parametrize(
+    ("web_server_config", "wifi_config", "expected"),
+    [
+        # AP only: embed the interface, the AP has no internet.
+        ({CONF_VERSION: 2}, AP_ONLY, True),
+        ({CONF_VERSION: 3}, AP_ONLY, True),
+        # Explicit setting always wins.
+        ({CONF_VERSION: 2, CONF_LOCAL: False}, AP_ONLY, False),
+        ({CONF_VERSION: 2, CONF_LOCAL: True}, STA_ONLY, True),
+        # AP fallback, no AP, no wifi, or version 1 (no local mode): hosted page.
+        ({CONF_VERSION: 2}, AP_FALLBACK, False),
+        ({CONF_VERSION: 2}, STA_ONLY, False),
+        ({CONF_VERSION: 2}, None, False),
+        ({CONF_VERSION: 1}, AP_ONLY, False),
+    ],
+)
+def test_serve_local(
+    web_server_config: dict, wifi_config: dict | None, expected: bool
+) -> None:
+    """The interface is embedded for AP only WiFi unless local is set explicitly."""
+    assert serve_local(web_server_config, wifi_config) is expected
+
+
+@pytest.mark.parametrize(
+    ("web_server_config", "full_config", "expected"),
+    [
+        # AP only: local is implied, web_server is the captive portal.
+        ({CONF_VERSION: 2}, {CONF_WIFI: AP_ONLY}, True),
+        # Captive portal probes only work on port 80.
+        ({CONF_VERSION: 2, CONF_PORT: 8080}, {CONF_WIFI: AP_ONLY}, False),
+        # AP fallback needs an explicit local: true to be captive.
+        ({CONF_VERSION: 2}, {CONF_WIFI: AP_FALLBACK}, False),
+        ({CONF_VERSION: 2, CONF_LOCAL: True}, {CONF_WIFI: AP_FALLBACK}, True),
+        # captive_portal owns the role when configured.
+        ({CONF_VERSION: 2}, {CONF_WIFI: AP_ONLY, "captive_portal": {}}, False),
+        # No AP, no wifi, hosted page, or version 1: never captive.
+        ({CONF_VERSION: 2, CONF_LOCAL: True}, {CONF_WIFI: STA_ONLY}, False),
+        ({CONF_VERSION: 2, CONF_LOCAL: True}, {}, False),
+        ({CONF_VERSION: 2, CONF_LOCAL: False}, {CONF_WIFI: AP_ONLY}, False),
+        ({CONF_VERSION: 1}, {CONF_WIFI: AP_ONLY}, False),
+    ],
+)
+def test_serve_captive(
+    web_server_config: dict, full_config: dict, expected: bool
+) -> None:
+    web_server_config.setdefault(CONF_PORT, 80)
+    assert serve_captive(web_server_config, full_config) is expected
+
+
+@pytest.mark.parametrize(
+    ("web_server_config", "expect_warning"),
+    [
+        # Explicit local: false on an AP only device: the hosted page will stay blank.
+        ({CONF_VERSION: 2, CONF_PORT: 80, CONF_LOCAL: False}, True),
+        # Default: embedded and captive, nothing to warn about.
+        ({CONF_VERSION: 2, CONF_PORT: 80}, False),
+    ],
+)
+def test_final_validate_ap_mode_warns_for_hosted_page(
+    web_server_config: dict, expect_warning: bool, caplog: pytest.LogCaptureFixture
+) -> None:
+    token = fv.full_config.set({"web_server": web_server_config, CONF_WIFI: AP_ONLY})
+    try:
+        with caplog.at_level(logging.WARNING):
+            _final_validate_ap_mode(web_server_config)
+    finally:
+        fv.full_config.reset(token)
+    assert ("stays blank" in caplog.text) is expect_warning
+
+
+def test_final_validate_ap_mode_warns_for_non_default_port(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Captive portal detection needs port 80; other ports get a hint, not captive mode."""
+    config = {CONF_VERSION: 2, CONF_PORT: 8080}
+    token = fv.full_config.set({"web_server": config, CONF_WIFI: AP_ONLY})
+    try:
+        with caplog.at_level(logging.WARNING):
+            _final_validate_ap_mode(config)
+    finally:
+        fv.full_config.reset(token)
+    assert "cannot open automatically" in caplog.text
+    assert "http://192.168.4.1:8080/" in caplog.text
+
+
+def test_final_validate_ap_mode_port_warning_uses_manual_ip(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The manual URL in the port warning honors wifi.ap.manual_ip."""
+    from esphome.const import CONF_MANUAL_IP, CONF_STATIC_IP
+
+    config = {CONF_VERSION: 2, CONF_PORT: 8080}
+    wifi = {CONF_AP: {CONF_MANUAL_IP: {CONF_STATIC_IP: "10.0.0.1"}}}
+    token = fv.full_config.set({"web_server": config, CONF_WIFI: wifi})
+    try:
+        with caplog.at_level(logging.WARNING):
+            _final_validate_ap_mode(config)
+    finally:
+        fv.full_config.reset(token)
+    assert "http://10.0.0.1:8080/" in caplog.text
+
+
+@pytest.mark.parametrize(
+    ("wifi_config", "expected"),
+    [
+        # Explicit local: true on a fallback AP: announce the captive fallback role.
+        (AP_FALLBACK, "fallback access point"),
+        # Explicit local: true on AP only skips the implied-local info; still announce.
+        (AP_ONLY, "captive portal while the access point"),
+    ],
+)
+def test_final_validate_ap_mode_informs_explicit_local_captive(
+    wifi_config: dict, expected: str, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Explicit local: true logs that web_server becomes the captive portal."""
+    config = {CONF_VERSION: 2, CONF_PORT: 80, CONF_LOCAL: True}
+    token = fv.full_config.set({"web_server": config, CONF_WIFI: wifi_config})
+    try:
+        with caplog.at_level(logging.INFO):
+            _final_validate_ap_mode(config)
+    finally:
+        fv.full_config.reset(token)
+    assert expected in caplog.text