[core] Defer voluptuous and bundle imports out of the upload and logs fast path (#18093)

tests/unit_tests/fixtures/lazy_imports/_storage.py

@@ -0,0 +1,27 @@
+"""Shared storage-sidecar factory for the lazy-import fixture scripts."""
+
+from esphome.storage_json import StorageJSON
+
+
+def make_storage() -> StorageJSON:
+    """A minimal post-compile esp32 sidecar the upload/logs fast path accepts."""
+    return StorageJSON(
+        storage_version=1,
+        name="test",
+        friendly_name="Test",
+        comment=None,
+        esphome_version="2026.1.0",
+        src_version=1,
+        address="1.2.3.4",
+        web_port=None,
+        target_platform="ESP32S3",
+        build_path=None,
+        firmware_bin_path=None,
+        loaded_integrations=set(),
+        loaded_platforms=set(),
+        no_mdns=False,
+        framework="esp-idf",
+        core_platform="esp32",
+        area=None,
+        framework_version="5.3.1",
+    )

tests/unit_tests/fixtures/lazy_imports/storage_json_fast_path.py

@@ -7,32 +7,12 @@ in on argv, the ones found in sys.modules afterwards go out on stdout.
 import sys
 
 from _leak_report import print_leaked_modules
+from _storage import make_storage
 
 from esphome.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT
 from esphome.core import CORE, Version
-from esphome.storage_json import StorageJSON
 
-storage = StorageJSON(
-    storage_version=1,
-    name="test",
-    friendly_name="Test",
-    comment=None,
-    esphome_version="2026.1.0",
-    src_version=1,
-    address="1.2.3.4",
-    web_port=None,
-    target_platform="ESP32S3",
-    build_path=None,
-    firmware_bin_path=None,
-    loaded_integrations=set(),
-    loaded_platforms=set(),
-    no_mdns=False,
-    framework="esp-idf",
-    core_platform="esp32",
-    area=None,
-    framework_version="5.3.1",
-)
-storage.apply_to_core()
+make_storage().apply_to_core()
 
 # Fail loudly if the esp32 fast path stopped doing its work; otherwise an
 # empty leak list could just mean nothing ran. Explicit exits rather than

tests/unit_tests/fixtures/lazy_imports/upload_command_fast_path.py

@@ -0,0 +1,66 @@
+"""Run the upload command dispatch path and report which heavy modules loaded.
+
+Executed as a subprocess by test_lazy_imports.py: heavy module names come
+in on argv, the ones found in sys.modules afterwards go out on stdout.
+Covers both fast-path claims: the bundle suffix check in run_esphome reads
+BUNDLE_EXTENSION from esphome.const without importing esphome.bundle, and
+the real validated-config cache parse, include resolution included, stays
+voluptuous free.
+"""
+
+import os
+from pathlib import Path
+import sys
+import tempfile
+from unittest.mock import patch
+
+from _leak_report import print_leaked_modules
+from _storage import make_storage
+import yaml
+
+from esphome import __main__ as main_mod
+
+CONFIG_TEXT = "esphome:\n  name: t\n"
+
+# An ambient data-dir override would relocate the storage tree away
+# from the tmp config dir this fixture builds.
+os.environ.pop("ESPHOME_DATA_DIR", None)
+os.environ.pop("ESPHOME_IS_HA_ADDON", None)
+
+with tempfile.TemporaryDirectory() as _td:
+    tmp = Path(_td)
+    conf_path = tmp / "test.yaml"
+    conf_path.write_text(CONFIG_TEXT)
+
+    storage_dir = tmp / ".esphome" / "storage"
+    storage_dir.mkdir(parents=True)
+    # The cache is a top-level !include so loading it resolves an
+    # IncludeFile for real on the fast path. The sidecar is written to the
+    # layout ext_storage_path resolves once run_esphome sets
+    # CORE.config_path; going through CORE here would be circular.
+    (storage_dir / "inc.yaml").write_text(CONFIG_TEXT)
+    cache_path = storage_dir / "test.yaml.validated.yaml"
+    cache_path.write_text("!include inc.yaml\n")
+    os.utime(cache_path)  # keep the cache at least as fresh as the source
+    make_storage().save(storage_dir / "test.yaml.json")
+
+    dispatched = {}
+
+    def fake_upload(args, config):
+        dispatched["config"] = config
+        return 0
+
+    with patch.dict(main_mod.POST_CONFIG_ACTIONS, {"upload": fake_upload}):
+        exit_code = main_mod.run_esphome(
+            ["esphome", "upload", str(conf_path), "--device", "192.0.2.1"]
+        )
+
+    # Fail loudly if the fast path didn't do its work; otherwise an empty
+    # leak list could just mean nothing ran. Explicit exits rather than
+    # asserts so PYTHONOPTIMIZE in the ambient environment can't strip them.
+    if exit_code != 0:
+        sys.exit(f"run_esphome exited {exit_code} before dispatching upload")
+    if dispatched.get("config") != yaml.safe_load(CONFIG_TEXT):
+        sys.exit(f"cache include did not resolve through the fast path: {dispatched!r}")
+
+    print_leaked_modules()

tests/unit_tests/test_bundle.py

@@ -25,7 +25,6 @@ from esphome.bundle import (
     add_bundle_file,
     add_secret_scan_dir,
     extract_bundle,
-    is_bundle_path,
     prepare_bundle_for_compile,
     read_bundle_manifest,
     remap_bundle_path,
@@ -99,26 +98,6 @@ def _setup_config_dir(
     return config_dir
 
 
-# ---------------------------------------------------------------------------
-# is_bundle_path
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.parametrize(
-    ("filename", "expected"),
-    [
-        (f"my_device{BUNDLE_EXTENSION}", True),
-        (f"MY_DEVICE{BUNDLE_EXTENSION.upper()}", True),
-        ("my_device.yaml", False),
-        ("my_device.tar.gz", False),
-        ("my_device.zip", False),
-        ("", False),
-    ],
-)
-def test_is_bundle_path(filename: str, expected: bool) -> None:
-    assert is_bundle_path(Path(filename)) is expected
-
-
 # ---------------------------------------------------------------------------
 # _default_target_dir
 # ---------------------------------------------------------------------------

tests/unit_tests/test_lazy_imports.py

@@ -41,6 +41,11 @@ FAST_PATH_HEAVY_MODULES = HEAVY_MODULES + ("esphome.components.esp32",)
 # in the existence guard so a rename can't silently no-op its check.
 API_HEAVY_MODULES = ("aioesphomeapi",)
 
+# Heavy only for the single-config dispatch path: the bundle suffix
+# check reads BUNDLE_EXTENSION from esphome.const so an ordinary run
+# never pays for the bundle machinery and its tarfile chain.
+BUNDLE_HEAVY_MODULES = ("esphome.bundle", "tarfile")
+
 
 def _leaked_heavy_modules(module: str, extra: tuple[str, ...] = ()) -> str:
     """Import ``module`` in a subprocess and report the heavy modules it pulled.
@@ -77,13 +82,15 @@ def test_main_module_does_not_import_heavy_modules() -> None:
 
 def test_watched_heavy_modules_exist() -> None:
     """A renamed heavy module would silently disable the leak checks."""
-    for module in FAST_PATH_HEAVY_MODULES + API_HEAVY_MODULES:
+    for module in FAST_PATH_HEAVY_MODULES + API_HEAVY_MODULES + BUNDLE_HEAVY_MODULES:
         assert importlib.util.find_spec(module) is not None, (
             f"{module} no longer resolves; update the heavy-module lists"
         )
 
 
-def _leaked_from_fixture(fixture_path: Path, script_name: str) -> str:
+def _leaked_from_fixture(
+    fixture_path: Path, script_name: str, extra: tuple[str, ...] = ()
+) -> str:
     """Run a fixture script with the watched modules on argv.
 
     Running a script file drops the cwd from sys.path, so prepend the
@@ -95,7 +102,7 @@ def _leaked_from_fixture(fixture_path: Path, script_name: str) -> str:
         python_path = os.pathsep.join((python_path, ambient))
     env = os.environ | {"PYTHONPATH": python_path}
     result = subprocess.run(
-        [sys.executable, str(script), *FAST_PATH_HEAVY_MODULES],
+        [sys.executable, str(script), *FAST_PATH_HEAVY_MODULES, *extra],
         capture_output=True,
         text=True,
         env=env,
@@ -212,3 +219,34 @@ def test_has_mqtt_ip_lookup_does_not_import_mqtt() -> None:
         "The upload/logs fast path skips validation; importing the "
         "validation stack anyway defeats the validated-config cache."
     )
+
+
+def test_yaml_util_does_not_import_heavy_modules() -> None:
+    """``esphome.yaml_util`` parses the validated-config cache on the
+    upload/logs fast path; importing it must not pull in voluptuous.
+    """
+    leaked = _leaked_heavy_modules("esphome.yaml_util")
+    assert not leaked, (
+        f"esphome.yaml_util imports heavy modules at top level: {leaked}. "
+        "The upload/logs fast path skips validation; importing the "
+        "validation stack anyway defeats the validated-config cache."
+    )
+
+
+def test_upload_command_path_does_not_import_heavy_modules(
+    fixture_path: Path,
+) -> None:
+    """The single-config dispatch path checks the bundle suffix on every
+    run; reading it from esphome.const must not drag in esphome.bundle
+    and its tarfile chain.
+    """
+    leaked = _leaked_from_fixture(
+        fixture_path, "upload_command_fast_path.py", extra=BUNDLE_HEAVY_MODULES
+    )
+    assert not leaked, (
+        f"the upload dispatch path pulls in heavy modules: {leaked}. "
+        "An ordinary run only needs the bundle suffix constant, and the "
+        "cache parse must not resolve voluptuous; keep the esphome.bundle "
+        "import inside the branch that extracts one and the Invalid import "
+        "inside the branch that raises it."
+    )

tests/unit_tests/test_main.py

@@ -6307,7 +6307,6 @@ def test_run_esphome_bundle_detection(tmp_path: Path) -> None:
     extracted_yaml = tmp_path / "extracted" / "device.yaml"
 
     with (
-        patch("esphome.bundle.is_bundle_path", return_value=True) as mock_is_bundle,
         patch(
             "esphome.bundle.prepare_bundle_for_compile",
             return_value=extracted_yaml,
@@ -6316,7 +6315,6 @@ def test_run_esphome_bundle_detection(tmp_path: Path) -> None:
     ):
         result = run_esphome(["esphome", "compile", str(bundle_path)])
 
-    mock_is_bundle.assert_called_once()
     mock_prepare.assert_called_once_with(bundle_path)
     # read_config returns None → exit code 2
     assert result == 2
@@ -6328,13 +6326,11 @@ def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
     yaml_file.write_text("esphome:\n  name: test\n")
 
     with (
-        patch("esphome.bundle.is_bundle_path", return_value=False) as mock_is_bundle,
         patch("esphome.bundle.prepare_bundle_for_compile") as mock_prepare,
         patch("esphome.config.read_config", return_value=None),
     ):
         result = run_esphome(["esphome", "compile", str(yaml_file)])
 
-    mock_is_bundle.assert_called_once()
     mock_prepare.assert_not_called()
     assert result == 2