Merge remote-tracking branch 'origin/dev' into socket-lwip-raw-udp

# Conflicts:
#	esphome/components/socket/__init__.py

esphome/components/socket/__init__.py

@@ -187,7 +187,9 @@ async def to_code(config: ConfigType) -> None:
 # for the selected implementation.
 FILTER_SOURCE_FILES = filter_source_files_from_defines(
     {
+        "lwip_raw_common_impl.cpp": "USE_SOCKET_IMPL_LWIP_TCP",
         "lwip_raw_tcp_impl.cpp": "USE_SOCKET_IMPL_LWIP_TCP",
+        "lwip_raw_udp_impl.cpp": "USE_SOCKET_IMPL_LWIP_TCP",
         "bsd_sockets_impl.cpp": "USE_SOCKET_IMPL_BSD_SOCKETS",
         "lwip_sockets_impl.cpp": "USE_SOCKET_IMPL_LWIP_SOCKETS",
     }

esphome/components/socket/bsd_sockets_impl.h

@@ -144,6 +144,9 @@ class BSDSocketImpl {
 
   int get_fd() const { return this->fd_; }
 
+  /// UDP rx drop counter parity with LWIPRawUDPImpl; drops are not counted here.
+  uint16_t get_rx_dropped() const { return 0; }
+
  protected:
   // fd_ < 0 means "not open" — used both pre-open (initial state) and post-close. This
   // replaces a separate closed_ flag: close() sets fd_ = -1 after ::close(), and the

esphome/components/socket/headers.h

@@ -20,6 +20,16 @@
 
 #define IPPROTO_IP 0
 #define IPPROTO_TCP 6
+#define IPPROTO_UDP 17
+
+#define IP_ADD_MEMBERSHIP 3
+#define IP_DROP_MEMBERSHIP 4
+
+// NOLINTNEXTLINE(readability-identifier-naming)
+struct ip_mreq {
+  struct in_addr imr_multiaddr;
+  struct in_addr imr_interface;
+};
 
 #if LWIP_IPV6
 #define AF_INET6 10

esphome/components/socket/lwip_raw_common_impl.cpp

@@ -0,0 +1,109 @@
+#include "lwip_raw_common_impl.h"
+#include "esphome/core/defines.h"
+
+#ifdef USE_SOCKET_IMPL_LWIP_TCP
+
+#include <cerrno>
+#include <cstring>
+
+namespace esphome::socket {
+
+int lwip_ip_to_sockaddr(sa_family_t family, const ip_addr_t *ip, uint16_t port_host, struct sockaddr *name,
+                        socklen_t *addrlen) {
+  if (family == AF_INET) {
+    if (*addrlen < sizeof(struct sockaddr_in)) {
+      errno = EINVAL;
+      return -1;
+    }
+    auto *addr = reinterpret_cast<struct sockaddr_in *>(name);
+    addr->sin_family = AF_INET;
+    *addrlen = addr->sin_len = sizeof(struct sockaddr_in);
+    addr->sin_port = htons(port_host);
+    inet_addr_from_ip4addr(&addr->sin_addr, ip_2_ip4(ip));
+    return 0;
+  }
+#if LWIP_IPV6
+  if (family == AF_INET6) {
+    if (*addrlen < sizeof(struct sockaddr_in6)) {
+      errno = EINVAL;
+      return -1;
+    }
+    auto *addr = reinterpret_cast<struct sockaddr_in6 *>(name);
+    addr->sin6_family = AF_INET6;
+    *addrlen = addr->sin6_len = sizeof(struct sockaddr_in6);
+    addr->sin6_port = htons(port_host);
+    // AF_INET6 sockets may receive IPv4 packets; convert to IPv4-mapped IPv6.
+    if (IP_IS_V4(ip)) {
+      ip_addr_t mapped;
+      ip4_2_ipv4_mapped_ipv6(ip_2_ip6(&mapped), ip_2_ip4(ip));
+      inet6_addr_from_ip6addr(&addr->sin6_addr, ip_2_ip6(&mapped));
+    } else {
+      inet6_addr_from_ip6addr(&addr->sin6_addr, ip_2_ip6(ip));
+    }
+    return 0;
+  }
+#endif
+  errno = EAFNOSUPPORT;
+  return -1;
+}
+
+bool sockaddr_to_lwip(const struct sockaddr *addr, socklen_t addrlen, ip_addr_t *ip, uint16_t *port) {
+  // headers.h defines sockaddr and sockaddr_in with the same size, so this covers AF_INET
+  if (addrlen < sizeof(struct sockaddr))
+    return false;
+  // Zero the whole struct — the IPv6 zone byte would otherwise be stack garbage
+  memset(ip, 0, sizeof(*ip));
+  if (addr->sa_family == AF_INET) {
+    auto *addr4 = reinterpret_cast<const sockaddr_in *>(addr);
+    *port = ntohs(addr4->sin_port);
+    IP_SET_TYPE_VAL(*ip, IPADDR_TYPE_V4);
+    ip_2_ip4(ip)->addr = addr4->sin_addr.s_addr;
+    return true;
+  }
+#if LWIP_IPV6
+  if (addr->sa_family == AF_INET6) {
+    if (addrlen < sizeof(sockaddr_in6))
+      return false;
+    auto *addr6 = reinterpret_cast<const sockaddr_in6 *>(addr);
+    *port = ntohs(addr6->sin6_port);
+    IP_SET_TYPE_VAL(*ip, IPADDR_TYPE_V6);
+    memcpy(&ip_2_ip6(ip)->addr, &addr6->sin6_addr.un.u8_addr, 16);
+    // Unmap ::ffff:a.b.c.d so replies to recvfrom addresses route as IPv4
+    if (ip6_addr_isipv4mappedipv6(ip_2_ip6(ip))) {
+      unmap_ipv4_mapped_ipv6(ip_2_ip4(ip), ip_2_ip6(ip));
+      IP_SET_TYPE_VAL(*ip, IPADDR_TYPE_V4);
+    }
+    return true;
+  }
+#endif
+  return false;
+}
+
+bool sockaddr_to_lwip_bind(sa_family_t family, const struct sockaddr *addr, socklen_t addrlen, ip_addr_t *ip,
+                           uint16_t *port) {
+  if (!sockaddr_to_lwip(addr, addrlen, ip, port))
+    return false;
+#if LWIP_IPV6
+  // Only promote wildcard binds — a specific address must keep filtering
+  if (family == AF_INET6 && ip_addr_isany_val(*ip))
+    IP_SET_TYPE_VAL(*ip, IPADDR_TYPE_ANY);
+#endif
+  return true;
+}
+
+int lwip_bind_err(err_t err) {
+  if (err == ERR_OK)
+    return 0;
+  if (err == ERR_USE) {
+    errno = EADDRINUSE;
+  } else if (err == ERR_VAL) {
+    errno = EINVAL;
+  } else {
+    errno = EIO;
+  }
+  return -1;
+}
+
+}  // namespace esphome::socket
+
+#endif  // USE_SOCKET_IMPL_LWIP_TCP

esphome/components/socket/lwip_raw_common_impl.h

@@ -0,0 +1,53 @@
+#pragma once
+#include "esphome/core/defines.h"
+
+#ifdef USE_SOCKET_IMPL_LWIP_TCP
+
+#include "headers.h"
+#include "lwip/ip.h"
+
+namespace esphome::socket {
+
+// ---- LWIP thread safety ----
+//
+// On RP2040 (Pico W), arduino-pico sets PICO_CYW43_ARCH_THREADSAFE_BACKGROUND=1.
+// This means lwip callbacks (recv_fn, accept_fn, err_fn) run from a low-priority
+// user IRQ context, not the main loop (see low_priority_irq_handler() in pico-sdk
+// async_context_threadsafe_background.c). They can preempt main-loop code at any point.
+//
+// Without locking, this causes race conditions between recv_fn and read() on the
+// shared rx_buf_ pbuf chain — recv_fn calls pbuf_cat() while read() is freeing
+// nodes, leading to use-after-free and infinite-loop crashes. See esphome#10681.
+//
+// On ESP8266, lwip callbacks run from the SYS context which cooperates with user
+// code (CONT context) — they never preempt each other, so no locking is needed.
+//
+// esphome::LwIPLock is the platform-provided RAII guard (see helpers.h/helpers.cpp).
+// On RP2040, it acquires cyw43_arch_lwip_begin/end (WiFi) or ethernet_arch_lwip_begin/end
+// (Ethernet). On ESP8266, it's a no-op.
+//
+// Each .cpp file that needs locking defines its own LWIP_LOCK() macro:
+//   #define LWIP_LOCK() esphome::LwIPLock lwip_lock_guard
+// This is a per-TU convenience macro, not defined here to avoid macro leaking.
+
+/// Convert lwip ip_addr_t + host-order port to sockaddr, based on the socket's address family.
+/// TCP callers pass ntohs(pcb port) to preserve historical getpeername/getsockname output.
+int lwip_ip_to_sockaddr(sa_family_t family, const ip_addr_t *ip, uint16_t port_host, struct sockaddr *name,
+                        socklen_t *addrlen);
+
+/// Convert sockaddr to lwip ip_addr_t and host-order port.
+/// For IPv6, sets type to IPADDR_TYPE_V6 — correct for sendto destinations.
+/// Bind paths must use sockaddr_to_lwip_bind() instead.
+bool sockaddr_to_lwip(const struct sockaddr *addr, socklen_t addrlen, ip_addr_t *ip, uint16_t *port);
+
+/// sockaddr_to_lwip variant for bind: promotes AF_INET6 sockets to
+/// IPADDR_TYPE_ANY so they accept both IPv4 and IPv6 (dual-stack).
+bool sockaddr_to_lwip_bind(sa_family_t family, const struct sockaddr *addr, socklen_t addrlen, ip_addr_t *ip,
+                           uint16_t *port);
+
+/// Map lwip bind error to errno. Returns 0 on success, -1 on error with errno set.
+int lwip_bind_err(err_t err);
+
+}  // namespace esphome::socket
+
+#endif  // USE_SOCKET_IMPL_LWIP_TCP

esphome/components/socket/lwip_raw_tcp_impl.cpp

@@ -10,6 +10,7 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/wake.h"
 #include "esphome/core/log.h"
+#include "lwip_raw_common_impl.h"
 
 #ifdef USE_OTA_PLATFORM_ESPHOME
 extern "C" void esphome_wake_ota_component_any_context();
@@ -24,23 +25,9 @@ extern "C" void esphome_wake_ota_component_any_context();
 
 namespace esphome::socket {
 
-// ---- LWIP thread safety ----
-//
-// On RP2040 (Pico W), arduino-pico sets PICO_CYW43_ARCH_THREADSAFE_BACKGROUND=1.
-// This means lwip callbacks (recv_fn, accept_fn, err_fn) run from a low-priority
-// user IRQ context, not the main loop (see low_priority_irq_handler() in pico-sdk
-// async_context_threadsafe_background.c). They can preempt main-loop code at any point.
-//
-// Without locking, this causes race conditions between recv_fn and read() on the
-// shared rx_buf_ pbuf chain — recv_fn calls pbuf_cat() while read() is freeing
-// nodes, leading to use-after-free and infinite-loop crashes. See esphome#10681.
-//
-// On ESP8266, lwip callbacks run from the SYS context which cooperates with user
-// code (CONT context) — they never preempt each other, so no locking is needed.
-//
-// esphome::LwIPLock is the platform-provided RAII guard (see helpers.h/helpers.cpp).
-// On RP2040, it acquires cyw43_arch_lwip_begin/end (WiFi) or ethernet_arch_lwip_begin/end
-// (Ethernet). On ESP8266, it's a no-op.
+// LWIP thread safety — see lwip_raw_common_impl.h for full explanation.
+// esphome::LwIPLock is the platform-provided RAII guard.
+// On RP2040, it acquires cyw43_arch_lwip_begin/end. On ESP8266, it's a no-op.
 #define LWIP_LOCK() esphome::LwIPLock lwip_lock_guard  // NOLINT
 
 static const char *const TAG = "socket";
@@ -112,59 +99,14 @@ int LWIPRawCommon::bind(const struct sockaddr *name, socklen_t addrlen) {
     return -1;
   }
   ip_addr_t ip;
-  in_port_t port;
-#if LWIP_IPV6
-  if (this->family_ == AF_INET) {
-    if (addrlen < sizeof(sockaddr_in)) {
-      errno = EINVAL;
-      return -1;
-    }
-    auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
-    port = ntohs(addr4->sin_port);
-    ip.type = IPADDR_TYPE_V4;
-    ip.u_addr.ip4.addr = addr4->sin_addr.s_addr;
-    LWIP_LOG("tcp_bind(%p ip=%s port=%u)", this->pcb_, ip4addr_ntoa(&ip.u_addr.ip4), port);
-  } else if (this->family_ == AF_INET6) {
-    if (addrlen < sizeof(sockaddr_in6)) {
-      errno = EINVAL;
-      return -1;
-    }
-    auto *addr6 = reinterpret_cast<const sockaddr_in6 *>(name);
-    port = ntohs(addr6->sin6_port);
-    ip.type = IPADDR_TYPE_ANY;
-    memcpy(&ip.u_addr.ip6.addr, &addr6->sin6_addr.un.u8_addr, 16);
-    LWIP_LOG("tcp_bind(%p ip=%s port=%u)", this->pcb_, ip6addr_ntoa(&ip.u_addr.ip6), port);
-  } else {
+  uint16_t port;
+  if (!sockaddr_to_lwip_bind(this->family_, name, addrlen, &ip, &port)) {
     errno = EINVAL;
     return -1;
   }
-#else
-  if (this->family_ != AF_INET) {
-    errno = EINVAL;
-    return -1;
-  }
-  auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
-  port = ntohs(addr4->sin_port);
-  ip.addr = addr4->sin_addr.s_addr;
-  LWIP_LOG("tcp_bind(%p ip=%u port=%u)", this->pcb_, ip.addr, port);
-#endif
   err_t err = tcp_bind(this->pcb_, &ip, port);
-  if (err == ERR_USE) {
-    LWIP_LOG("  -> err ERR_USE");
-    errno = EADDRINUSE;
-    return -1;
-  }
-  if (err == ERR_VAL) {
-    LWIP_LOG("  -> err ERR_VAL");
-    errno = EINVAL;
-    return -1;
-  }
-  if (err != ERR_OK) {
-    LWIP_LOG("  -> err %d", err);
-    errno = EIO;
-    return -1;
-  }
-  return 0;
+  LWIP_LOG("  -> err %d", err);
+  return lwip_bind_err(err);
 }
 
 int LWIPRawCommon::close() {
@@ -349,43 +291,8 @@ int LWIPRawCommon::setsockopt(int level, int optname, const void *optval, sockle
 }
 
 int LWIPRawCommon::ip2sockaddr_(ip_addr_t *ip, uint16_t port, struct sockaddr *name, socklen_t *addrlen) {
-  if (this->family_ == AF_INET) {
-    if (*addrlen < sizeof(struct sockaddr_in)) {
-      errno = EINVAL;
-      return -1;
-    }
-
-    struct sockaddr_in *addr = reinterpret_cast<struct sockaddr_in *>(name);
-    addr->sin_family = AF_INET;
-    *addrlen = addr->sin_len = sizeof(struct sockaddr_in);
-    addr->sin_port = port;
-    inet_addr_from_ip4addr(&addr->sin_addr, ip_2_ip4(ip));
-    return 0;
-  }
-#if LWIP_IPV6
-  else if (this->family_ == AF_INET6) {
-    if (*addrlen < sizeof(struct sockaddr_in6)) {
-      errno = EINVAL;
-      return -1;
-    }
-
-    struct sockaddr_in6 *addr = reinterpret_cast<struct sockaddr_in6 *>(name);
-    addr->sin6_family = AF_INET6;
-    *addrlen = addr->sin6_len = sizeof(struct sockaddr_in6);
-    addr->sin6_port = port;
-
-    // AF_INET6 sockets are bound to IPv4 as well, so we may encounter IPv4 addresses that must be converted to IPv6.
-    if (IP_IS_V4(ip)) {
-      ip_addr_t mapped;
-      ip4_2_ipv4_mapped_ipv6(ip_2_ip6(&mapped), ip_2_ip4(ip));
-      inet6_addr_from_ip6addr(&addr->sin6_addr, ip_2_ip6(&mapped));
-    } else {
-      inet6_addr_from_ip6addr(&addr->sin6_addr, ip_2_ip6(ip));
-    }
-    return 0;
-  }
-#endif
-  return -1;
+  // lwip pcb ports are host order; ntohs preserves historical byte-swapped sin_port output
+  return lwip_ip_to_sockaddr(this->family_, ip, ntohs(port), name, addrlen);
 }
 
 // ---- LWIPRawImpl methods ----
@@ -887,11 +794,11 @@ err_t LWIPRawListenImpl::accept_fn_(struct tcp_pcb *newpcb, err_t err) {
   return ERR_OK;
 }
 
-// ---- Factory functions ----
+// ---- TCP Factory functions ----
 
 std::unique_ptr<Socket> socket(int domain, int type, int protocol) {
   if (type != SOCK_STREAM) {
-    ESP_LOGE(TAG, "UDP sockets not supported on this platform, use WiFiUDP");
+    ESP_LOGE(TAG, "Use socket_udp() for UDP sockets on this platform");
     errno = EPROTOTYPE;
     return nullptr;
   }
@@ -911,7 +818,7 @@ std::unique_ptr<Socket> socket_loop_monitored(int domain, int type, int protocol
 
 std::unique_ptr<ListenSocket> socket_listen(int domain, int type, int protocol) {
   if (type != SOCK_STREAM) {
-    ESP_LOGE(TAG, "UDP sockets not supported on this platform, use WiFiUDP");
+    ESP_LOGE(TAG, "Use socket_udp() for UDP sockets on this platform");
     errno = EPROTOTYPE;
     return nullptr;
   }

esphome/components/socket/lwip_raw_udp_impl.cpp

@@ -0,0 +1,328 @@
+#include "socket.h"
+#include "esphome/core/defines.h"
+
+#ifdef USE_SOCKET_IMPL_LWIP_TCP
+
+#include <cerrno>
+#include <cstring>
+
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+#include "esphome/core/wake.h"
+#include "lwip_raw_common_impl.h"
+
+#include "lwip/igmp.h"
+#include "lwip/pbuf.h"
+#include "lwip/udp.h"
+
+namespace esphome::socket {
+
+// LWIP thread safety — see lwip_raw_common_impl.h for full explanation.
+// esphome::LwIPLock is the platform-provided RAII guard.
+// On RP2040, it acquires cyw43_arch_lwip_begin/end. On ESP8266, it's a no-op.
+#define LWIP_LOCK() esphome::LwIPLock lwip_lock_guard  // NOLINT
+
+// ---- LWIPRawUDPSendImpl (send-only) methods ----
+
+LWIPRawUDPSendImpl::~LWIPRawUDPSendImpl() {
+  // Guard avoids acquiring the lwip lock when already closed
+  if (this->pcb_ != nullptr)
+    this->close();
+}
+
+int LWIPRawUDPSendImpl::bind(const struct sockaddr *name, socklen_t addrlen) {
+  LWIP_LOCK();
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+  if (name == nullptr) {
+    errno = EINVAL;
+    return -1;
+  }
+  ip_addr_t ip;
+  uint16_t port;
+  if (!sockaddr_to_lwip_bind(this->family_, name, addrlen, &ip, &port)) {
+    errno = EINVAL;
+    return -1;
+  }
+  return lwip_bind_err(udp_bind(this->pcb_, &ip, port));
+}
+
+int LWIPRawUDPSendImpl::close() {
+  LWIP_LOCK();
+  return this->close_internal_locked_();
+}
+
+int LWIPRawUDPSendImpl::close_internal_locked_() {
+  // Caller must hold LWIP_LOCK
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+  udp_remove(this->pcb_);
+  this->pcb_ = nullptr;
+  return 0;
+}
+
+int LWIPRawUDPSendImpl::ip2sockaddr_(const ip_addr_t *ip, uint16_t port, struct sockaddr *name, socklen_t *addrlen) {
+  // UDP recv callback provides port in host byte order
+  return lwip_ip_to_sockaddr(this->family_, ip, port, name, addrlen);
+}
+
+ssize_t LWIPRawUDPSendImpl::sendto(const void *buf, size_t len, int flags, const struct sockaddr *dest_addr,
+                                   socklen_t addrlen) {
+  (void) flags;  // Flags (MSG_DONTWAIT, etc.) are ignored; raw lwip is always non-blocking
+  if (buf == nullptr || dest_addr == nullptr) {
+    errno = EINVAL;
+    return -1;
+  }
+
+  // pbuf_alloc takes u16_t length; reject oversized packets
+  if (len > UINT16_MAX) {
+    errno = EMSGSIZE;
+    return -1;
+  }
+
+  ip_addr_t dst_ip;
+  uint16_t dst_port;
+  if (!sockaddr_to_lwip(dest_addr, addrlen, &dst_ip, &dst_port)) {
+    errno = EINVAL;
+    return -1;
+  }
+
+  LWIP_LOCK();
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+
+  // Allocate pbuf and copy data
+  struct pbuf *pb = pbuf_alloc(PBUF_TRANSPORT, (uint16_t) len, PBUF_RAM);
+  if (pb == nullptr) {
+    errno = ENOMEM;
+    return -1;
+  }
+  memcpy(pb->payload, buf, len);
+
+  err_t err = udp_sendto(this->pcb_, pb, &dst_ip, dst_port);
+  pbuf_free(pb);
+
+  if (err != ERR_OK) {
+    errno = err == ERR_MEM ? ENOMEM : EIO;
+    return -1;
+  }
+  return (ssize_t) len;
+}
+
+int LWIPRawUDPSendImpl::setsockopt(int level, int optname, const void *optval, socklen_t optlen) {
+  LWIP_LOCK();
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+  if (level == SOL_SOCKET && optname == SO_REUSEADDR) {
+    if (optval == nullptr || optlen < sizeof(int)) {
+      errno = EINVAL;
+      return -1;
+    }
+    // Effective only where lwip is built with SO_REUSE=1 (ESP8266 yes, RP2040 currently no)
+    if (*reinterpret_cast<const int *>(optval)) {
+      ip_set_option(this->pcb_, SOF_REUSEADDR);
+    } else {
+      ip_reset_option(this->pcb_, SOF_REUSEADDR);
+    }
+    return 0;
+  }
+  if (level == SOL_SOCKET && optname == SO_BROADCAST) {
+    if (optval == nullptr || optlen < sizeof(int)) {
+      errno = EINVAL;
+      return -1;
+    }
+    int val = *reinterpret_cast<const int *>(optval);
+    if (val) {
+      ip_set_option(this->pcb_, SOF_BROADCAST);
+    } else {
+      ip_reset_option(this->pcb_, SOF_BROADCAST);
+    }
+    return 0;
+  }
+  if (level == IPPROTO_IP && (optname == IP_ADD_MEMBERSHIP || optname == IP_DROP_MEMBERSHIP)) {
+    if (optval == nullptr || optlen < sizeof(struct ip_mreq)) {
+      errno = EINVAL;
+      return -1;
+    }
+    auto *mreq = reinterpret_cast<const struct ip_mreq *>(optval);
+    ip4_addr_t multiaddr{mreq->imr_multiaddr.s_addr};
+    ip4_addr_t ifaddr{mreq->imr_interface.s_addr};
+    err_t err =
+        optname == IP_ADD_MEMBERSHIP ? igmp_joingroup(&ifaddr, &multiaddr) : igmp_leavegroup(&ifaddr, &multiaddr);
+    if (err != ERR_OK) {
+      errno = EIO;
+      return -1;
+    }
+    return 0;
+  }
+  errno = ENOPROTOOPT;
+  return -1;
+}
+
+int LWIPRawUDPSendImpl::getsockopt(int level, int optname, void *optval, socklen_t *optlen) {
+  LWIP_LOCK();
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+  if (level == SOL_SOCKET && optname == SO_REUSEADDR) {
+    if (optval == nullptr || optlen == nullptr || *optlen < sizeof(int)) {
+      errno = EINVAL;
+      return -1;
+    }
+    *reinterpret_cast<int *>(optval) = ip_get_option(this->pcb_, SOF_REUSEADDR) ? 1 : 0;
+    *optlen = sizeof(int);
+    return 0;
+  }
+  errno = ENOPROTOOPT;
+  return -1;
+}
+
+int LWIPRawUDPSendImpl::setblocking(bool blocking) {
+  if (blocking) {
+    // blocking operation not supported on raw lwip
+    errno = EINVAL;
+    return -1;
+  }
+  return 0;
+}
+
+// ---- LWIPRawUDPImpl methods ----
+
+LWIPRawUDPImpl::LWIPRawUDPImpl(sa_family_t family, struct udp_pcb *pcb) : LWIPRawUDPSendImpl(family, pcb) {
+  // Registered here (not in bind) so unbound client sockets can receive replies
+  udp_recv(this->pcb_, LWIPRawUDPImpl::s_recv_fn, this);
+}
+
+LWIPRawUDPImpl::~LWIPRawUDPImpl() {
+  // Flush rx queue and unregister callback before base destructor removes pcb
+  if (this->pcb_ != nullptr)
+    this->close();
+}
+
+int LWIPRawUDPImpl::close() {
+  LWIP_LOCK();
+  // Unregister recv callback before removing pcb
+  if (this->pcb_ != nullptr) {
+    udp_recv(this->pcb_, nullptr, nullptr);
+  }
+  // Flush queued rx packets; slots within rx_count_ always hold a live pbuf
+  for (; this->rx_count_ > 0; this->rx_count_--) {
+    pbuf_free(this->rx_queue_[this->rx_read_idx_].pb);
+    this->rx_read_idx_ = (this->rx_read_idx_ + 1) & UDP_RX_MASK;
+  }
+  // close_internal_locked_() returns EBADF if already closed, which is fine from destructor
+  return this->close_internal_locked_();
+}
+
+ssize_t LWIPRawUDPImpl::read(void *buf, size_t len) { return this->recvfrom(buf, len, nullptr, nullptr); }
+
+ssize_t LWIPRawUDPImpl::recvfrom(void *buf, size_t len, struct sockaddr *src_addr, socklen_t *addrlen) {
+  if (buf == nullptr && len > 0) {
+    errno = EINVAL;
+    return -1;
+  }
+  LWIP_LOCK();
+  if (this->pcb_ == nullptr) {
+    errno = EBADF;
+    return -1;
+  }
+  if (this->rx_count_ == 0) {
+    errno = EWOULDBLOCK;
+    return -1;
+  }
+
+  auto &pkt = this->rx_queue_[this->rx_read_idx_];
+  // On address conversion failure, still consume the packet — the failure is
+  // deterministic (family_ and *addrlen), so keeping it would wedge the queue
+  ssize_t ret = -1;
+  if (src_addr == nullptr || addrlen == nullptr ||
+      this->ip2sockaddr_(&pkt.src_addr, pkt.src_port, src_addr, addrlen) == 0) {
+    ret = (ssize_t) std::min(len, (size_t) pkt.pb->tot_len);
+    pbuf_copy_partial(pkt.pb, buf, ret, 0);
+  }
+  pbuf_free(pkt.pb);
+  this->rx_read_idx_ = (this->rx_read_idx_ + 1) & UDP_RX_MASK;
+  this->rx_count_--;
+  return ret;
+}
+
+void LWIPRawUDPImpl::s_recv_fn(void *arg, struct udp_pcb *pcb, struct pbuf *p, const ip_addr_t *addr, u16_t port) {
+  auto *self = reinterpret_cast<LWIPRawUDPImpl *>(arg);
+  self->recv_fn_(p, addr, port);
+}
+
+// LWIP CALLBACK — runs from IRQ context on RP2040 (low-priority user IRQ).
+// No heap allocation allowed — malloc is not IRQ-safe (see #14687).
+// No LWIP_LOCK() needed — lwip core already holds the async_context lock.
+void LWIPRawUDPImpl::recv_fn_(struct pbuf *p, const ip_addr_t *addr, u16_t port) {
+  if (p == nullptr)
+    return;
+
+  // Check if queue is full
+  if (this->rx_count_ >= UDP_RX_QUEUE_SIZE) {
+    // Drop packet — queue full. Can't log from IRQ context, so count it
+    // (saturating) for consumers to surface via get_rx_dropped().
+    if (this->rx_dropped_ != UINT16_MAX)
+      this->rx_dropped_++;
+    pbuf_free(p);
+    return;
+  }
+
+  // Enqueue the packet
+  uint8_t write_idx = (this->rx_read_idx_ + this->rx_count_) & UDP_RX_MASK;
+  auto &slot = this->rx_queue_[write_idx];
+  slot.pb = p;
+  slot.src_addr = *addr;
+  slot.src_port = port;
+  this->rx_count_++;
+
+  esphome::wake_loop_any_context();
+}
+
+// ---- UDP Factory functions ----
+
+static struct udp_pcb *new_udp_pcb(int domain) {
+#if LWIP_IPV6
+  return udp_new_ip_type(domain == AF_INET6 ? IPADDR_TYPE_ANY : IPADDR_TYPE_V4);
+#else
+  return udp_new();
+#endif
+}
+
+std::unique_ptr<UDPSendSocket> socket_udp_send(int domain, int protocol) {
+  (void) protocol;  // Raw lwip UDP ignores protocol; kept for API compatibility
+  LWIP_LOCK();
+  auto *pcb = new_udp_pcb(domain);
+  if (pcb == nullptr) {
+    errno = ENOMEM;
+    return nullptr;
+  }
+  return make_unique<LWIPRawUDPSendImpl>((sa_family_t) domain, pcb);
+}
+
+std::unique_ptr<UDPSocket> socket_udp(int domain, int protocol) {
+  (void) protocol;  // Raw lwip UDP ignores protocol; kept for API compatibility
+  LWIP_LOCK();
+  auto *pcb = new_udp_pcb(domain);
+  if (pcb == nullptr) {
+    errno = ENOMEM;
+    return nullptr;
+  }
+  // Ctor registers the recv callback under the lock held here
+  return make_unique<LWIPRawUDPImpl>((sa_family_t) domain, pcb);
+}
+
+#undef LWIP_LOCK
+
+}  // namespace esphome::socket
+
+#endif  // USE_SOCKET_IMPL_LWIP_TCP

esphome/components/socket/lwip_raw_udp_impl.h

@@ -0,0 +1,113 @@
+#pragma once
+#include "esphome/core/defines.h"
+
+#ifdef USE_SOCKET_IMPL_LWIP_TCP
+
+#include <array>
+#include <cstdint>
+
+#include "headers.h"
+#include "lwip/ip.h"
+#include "lwip/udp.h"
+
+namespace esphome::socket {
+
+/// Send-only UDP socket implementation for LWIP raw API.
+/// Non-virtual, concrete type. Uses lwip/udp.h raw API.
+/// No receive capability — use LWIPRawUDPImpl for sockets that need to receive.
+class LWIPRawUDPSendImpl {
+ public:
+  /// The pcb is allocated by the factory (like the TCP impl); never null here.
+  LWIPRawUDPSendImpl(sa_family_t family, struct udp_pcb *pcb) : pcb_(pcb), family_(family) {}
+  ~LWIPRawUDPSendImpl();
+  LWIPRawUDPSendImpl(const LWIPRawUDPSendImpl &) = delete;
+  LWIPRawUDPSendImpl &operator=(const LWIPRawUDPSendImpl &) = delete;
+
+  int bind(const struct sockaddr *name, socklen_t addrlen);
+  int close();
+
+  /// Send a UDP packet to the specified destination.
+  ssize_t sendto(const void *buf, size_t len, int flags, const struct sockaddr *dest_addr, socklen_t addrlen);
+
+  int setsockopt(int level, int optname, const void *optval, socklen_t optlen);
+  int getsockopt(int level, int optname, void *optval, socklen_t *optlen);
+
+  int setblocking(bool blocking);
+
+  bool ready() const { return false; }
+  int get_fd() const { return -1; }
+
+ protected:
+  /// Convert lwip ip_addr_t and port to sockaddr.
+  int ip2sockaddr_(const ip_addr_t *ip, uint16_t port, struct sockaddr *name, socklen_t *addrlen);
+
+  /// Shared close logic — removes the udp pcb. Caller must hold LWIP_LOCK.
+  int close_internal_locked_();
+
+  struct udp_pcb *pcb_;
+  sa_family_t family_;
+};
+
+/// UDP socket with receive support for LWIP raw API.
+/// Extends LWIPRawUDPSendImpl with a fixed-size ring buffer for incoming packets.
+/// Inheritance is private (base dtor is non-virtual; converting to a base
+/// pointer would leak queued pbufs on destruction).
+class LWIPRawUDPImpl : private LWIPRawUDPSendImpl {
+ public:
+  /// Caller (the factory) must hold the lwip lock; registers the recv callback.
+  LWIPRawUDPImpl(sa_family_t family, struct udp_pcb *pcb);
+  ~LWIPRawUDPImpl();
+
+  using LWIPRawUDPSendImpl::bind;
+  using LWIPRawUDPSendImpl::get_fd;
+  using LWIPRawUDPSendImpl::getsockopt;
+  using LWIPRawUDPSendImpl::sendto;
+  using LWIPRawUDPSendImpl::setblocking;
+  using LWIPRawUDPSendImpl::setsockopt;
+
+  /// Close the socket, flushing any queued rx packets first.
+  int close();
+
+  /// Read the next queued packet, discarding source address info.
+  /// If buf is smaller than the packet, data is silently truncated (returns bytes copied).
+  /// Note: unlike POSIX MSG_TRUNC, this does not return the original packet length on truncation.
+  ssize_t read(void *buf, size_t len);
+  /// Read the next queued packet and return the source address.
+  /// If buf is smaller than the packet, data is silently truncated (returns bytes copied).
+  /// Note: unlike POSIX MSG_TRUNC, this does not return the original packet length on truncation.
+  ssize_t recvfrom(void *buf, size_t len, struct sockaddr *src_addr, socklen_t *addrlen);
+
+  /// Returns true if there are packets available to read.
+  /// Intentionally unlocked — same rationale as LWIPRawImpl::ready().
+  bool ready() const { return this->rx_count_ > 0; }
+
+  /// Number of packets dropped because the rx queue was full (saturating).
+  uint16_t get_rx_dropped() const { return this->rx_dropped_; }
+
+ protected:
+  static void s_recv_fn(void *arg, struct udp_pcb *pcb, struct pbuf *p, const ip_addr_t *addr, u16_t port);
+  void recv_fn_(struct pbuf *p, const ip_addr_t *addr, u16_t port);
+
+  /// Ring buffer for received UDP packets.
+  /// Both producer (recv callback) and consumer (main loop) are serialized by the
+  /// lwip lock — the callback runs under lwip core lock, and consumer methods hold
+  /// LWIP_LOCK(). All 4 slots are usable (no wasted slot for full/empty distinction).
+  /// No heap allocation in the recv callback — packets are dropped if the queue is full.
+  static constexpr uint8_t UDP_RX_QUEUE_SIZE = 4;
+  static constexpr uint8_t UDP_RX_MASK = UDP_RX_QUEUE_SIZE - 1;
+  static_assert((UDP_RX_QUEUE_SIZE & UDP_RX_MASK) == 0, "UDP_RX_QUEUE_SIZE must be power of 2");
+  // Fields are written by recv_fn_ before rx_count_ makes a slot visible
+  struct UDPRxPacket {
+    ip_addr_t src_addr;
+    struct pbuf *pb;
+    uint16_t src_port;
+  };
+  std::array<UDPRxPacket, UDP_RX_QUEUE_SIZE> rx_queue_{};
+  uint16_t rx_dropped_{0};
+  uint8_t rx_read_idx_{0};
+  uint8_t rx_count_{0};
+};
+
+}  // namespace esphome::socket
+
+#endif  // USE_SOCKET_IMPL_LWIP_TCP

esphome/components/socket/lwip_sockets_impl.h

@@ -84,6 +84,9 @@ class LwIPSocketImpl {
 
   int get_fd() const { return this->fd_; }
 
+  /// UDP rx drop counter parity with LWIPRawUDPImpl; drops are not counted here.
+  uint16_t get_rx_dropped() const { return 0; }
+
  protected:
   // fd_ < 0 means "not open" — used both pre-open (initial state) and post-close. This
   // replaces a separate closed_ flag: close() sets fd_ = -1 after lwip_close(), and the

esphome/components/socket/socket.cpp

@@ -116,25 +116,7 @@ size_t format_sockaddr_to(const struct sockaddr *addr_ptr, socklen_t len, std::s
   return 0;
 }
 
-std::unique_ptr<Socket> socket_ip(int type, int protocol) {
-#if USE_NETWORK_IPV6
-  return socket(AF_INET6, type, protocol);
-#else
-  return socket(AF_INET, type, protocol);
-#endif /* USE_NETWORK_IPV6 */
-}
-
-#ifdef USE_SOCKET_IMPL_LWIP_TCP
-// LWIP_TCP has separate Socket/ListenSocket types — needs out-of-line factory.
-// BSD and LWIP_SOCKETS define this inline in socket.h.
-std::unique_ptr<ListenSocket> socket_ip_loop_monitored(int type, int protocol) {
-#if USE_NETWORK_IPV6
-  return socket_listen_loop_monitored(AF_INET6, type, protocol);
-#else
-  return socket_listen_loop_monitored(AF_INET, type, protocol);
-#endif /* USE_NETWORK_IPV6 */
-}
-#endif
+std::unique_ptr<Socket> socket_ip(int type, int protocol) { return socket(IP_DOMAIN, type, protocol); }
 
 socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_address, uint16_t port) {
 #if USE_NETWORK_IPV6

esphome/components/socket/socket.h

@@ -20,6 +20,7 @@
 #include "lwip_sockets_impl.h"
 #elif defined(USE_SOCKET_IMPL_LWIP_TCP)
 #include "lwip_raw_tcp_impl.h"
+#include "lwip_raw_udp_impl.h"
 #endif
 
 namespace esphome::socket {
@@ -27,17 +28,32 @@ namespace esphome::socket {
 // Type aliases — only one implementation is active per build.
 // Socket is the concrete type for connected sockets.
 // ListenSocket is the concrete type for listening/server sockets.
-// On BSD and LWIP_SOCKETS, both aliases resolve to the same type.
+// UDPSocket is the concrete type for UDP sockets (send + receive).
+// UDPSendSocket is the concrete type for send-only UDP sockets.
+// On BSD and LWIP_SOCKETS, all aliases resolve to the same type.
 // On LWIP_TCP, they are different types (no virtual dispatch between them).
 #ifdef USE_SOCKET_IMPL_BSD_SOCKETS
 using Socket = BSDSocketImpl;
 using ListenSocket = BSDSocketImpl;
+using UDPSendSocket = BSDSocketImpl;
+using UDPSocket = BSDSocketImpl;
 #elif defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
 using Socket = LwIPSocketImpl;
 using ListenSocket = LwIPSocketImpl;
+using UDPSendSocket = LwIPSocketImpl;
+using UDPSocket = LwIPSocketImpl;
 #elif defined(USE_SOCKET_IMPL_LWIP_TCP)
 using Socket = LWIPRawImpl;
 using ListenSocket = LWIPRawListenImpl;
+using UDPSendSocket = LWIPRawUDPSendImpl;
+using UDPSocket = LWIPRawUDPImpl;
+#endif
+
+// Domain used by the socket_ip_* helpers: newest available IP domain.
+#if USE_NETWORK_IPV6
+inline constexpr int IP_DOMAIN = AF_INET6;
+#else
+inline constexpr int IP_DOMAIN = AF_INET;
 #endif
 
 #ifdef USE_LWIP_FAST_SELECT
@@ -104,6 +120,36 @@ std::unique_ptr<Socket> socket_ip(int type, int protocol);
 /// File descriptors >= FD_SETSIZE will not be monitored and will log an error.
 std::unique_ptr<Socket> socket_loop_monitored(int domain, int type, int protocol);
 
+/// Create a send-only UDP socket (socket_udp_send), a UDP socket with receive
+/// support (socket_udp), or a UDP socket monitored for data in the main loop
+/// (socket_udp_loop_monitored).
+#ifdef USE_SOCKET_IMPL_LWIP_TCP
+std::unique_ptr<UDPSendSocket> socket_udp_send(int domain, int protocol);
+std::unique_ptr<UDPSocket> socket_udp(int domain, int protocol);
+// Wake is built into the recv callback, so monitoring needs nothing extra.
+inline std::unique_ptr<UDPSocket> socket_udp_loop_monitored(int domain, int protocol) {
+  return socket_udp(domain, protocol);
+}
+#else
+inline std::unique_ptr<UDPSendSocket> socket_udp_send(int domain, int protocol) {
+  return esphome::socket::socket(domain, SOCK_DGRAM, protocol);
+}
+inline std::unique_ptr<UDPSocket> socket_udp(int domain, int protocol) {
+  return esphome::socket::socket(domain, SOCK_DGRAM, protocol);
+}
+// Registers the socket with the Application's select() loop.
+inline std::unique_ptr<UDPSocket> socket_udp_loop_monitored(int domain, int protocol) {
+  return socket_loop_monitored(domain, SOCK_DGRAM, protocol);
+}
+#endif
+
+/// socket_udp* variants using the newest available IP domain.
+inline std::unique_ptr<UDPSendSocket> socket_ip_udp_send(int protocol) { return socket_udp_send(IP_DOMAIN, protocol); }
+inline std::unique_ptr<UDPSocket> socket_ip_udp(int protocol) { return socket_udp(IP_DOMAIN, protocol); }
+inline std::unique_ptr<UDPSocket> socket_ip_udp_loop_monitored(int protocol) {
+  return socket_udp_loop_monitored(IP_DOMAIN, protocol);
+}
+
 /// Create a listening socket of the given domain, type and protocol.
 /// Create a listening socket and monitor it for data in the main loop.
 /// Create a listening socket in the newest available IP domain and monitor it.
@@ -111,7 +157,6 @@ std::unique_ptr<Socket> socket_loop_monitored(int domain, int type, int protocol
 // LWIP_TCP has separate Socket/ListenSocket types — needs distinct factory functions.
 std::unique_ptr<ListenSocket> socket_listen(int domain, int type, int protocol);
 std::unique_ptr<ListenSocket> socket_listen_loop_monitored(int domain, int type, int protocol);
-std::unique_ptr<ListenSocket> socket_ip_loop_monitored(int type, int protocol);
 #else
 // BSD and LWIP_SOCKETS: Socket == ListenSocket, so listen variants just delegate.
 inline std::unique_ptr<ListenSocket> socket_listen(int domain, int type, int protocol) {
@@ -120,14 +165,10 @@ inline std::unique_ptr<ListenSocket> socket_listen(int domain, int type, int pro
 inline std::unique_ptr<ListenSocket> socket_listen_loop_monitored(int domain, int type, int protocol) {
   return socket_loop_monitored(domain, type, protocol);
 }
+#endif
 inline std::unique_ptr<ListenSocket> socket_ip_loop_monitored(int type, int protocol) {
-#if USE_NETWORK_IPV6
-  return socket_loop_monitored(AF_INET6, type, protocol);
-#else
-  return socket_loop_monitored(AF_INET, type, protocol);
-#endif
+  return socket_listen_loop_monitored(IP_DOMAIN, type, protocol);
 }
-#endif
 
 /// Set a sockaddr to the specified address and port for the IP version used by socket_ip().
 /// @param addr Destination sockaddr structure

script/helpers.py

@@ -1104,6 +1104,10 @@ 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
@@ -1123,6 +1127,19 @@ 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/integration/README.md

@@ -21,6 +21,13 @@ 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,17 +4,22 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import AsyncGenerator, Callable, Generator
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress
 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
@@ -23,7 +28,13 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
-from esphome.helpers import get_usable_cpu_count
+from esphome.helpers import (
+    get_usable_cpu_count,
+    read_file,
+    rmtree,
+    write_file,
+    write_file_if_changed,
+)
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -56,6 +67,21 @@ 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()
@@ -78,7 +104,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(Path(__file__).resolve().parent.parent.parent)
+    repo_root = str(REPO_ROOT)
     existing = env.get("PYTHONPATH")
     env["PYTHONPATH"] = f"{repo_root}{os.pathsep}{existing}" if existing else repo_root
     return env
@@ -88,8 +114,7 @@ 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.
-    # CI caches parts of this path; keep in sync with ci.yml integration-tests.
-    test_cache_dir = Path.home() / ".esphome-integration-tests"
+    test_cache_dir = INTEGRATION_TESTS_ROOT
     cache_dir = test_cache_dir / "platformio"
 
     # Use a lock file in the home directory to ensure only one process initializes the cache
@@ -112,7 +137,9 @@ 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())
+                config_path.write_text(
+                    fixture_path.read_text(encoding="utf-8"), encoding="utf-8"
+                )
 
                 # Run compilation to populate the cache
                 # We must succeed here to avoid race conditions where multiple
@@ -162,13 +189,6 @@ def integration_test_dir() -> Generator[Path]:
         yield Path(tmpdir)
 
 
-@pytest.fixture
-def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
-    """Host preferences persist per device name; give the test its own so a
-    provisioned key never leaks into another run."""
-    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
-
-
 @pytest.fixture
 def reserved_tcp_port() -> Generator[tuple[int, socket.socket]]:
     """Reserve an unused TCP port by holding the socket open."""
@@ -188,21 +208,29 @@ 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."""
-    # 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]
+    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]
 
     # Load the fixture file
-    fixture_path = Path(__file__).parent / "fixtures" / f"{base_name}.yaml"
+    fixture_path = FIXTURES_DIR / 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, fixture_path.read_text)
+    content = await loop.run_in_executor(None, read_file, fixture_path)
 
     # Replace the port in the config if it contains api section
     if "api:" in content:
@@ -226,11 +254,13 @@ 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(
-            Path(__file__).parent / "fixtures" / "external_components"
-        )
+        external_components_path = str(FIXTURES_DIR / "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
 
 
@@ -240,24 +270,218 @@ async def write_yaml_config(
 ) -> AsyncGenerator[ConfigWriter]:
     """Write YAML configuration to a file."""
     # Get the test name for default filename
-    test_name = request.node.name
-    base_name = test_name.replace("test_", "").split("[")[0]
+    base_name = request.node.name.replace("test_", "").partition("[")[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, config_path.write_text, content)
+        await loop.run_in_executor(None, write_file, config_path, 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."""
 
@@ -265,66 +489,96 @@ 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()
 
-        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)}
+        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
             )
-            if config is None:
-                raise RuntimeError(f"Failed to read config from {config_path}")
 
-            # 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
+        # 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
 
     yield _compile
 

tests/integration/fixtures/sensor_filters_batch_window.yaml

@@ -1,58 +0,0 @@
-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

@@ -1,111 +0,0 @@
-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

@@ -1,88 +0,0 @@
-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

@@ -1,106 +0,0 @@
-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

@@ -1,95 +0,0 @@
-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

@@ -1,97 +0,0 @@
-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

@@ -0,0 +1,233 @@
+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_mesh.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-server-contro
+  name: uart-mock-modbus-mesh
 
 host:
 api:
@@ -17,13 +17,14 @@ uart:
   baud_rate: 115200
   port: /dev/null
 
+# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed read-only
+# registers, addr 5 = the read/write 0x17 target, addr 2/3 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:
@@ -31,35 +32,68 @@ uart_mock:
           - uart_mock.inject_rx:
               id: virtual_uart_controller
               data: !lambda return data;
-  - id: virtual_uart_controller
+          - 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
+    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
+    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"
 
 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_controller
+    id: virtual_modbus_client
     role: client
     turnaround_time: 10ms
 
 modbus_controller:
   - address: 1
-    modbus_id: virtual_modbus_controller
+    modbus_id: virtual_modbus_client
     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
 
 modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
-    id: modbus_server_1
     registers:
       - address: 0x01
         value_type: U_WORD
@@ -103,6 +137,34 @@ modbus_server:
       - address: 0x28
         value_type: FP32_R
         read_lambda: return 3.14;
+  - 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;
 
 sensor:
   - platform: modbus_controller
@@ -195,9 +257,46 @@ 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
 
 button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    # This test does not have anything to start (mock is autostart)
+    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]);
+                  }

tests/integration/fixtures/uart_mock_modbus_register_offset.yaml

@@ -1,138 +0,0 @@
-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

@@ -1,124 +0,0 @@
-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_multiple.yaml

@@ -1,116 +0,0 @@
-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_injected.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-srv-rw
+  name: uart-mock-modbus-srv-injected
 
 host:
 api:
@@ -17,6 +17,8 @@ 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
@@ -25,18 +27,31 @@ uart_mock:
     auto_start: false
     debug:
     injections:
-      # 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, 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.
       - delay: 100ms
         inject_rx:
           [0x01, 0x17, 0x00, 0x01, 0x00, 0x02, 0x00, 0x01, 0x00, 0x01, 0x02, 0x12, 0x34, 0x49, 0xD8]
-      # FC 0x17: write reg 0x0003 = 0x5678 (qty 1), then read reg 0x0003 (qty 1) -
+      # FC 0x17: write reg 0x0006 = 0x5678 (qty 1), then read reg 0x0006 (qty 1) -
       # a write and read targeting a different register block.
       - delay: 100ms
         inject_rx:
-          [0x01, 0x17, 0x00, 0x03, 0x00, 0x01, 0x00, 0x03, 0x00, 0x01, 0x02, 0x56, 0x78, 0x9B, 0x10]
+          [0x01, 0x17, 0x00, 0x06, 0x00, 0x01, 0x00, 0x06, 0x00, 0x01, 0x02, 0x56, 0x78, 0x8B, 0x55]
 
 globals:
   - id: stored_1
@@ -70,8 +85,18 @@ 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));
@@ -80,8 +105,22 @@ 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/prefs/<app_name>.prefs`` using a simple binary layout that
+``$ESPHOME_PREFDIR/<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,13 +11,21 @@ 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."""
-    return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
+    """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"
 
 
 def clear_host_prefs(device_name: str) -> None:

tests/integration/test_api_zero_psk_provisioning.py

@@ -24,7 +24,6 @@ from .types import (
     RunCompiledFunction,
 )
 
-pytestmark = pytest.mark.usefixtures("isolated_preferences")
 NEW_KEY = PROVISIONING_PSK
 
 

tests/integration/test_host_preferences_suspend_resume.py

@@ -41,15 +41,6 @@ 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,
@@ -58,7 +49,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
+    pref_file = isolated_preferences / f"{DEVICE_NAME}.prefs"
 
     loop = asyncio.get_running_loop()
     saved_in_memory = loop.create_future()

tests/integration/test_light_initial_state.py

@@ -11,14 +11,6 @@ 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

@@ -173,6 +173,7 @@ 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,
@@ -203,6 +204,7 @@ 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,
@@ -231,8 +233,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 0x0003
-            "rw_read_3": 22136,  # reg 0x0003 reads back the just-written value
+            "rw_write_3": 22136,  # 0x5678 written to reg 0x0006
+            "rw_read_3": 22136,  # reg 0x0006 reads back the just-written value
         }
     )
 
@@ -241,7 +243,8 @@ async def test_uart_mock_modbus_server_read_write(
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
-        await tracker.await_all(futures)
+        # The FC 0x17 injections fire last, behind four earlier 100ms delays
+        await tracker.await_all(futures, timeout=4.0)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
@@ -296,6 +299,7 @@ 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,
@@ -485,6 +489,7 @@ 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 +500,7 @@ async def test_uart_mock_modbus_server_controller_multiple(
 
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
+    expected_values = {"multi_reg_a": 919, "multi_reg_b": 929}
     tracker = SensorTracker(list(expected_values.keys()))
     futures = tracker.expect_all(expected_values)
 
@@ -706,6 +711,7 @@ 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,
@@ -932,6 +938,7 @@ 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,
@@ -947,9 +954,7 @@ 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", "srv_read_1", "client_read_0", "client_read_1"]
-    )
+    tracker = SensorTracker(["srv_write_1", "client_read_0", "client_read_1"])
     futures = tracker.expect_all(
         {
             "srv_write_1": 4660,  # server wrote 0x1234 to reg 0x0001
@@ -967,6 +972,7 @@ 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,
@@ -1022,6 +1028,7 @@ 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,
@@ -1058,6 +1065,7 @@ 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,6 +1121,7 @@ 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,6 +2122,34 @@ 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)]