Derive both raw swap oracles from one helper

script/ci-custom.py

@@ -3,7 +3,6 @@
 import argparse
 import codecs
 import collections
-from collections.abc import Iterator
 import fnmatch
 import functools
 import os.path
@@ -1121,56 +1120,7 @@ def lint_no_std_bind(fname, match):
     )
 
 
-LOG_CALL_START_RE = re.compile(r"ESP_LOG\w+\s*\(")
-# Comments, raw/plain string literals and single char literals are consumed whole so ; ( ) ? :
-# inside them are never seen. A char literal is exactly one (escaped) char so a digit separator
-# like 1'000'000 cannot open one.
-CPP_COMMENT_RE = r"//[^\n]*|/\*.*?\*/"
-CPP_SKIP_RE = (
-    CPP_COMMENT_RE
-    + r'|R"(?P<raw_delim>[^(\s]*)\(.*?\)(?P=raw_delim)"|"(?:[^"\\]|\\.)*"|\'(?:[^\'\\\n]|\\.)\''
-)
-LOG_CALL_TOKEN_RE = re.compile(CPP_SKIP_RE + r"|[()]", re.DOTALL)
-# The last alternative matches a ? or : followed (after spaces or comments) by an opening quote,
-# i.e. a string literal used as a ternary branch.
-LOG_TERNARY_LITERAL_RE = re.compile(
-    CPP_SKIP_RE + r"|[?:](?:\s|" + CPP_COMMENT_RE + r')*(?=")', re.DOTALL
-)
-# A bare NOLINT; a clang-tidy NOLINT(check-name) is aimed at a different tool.
-NOLINT_RE = re.compile(r"\bNOLINT\b(?!\()")
-
-
-def _line_col(content: str, pos: int) -> tuple[int, int]:
-    """1-based line and column of an offset in content."""
-    return content.count("\n", 0, pos) + 1, pos - content.rfind("\n", 0, pos)
-
-
-def _iter_log_calls(content: str) -> Iterator[tuple[int, str | None]]:
-    """Yield (start, text) for every ESP_LOG*(...) call, text running to the matching close paren.
-    text is None when no matching paren exists so callers can report the call instead of skipping it."""
-    for head in LOG_CALL_START_RE.finditer(content):
-        depth = 1
-        for tok in LOG_CALL_TOKEN_RE.finditer(content, head.end()):
-            if tok.group(0) == "(":
-                depth += 1
-            elif tok.group(0) == ")":
-                depth -= 1
-                if depth == 0:
-                    yield head.start(), content[head.start() : tok.end()]
-                    break
-        else:
-            yield head.start(), None
-
-
-def _unbalanced_log_call_error(content: str, pos: int) -> tuple[int, int, str]:
-    lineno, col = _line_col(content, pos)
-    return (
-        lineno,
-        col,
-        "ESP_LOG call has no matching closing parenthesis, so it cannot be checked.",
-    )
-
-
+LOG_MULTILINE_RE = re.compile(r"ESP_LOG\w+\s*\(.*?;", re.DOTALL)
 LOG_BAD_CONTINUATION_RE = re.compile(r'\\n(?:[^ \\"\r\n\t]|"\s*\n\s*"[^ \\])')
 LOG_PERCENT_S_CONTINUATION_RE = re.compile(r'\\n(?:%s|"\s*\n\s*"%s)')
 
@@ -1178,16 +1128,16 @@ LOG_PERCENT_S_CONTINUATION_RE = re.compile(r'\\n(?:%s|"\s*\n\s*"%s)')
 @lint_content_check(include=cpp_include)
 def lint_log_multiline_continuation(fname, content):
     errs = []
-    for log_start, log_text in _iter_log_calls(content):
-        if log_text is None:
-            errs.append(_unbalanced_log_call_error(content, log_start))
-            continue
+    for log_match in LOG_MULTILINE_RE.finditer(content):
+        log_text = log_match.group(0)
         for bad_match in LOG_BAD_CONTINUATION_RE.finditer(log_text):
             # %s may expand to a whitespace prefix at runtime, skip those
             if LOG_PERCENT_S_CONTINUATION_RE.match(log_text, bad_match.start()):
                 continue
             # Calculate line number from position in full content
