[core] Defer stdlib imports out of the upload and logs fast path (#18105)

tests/unit_tests/fixtures/lazy_imports/upload_command_fast_path.py

@@ -18,7 +18,12 @@ from _leak_report import print_leaked_modules
 from _storage import make_storage
 import yaml
 
-from esphome import __main__ as main_mod
+# Everything imported past this point is the code under test; the pop
+# below must only drop what the setup itself preloaded, or it would
+# hide modules the dispatch chain pulls in (tarfile has no other guard).
+_FIXTURE_PRELOADED = frozenset(sys.modules)
+
+from esphome import __main__ as main_mod  # noqa: E402
 
 CONFIG_TEXT = "esphome:\n  name: t\n"
 
@@ -50,6 +55,17 @@ with tempfile.TemporaryDirectory() as _td:
         dispatched["config"] = config
         return 0
 
+    # This setup pre-imports some watched stdlib modules (tempfile above,
+    # write_file inside make_storage().save(), unittest.mock -> asyncio ->
+    # subprocess). Drop exactly those so only a genuine dispatch-time
+    # re-import is reported; live objects keep their references, so
+    # cleanup still works. Module-level re-imports are out of reach here
+    # (esphome.__main__ is already loaded) — the bare-import check in
+    # test_lazy_imports owns that contract.
+    for module in sys.argv[1:]:
+        if module in _FIXTURE_PRELOADED:
+            sys.modules.pop(module, None)
+
     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"]

tests/unit_tests/test_lazy_imports.py

@@ -46,6 +46,22 @@ API_HEAVY_MODULES = ("aioesphomeapi",)
 # never pays for the bundle machinery and its tarfile chain.
 BUNDLE_HEAVY_MODULES = ("esphome.bundle", "tarfile")
 
+# Stdlib modules deferred out of the dispatch fast path: a cache-hit
+# upload/logs run never writes a file (tempfile), spawns a process
+# (subprocess), parses a URL (urllib.parse), or prints a serial
+# permission hint (getpass). shutil is deferred too but unwatchable:
+# argparse imports it from every add_argument on py3.14. urllib.parse
+# is only watchable on 3.13+ where pathlib stopped importing it.
+STDLIB_FAST_PATH_MODULES = (
+    "tempfile",
+    "subprocess",
+    "getpass",
+    # Pins the module-level contract only: PyYAML's constructor loads
+    # datetime during the cache parse until the JSON cache lands.
+    "datetime",
+    *(("urllib.parse",) if sys.version_info >= (3, 13) else ()),
+)
+
 
 def _leaked_heavy_modules(module: str, extra: tuple[str, ...] = ()) -> str:
     """Import ``module`` in a subprocess and report the heavy modules it pulled.
@@ -70,8 +86,14 @@ def _leaked_heavy_modules(module: str, extra: tuple[str, ...] = ()) -> str:
 
 
 def test_main_module_does_not_import_heavy_modules() -> None:
-    """A bare ``import esphome.__main__`` must not drag in validation/codegen."""
-    leaked = _leaked_heavy_modules("esphome.__main__")
+    """A bare ``import esphome.__main__`` must not drag in validation/codegen.
+
+    The stdlib watch list rides along here because this check runs in a
+    clean subprocess: a module-level re-import anywhere on the chain is
+    caught, which the dispatch fixture (whose setup pre-imports them and
+    pops before dispatch) structurally cannot do.
+    """
+    leaked = _leaked_heavy_modules("esphome.__main__", extra=STDLIB_FAST_PATH_MODULES)
     assert not leaked, (
         f"esphome.__main__ imports heavy modules at top level: {leaked}. "
         "Import them lazily inside the command that needs them instead; "
@@ -82,7 +104,12 @@ 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 + BUNDLE_HEAVY_MODULES:
+    for module in (
+        FAST_PATH_HEAVY_MODULES
+        + API_HEAVY_MODULES
+        + BUNDLE_HEAVY_MODULES
+        + STDLIB_FAST_PATH_MODULES
+    ):
         assert importlib.util.find_spec(module) is not None, (
             f"{module} no longer resolves; update the heavy-module lists"
         )
@@ -241,12 +268,15 @@ def test_upload_command_path_does_not_import_heavy_modules(
     and its tarfile chain.
     """
     leaked = _leaked_from_fixture(
-        fixture_path, "upload_command_fast_path.py", extra=BUNDLE_HEAVY_MODULES
+        fixture_path,
+        "upload_command_fast_path.py",
+        extra=BUNDLE_HEAVY_MODULES + STDLIB_FAST_PATH_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."
+        "import inside the branch that extracts one, the Invalid import "
+        "inside the branch that raises it, and the deferred stdlib "
+        "imports inside the write/spawn/serial helpers that use them."
     )

tests/unit_tests/test_main.py

@@ -27,6 +27,7 @@ from esphome.__main__ import (
     _unresolved_default_error,
     _validate_bootloader_binary,
     _validate_partition_table_binary,
+    check_permissions,
     choose_upload_log_host,
     command_analyze_memory,
     command_bundle,
@@ -6715,3 +6716,27 @@ def test_command_idedata_esp_idf_no_build_errors() -> None:
         result = command_idedata(MagicMock(), CORE.config)
 
     assert result == 1
+
+
+@pytest.mark.skipif(
+    os.name != "posix", reason="serial permission checks are posix-only"
+)
+def test_check_permissions_missing_port() -> None:
+    """A nonexistent serial port raises the does-not-exist guidance."""
+    with (
+        patch("os.access", return_value=False),
+        pytest.raises(EsphomeError, match="serial port does not exist"),
+    ):
+        check_permissions("/dev/ttyUSB99")
+
+
+@pytest.mark.skipif(
+    os.name != "posix", reason="serial permission checks are posix-only"
+)
+def test_check_permissions_unreadable_port() -> None:
+    """An existing but unreadable serial port raises the dialout guidance."""
+    with (
+        patch("os.access", side_effect=lambda _path, mode: mode == os.F_OK),
+        pytest.raises(EsphomeError, match="read or write permission"),
+    ):
+        check_permissions("/dev/ttyUSB99")

tests/unit_tests/test_util.py

@@ -561,7 +561,7 @@ def test_run_external_process_line_callbacks() -> None:
             return "PROCESS CALLBACK\n"
         return None
 
-    with patch("esphome.util.subprocess.run") as mock_run:
+    with patch("subprocess.run") as mock_run:
 
         def run_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
             # Simulate subprocess writing to the stdout RedirectText
@@ -635,7 +635,7 @@ def test_detect_rp2040_bootsel_found() -> None:
     """Test BOOTSEL device detection when device is present."""
     mock_result = MagicMock()
     mock_result.stdout = b"Device Information\n type: RP2040\n"
-    with patch("esphome.util.subprocess.run", return_value=mock_result):
+    with patch("subprocess.run", return_value=mock_result):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 1
     assert result.permission_error is False
@@ -645,7 +645,7 @@ def test_detect_rp2040_bootsel_multiple() -> None:
     """Test BOOTSEL detection with multiple devices."""
     mock_result = MagicMock()
     mock_result.stdout = b"type: RP2040\ntype: RP2350\n"
-    with patch("esphome.util.subprocess.run", return_value=mock_result):
+    with patch("subprocess.run", return_value=mock_result):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 2
     assert result.permission_error is False
@@ -658,7 +658,7 @@ def test_detect_rp2040_bootsel_none() -> None:
         b"No accessible RP2040/RP2350 devices in BOOTSEL mode were found.\n"
     )
     mock_result.stderr = b""
-    with patch("esphome.util.subprocess.run", return_value=mock_result):
+    with patch("subprocess.run", return_value=mock_result):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 0
     assert result.permission_error is False
@@ -675,7 +675,7 @@ def test_detect_rp2040_bootsel_permission_error() -> None:
         b"but picotool was unable to connect. "
         b"Maybe try 'sudo' or check your permissions.\n"
     )