-            lineno, col = _line_col(content, log_start + bad_match.start())
+            abs_pos = log_match.start() + bad_match.start()
+            lineno = content.count("\n", 0, abs_pos) + 1
+            col = abs_pos - content.rfind("\n", 0, abs_pos)
             errs.append(
                 (
                     lineno,
@@ -1205,90 +1155,6 @@ def lint_log_multiline_continuation(fname, content):
     return errs
 
 
-def _find_ternary_literals(text: str) -> Iterator[tuple[int, str]]:
-    """Yield (offset, literal) for every string literal used as a ternary branch."""
-    branch = False
-    for m in LOG_TERNARY_LITERAL_RE.finditer(text):
-        tok = m.group(0)
-        # An empty literal is merged with every other string's terminator, so it costs no RAM,
-        # while a PSTR("") would add its own flash array; leave it alone.
-        if branch and tok[0] == '"' and tok != '""':
-            yield m.start(), tok
-        branch = tok[0] in "?:"
-
-
-# LOG_STR_LITERAL is a no op everywhere except ESP8266, so code that never builds there is skipped
-# to avoid churn: platform specific sources and components for ESP32, LibreTiny, RP2 and Zephyr only.
-# A component belongs here only if it has no tests/components/<name>/test.esp8266-ard.yaml.
-LOG_LITERAL_LINT_EXCLUDE = [
-    "*_esp32.cpp",
-    "*_esp32_*.cpp",
-    "*_esp_idf.cpp",
-    "*_rmt.cpp",
-    "*_zephyr.cpp",
-    "*_bk72xx.cpp",
-    "*_libretiny.cpp",
-    "*_pico_w.cpp",
-    "*_host.cpp",
-    "esphome/components/esp32*/*",
-    "esphome/components/bk72xx*/*",
-    "esphome/components/ln882h*/*",
-    "esphome/components/ln882x*/*",
-    "esphome/components/rp2*/*",
-    "esphome/components/zephyr*/*",
-    "esphome/components/host/*",
-    "esphome/components/libretiny*/*",
-    "esphome/components/bluetooth_proxy/*",
-    "esphome/components/bluetooth_connection/*",
-    "esphome/components/ble_client/*",
-    "esphome/components/bedjet/*",
-    "esphome/components/anova/*",
-    "esphome/components/xiaomi_ble/*",
-    "esphome/components/bthome_mithermometer/*",
-    "esphome/components/usb_host/*",
-    "esphome/components/zigbee/*",
-    "esphome/components/lvgl/*",
-    # Test fixtures and host only unit tests - not production embedded code
-    "tests/integration/fixtures/*",
-    "tests/components/*",
-]
-
-
-@lint_content_check(include=cpp_include, exclude=LOG_LITERAL_LINT_EXCLUDE)
-def lint_log_no_bare_literal_ternary(
-    fname: Path, content: str
-) -> list[tuple[int, int, str]]:
-    errs = []
-    for log_start, log_text in _iter_log_calls(content):
-        if log_text is None:
-            continue  # reported by lint_log_multiline_continuation, which sees every file
-        # A NOLINT anywhere on the lines the call spans silences every branch in it
-        first_line = content.rfind("\n", 0, log_start) + 1
-        last_line = content.find("\n", log_start + len(log_text))
-        if NOLINT_RE.search(
-            content[first_line : last_line if last_line != -1 else None]
-        ):
-            continue
-        for offset, literal in _find_ternary_literals(log_text):
-            lineno, col = _line_col(content, log_start + offset)
-            errs.append(
-                (
-                    lineno,
-                    col,
-                    (
-                        "String literal used as a ternary branch in a log call. On ESP8266 the "
-                        "log macro moves the format string to flash, but bare literal arguments "
-                        "stay in RAM. Wrap each branch passed straight to the log call in "
-                        f"{highlight('LOG_STR_LITERAL(...)')}:\n"
-                        f"  Before: {highlight(literal)}\n"
-                        f"  After:  {highlight(f'LOG_STR_LITERAL({literal})')}\n"
-                        f"(If strictly necessary, add `{highlight('// NOLINT')}` to the end of the line)"
-                    ),
-                )
-            )
-    return errs
-
-
 @lint_content_find_check(
     "ESP_LOG",
     include=["*.h", "*.tcc"],

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

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

tests/integration/fixtures/uart_mock_modbus_server_controller_bits.yaml

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

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -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/integration_test_durations.json

@@ -1,142 +1,143 @@
 {
-  "tests/integration/test_action_concurrent_reentry.py": 45.23,
-  "tests/integration/test_addressable_light_transition.py": 74.47,
-  "tests/integration/test_alarm_control_panel_state_transitions.py": 74.1,
-  "tests/integration/test_api_action_metadata.py": 62.1,
-  "tests/integration/test_api_action_responses.py": 71.08,
-  "tests/integration/test_api_action_timeout.py": 21.64,
-  "tests/integration/test_api_conditional_memory.py": 13.72,
-  "tests/integration/test_api_custom_services.py": 24.16,
-  "tests/integration/test_api_get_time_response_timezone.py": 23.48,
-  "tests/integration/test_api_homeassistant.py": 37.87,
-  "tests/integration/test_api_homeassistant_action_no_subscriber.py": 14.38,
-  "tests/integration/test_api_list_entities_backpressure.py": 26.85,
-  "tests/integration/test_api_message_size_batching.py": 33.36,
-  "tests/integration/test_api_reboot_timeout.py": 13.63,
-  "tests/integration/test_api_string_lambda.py": 25.04,
-  "tests/integration/test_api_vv_logging.py": 16.6,
-  "tests/integration/test_api_zero_psk_provisioning.py": 43.14,
-  "tests/integration/test_areas_and_devices.py": 25.98,
-  "tests/integration/test_automation_wait_actions.py": 21.91,
-  "tests/integration/test_automations.py": 42.43,
-  "tests/integration/test_batch_delay_zero_rapid_transitions.py": 16.65,
-  "tests/integration/test_binary_sensor_autorepeat_filter.py": 28.67,
-  "tests/integration/test_binary_sensor_invalidate_state.py": 23.69,
-  "tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 22.99,
-  "tests/integration/test_build_info.py": 24.96,
-  "tests/integration/test_camera_mock.py": 14.47,
-  "tests/integration/test_climate_control_action.py": 31.07,
-  "tests/integration/test_climate_custom_modes.py": 28.59,
-  "tests/integration/test_continuation_actions.py": 14.96,
-  "tests/integration/test_cover_control_action.py": 26.14,
-  "tests/integration/test_crc8_helper.py": 10.92,
-  "tests/integration/test_device_id_in_state.py": 64.97,
-  "tests/integration/test_duplicate_entities.py": 30.81,
-  "tests/integration/test_entity_icon.py": 32.85,
-  "tests/integration/test_fan_turn_on_action.py": 24.91,
-  "tests/integration/test_fnv1_hash_object_id.py": 12.54,
-  "tests/integration/test_fnv1a_hash.py": 21.8,
-  "tests/integration/test_gpio_expander_cache.py": 5.2,
-  "tests/integration/test_host_logger_thread_safety.py": 21.7,
-  "tests/integration/test_host_mode_basic.py": 13.62,
-  "tests/integration/test_host_mode_batch_delay.py": 14.56,
-  "tests/integration/test_host_mode_climate_basic_state.py": 30.95,
-  "tests/integration/test_host_mode_climate_control.py": 29.06,
-  "tests/integration/test_host_mode_empty_string_options.py": 27.22,
-  "tests/integration/test_host_mode_entity_fields.py": 30.95,
-  "tests/integration/test_host_mode_fan_preset.py": 14.44,
-  "tests/integration/test_host_mode_many_entities.py": 54.13,
-  "tests/integration/test_host_mode_many_entities_multiple_connections.py": 32.17,
-  "tests/integration/test_host_mode_noise_encryption.py": 42.77,
-  "tests/integration/test_host_mode_reconnect.py": 4.06,
-  "tests/integration/test_host_mode_sensor.py": 13.47,
-  "tests/integration/test_host_ota.py": 21.4,
-  "tests/integration/test_host_preferences.py": 25.43,
-  "tests/integration/test_host_preferences_suspend_resume.py": 19.2,
-  "tests/integration/test_improv_serial_uart.py": 31.52,
-  "tests/integration/test_large_message_batching.py": 15.64,
-  "tests/integration/test_legacy_area.py": 22.63,
-  "tests/integration/test_legacy_climate_compat.py": 26.13,
-  "tests/integration/test_legacy_fan_compat.py": 24.05,
-  "tests/integration/test_light_automations.py": 30.86,
-  "tests/integration/test_light_binary_effect_off_phase.py": 23.19,
-  "tests/integration/test_light_calls.py": 32.35,
-  "tests/integration/test_light_constant_brightness.py": 29.89,
-  "tests/integration/test_light_control_action.py": 29.06,
-  "tests/integration/test_light_dim_relative_action.py": 29.61,
-  "tests/integration/test_light_effect_zero_brightness.py": 18.68,
-  "tests/integration/test_light_initial_state.py": 24.49,
-  "tests/integration/test_light_toggle_action.py": 26.46,
-  "tests/integration/test_lock_automations.py": 23.28,
-  "tests/integration/test_logger_buffered_recursion_guard.py": 24.29,
-  "tests/integration/test_loop_disable_enable.py": 45.28,
-  "tests/integration/test_loop_interval_decoupling.py": 28.35,
-  "tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.97,
-  "tests/integration/test_micros_to_millis.py": 20.79,
-  "tests/integration/test_multi_click_trigger.py": 26.2,
-  "tests/integration/test_multi_device_preferences.py": 16.87,
-  "tests/integration/test_noise_encryption_key_protection.py": 77.05,
-  "tests/integration/test_object_id_api_verification.py": 73.51,
-  "tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 62.33,
-  "tests/integration/test_object_id_no_friendly_name.py": 43.47,
-  "tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 32.21,
-  "tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 56.86,
-  "tests/integration/test_online_image_bmp.py": 50.9,
-  "tests/integration/test_oversized_payloads.py": 53.2,
-  "tests/integration/test_preference_key_stability.py": 26.09,
-  "tests/integration/test_runtime_stats.py": 18.34,
-  "tests/integration/test_safe_mode_loop_runs.py": 10.07,
-  "tests/integration/test_scheduler_blocking_warning.py": 40.91,
-  "tests/integration/test_scheduler_bulk_cleanup.py": 23.14,
-  "tests/integration/test_scheduler_defer_cancel.py": 24.54,
-  "tests/integration/test_scheduler_defer_cancel_regular.py": 13.48,
-  "tests/integration/test_scheduler_defer_fifo_simple.py": 26.86,
-  "tests/integration/test_scheduler_defer_stress.py": 27.23,
-  "tests/integration/test_scheduler_heap_stress.py": 24.02,
-  "tests/integration/test_scheduler_internal_id_no_collision.py": 24.57,
-  "tests/integration/test_scheduler_interval_reschedule.py": 13.12,
-  "tests/integration/test_scheduler_interval_zero_coerced.py": 22.91,
-  "tests/integration/test_scheduler_null_name.py": 23.46,
-  "tests/integration/test_scheduler_numeric_id_test.py": 24.54,
-  "tests/integration/test_scheduler_pool.py": 25.0,
-  "tests/integration/test_scheduler_rapid_cancellation.py": 14.68,
-  "tests/integration/test_scheduler_recursive_timeout.py": 25.35,
-  "tests/integration/test_scheduler_removed_item_race.py": 26.19,
-  "tests/integration/test_scheduler_self_keyed.py": 23.43,
-  "tests/integration/test_scheduler_simultaneous_callbacks.py": 22.16,
-  "tests/integration/test_scheduler_string_test.py": 15.22,
-  "tests/integration/test_script_array_params.py": 14.67,
-  "tests/integration/test_script_delay_params.py": 15.65,
-  "tests/integration/test_script_queued.py": 24.93,
-  "tests/integration/test_script_queued_idle_loop.py": 5.04,
-  "tests/integration/test_script_wait_on_boot.py": 13.08,
-  "tests/integration/test_select_stringref_trigger.py": 29.6,
-  "tests/integration/test_sensor_filters_delta.py": 28.01,
-  "tests/integration/test_sensor_filters_ring_buffer.py": 25.04,
-  "tests/integration/test_sensor_filters_sliding_window.py": 71.5,
-  "tests/integration/test_sensor_filters_value_list.py": 16.94,
-  "tests/integration/test_sensor_timeout_filter.py": 29.48,
-  "tests/integration/test_socket_wake_gate_tcp.py": 20.36,
-  "tests/integration/test_status_flags.py": 37.42,
-  "tests/integration/test_strftime_to.py": 22.61,
-  "tests/integration/test_syslog.py": 16.34,
-  "tests/integration/test_template_alarm_control_panel_many_sensors.py": 29.81,
-  "tests/integration/test_template_text_save.py": 25.43,
-  "tests/integration/test_text_command.py": 23.34,
-  "tests/integration/test_text_sensor_raw_state.py": 69.57,
-  "tests/integration/test_uart_mock_ld2410.py": 37.95,
-  "tests/integration/test_uart_mock_ld2412.py": 93.22,
-  "tests/integration/test_uart_mock_ld2420.py": 43.24,
-  "tests/integration/test_uart_mock_ld2450.py": 31.75,
-  "tests/integration/test_uart_mock_modbus.py": 667.4,
-  "tests/integration/test_udp.py": 9.38,
-  "tests/integration/test_use_address_runtime.py": 37.05,
-  "tests/integration/test_valve_control_action.py": 24.47,
-  "tests/integration/test_varint_five_byte_device_id.py": 25.03,
-  "tests/integration/test_wait_until_mid_loop_timing.py": 23.73,
-  "tests/integration/test_wait_until_on_boot.py": 9.16,
-  "tests/integration/test_wait_until_ordering.py": 13.3,
-  "tests/integration/test_wait_until_reentrant_restart.py": 25.23,
-  "tests/integration/test_wake_loop_forces_phase_b.py": 23.34,
-  "tests/integration/test_water_heater_template.py": 17.67
+  "tests/integration/test_action_concurrent_reentry.py": 57.91,
+  "tests/integration/test_addressable_light_transition.py": 21.25,
+  "tests/integration/test_alarm_control_panel_state_transitions.py": 70.71,
+  "tests/integration/test_api_action_metadata.py": 66.6,
+  "tests/integration/test_api_action_responses.py": 36.1,
+  "tests/integration/test_api_action_timeout.py": 68.86,
+  "tests/integration/test_api_conditional_memory.py": 15.48,
+  "tests/integration/test_api_custom_services.py": 18.77,
+  "tests/integration/test_api_get_time_response_timezone.py": 21.08,
+  "tests/integration/test_api_homeassistant.py": 65.59,
+  "tests/integration/test_api_homeassistant_action_no_subscriber.py": 18.44,
+  "tests/integration/test_api_homeassistant_binary_sensor_initial_state.py": 15.05,
+  "tests/integration/test_api_list_entities_backpressure.py": 13.88,
+  "tests/integration/test_api_message_size_batching.py": 29.98,
+  "tests/integration/test_api_reboot_timeout.py": 16.05,
+  "tests/integration/test_api_string_lambda.py": 15.31,
+  "tests/integration/test_api_vv_logging.py": 19.28,
+  "tests/integration/test_api_zero_psk_provisioning.py": 31.5,
+  "tests/integration/test_areas_and_devices.py": 24.95,
+  "tests/integration/test_automation_wait_actions.py": 20.92,
+  "tests/integration/test_automations.py": 35.19,
+  "tests/integration/test_batch_delay_zero_rapid_transitions.py": 17.99,
+  "tests/integration/test_binary_sensor_autorepeat_filter.py": 20.39,
+  "tests/integration/test_binary_sensor_invalidate_state.py": 18.41,
+  "tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 24.69,
+  "tests/integration/test_build_info.py": 18.7,
+  "tests/integration/test_camera_mock.py": 16.23,
+  "tests/integration/test_climate_control_action.py": 21.14,
+  "tests/integration/test_climate_custom_modes.py": 20.74,
+  "tests/integration/test_continuation_actions.py": 16.81,
+  "tests/integration/test_cover_control_action.py": 20.34,
+  "tests/integration/test_crc8_helper.py": 9.36,
+  "tests/integration/test_device_id_in_state.py": 44.67,
+  "tests/integration/test_duplicate_entities.py": 23.58,
+  "tests/integration/test_entity_icon.py": 34.35,
+  "tests/integration/test_fan_turn_on_action.py": 24.23,
+  "tests/integration/test_fnv1_hash_object_id.py": 16.21,
+  "tests/integration/test_fnv1a_hash.py": 13.38,
+  "tests/integration/test_gpio_expander_cache.py": 13.06,
+  "tests/integration/test_host_logger_thread_safety.py": 23.66,
+  "tests/integration/test_host_mode_basic.py": 8.01,
+  "tests/integration/test_host_mode_batch_delay.py": 21.0,
+  "tests/integration/test_host_mode_climate_basic_state.py": 22.14,
+  "tests/integration/test_host_mode_climate_control.py": 19.39,
+  "tests/integration/test_host_mode_empty_string_options.py": 21.76,
+  "tests/integration/test_host_mode_entity_fields.py": 29.61,
+  "tests/integration/test_host_mode_fan_preset.py": 20.01,
+  "tests/integration/test_host_mode_many_entities.py": 39.08,
+  "tests/integration/test_host_mode_many_entities_multiple_connections.py": 23.92,
+  "tests/integration/test_host_mode_noise_encryption.py": 42.42,
+  "tests/integration/test_host_mode_reconnect.py": 3.41,
+  "tests/integration/test_host_mode_sensor.py": 22.96,
+  "tests/integration/test_host_ota.py": 29.5,
+  "tests/integration/test_host_preferences.py": 16.06,
+  "tests/integration/test_host_preferences_suspend_resume.py": 18.71,
+  "tests/integration/test_improv_serial_uart.py": 20.22,
+  "tests/integration/test_large_message_batching.py": 26.56,
+  "tests/integration/test_legacy_area.py": 22.72,
+  "tests/integration/test_legacy_climate_compat.py": 14.13,
+  "tests/integration/test_legacy_fan_compat.py": 14.33,
+  "tests/integration/test_light_automations.py": 18.81,
+  "tests/integration/test_light_binary_effect_off_phase.py": 8.38,
+  "tests/integration/test_light_calls.py": 21.88,
+  "tests/integration/test_light_constant_brightness.py": 59.45,
+  "tests/integration/test_light_control_action.py": 31.91,
+  "tests/integration/test_light_dim_relative_action.py": 14.43,
+  "tests/integration/test_light_effect_zero_brightness.py": 25.05,
+  "tests/integration/test_light_initial_state.py": 18.97,
+  "tests/integration/test_light_toggle_action.py": 17.44,
+  "tests/integration/test_lock_automations.py": 18.9,
+  "tests/integration/test_logger_buffered_recursion_guard.py": 18.2,
+  "tests/integration/test_loop_disable_enable.py": 63.35,
+  "tests/integration/test_loop_interval_decoupling.py": 17.7,
+  "tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.56,
+  "tests/integration/test_micros_to_millis.py": 15.89,
+  "tests/integration/test_multi_click_trigger.py": 17.23,
+  "tests/integration/test_multi_device_preferences.py": 19.4,
+  "tests/integration/test_noise_encryption_key_protection.py": 72.59,
+  "tests/integration/test_object_id_api_verification.py": 19.22,
+  "tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 16.77,
+  "tests/integration/test_object_id_no_friendly_name.py": 45.8,
+  "tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 86.73,
+  "tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 40.4,
+  "tests/integration/test_online_image_bmp.py": 37.24,
+  "tests/integration/test_oversized_payloads.py": 55.75,
+  "tests/integration/test_preference_key_stability.py": 25.49,
+  "tests/integration/test_runtime_stats.py": 29.81,
+  "tests/integration/test_safe_mode_loop_runs.py": 6.26,
+  "tests/integration/test_scheduler_blocking_warning.py": 37.98,
+  "tests/integration/test_scheduler_bulk_cleanup.py": 18.67,
+  "tests/integration/test_scheduler_defer_cancel.py": 18.46,
+  "tests/integration/test_scheduler_defer_cancel_regular.py": 16.34,
+  "tests/integration/test_scheduler_defer_fifo_simple.py": 18.26,
+  "tests/integration/test_scheduler_defer_stress.py": 17.74,
+  "tests/integration/test_scheduler_heap_stress.py": 3.89,
+  "tests/integration/test_scheduler_internal_id_no_collision.py": 20.01,
+  "tests/integration/test_scheduler_interval_reschedule.py": 16.29,
+  "tests/integration/test_scheduler_interval_zero_coerced.py": 16.09,
+  "tests/integration/test_scheduler_null_name.py": 14.69,
+  "tests/integration/test_scheduler_numeric_id_test.py": 17.08,
+  "tests/integration/test_scheduler_pool.py": 19.88,
+  "tests/integration/test_scheduler_rapid_cancellation.py": 4.42,
+  "tests/integration/test_scheduler_recursive_timeout.py": 4.3,
+  "tests/integration/test_scheduler_removed_item_race.py": 15.49,
+  "tests/integration/test_scheduler_self_keyed.py": 25.77,
+  "tests/integration/test_scheduler_simultaneous_callbacks.py": 14.84,
+  "tests/integration/test_scheduler_string_test.py": 15.42,
+  "tests/integration/test_script_array_params.py": 12.73,
+  "tests/integration/test_script_delay_params.py": 12.69,
+  "tests/integration/test_script_queued.py": 20.38,
+  "tests/integration/test_script_queued_idle_loop.py": 25.06,
+  "tests/integration/test_script_wait_on_boot.py": 15.67,
+  "tests/integration/test_select_stringref_trigger.py": 19.48,
+  "tests/integration/test_sensor_filters_delta.py": 27.62,
+  "tests/integration/test_sensor_filters_ring_buffer.py": 20.27,
+  "tests/integration/test_sensor_filters_sliding_window.py": 56.28,
+  "tests/integration/test_sensor_filters_value_list.py": 20.6,
+  "tests/integration/test_sensor_timeout_filter.py": 22.21,
+  "tests/integration/test_socket_wake_gate_tcp.py": 16.37,
+  "tests/integration/test_status_flags.py": 29.68,
+  "tests/integration/test_strftime_to.py": 17.42,
+  "tests/integration/test_syslog.py": 18.39,
+  "tests/integration/test_template_alarm_control_panel_many_sensors.py": 25.61,
+  "tests/integration/test_template_text_save.py": 19.16,
+  "tests/integration/test_text_command.py": 16.43,
+  "tests/integration/test_text_sensor_raw_state.py": 17.19,
+  "tests/integration/test_uart_mock_ld2410.py": 37.0,
+  "tests/integration/test_uart_mock_ld2412.py": 40.82,
+  "tests/integration/test_uart_mock_ld2420.py": 32.7,
+  "tests/integration/test_uart_mock_ld2450.py": 32.84,
+  "tests/integration/test_uart_mock_modbus.py": 548.87,
+  "tests/integration/test_udp.py": 16.67,
+  "tests/integration/test_use_address_runtime.py": 27.26,
+  "tests/integration/test_valve_control_action.py": 24.58,
+  "tests/integration/test_varint_five_byte_device_id.py": 22.5,
+  "tests/integration/test_wait_until_mid_loop_timing.py": 22.05,
+  "tests/integration/test_wait_until_on_boot.py": 10.37,
+  "tests/integration/test_wait_until_ordering.py": 18.23,
+  "tests/integration/test_wait_until_reentrant_restart.py": 19.35,
+  "tests/integration/test_wake_loop_forces_phase_b.py": 17.83,
+  "tests/integration/test_water_heater_template.py": 25.7
 }

tests/integration/test_api_zero_psk_provisioning.py

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

tests/integration/test_host_preferences_suspend_resume.py

@@ -41,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

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

@@ -4,16 +4,12 @@ The rule flags an if/else/for/while whose only body is an unbraced ESP_LOG*() ca
 empty statement -- and a -Wempty-body warning -- once the log level compiles the macro out). These
 tests pin the comment/string/raw-string masker, the accepted control-statement shapes, and the
 NOLINT escape hatch at both placements a contributor would try.
-
-Also covers the ESP_LOG call scanner (_iter_log_calls) and the bare-literal-ternary lint.
 """
 
 import importlib.util
 from pathlib import Path
 import sys
 
-import pytest
-
 SCRIPT_DIR = (Path(__file__).parent / ".." / ".." / "script").resolve()
 sys.path.insert(0, str(SCRIPT_DIR))
 _spec = importlib.util.spec_from_file_location("ci_custom", SCRIPT_DIR / "ci-custom.py")
@@ -149,125 +145,3 @@ def test_nolint_at_end_of_log_line_suppresses() -> None:
 
 def test_nolint_on_control_line_suppresses() -> None:
     assert not _lint("if (x)  // NOLINT\n  ESP_LOGD(t);\n")
-
-
-# --- ESP_LOG call scanner and bare-literal-ternary lint ---
-
-
-def _calls(content: str) -> list[str | None]:
-    return [text for _, text in ci_custom._iter_log_calls(content)]
-
-
-def _ternary_errors(content: str) -> list[tuple[int, int]]:
-    errs = ci_custom.lint_log_no_bare_literal_ternary(Path("x.cpp"), content)
-    return [(line, col) for line, col, _ in errs]
-
-
-@pytest.mark.parametrize(
-    "content",
-    [
-        'ESP_LOGD(TAG, "a ) b ( c; d")',
-        'ESP_LOGD(TAG, "quote \\" inside")',
-        "ESP_LOGD(TAG, \"%s\", format_hex_pretty(x, '-', false).c_str())",
-        "ESP_LOGD(TAG, \"%c%c\", '(', ')')",
-        "ESP_LOGD(TAG, \"%d\", 1'000'000)",
-        'ESP_LOGD(TAG,  // it\'s a comment with ) and (\n         "x")',
-        'ESP_LOGD(TAG, /* :) */ "x")',
-        'ESP_LOGD(TAG, "%s", R"(say "hi" :) )")',
-        'ESP_LOGD(TAG, "%s", R"x(a)"b)x")',
-    ],
-)
-def test_iter_log_calls_spans_whole_call(content: str) -> None:
-    calls = _calls(content + ";\nint other = (1);")
-    assert calls == [content]
-
-
-def test_iter_log_calls_reports_unbalanced_call_once() -> None:
-    content = 'ESP_LOGD(TAG, "x";\nvoid f();'
-    assert _calls(content) == [None]
-    errs = ci_custom.lint_log_multiline_continuation(Path("x.cpp"), content)
-    assert len(errs) == 1
-    assert errs[0][:2] == (1, 1)
-    assert "no matching closing parenthesis" in errs[0][2]
-    assert _ternary_errors(content) == []
-
-
-@pytest.mark.parametrize(
-    ("content", "expected"),
-    [
-        # A ; inside the format string no longer cuts the call short
-        ('ESP_LOGD(TAG, "a; b\\nc %s", x);', [(1, 20)]),
-        # A \n%s continuation is exempt since %s may expand to leading whitespace
-        ('ESP_LOGD(TAG, "a\\n%s", x);', []),
-        ('ESP_LOGD(TAG, "a\\n  b");', []),
-    ],
-)
-def test_multiline_continuation_detection(
-    content: str, expected: list[tuple[int, int]]
-) -> None:
-    errs = ci_custom.lint_log_multiline_continuation(Path("x.cpp"), content)
-    assert [(line, col) for line, col, _ in errs] == expected
-
-
-def test_exclusion_list_only_names_components_without_esp8266_tests() -> None:
-    root = Path(__file__).parent / ".." / ".."
-    for pattern in ci_custom.LOG_LITERAL_LINT_EXCLUDE:
-        if not pattern.startswith("esphome/components/"):
-            continue
-        prefix = pattern.removeprefix("esphome/components/").split("/")[0]
-        comps = list((root / "esphome" / "components").glob(prefix))
-        assert comps, f"{pattern!r} matches no component"
-        for comp in comps:
-            test = root / "tests" / "components" / comp.name / "test.esp8266-ard.yaml"
-            assert not test.exists(), (
-                f"{comp.name} builds for ESP8266, drop {pattern!r}"
-            )
-
-
-def test_unbalanced_calls_are_reported_by_a_check_that_sees_every_file() -> None:
-    # lint_log_no_bare_literal_ternary skips unbalanced calls and relies on this
-    checks = {c["func"].__name__: c for c in ci_custom.LINT_CONTENT_CHECKS}
-    continuation = checks["lint_log_multiline_continuation"]
-    ternary = checks["lint_log_no_bare_literal_ternary"]
-    assert continuation["exclude"] == []
-    assert continuation["include"] == ternary["include"]
-
-
-@pytest.mark.parametrize(
-    ("content", "expected"),
-    [
-        ('ESP_LOGD(TAG, "%s", x ? "on" : "off");', [(1, 25), (1, 32)]),
-        (
-            'ESP_LOGD(TAG, "%s", x ? LOG_STR_LITERAL("on") : LOG_STR_LITERAL("off"));',
-            [],
-        ),
-        ('ESP_LOGD(TAG, "%s", x ? LOG_STR_LITERAL("on") : "off");', [(1, 49)]),
-        ('ESP_LOGD(TAG, "%s", x ? "on" : "");', [(1, 25)]),
-        (
-            'ESP_LOGD(TAG, "%s",\n         x ? "yes"\n           : "no");',
-            [(2, 14), (3, 14)],
-        ),
-        ("ESP_LOGD(TAG, \"%c\", x ? '1' : '0');", []),
-        ('ESP_LOGD(TAG, "a ? b : c %s", x ? "on" : "off");', [(1, 35), (1, 42)]),
-        ('ESP_LOGD(TAG, "x:" "y %s", p);', []),
-        ('ESP_LOGD(TAG, "%s", x ? "on" : "off");  // NOLINT', []),
-        ('ESP_LOGD(TAG, "%s",\n         x ? "yes"\n           : "no");  // NOLINT', []),
-        ('ESP_LOGD(TAG, "%s", x ? /* c */ "on" : "off");', [(1, 33), (1, 40)]),
-        (
-            'ESP_LOGD(TAG, "%s",\n         x ? "on"  // NOLINT(some-clang-check)\n           : "off");',
-            [(2, 14), (3, 14)],
-        ),
-    ],
-)
-def test_ternary_literal_detection(
-    content: str, expected: list[tuple[int, int]]
-) -> None:
-    assert _ternary_errors(content) == expected
-
-
-def test_ternary_error_message_names_the_literal() -> None:
-    errs = ci_custom.lint_log_no_bare_literal_ternary(
-        Path("x.cpp"), 'ESP_LOGD(TAG, "%s", x ? "enabled" : LOG_STR_LITERAL("off"));'
-    )
-    assert len(errs) == 1
-    assert 'LOG_STR_LITERAL("enabled")' in errs[0][2]

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)]