-    with patch("esphome.util.subprocess.run", return_value=mock_result):
+    with patch("subprocess.run", return_value=mock_result):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 0
     assert result.permission_error is True
@@ -686,7 +686,7 @@ def test_detect_rp2040_bootsel_libusb_access_error() -> None:
     mock_result = MagicMock()
     mock_result.stdout = b""
     mock_result.stderr = b"LIBUSB_ERROR_ACCESS\n"
-    with patch("esphome.util.subprocess.run", return_value=mock_result):
+    with patch("subprocess.run", return_value=mock_result):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 0
     assert result.permission_error is True
@@ -694,7 +694,7 @@ def test_detect_rp2040_bootsel_libusb_access_error() -> None:
 
 def test_detect_rp2040_bootsel_oserror() -> None:
     """Test BOOTSEL detection handles OSError."""
-    with patch("esphome.util.subprocess.run", side_effect=OSError("not found")):
+    with patch("subprocess.run", side_effect=OSError("not found")):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")
     assert result.device_count == 0
     assert result.permission_error is False
@@ -703,7 +703,7 @@ def test_detect_rp2040_bootsel_oserror() -> None:
 def test_detect_rp2040_bootsel_timeout() -> None:
     """Test BOOTSEL detection handles timeout."""
     with patch(
-        "esphome.util.subprocess.run",
+        "subprocess.run",
         side_effect=subprocess.TimeoutExpired("picotool", 10),
     ):
         result = util.detect_rp2040_bootsel("/usr/bin/picotool")