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

tests/unit_tests/build_helpers/test_ccache.py

@@ -0,0 +1,122 @@
+"""Tests for the shared ccache policy in esphome.build_helpers.ccache."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from esphome.build_helpers import ccache
+
+
+def test_resolve_opt_out() -> None:
+    with patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_no_binary(caplog: pytest.LogCaptureFixture) -> None:
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch("shutil.which", return_value=None),
+    ):
+        assert ccache.resolve_ccache_path() is None
+    assert "no ccache binary" not in caplog.text
+
+
+def test_resolve_probe_failure() -> None:
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")),
+    ):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_explicit_skips_probe_and_warns_missing(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch.object(ccache, "_ccache_runs", side_effect=AssertionError),
+    ):
+        assert ccache.resolve_ccache_path() == "/usr/bin/ccache"
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
+        patch("shutil.which", return_value=None),
+    ):
+        assert ccache.resolve_ccache_path() is None
+    assert "no ccache binary is on PATH" in caplog.text
+
+
+def test_probe_spawns_with_close_fds_false() -> None:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
+        assert ccache._ccache_runs("/usr/bin/ccache") is True
+    assert mock_run.call_args.kwargs["close_fds"] is False
+
+
+def test_defaults_env(tmp_path: Path) -> None:
+    with (
+        patch("esphome.core.CORE", SimpleNamespace(build_path=tmp_path / "b")),
+        patch.dict(os.environ, {"CCACHE_NOHASHDIR": "false"}, clear=True),
+    ):
+        env = ccache.ccache_defaults_env(tmp_path / "cache")
+    assert env["CCACHE_DIR"] == str(tmp_path / "cache")
+    assert env["CCACHE_DEPEND"] == "1"
+    assert "CCACHE_NOHASHDIR" not in env  # user value respected
+
+
+def test_defaults_env_requires_build_path() -> None:
+    with (
+        patch("esphome.core.CORE", SimpleNamespace(build_path=None)),
+        pytest.raises(ValueError, match="build_path"),
+    ):
+        ccache.ccache_defaults_env(Path("/x"))
+
+
+@pytest.mark.parametrize("value", ["no", "off", "false", "0"])
+def test_resolve_opt_out_synonyms(value: str) -> None:
+    """Every recognized falsy spelling disables ccache."""
+    with patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": value}):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_unrecognized_value_warns_and_probes(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An unparsable ESPHOME_CCACHE_ENABLE is treated as unset: it must not
+    silently enable ccache or skip the runnability probe."""
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "enabled"}),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch.object(ccache, "_ccache_runs", return_value=False) as mock_probe,
+    ):
+        assert ccache.resolve_ccache_path() is None
+    mock_probe.assert_called_once()
+    assert "unrecognized ESPHOME_CCACHE_ENABLE" in caplog.text
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("1", True),
+        ("enable", True),
+        ("ON", True),
+        ("0", False),
+        ("disable", False),
+        ("Off", False),
+        ("maybe", None),
+        # ENV KNOB= (Docker/CI) has always read as a disable
+        ("", False),
+        ("   ", False),
+    ],
+)
+def test_parse_enable_env_spelling_tables(
+    monkeypatch: pytest.MonkeyPatch, raw: str, expected: bool | None
+) -> None:
+    """cv.boolean's spelling tables plus the 1/0 env convention."""
+    monkeypatch.setenv("ESPHOME_CCACHE_ENABLE", raw)
+    assert ccache.parse_enable_env("ESPHOME_CCACHE_ENABLE") is expected

tests/unit_tests/build_helpers/test_ninja.py

@@ -0,0 +1,143 @@
+"""Tests for esphome.build_helpers.ninja."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import subprocess
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.build_helpers import ninja as ninja_helper
+from esphome.core import EsphomeError
+
+
+def test_find_ninja_prefers_path(tmp_path: Path) -> None:
+    with (
+        patch("shutil.which", return_value=str(tmp_path / "ninja")),
+        patch.object(ninja_helper, "_ninja_runs", return_value=True),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / "ninja"
+
+
+def test_find_ninja_falls_back_to_wheel(tmp_path: Path) -> None:
+    """Without a PATH entry, the ninja PyPI wheel's binary is used."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / binary_name
+
+
+def test_find_ninja_package_not_installed() -> None:
+    """A missing ninja package raises the actionable message, not ImportError."""
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": None}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()
+
+
+def test_find_ninja_missing_everywhere(tmp_path: Path) -> None:
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()
+
+
+def test_escape_ninja_specials() -> None:
+    assert ninja_helper.escape("a b:c$d") == "a$ b$:c$$d"
+
+
+def _q(tok: str) -> str:
+    """The platform's shell_token quote wrapper (argv rule on Windows)."""
+    return f'"{tok}"' if os.name == "nt" else f"'{tok}'"
+
+
+def test_quote_arg_windows_argv_rule() -> None:
+    # Backslash runs double only before a quote (subprocess.list2cmdline rule)
+    assert ninja_helper.quote_arg('-DX=a\\"b c') == '"-DX=a\\\\\\"b c"'
+    assert ninja_helper.quote_arg("a b\\") == '"a b\\\\"'
+
+
+def test_shell_token_quotes_only_when_needed() -> None:
+    assert ninja_helper.shell_token("-Os") == "-Os"
+    assert ninja_helper.shell_token("-DP=C:\\x y") == _q("-DP=C:\\x y")
+    assert ninja_helper.shell_token("plain", force=True) == _q("plain")
+
+
+def test_shell_token_quotes_shell_metacharacters() -> None:
+    """Tokens like -DMASK=(1<<3) must not reach /bin/sh -c bare."""
+    assert ninja_helper.shell_token("-DMASK=(1<<3)") == _q("-DMASK=(1<<3)")
+    assert ninja_helper.shell_token("-DX=a;b") == _q("-DX=a;b")
+    assert ninja_helper.shell_token("-DX=$HOME") == _q("-DX=$$HOME")
+
+
+def test_shell_token_posix_roundtrips_through_sh() -> None:
+    """Backslash runs, $, backticks, and quotes must reach the compiler
+    exactly as lexed once ninja un-doubles $$ and /bin/sh strips quotes."""
+
+    if sys.platform == "win32":
+        pytest.skip("POSIX sh quoting")
+    for tok in ("-DP=a\\\\b", "-DX=$VAR", "-DY=`date`", "-DZ=it's", '-DC="q"'):
+        quoted = ninja_helper.shell_token(tok).replace("$$", "$")
+        out = subprocess.run(
+            ["/bin/sh", "-c", f'printf "%s" {quoted}'],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        assert out.stdout == tok
+
+
+def test_quote_path_force_quotes() -> None:
+    assert ninja_helper.quote_path(Path("a b")) == _q("a b")
+    assert ninja_helper.quote_path("simple") == _q("simple")
+
+
+def test_shell_token_empty_token_is_quoted() -> None:
+    """An empty argv element must survive as an explicit pair of quotes."""
+    assert ninja_helper.shell_token("") == _q("")
+
+
+def test_find_ninja_probes_path_hit(tmp_path: Path) -> None:
+    """A broken PATH shim falls back to the wheel instead of failing every
+    build later."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value="/broken/ninja"),
+        patch.object(ninja_helper, "_ninja_runs", return_value=False),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / binary_name
+
+
+def test_ninja_probe_failure_warns(caplog: pytest.LogCaptureFixture) -> None:
+    with patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")):
+        assert ninja_helper._ninja_runs("/broken/ninja") is False
+    assert "failed to run" in caplog.text
+
+
+def test_ninja_probe_success() -> None:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
+        assert ninja_helper._ninja_runs("/usr/bin/ninja") is True
+    assert mock_run.call_args.kwargs["close_fds"] is False
+
+
+def test_shell_token_windows_branch_uses_argv_rule() -> None:
+    """The nt branch quotes with the CreateProcess argv rule (the ubuntu
+    coverage run never takes it naturally)."""
+    with patch.object(os, "name", "nt"):
+        assert ninja_helper.shell_token("a b") == '"a b"'
+        assert ninja_helper.shell_token("", force=True) == '""'

tests/unit_tests/components/api/test_api_protobuf_generator.py

@@ -15,7 +15,12 @@ import pytest
 
 sys.path.insert(0, str(Path(__file__).parents[4] / "script" / "api_protobuf"))
 
-from api_protobuf import _make_ifdef_line, get_varint64_ifdef  # noqa: E402
+from api_protobuf import (  # noqa: E402
+    MAX_MESSAGE_ID,
+    _make_ifdef_line,
+    get_varint64_ifdef,
+    validate_message_id,
+)
 from google.protobuf import descriptor_pb2  # noqa: E402
 
 
@@ -91,3 +96,14 @@ def test_make_ifdef_line_conjunction_and_negation() -> None:
     assert (
         _make_ifdef_line("USE_X && !USE_Y") == "#if defined(USE_X) && !defined(USE_Y)"
     )
+
+
+def test_message_id_at_maximum_is_accepted() -> None:
+    # 16383 is the largest ID whose plaintext type varint fits the 2 bytes
+    # budgeted in HEADER_PADDING.
+    validate_message_id(MAX_MESSAGE_ID, "MaxMessage")
+
+
+def test_message_id_above_maximum_is_rejected() -> None:
+    with pytest.raises(ValueError, match="exceeds the plaintext"):
+        validate_message_id(MAX_MESSAGE_ID + 1, "TooBigMessage")

tests/unit_tests/components/esp8266/test_boards.py

@@ -0,0 +1,33 @@
+"""Tests for the per-board linker-script rule."""
+
+import pytest
+
+from esphome.components.esp8266 import _choose_ld_script
+from esphome.components.esp8266.boards import BOARDS, board_ld_script
+import esphome.config_validation as cv
+from esphome.core import EsphomeError
+
+
+def test_d1_wroom_02_keeps_its_shipped_layout() -> None:
+    """The override must survive a BOARDS regeneration or key typo: the
+    2m.ld default moves _FS_end and the preferences sector on deployed
+    devices."""
+    assert board_ld_script(BOARDS["d1_wroom_02"]) == "eagle.flash.2m64.ld"
+
+
+def test_default_boards_use_the_flash_size_layout() -> None:
+    assert board_ld_script(BOARDS["d1_mini"]) == "eagle.flash.4m.ld"
+    assert board_ld_script(BOARDS["esp01_1m"]) == "eagle.flash.1m.ld"
+
+
+def test_choose_ld_script_paths() -> None:
+    """Old cores get the size default, overriding boards hard-error there
+    (a substituted layout would wipe flash-backed state), modern cores
+    honor the override."""
+    assert _choose_ld_script("nodemcuv2", cv.Version(2, 3, 0)) is None
+    assert _choose_ld_script("nodemcuv2", cv.Version(2, 4, 2)) == "eagle.flash.4m.ld"
+    assert _choose_ld_script("d1_wroom_02", cv.Version(2, 7, 4)) == (
+        "eagle.flash.2m64.ld"
+    )
+    with pytest.raises(EsphomeError, match="cannot honor"):
+        _choose_ld_script("d1_wroom_02", cv.Version(2, 4, 2))

tests/unit_tests/components/esp8266/test_build_surgery.py

@@ -0,0 +1,145 @@
+"""Tests for the linker-script surgery shared with the native toolchain."""
+
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+import sys
+
+import pytest
+
+from esphome.components.esp8266 import build_surgery
+from esphome.components.esp8266.boards import BOARDS, ESP8266_BOARD_BUILD
+from esphome.components.esp8266.build_surgery import (
+    RATETABLE_RULE,
+    apply_testing_memory_patches,
+    relocate_ratetable,
+    segment_length,
+)
+
+_COMMON_LD_SNIPPET = """\
+  .dport0.data : ALIGN(4)
+  {
+    _dport0_data_start = ABSOLUTE(.);
+  } >dport0_0_seg :dport0_0_phdr
+  .data : ALIGN(4)
+  {
+    _data_start = ABSOLUTE(.);
+    *(.data)
+  } >dram0_0_seg :dram0_0_phdr
+"""
+
+# Shaped like the real SDK flash ld scripts: no iram1_0_seg (that lives in
+# the generated common ld only)
+_FLASH_LD_SNIPPET = """\
+MEMORY
+{
+  dport0_0_seg :                        org = 0x3FF00000, len = 0x10
+  dram0_0_seg :                         org = 0x3FFE8000, len = 0x14000
+  irom0_0_seg :                         org = 0x40201010, len = 0xfeff0
+}
+"""
+
+# Shaped like the preprocessed common ld: MMU_IRAM_SIZE expands with a ul
+# suffix the patcher must leave in place
+_COMMON_LD_MEMORY_SNIPPET = """\
+MEMORY
+{
+  iram1_0_seg :                         org = 0x40100000, len = 0x8000ul
+}
+"""
+
+
+def test_relocate_ratetable_inserts_after_data_start() -> None:
+    patched = relocate_ratetable(_COMMON_LD_SNIPPET)
+    assert RATETABLE_RULE in patched
+    # Inserted after the .data section's anchor, not the .dport0.data one
+    # (whose closing brace bounds the decoy block)
+    assert RATETABLE_RULE not in patched[: patched.index("} >dport0_0_seg")]
+    assert patched.index(RATETABLE_RULE) < patched.index("*(.data)")
+    # Idempotent on an already-patched script
+    assert relocate_ratetable(patched) == patched
+
+
+def test_relocate_ratetable_requires_anchor() -> None:
+    with pytest.raises(RuntimeError, match="_data_start"):
+        relocate_ratetable("SECTIONS { }")
+
+
+def test_testing_memory_patches_enlarge_segments() -> None:
+    patched = apply_testing_memory_patches(
+        _FLASH_LD_SNIPPET, ("dram0_0_seg", "irom0_0_seg")
+    )
+    assert segment_length(patched, "dram0_0_seg") == 0x200000
+    assert segment_length(patched, "irom0_0_seg") == 0x2000000
+    # Untouched segments keep their sizes
+    assert segment_length(patched, "dport0_0_seg") == 0x10
+
+
+def test_testing_memory_patches_keep_ul_suffix() -> None:
+    """The common ld's preprocessed sizes carry a ul suffix; the patch must
+    replace only the hex digits, as testing_mode.py.script does."""
+    patched = apply_testing_memory_patches(_COMMON_LD_MEMORY_SNIPPET, ("iram1_0_seg",))
+    assert "len = 0x200000ul" in patched
+    assert segment_length(patched, "iram1_0_seg") == 0x200000
+
+
+def test_segment_length_requires_whole_name() -> None:
+    """A name must match its own line, never inside a longer segment name."""
+    assert segment_length(_FLASH_LD_SNIPPET, "ram0_0_seg") is None
+
+
+def test_testing_memory_patches_unknown_segment_raises() -> None:
+    with pytest.raises(RuntimeError, match="Unknown testing-mode segment"):
+        apply_testing_memory_patches("MEMORY { }", ("bogus_seg",))
+
+
+def test_segment_length() -> None:
+    assert segment_length(_FLASH_LD_SNIPPET, "irom0_0_seg") == 0xFEFF0
+    assert segment_length(_FLASH_LD_SNIPPET, "missing_seg") is None
+
+
+def test_testing_memory_patches_missing_segment_raises() -> None:
+    """A named segment the patch could not find raises instead of silently
+    keeping the real memory limits."""
+    with pytest.raises(RuntimeError, match="dram0_0_seg"):
+        apply_testing_memory_patches("MEMORY { }", ("dram0_0_seg",))
+
+
+def test_board_build_covers_every_board() -> None:
+    """Every supported board has native build metadata (the table may carry
+    extras that BOARDS does not expose)."""
+    assert set(BOARDS) <= set(ESP8266_BOARD_BUILD)
+
+
+def test_surgery_fingerprint_is_stable_and_sensitive(tmp_path) -> None:
+    """The properties the linker-script cache depends on: the fingerprint is
+    stable across calls and changes when the module's source changes."""
+
+    first = build_surgery.surgery_fingerprint()
+    assert first == build_surgery.surgery_fingerprint()
+    assert len(first) == 64
+    int(first, 16)  # sha256 hex digest
+
+    # A modified copy of the module must fingerprint differently
+    copy = tmp_path / "build_surgery_variant.py"
+    copy.write_text(
+        Path(build_surgery.__file__).read_text(encoding="utf-8")
+        + "\nEXTRA_BEHAVIORAL_INPUT = 1\n",
+        encoding="utf-8",
+    )
+    spec = importlib.util.spec_from_file_location("build_surgery_variant", copy)
+    variant = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = variant
+    try:
+        spec.loader.exec_module(variant)
+        assert variant.surgery_fingerprint() != first
+    finally:
+        del sys.modules[spec.name]
+
+
+def test_testing_memory_patches_present_but_unselected_raises() -> None:
+    """A known segment left off the caller's list must fail, not silently
+    keep its real memory limit."""
+    with pytest.raises(RuntimeError, match="not selected"):
+        apply_testing_memory_patches(_FLASH_LD_SNIPPET, ("dram0_0_seg",))

tests/unit_tests/components/lvgl/test_table_codegen.py

@@ -0,0 +1,206 @@
+"""Tests for the LVGL table widget's C++ code generation."""
+
+from __future__ import annotations
+
+import pytest
+
+from esphome.automation import ACTION_REGISTRY
+from esphome.components.lvgl.defines import set_widgets_completed
+from esphome.components.lvgl.lvcode import LvContext
+from esphome.components.lvgl.schemas import container_schema
+from esphome.components.lvgl.trigger import generate_triggers
+from esphome.components.lvgl.widgets import Widget, widget_to_code
+from esphome.components.lvgl.widgets.table import table_spec
+from esphome.const import (
+    CONF_AUTOMATION_ID,
+    CONF_ON_VALUE,
+    CONF_THEN,
+    CONF_TRIGGER_ID,
+    CONF_TYPE_ID,
+)
+from esphome.core import CORE, ID
+from esphome.cpp_generator import MockObj, TemplateArguments
+from esphome.yaml_util import make_data_base
+
+
+async def _create_table(raw_config: dict) -> Widget:
+    """Validate `raw_config` as a table widget and generate its creation code."""
+    config = container_schema(table_spec)(raw_config)
+    parent = MockObj("parent_obj")
+    async with LvContext():
+        return await widget_to_code(config, table_spec, parent)
+
+
+def _statements() -> list[str]:
+    return [str(s) for s in CORE.main_statements]
+
+
+@pytest.mark.asyncio
+async def test_create_table_sets_row_and_column_count(setup_core) -> None:
+    await _create_table(
+        {"id": "table_counts", "rows": [["Name", "Value"], ["Temp", "22.5"]]}
+    )
+    statements = _statements()
+    assert any("lv_table_set_row_count(table_counts->obj, 2)" in s for s in statements)
+    assert any(
+        "lv_table_set_column_count(table_counts->obj, 2)" in s for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_create_table_writes_cell_values(setup_core) -> None:
+    await _create_table({"id": "table_cells", "rows": [["Name", "Value"]]})
+    statements = _statements()
+    assert any(
+        'lv_table_set_cell_value(table_cells->obj, 0, 0, "Name")' in s
+        for s in statements
+    )
+    assert any(
+        'lv_table_set_cell_value(table_cells->obj, 0, 1, "Value")' in s
+        for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_create_table_sets_cell_control_flags(setup_core) -> None:
+    await _create_table(
+        {
+            "id": "table_ctrl",
+            "rows": [
+                {
+                    "cells": [
+                        {"text": "wide", "merge_right": True},
+                        {"text": "cropped", "text_crop": True},
+                    ]
+                }
+            ],
+        }
+    )
+    statements = _statements()
+    assert any(
+        "lv_table_set_cell_ctrl(table_ctrl->obj, 0, 0, LV_TABLE_CELL_CTRL_MERGE_RIGHT)"
+        in s
+        for s in statements
+    )
+    assert any(
+        "lv_table_set_cell_ctrl(table_ctrl->obj, 0, 1, LV_TABLE_CELL_CTRL_TEXT_CROP)"
+        in s
+        for s in statements
+    )
+    # text_crop omitted for cell 0: no clear_cell_ctrl() should be emitted.
+    assert not any(
+        "table_ctrl->obj, 0, 0, LV_TABLE_CELL_CTRL_TEXT_CROP" in s for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_pixel_column_width_calls_lvgl_directly(setup_core) -> None:
+    await _create_table({"id": "table_px", "columns": [{"width": 96}]})
+    statements = _statements()
+    assert any(
+        "lv_table_set_column_width(table_px->obj, 0, 96)" in s for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_percent_column_width_uses_the_dynamic_helper(setup_core) -> None:
+    """Regression test: lv_table_set_column_width() only accepts a literal
+    pixel count, so a percentage width must not be passed to it directly -
+    it has to go through the LvTableType helper that recomputes it at
+    runtime from the table's actual content width.
+    """
+    await _create_table({"id": "table_pct", "columns": [{"width": "40%"}]})
+    statements = _statements()
+    assert any("table_pct->init_column_pct(1)" in s for s in statements)
+    assert any("table_pct->add_column_width_pct(0, 40)" in s for s in statements)
+    assert not any(
+        "lv_table_set_column_width(table_pct->obj, 0" in s for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_selected_cell_with_both_indices(setup_core) -> None:
+    await _create_table(
+        {"id": "table_sel_both", "selected_row": 1, "selected_column": 2}
+    )
+    statements = _statements()
+    assert any(
+        "lv_table_set_selected_cell(table_sel_both->obj, 1, 2)" in s for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_selected_cell_with_only_row_selects_whole_row(setup_core) -> None:
+    await _create_table({"id": "table_sel_row", "selected_row": 1})
+    statements = _statements()
+    assert any(
+        "lv_table_set_selected_cell(table_sel_row->obj, 1, LV_TABLE_CELL_NONE)" in s
+        for s in statements
+    )
+
+
+@pytest.mark.asyncio
+async def test_selected_cell_omitted_entirely_when_not_configured(
+    setup_core,
+) -> None:
+    await _create_table({"id": "table_no_selection", "rows": [["a"]]})
+    statements = _statements()
+    assert not any("lv_table_set_selected_cell" in s for s in statements)
+
+
+@pytest.mark.asyncio
+async def test_cell_update_action_writes_only_the_given_fields(setup_core) -> None:
+    await _create_table({"id": "table_update", "rows": [["a", "b"], ["c", "d"]]})
+    set_widgets_completed(True)
+    # Only inspect statements emitted by the action below, not by creation.
+    before = len(_statements())
+
+    entry = ACTION_REGISTRY["lvgl.table.cell.update"]
+    config = entry.schema(
+        {"id": "table_update", "row": 1, "column": 1, "text": "new value"}
+    )
+    action_id = ID("test_cell_update_action", is_declaration=True, type=entry.type_id)
+    await entry.coroutine_fun(config, action_id, TemplateArguments(), [])
+
+    statements = _statements()[before:]
+    assert any(
+        'lv_table_set_cell_value(table_update->obj, 1, 1, "new value")' in s
+        for s in statements
+    )
+    # Neither control flag was specified, so neither call should be emitted.
+    assert not any("LV_TABLE_CELL_CTRL" in s for s in statements)
+
+
+@pytest.mark.asyncio
+async def test_on_value_registers_a_value_changed_event_callback(setup_core) -> None:
+    config = container_schema(table_spec)(
+        {
+            "id": "table_on_value",
+            "rows": [["a"]],
+            "on_value": [
+                {"lambda": make_data_base("id(table_on_value).get_selected_row();")}
+            ],
+        }
+    )
+    # Auto-generated IDs (trigger/automation/action) are normally resolved to
+    # unique names by esphome's full config pass before code generation; do
+    # that by hand here since this test only exercises the widget/trigger
+    # codegen slice in isolation.
+    automation_conf = config[CONF_ON_VALUE][0]
+    automation_conf[CONF_TRIGGER_ID].resolve([])
+    automation_conf[CONF_AUTOMATION_ID].resolve([])
+    automation_conf[CONF_THEN][0][CONF_TYPE_ID].resolve([])
+
+    parent = MockObj("parent_obj")
+    async with LvContext():
+        await widget_to_code(config, table_spec, parent)
+        set_widgets_completed(True)
+        await generate_triggers()
+
+    statements = _statements()
+    assert any(
+        "table_on_value->obj" in s
+        and "add_event_cb" in s
+        and "LV_EVENT_VALUE_CHANGED" in s
+        for s in statements
+    )

tests/unit_tests/components/lvgl/test_table_config.py

@@ -0,0 +1,142 @@
+"""Tests for the LVGL table widget's configuration validation."""
+
+from __future__ import annotations
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.automation import ACTION_REGISTRY
+from esphome.components.lvgl.widgets.table import (
+    CONF_MERGE_RIGHT,
+    CONF_TEXT_CROP,
+    TABLE_SCHEMA,
+)
+
+
+def test_minimal_config_is_valid() -> None:
+    assert TABLE_SCHEMA({}) == {}
+
+
+def test_row_shorthand_expands_to_plain_cells() -> None:
+    config = TABLE_SCHEMA({"rows": [["Name", "Value"]]})
+    [row] = config["rows"]
+    assert row["cells"] == [{"text": "Name"}, {"text": "Value"}]
+
+
+def test_row_dict_form_with_cell_overrides() -> None:
+    config = TABLE_SCHEMA(
+        {
+            "rows": [
+                {
+                    "cells": [
+                        "Temp",
+                        {"text": "22.5", "text_crop": True, "merge_right": True},
+                    ]
+                }
+            ]
+        }
+    )
+    [row] = config["rows"]
+    assert row["cells"][0] == {"text": "Temp"}
+    assert row["cells"][1] == {
+        "text": "22.5",
+        "merge_right": True,
+        "text_crop": True,
+    }
+
+
+def test_row_count_defaults_are_not_injected_by_the_schema() -> None:
+    # Inference of row/column counts from `rows` happens at code generation
+    # time, not during validation - the schema should leave them unset.
+    config = TABLE_SCHEMA({"rows": [["a", "b"], ["c"]]})
+    assert "row_count" not in config
+    assert "column_count" not in config
+
+
+def test_explicit_row_and_column_count_are_kept() -> None:
+    config = TABLE_SCHEMA({"row_count": 5, "column_count": 3})
+    assert config["row_count"] == 5
+    assert config["column_count"] == 3
+
+
+def test_row_count_too_small_for_given_rows_raises() -> None:
+    with pytest.raises(cv.Invalid, match="row_count"):
+        TABLE_SCHEMA({"rows": [["a"], ["b"], ["c"]], "row_count": 2})
+
+
+def test_column_count_too_small_for_given_cells_raises() -> None:
+    with pytest.raises(cv.Invalid, match="column_count"):
+        TABLE_SCHEMA({"rows": [["a", "b", "c"]], "column_count": 2})
+
+
+def test_columns_list_longer_than_column_count_raises() -> None:
+    with pytest.raises(cv.Invalid, match="columns"):
+        TABLE_SCHEMA(
+            {
+                "column_count": 1,
+                "columns": [{"width": 10}, {"width": 20}],
+            }
+        )
+
+
+def test_columns_list_matching_inferred_column_count_is_valid() -> None:
+    config = TABLE_SCHEMA(
+        {
+            "rows": [["a", "b"]],
+            "columns": [{"width": 10}, {"width": 20}],
+        }
+    )
+    assert [c["width"] for c in config["columns"]] == [10, 20]
+
+
+@pytest.mark.parametrize(
+    ("width", "expected"),
+    [
+        (100, 100),
+        ("50%", 0.5),
+        ("32px", 32),
+    ],
+)
+def test_column_width_accepts_pixels_and_percent(width, expected) -> None:
+    config = TABLE_SCHEMA({"columns": [{"width": width}]})
+    assert config["columns"][0]["width"] == expected
+
+
+def test_columns_percent_widths_summing_over_100_percent_raises() -> None:
+    with pytest.raises(cv.Invalid, match="columns"):
+        TABLE_SCHEMA({"columns": [{"width": "60%"}, {"width": "50%"}]})
+
+
+def test_columns_percent_widths_summing_to_100_percent_is_valid() -> None:
+    config = TABLE_SCHEMA({"columns": [{"width": "60%"}, {"width": "40%"}]})
+    assert [c["width"] for c in config["columns"]] == [0.6, 0.4]
+
+
+def test_columns_mixed_pixel_and_percent_widths_ignore_pixels_in_the_total() -> None:
+    # Pixel widths aren't part of the percentage budget, so they shouldn't
+    # count towards the 100% limit.
+    config = TABLE_SCHEMA(
+        {"columns": [{"width": 200}, {"width": "80%"}, {"width": "20%"}]}
+    )
+    assert [c["width"] for c in config["columns"]] == [200, 0.8, 0.2]
+
+
+def test_selected_row_and_selected_column_are_independently_optional() -> None:
+    config = TABLE_SCHEMA({"selected_row": 1})
+    assert config["selected_row"] == 1
+    assert "selected_column" not in config
+
+
+def test_cell_update_action_requires_at_least_one_field() -> None:
+    entry = ACTION_REGISTRY["lvgl.table.cell.update"]
+    with pytest.raises(cv.Invalid):
+        entry.schema({"id": "some_table", "row": 0, "column": 0})
+
+
+def test_cell_update_action_accepts_a_single_field() -> None:
+    entry = ACTION_REGISTRY["lvgl.table.cell.update"]
+    config = entry.schema(
+        {"id": "some_table", "row": 0, "column": 0, "merge_right": True}
+    )
+    assert config[CONF_MERGE_RIGHT] is True
+    assert CONF_TEXT_CROP not in config

tests/unit_tests/core/test_config.py

@@ -1285,6 +1285,7 @@ async def test_add_platformio_options_native_idf(
     await config._add_platformio_options(
         {
             "build_flags": "-DSINGLE_FLAG",  # string and list forms both valid
+            "build_unflags": ["-Os"],
             "lib_deps": ["bblanchon/ArduinoJson@7.4.2"],
             "lib_ignore": "libsodium",
             "upload_speed": "115200",
@@ -1294,6 +1295,7 @@ async def test_add_platformio_options_native_idf(
 
     assert "-DSINGLE_FLAG" in CORE.build_flags
     assert "ArduinoJson" in CORE.platformio_libraries
+    assert "-Os" in CORE.build_unflags
     # lib_ignore is stored (listified) for generate_idf_components to read;
     # nothing else lands in platformio_options on the native toolchain.
     assert CORE.platformio_options == {"lib_ignore": ["libsodium"]}
@@ -1389,3 +1391,50 @@ def test_esphome_build_internals_are_yaml_only() -> None:
         assert markers[field].visibility is cv.Visibility.ADVANCED, field
     # A regular device-config field stays on the main form.
     assert markers[CONF_NAME_ADD_MAC_SUFFIX].visibility is None
+
+
+@pytest.mark.asyncio
+async def test_add_platformio_options_native_arduino(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The native ESP8266 Arduino toolchain honors board_build.f_cpu (a
+    real-world overclock knob) and warns about the rest like native IDF."""
+    CORE.toolchain = Toolchain.ARDUINO
+    CORE.data[KEY_CORE] = {
+        KEY_TARGET_PLATFORM: "esp8266",
+        KEY_TARGET_FRAMEWORK: "arduino",
+    }
+
+    await config._add_platformio_options(
+        {
+            "board_build.f_cpu": "160000000L",
+            # The schema also permits the list form; the last value wins
+            # and reaches the generator as a scalar
+            "board_build.ldscript": ["eagle.flash.2m.ld", "eagle.flash.4m2m.ld"],
+            "board_build.filesystem": "littlefs",
+            "upload_speed": "115200",
+        }
+    )
+
+    assert CORE.platformio_options["board_build.f_cpu"] == "160000000L"
+    assert CORE.platformio_options["board_build.ldscript"] == "eagle.flash.4m2m.ld"
+    assert "board_build.f_cpu is ignored" not in caplog.text
+    assert "board_build.ldscript is ignored" not in caplog.text
+    assert (
+        "esphome->platformio_options->board_build.filesystem is ignored" in caplog.text
+    )
+    # An empty list for an honored key is not a scalar; it falls through
+    # to the ignored-option warning instead of an IndexError
+    await config._add_platformio_options({"board_build.ldscript": []})
+    assert "board_build.ldscript is ignored" in caplog.text
+    assert "'arduino' toolchain" in caplog.text
+    assert "upload_speed" not in caplog.text
+
+
+def test_esp8266_rejects_unsupported_cli_toolchain() -> None:
+    """Until the native backend lands, ESP8266 serves only PlatformIO."""
+    from esphome.components.esp8266 import CONFIG_SCHEMA
+
+    CORE.toolchain = Toolchain.ARDUINO
+    with pytest.raises(cv.Invalid, match="Unsupported toolchain 'arduino'"):
+        CONFIG_SCHEMA({"board": "nodemcuv2"})

tests/unit_tests/test_arduino8266_framework.py

@@ -0,0 +1,170 @@
+"""Tests for esphome.arduino8266.framework (downloads and environment)."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.arduino8266 import framework
+import esphome.config_validation as cv
+from esphome.core import CORE, EsphomeError
+
+
+@pytest.fixture(autouse=True)
+def _build_path(tmp_path: Path) -> None:
+    CORE.build_path = tmp_path
+
+
+def test_framework_package_version() -> None:
+    assert framework.framework_package_version(cv.Version(3, 1, 2)) == "3.30102.0"
+    assert framework.framework_package_version(cv.Version(3, 2, 0)) == "3.30200.0"
+    # 2.6.3+ cores use the same package-major-3 encoding (PlatformIO path)
+    assert framework.framework_package_version(cv.Version(2, 7, 4)) == "3.20704.0"
+    # A future major bump needs its own encoding, not a doomed registry lookup
+    with pytest.raises(EsphomeError, match="not supported yet"):
+        framework.framework_package_version(cv.Version(4, 0, 0))
+    # The boundary matches the PlatformIO era guard; a 2.6.2 pre-release
+    # keeps this encoding
+    with pytest.raises(EsphomeError, match="older package encoding"):
+        framework.framework_package_version(cv.Version(2, 6, 2))
+    assert framework.framework_package_version(cv.Version(2, 6, 2, "b1")) == "3.20602.0"
+    assert framework.framework_package_version(cv.Version(2, 6, 3)) == "3.20603.0"
+
+
+def test_format_framework_arduino_version_pins_all_series() -> None:
+    """The esp8266 component's PIO source formatter across every encoding
+    era, including the 4.x rejection it now shares with the installer."""
+    from esphome.components.esp8266 import _format_framework_arduino_version as fmt
+
+    assert fmt(cv.Version(2, 4, 1)) == "~1.20401.0"
+    assert fmt(cv.Version(2, 6, 2)) == "~2.20602.0"
+    assert fmt(cv.Version(2, 7, 4)) == "~3.20704.0"
+    assert fmt(cv.Version(3, 1, 2)) == "~3.30102.0"
+    # Anchored to the framework version line, not a bare EsphomeError
+    with pytest.raises(cv.Invalid, match="not supported yet") as excinfo:
+        fmt(cv.Version(4, 0, 0))
+    assert excinfo.value.path == ["version"]
+
+
+def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}):
+        assert framework.get_arduino8266_tools_path() == tmp_path.resolve()
+    # A blank prefix must be treated as unset, not as the CWD
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": "  "}):
+        path = framework.get_arduino8266_tools_path()
+    assert path.name == "arduino8266"
+    assert path != Path.cwd()
+
+
+def test_check_and_install_returns_paths(tmp_path: Path) -> None:
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "install_package") as mock_install,
+        patch.object(framework, "prefetch_packages") as mock_prefetch,
+        patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
+    ):
+        paths = framework.check_and_install(cv.Version(3, 1, 2))
+    assert paths.framework == tmp_path / "frameworks" / "3.30102.0"
+    assert paths.toolchain == tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION
+    assert paths.ninja == tmp_path / "ninja"
+    assert mock_install.call_count == 2
+    # Full argument pinning: a copy-paste swap between the two near-identical
+    # calls (mirrors, destination) must not stay green
+    fw_call, tc_call = mock_install.call_args_list
+    assert fw_call.args == (
+        framework.FRAMEWORK_PACKAGE,
+        "3.30102.0",
+        tmp_path / "frameworks" / "3.30102.0",
+        framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+        tmp_path / "downloads",
+    )
+    assert fw_call.kwargs["expect"] == ("cores/esp8266", "tools/sdk", "libraries")
+    assert tc_call.args == (
+        framework.TOOLCHAIN_PACKAGE,
+        framework.TOOLCHAIN_VERSION,
+        tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
+        framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+        tmp_path / "downloads",
+    )
+    assert tc_call.kwargs["expect"] == ("bin", "xtensa-lx106-elf")
+    # The prefetch sees the same package specs as the installs
+    assert mock_prefetch.call_args.args == (
+        [
+            (
+                framework.FRAMEWORK_PACKAGE,
+                "3.30102.0",
+                tmp_path / "frameworks" / "3.30102.0",
+                framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+            ),
+            (
+                framework.TOOLCHAIN_PACKAGE,
+                framework.TOOLCHAIN_VERSION,
+                tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
+                framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+            ),
+        ],
+        tmp_path / "downloads",
+    )
+
+
+def test_get_build_env_prepends_toolchain_bin(tmp_path: Path) -> None:
+    with patch.object(framework, "ccache_env", return_value={"CCACHE_DIR": "x"}):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"].startswith(str(tmp_path / "bin") + os.pathsep)
+    assert env["CCACHE_DIR"] == "x"
+
+
+def test_ccache_env(tmp_path: Path) -> None:
+    assert framework.ccache_env(None) == {}
+    with patch.dict(os.environ, {"CCACHE_NOHASHDIR": "false"}, clear=True):
+        env = framework.ccache_env("/usr/bin/ccache")
+    # User-set values are respected; the rest get defaults
+    assert "CCACHE_NOHASHDIR" not in env
+    assert env["CCACHE_DEPEND"] == "1"
+    assert env["CCACHE_BASEDIR"] == str(Path(CORE.build_path).resolve())
+    assert env["CCACHE_DIR"].endswith("ccache")
+
+
+def test_check_and_install_rejects_old_core(tmp_path: Path) -> None:
+    """Calling the installer below the floor fails before any download."""
+    with pytest.raises(EsphomeError, match=">= 3.1.1"):
+        framework.check_and_install(cv.Version(3, 0, 2))
+
+
+def test_get_build_env_without_path_has_no_empty_entry(tmp_path: Path) -> None:
+    """An absent PATH must not leave a trailing separator (an empty entry
+    means the current directory to the shell)."""
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch.object(framework, "ccache_env", return_value={}),
+    ):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"] == str(tmp_path / "bin")
+    with (
+        patch.dict(
+            os.environ, {"PATH": f"/usr/bin{os.pathsep}{os.pathsep}/bin"}, clear=True
+        ),
+        patch.object(framework, "ccache_env", return_value={}),
+    ):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"].split(os.pathsep) == [str(tmp_path / "bin"), "/usr/bin", "/bin"]
+
+
+def test_ccache_env_accepts_a_preresolved_path() -> None:
+    """The caller resolves ccache once and threads it through; None means
+    resolved-and-disabled."""
+    with patch.dict(os.environ, {}, clear=True):
+        assert framework.ccache_env(None) == {}
+        env = framework.ccache_env("/usr/bin/ccache")
+    assert env["CCACHE_DIR"].endswith("ccache")
+
+
+def test_toolchain_tool_layout(tmp_path: Path) -> None:
+    """One owner for the bin/xtensa-lx106-elf-<name> layout."""
+    tool = framework.toolchain_tool(tmp_path, "addr2line")
+    assert tool.parent == tmp_path / "bin"
+    assert tool.name.startswith("xtensa-lx106-elf-addr2line")
+    assert (tool.suffix == ".exe") is (os.name == "nt")

tests/unit_tests/test_compiled_config.py

@@ -68,6 +68,7 @@ def _write_storage(
     esp_platform: str | None = "ESP32",
     core_platform: str | None = "esp32",
     build_path: str | None = "/build/lite_test",
+    toolchain: str | None = None,
 ) -> None:
     """Write a vanilla StorageJSON sidecar for the cache tests."""
     storage_path.parent.mkdir(parents=True, exist_ok=True)
@@ -88,6 +89,7 @@ def _write_storage(
         "no_mdns": False,
         "framework": "arduino",
         "core_platform": core_platform,
+        "toolchain": toolchain,
     }
     storage_path.write_text(json.dumps(data), encoding="utf-8")
 
@@ -629,6 +631,35 @@ def test_save_compiled_config_and_sidecar_builds_real_sidecar(tmp_path: Path) ->
     assert load_compiled_config(yaml_path) is not None
 
 
+@pytest.mark.parametrize(
+    ("sidecar_toolchain", "saved"),
+    [
+        ("esp-idf", False),
+        ("platformio", True),
+        (None, True),  # legacy sidecar without the field: guard is inert
+    ],
+)
+def test_save_compiled_config_and_sidecar_toolchain_mismatch(
+    tmp_path: Path, sidecar_toolchain: str | None, saved: bool
+) -> None:
+    """A config validated under a different toolchain than the compile's
+    must not overwrite the cache."""
+    yaml_path = _bare_yaml(tmp_path)
+    _prime_core(tmp_path)
+    CORE.config = {CONF_ESPHOME: {CONF_NAME: "lite_test"}}
+    CORE.toolchain = Toolchain.PLATFORMIO
+    _write_storage(
+        tmp_path / ".esphome" / "storage" / "lite_test.yaml.json",
+        toolchain=sidecar_toolchain,
+    )
+
+    save_compiled_config_and_sidecar(CORE.config)
+
+    cache = tmp_path / ".esphome" / "storage" / "lite_test.yaml.validated.json"
+    assert cache.exists() is saved
+    assert (load_compiled_config(yaml_path) is not None) is saved
+
+
 @pytest.mark.parametrize("command", ["upload", "logs"])
 def test_run_esphome_upload_and_logs_refresh_cache_on_fallback(
     tmp_path: Path, command: str

tests/unit_tests/test_config_validation.py

@@ -1,3 +1,4 @@
+import importlib
 import json
 import logging
 from pathlib import Path
@@ -48,6 +49,7 @@ from esphome.const import (
     TYPE_GIT,
     TYPE_LOCAL,
     Framework,
+    Toolchain,
 )
 from esphome.core import (
     CORE,
@@ -2563,6 +2565,52 @@ def test_returning_lambda_no_return() -> None:
         cv.returning_lambda(Lambda("int x = 5;"))
 
 
+def test_returning_lambda_return_only_in_comment() -> None:
+    with pytest.raises(Invalid, match="return statement"):
+        cv.returning_lambda(Lambda("// return 5;\nint x = 5;"))
+
+
+def test_returning_lambda_missing_semicolon_is_accepted() -> None:
+    """A forgotten semicolon is left for the C++ compiler to report."""
+    assert isinstance(cv.returning_lambda(Lambda("return x")), Lambda)
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("return 5;", True),
+        ("if (x) { return x; } return 0;", True),
+        ("if (x) return 1; else return 0;", True),
+        ("switch (x) { case 0: return 1; }", True),
+        # a semicolon means code: any return keyword counts
+        ("return not x;", True),
+        ("return a and b;", True),
+        ("please return the sensor; then wait", True),
+        # a forgotten semicolon is still lambda source; the compiler reports it
+        ("return id(x).state", True),
+        ("return x", True),
+        ("return 5", True),
+        ("return not x", True),
+        # accepted: a one-word tail is indistinguishable from 'return x'
+        ("return soon", True),
+        ("Alert: return home", True),
+        ("static value", False),
+        ("no returns here", False),
+        ("the_return_value", False),
+        # without a semicolon, prose is not lambda source
+        ("please return the item", False),
+        ("return to sender", False),
+        ("return a and b", False),
+        # return only inside a comment is not a return statement
+        ("// return 5;\nint x = 5;", False),
+        ("/* return 5; */ int x = 5;", False),
+        ("return 5; // done", True),
+    ],
+)
+def test_looks_like_returning_lambda(value: str, expected: bool) -> None:
+    assert cv.looks_like_returning_lambda(value) is expected
+
+
 # ---------------------------------------------------------------------------
 # dimensions
 # ---------------------------------------------------------------------------
@@ -3165,3 +3213,46 @@ def test_file__remapped_path_is_directory_raises(setup_core: Path) -> None:
 
     with pytest.raises(Invalid, match="is not a file"):
         cv.file_("/original/config/headers")
+
+
+def test_require_platformio_toolchain() -> None:
+    """Platforms with only the PlatformIO backend reject other toolchains."""
+    validator = cv.require_platformio_toolchain("RP2")
+    CORE.toolchain = None
+    config: dict = {}
+    assert validator(config) is config
+    assert CORE.toolchain == Toolchain.PLATFORMIO
+
+    CORE.toolchain = Toolchain.ARDUINO
+    with pytest.raises(Invalid, match="Unsupported toolchain 'arduino' for RP2"):
+        validator(config)
+
+
+def test_check_supported_toolchain_unresolved_is_an_ordering_bug() -> None:
+    """Calling the check before resolution fails naming the ordering bug,
+    not a user-facing unsupported-toolchain error."""
+    CORE.toolchain = None
+    with pytest.raises(Invalid, match="not resolved before RP2 validation"):
+        cv._check_supported_toolchain("RP2", (Toolchain.PLATFORMIO,))
+
+
+@pytest.mark.parametrize(
+    ("platform", "minimal_config"),
+    [
+        ("host", {}),
+        ("rp2", {"board": "rpipicow"}),
+        ("bk72xx", {"board": "generic-bk7231n-qfn32-tuya"}),
+        ("rtl87xx", {"board": "generic-rtl8710bn-2mb-788k"}),
+        ("ln882x", {"board": "generic-ln882h"}),
+        # The legacy stub platform must reject too, not just the chip families
+        ("libretiny", {}),
+    ],
+)
+def test_every_platformio_only_platform_rejects_arduino_toolchain(
+    platform: str, minimal_config: dict
+) -> None:
+    """A platform that cannot serve a CLI toolchain rejects it at validation."""
+    module = importlib.import_module(f"esphome.components.{platform}")
+    CORE.toolchain = Toolchain.ARDUINO
+    with pytest.raises(Invalid, match="Unsupported toolchain 'arduino'"):
+        module.CONFIG_SCHEMA(dict(minimal_config))

tests/unit_tests/test_core.py

@@ -958,6 +958,24 @@ class TestEsphomeCore:
         target.toolchain = const.Toolchain.ESP_IDF
         assert target.using_toolchain_sdk_nrf is False
 
+    def test_using_toolchain_arduino(self, target):
+        """A toolchain choice, distinct from the arduino target framework."""
+        target.toolchain = const.Toolchain.ARDUINO
+        assert target.using_toolchain_arduino is True
+        target.toolchain = const.Toolchain.PLATFORMIO
+        assert target.using_toolchain_arduino is False
+
+    def test_using_native_toolchain(self, target):
+        """True exactly for the toolchains that never read platformio.ini."""
+        target.toolchain = const.Toolchain.ESP_IDF
+        assert target.using_native_toolchain is True
+        target.toolchain = const.Toolchain.ARDUINO
+        assert target.using_native_toolchain is True
+        target.toolchain = const.Toolchain.PLATFORMIO
+        assert target.using_native_toolchain is False
+        target.toolchain = const.Toolchain.SDK_NRF
+        assert target.using_native_toolchain is False
+
     def test_add_library__extracts_short_name_from_path(self, target):
         """Test add_library extracts short name from library paths like owner/lib."""
         target.data[const.KEY_CORE] = {

tests/unit_tests/test_espidf_framework.py

@@ -1560,13 +1560,14 @@ def test_get_framework_env_without_python_env_uses_os_path(tmp_path: Path) -> No
 
 def _ccache_patches(tmp_path: Path, which: str | None, build_path: Path | None):
     return (
-        patch("esphome.espidf.framework.shutil.which", return_value=which),
+        patch("esphome.espidf.framework.resolve_ccache_path", return_value=which),
         patch(
             "esphome.espidf.framework.get_idf_tools_path",
             return_value=tmp_path / "tools",
         ),
+        # ccache_defaults_env (build_helpers.ccache) reads CORE at call time
         patch(
-            "esphome.espidf.framework.CORE",
+            "esphome.core.CORE",
             SimpleNamespace(build_path=build_path),
         ),
     )
@@ -1587,7 +1588,8 @@ def test_ccache_env_disabled_when_binary_missing(tmp_path: Path) -> None:
     # build_path is None here too: a disabled cache must not require it.
     p1, p2, p3 = _ccache_patches(tmp_path, None, None)
     with patch.dict("os.environ", {}, clear=True), p1, p2, p3:
-        assert _ccache_env() == {}
+        # Canonical off, so an inherited/unparsable value cannot enable it
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
 
 
 def test_ccache_env_opt_out_via_env(tmp_path: Path) -> None:
@@ -1595,18 +1597,111 @@ def test_ccache_env_opt_out_via_env(tmp_path: Path) -> None:
     # short-circuits before build_path is needed.
     p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", None)
     with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "0"}, clear=True), p1, p2, p3:
-        assert _ccache_env() == {}
+        # The canonical off spelling is exported: the raw value is inherited
+        # by idf.py, where a spelling like "disable" would read as truthy
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
 
 
-def test_ccache_env_opt_in_without_binary(tmp_path: Path) -> None:
-    # Explicit IDF_CCACHE_ENABLE=1 forces it on without probing PATH. It's
-    # already in the environment, so it isn't re-emitted, but the rest is.
+def test_ccache_env_opt_in_without_binary(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Explicit IDF_CCACHE_ENABLE=1 forces it on; without a usable binary
+    # idf.py silently skips ccache, so this branch must say so out loud.
     p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
     with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "1"}, clear=True), p1, p2, p3:
         env = _ccache_env()
-    assert "IDF_CCACHE_ENABLE" not in env
+    assert env["IDF_CCACHE_ENABLE"] == "1"
     assert env["CCACHE_DIR"] == str(tmp_path / "tools" / "ccache")
     assert env["CCACHE_DEPEND"] == "1"
+    assert "no ccache binary is on PATH" in caplog.text
+
+
+def test_ccache_env_opt_in_with_working_binary(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Forced on with a working binary: no warning fires at all.
+    ccache = tmp_path / "ccache"
+    ccache.touch()
+    p1, p2, p3 = _ccache_patches(tmp_path, str(ccache), tmp_path / "build")
+    with (
+        patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "1"}, clear=True),
+        patch("esphome.espidf.framework.shutil.which", return_value=str(ccache)),
+        patch("esphome.espidf.framework.tool_version_runs", return_value=True),
+        p1,
+        p2,
+        p3,
+    ):
+        env = _ccache_env()
+    assert env["IDF_CCACHE_ENABLE"] == "1"
+    assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
+
+
+def test_ccache_env_opt_in_with_rejected_binary(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Forced on with a present-but-rejected binary: idf.py does its own
+    # PATH lookup and uses it anyway; the warning must say so, not claim
+    # the build runs without ccache.
+    # A present but non-executable file: the real probe fails and logs
+    # the forced-on message (patching the probe would silence it)
+    broken = tmp_path / "broken-ccache"
+    broken.touch()
+    p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
+    with (
+        patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "1"}, clear=True),
+        patch("esphome.espidf.framework.shutil.which", return_value=str(broken)),
+        p1,
+        p2,
+        p3,
+    ):
+        env = _ccache_env()
+    assert env["IDF_CCACHE_ENABLE"] == "1"
+    assert "idf.py will use it anyway" in caplog.text
+    # Exactly one story: the resolver's contradictory "compiling without
+    # ccache" must not precede it
+    assert "compiling without ccache" not in caplog.text
+
+
+def test_ccache_env_honors_shared_esphome_opt_out(tmp_path: Path) -> None:
+    """ESPHOME_CCACHE_ENABLE=0 disables ccache here too; the shared policy
+    must not apply to every backend except this one."""
+    _p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    env_vars = {"ESPHOME_CCACHE_ENABLE": "0", "PATH": "/usr/bin"}
+    with patch.dict("os.environ", env_vars, clear=True), p2, p3:
+        # The real resolver runs so the opt-out parse is exercised
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
+
+
+@pytest.mark.parametrize("value", ["off", "no"])
+def test_ccache_env_idf_knob_parses_strictly(tmp_path: Path, value: str) -> None:
+    """IDF_CCACHE_ENABLE uses the same strict table as the shared knob, so
+    "off" disables instead of reading as truthy."""
+    p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": value}, clear=True), p1, p2, p3:
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
+
+
+def test_ccache_env_idf_knob_unrecognized_warns_and_defers(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unparsable IDF_CCACHE_ENABLE warns, defers to the shared resolver,
+    and is not forwarded to idf.py as truthy."""
+    p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    env_vars = {"IDF_CCACHE_ENABLE": "enabled"}
+    with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
+        env = _ccache_env()
+    assert "unrecognized IDF_CCACHE_ENABLE" in caplog.text
+    assert env["IDF_CCACHE_ENABLE"] == "1"
+
+
+def test_ccache_env_idf_knob_wins_over_shared_opt_out(tmp_path: Path) -> None:
+    """IDF_CCACHE_ENABLE=1 takes precedence over ESPHOME_CCACHE_ENABLE=0."""
+    p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
+    env_vars = {"IDF_CCACHE_ENABLE": "1", "ESPHOME_CCACHE_ENABLE": "0"}
+    with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
+        env = _ccache_env()
+    assert env["CCACHE_DIR"] == str(tmp_path / "tools" / "ccache")
+    assert env["IDF_CCACHE_ENABLE"] == "1"
 
 
 def test_ccache_env_preserves_user_overrides(tmp_path: Path) -> None:

tests/unit_tests/test_framework_helpers.py

@@ -2278,3 +2278,37 @@ class TestGetProjectCxxCompileFlags:
     def test_empty_flags(self) -> None:
         with patch("esphome.core.CORE", _make_core_cxx(set())):
             assert get_project_cxx_compile_flags() == []
+
+
+@pytest.mark.parametrize(
+    ("platform", "input_path", "expected"),
+    [
+        # win32: drive-letter extended-length prefix is stripped
+        (
+            "win32",
+            "\\\\?\\C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+        ),
+        # win32: UNC extended-length prefix is translated to a regular UNC path
+        (
+            "win32",
+            "\\\\?\\UNC\\server\\share\\python.exe",
+            "\\\\server\\share\\python.exe",
+        ),
+        # win32: paths without the prefix are returned unchanged
+        (
+            "win32",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+        ),
+        # non-win32: prefix is left alone (no-op)
+        ("linux", "\\\\?\\C:\\python.exe", "\\\\?\\C:\\python.exe"),
+        ("darwin", "/usr/bin/python3", "/usr/bin/python3"),
+    ],
+)
+def test_strip_win_long_path_prefix(
+    platform: str, input_path: str, expected: str
+) -> None:
+    r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
+    with patch("esphome.framework_helpers.sys.platform", platform):
+        assert framework_helpers.strip_win_long_path_prefix(input_path) == expected

tests/unit_tests/test_main.py

@@ -7195,6 +7195,42 @@ def test_compile_program_espidf_idedata_none_warns(
     assert "No idedata was generated" in caplog.text
 
 
+def test_cli_toolchain_skips_the_validated_config_cache(tmp_path: Path) -> None:
+    """An explicit --toolchain must run the per-platform validators, so the
+    upload/logs fast path becomes a cache miss."""
+    conf = tmp_path / "device.yaml"
+    conf.write_text("esphome:\n  name: t\n")
+    argv = ["esphome", "--toolchain", "arduino", "logs", str(conf)]
+    with (
+        patch("esphome.compiled_config.load_compiled_config") as mock_cache,
+        patch("esphome.config.read_config", return_value=None) as mock_read,
+    ):
+        assert run_esphome(argv) == 2
+    mock_cache.assert_not_called()
+    mock_read.assert_called_once()
+
+
+def test_cli_toolchain_still_refreshes_the_validated_config_cache(
+    tmp_path: Path,
+) -> None:
+    """An explicit --toolchain gates only the cache read; with a matching
+    sidecar the freshly validated config is still saved."""
+    conf = tmp_path / "device.yaml"
+    conf.write_text("esphome:\n  name: t\n")
+    argv = ["esphome", "--toolchain", "platformio", "logs", str(conf)]
+    with (
+        patch("esphome.compiled_config.load_compiled_config") as mock_load,
+        patch("esphome.config.read_config", return_value={CONF_ESPHOME: {}}),
+        patch("esphome.compiled_config.save_compiled_config_and_sidecar") as mock_save,
+        patch.dict(
+            "esphome.__main__.POST_CONFIG_ACTIONS", {"logs": Mock(return_value=0)}
+        ),
+    ):
+        assert run_esphome(argv) == 0
+    mock_load.assert_not_called()
+    mock_save.assert_called_once()
+
+
 @pytest.mark.asyncio
 async def test_wrap_to_code_comment_is_insertion_order_independent() -> None:
     """The config comment dumps with sorted keys: voluptuous fills schema

tests/unit_tests/test_nrf52_framework.py

@@ -7,8 +7,10 @@ import sys
 from types import SimpleNamespace
 from unittest.mock import patch
 
+import platformdirs
 import pytest
 
+from esphome.components.nrf52 import _resolve_toolchain
 from esphome.components.nrf52.framework import (
     _PLATFORMIO_PENV_REQUIREMENTS,
     _REQUIREMENTS,
@@ -22,8 +24,9 @@ from esphome.components.nrf52.framework import (
     get_sdk_nrf_tools_path,
     setup_platformio_python_env,
 )
+import esphome.config_validation as cv
 from esphome.config_validation import Version
-from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
+from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION, Toolchain
 from esphome.core import CORE, EsphomeError
 from esphome.framework_helpers import get_python_env_executable_path
 
@@ -560,7 +563,6 @@ def testget_tools_path_blank_env_falls_back_to_default(
     Path("") would resolve to the working directory, which clean-all could
     then delete by accident.
     """
-    import platformdirs
 
     monkeypatch.setenv("ESPHOME_SDK_NRF_PREFIX", value)
     expected = (
@@ -572,7 +574,6 @@ def testget_tools_path_blank_env_falls_back_to_default(
 def testget_tools_path_default_is_global_cache(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    import platformdirs
 
     monkeypatch.delenv("ESPHOME_SDK_NRF_PREFIX", raising=False)
     expected = (
@@ -621,3 +622,11 @@ def test_needs_venv_rebuild_on_dangling_interpreter_symlink(tmp_path: Path) -> N
     assert not python.exists()
 
     assert _needs_venv_rebuild(python, sentinel, "abc123")
+
+
+def test_resolve_toolchain_rejects_unsupported() -> None:
+    """A --toolchain nRF52 cannot serve fails instead of degrading silently."""
+
+    CORE.toolchain = Toolchain.ARDUINO
+    with pytest.raises(cv.Invalid, match="Unsupported toolchain 'arduino'"):
+        _resolve_toolchain({})

tests/unit_tests/test_platformio_registry.py

@@ -0,0 +1,725 @@
+"""Tests for esphome.platformio.registry (PIO-registry package installs)."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.core import EsphomeError
+from esphome.platformio import registry
+
+
+def test_registry_download_resolves_once_per_process() -> None:
+    """The prefetch and the install share one metadata resolve per package."""
+    calls: list[dict] = []
+    payload = {
+        "versions": [
+            {
+                "name": "1.0.0",
+                "files": [
+                    {
+                        "download_url": "http://x/pkg.tar.gz",
+                        "checksum": {"sha256": "ab" * 32},
+                        "size": 5,
+                    }
+                ],
+            }
+        ]
+    }
+
+    def fake_request(method, url, **kwargs):
+        calls.append(url)
+        return _http_response(json.dumps(payload))
+
+    with patch.object(registry, "http_request", side_effect=fake_request):
+        first = registry.registry_download("o/pkg", "1.0.0")
+        second = registry.registry_download("o/pkg", "1.0.0")
+    assert first == second
+    assert len(calls) == 1
+
+
+@pytest.fixture(autouse=True)
+def _fresh_registry_cache():
+    # registry_download memoizes per process; tests reuse package names
+    registry.registry_download.cache_clear()
+    yield
+    registry.registry_download.cache_clear()
+
+
+@pytest.mark.parametrize(
+    ("system", "machine", "expected"),
+    [
+        ("Darwin", "arm64", "darwin_arm64"),
+        ("Darwin", "x86_64", "darwin_x86_64"),
+        ("Windows", "AMD64", "windows_amd64"),
+        # Deviation from upstream: auto-mapped to the emulated-x86 packages
+        ("Windows", "ARM64", "windows_amd64"),
+        ("Windows", "x86", "windows_x86"),
+        ("Linux", "x86_64", "linux_x86_64"),
+        ("Linux", "aarch64", "linux_aarch64"),
+        ("Linux", "i686", "linux_i686"),
+        ("Linux", "armv7l", "linux_armv7l"),
+        # Unknown hosts pass through like upstream; the registry lookup
+        # then fails naming the tag
+        ("FreeBSD", "amd64", "freebsd_amd64"),
+    ],
+)
+def test_get_systype(system: str, machine: str, expected: str) -> None:
+    with (
+        patch("platform.system", return_value=system),
+        patch("platform.machine", return_value=machine),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == expected
+
+
+def test_get_systype_env_override() -> None:
+    """PLATFORMIO_SYSTEM_TYPE wins, exactly as in upstream get_systype()."""
+    with patch.dict(os.environ, {"PLATFORMIO_SYSTEM_TYPE": "windows_amd64"}):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def test_get_systype_aarch64_32bit_userland() -> None:
+    """A 32-bit userland on a 64-bit arm kernel gets armv7l binaries."""
+    with (
+        patch("platform.system", return_value="Linux"),
+        patch("platform.machine", return_value="aarch64"),
+        patch("platform.architecture", return_value=("32bit", "")),
+    ):
+        assert registry.get_systype() == "linux_armv7l"
+
+
+def test_get_systype_windows_empty_machine() -> None:
+    """An empty machine string falls back to the architecture bits."""
+    with (
+        patch("platform.system", return_value="Windows"),
+        patch("platform.machine", return_value=""),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def _http_response(text: str) -> MagicMock:
+    resp = MagicMock()
+    resp.text = text
+    resp.raise_for_status.return_value = None
+    return resp
+
+
+def _registry_response(files: list[dict]):
+    """Patch the consolidated HTTP path to serve a canned registry response."""
+    payload = {"versions": [{"name": "1.0.0", "files": files}]}
+    return patch.object(
+        registry, "http_request", return_value=_http_response(json.dumps(payload))
+    )
+
+
+def test_registry_download_uses_shared_http_path() -> None:
+    """The metadata fetch delegates to the consolidated http_request path;
+    request failures surface as a named EsphomeError."""
+    import requests as req
+
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            side_effect=req.exceptions.ConnectionError("registry down"),
+        ) as mock_request,
+        pytest.raises(EsphomeError, match="Could not fetch registry metadata"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+    (method, url), _ = mock_request.call_args
+    assert method == "GET"
+    assert url == registry._REGISTRY_URL.format(package="pkg")
+
+
+def test_registry_download_invalid_json_is_clean() -> None:
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response("<html>not json</html>"),
+        ),
+        pytest.raises(EsphomeError, match="invalid JSON"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_matches_system() -> None:
+    with (
+        _registry_response(
+            [
+                {"system": ["windows_amd64"], "download_url": "http://x/win"},
+                {
+                    "system": ["linux_x86_64"],
+                    "download_url": "http://x/linux",
+                    "checksum": {"sha256": "abc123"},
+                    "size": 42,
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/linux",
+            "abc123",
+            42,
+        )
+
+
+def test_registry_download_bare_string_system() -> None:
+    """A bare-string system tag is an exact match, not a substring test."""
+    with (
+        _registry_response(
+            [
+                {"system": "linux_x86", "download_url": "http://x/x86"},
+                {
+                    "system": "linux_x86_64",
+                    "download_url": "http://x/x86_64",
+                    "checksum": {"sha256": "abc"},
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0")[0] == "http://x/x86_64"
+
+
+def test_registry_download_wildcard_system() -> None:
+    with _registry_response(
+        [
+            {
+                "system": "*",
+                "download_url": "http://x/any",
+                "checksum": {"sha256": "abc"},
+                "size": 7,
+            }
+        ]
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/any",
+            "abc",
+            7,
+        )
+
+
+def test_registry_download_missing_checksum_raises() -> None:
+    """An unverifiable archive is refused, never silently extracted."""
+    with (
+        _registry_response([{"system": "*", "download_url": "http://x/any"}]),
+        pytest.raises(EsphomeError, match="no sha256"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_no_system_match() -> None:
+    with (
+        _registry_response(
+            [{"system": ["windows_amd64"], "download_url": "http://x/win"}]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_version_not_found() -> None:
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response(
+                json.dumps({"versions": [{"name": "2.0.0", "files": []}]})
+            ),
+        ),
+        pytest.raises(EsphomeError, match="not found"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    (dest / "payload").mkdir(parents=True)
+    (dest / ".esphome_extracted").touch()
+    with patch.object(registry, "download_from_mirrors") as mock_download:
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+    mock_download.assert_not_called()
+
+
+def test_install_package_marker_hit_rechecks_layout(tmp_path: Path) -> None:
+    """A marked install that later lost files fails by name instead of
+    surfacing as an opaque toolchain error."""
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    (dest / ".esphome_extracted").touch()
+    with pytest.raises(EsphomeError, match="missing the expected payload"):
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+
+
+def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
+    with (
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        # Extraction is expected to create the directory
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, mirrors, tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_download.call_args[0][0] is mirrors
+    assert mock_download.call_args[0][1] == {
+        "VERSION": "1.0.0",
+        "SYSTEM": "linux_x86_64",
+    }
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
+    """The registry path downloads with the registry's sha256 and size."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(
+            registry,
+            "registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", 42),
+        ),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
+    assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
+
+
+def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
+    """The success marker is only written when the extracted tree is usable."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="missing the expected bin"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert not (dest / ".esphome_extracted").exists()
+
+
+def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
+    """A concurrent install finishing while we wait for the lock is detected."""
+    dest = tmp_path / "pkg"
+    marker = dest / ".esphome_extracted"
+
+    @contextmanager
+    def _fake_lock(*_a, **_kw):
+        dest.mkdir(parents=True, exist_ok=True)
+        marker.touch()
+        yield
+
+    with (
+        patch("filelock.FileLock", _fake_lock),
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "rmdir") as mock_rmdir,
+    ):
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("payload",)
+        )
+    mock_download.assert_not_called()
+    mock_rmdir.assert_not_called()
+
+
+def test_install_package_uses_hard_lock(tmp_path: Path) -> None:
+    """The install lock must never degrade to a soft (existence) lock."""
+    dest = tmp_path / "pkg"
+    with (
+        patch("filelock.FileLock") as mock_lock,
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True, exist_ok=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_lock.call_args.kwargs["fallback_to_soft"] is False
+
+
+def test_registry_download_empty_system_list_does_not_match() -> None:
+    """An explicitly empty system list must not act as a wildcard."""
+    with (
+        _registry_response([{"system": [], "download_url": "http://x/any"}]),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_unexpected_payload_is_named() -> None:
+    """An error envelope without a versions list is not 'version not found'."""
+
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response(json.dumps({"message": "rate limited"})),
+        ),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_missing_system_key_matches_any() -> None:
+    """A file with no system key at all serves every host."""
+    with _registry_response(
+        [{"download_url": "http://x/any", "checksum": {"sha256": "abc"}, "size": 1}]
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == ("http://x/any", "abc", 1)
+
+
+def test_registry_download_missing_files_list_is_named() -> None:
+    """A version entry without a files list is an unexpected payload, not a
+    missing platform build."""
+    with (
+        _registry_response(None),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_missing_download_url_is_named() -> None:
+    with (
+        _registry_response([{"system": "*", "checksum": {"sha256": "abc"}, "size": 1}]),
+        pytest.raises(EsphomeError, match="no download URL"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_install_package_empty_expect_rejected(tmp_path: Path) -> None:
+    """Layout validation is the only guard before marker.touch(), so an
+    empty expect is a caller bug, not a lenient install."""
+    with pytest.raises(ValueError, match="non-empty expect"):
+        registry.install_package(
+            "pkg", "1.0.0", tmp_path / "pkg", [], tmp_path / "dl", expect=()
+        )
+
+
+def test_registry_download_non_dict_version_entry_is_named() -> None:
+    """A versions list of bare strings is an unexpected payload, not an
+    AttributeError traceback."""
+
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response(json.dumps({"versions": ["1.0.0", "2.0.0"]})),
+        ),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_dict_file_entry_is_named() -> None:
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response(
+                json.dumps({"versions": [{"name": "1.0.0", "files": ["a.tar.gz"]}]})
+            ),
+        ),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_dict_payload_is_named() -> None:
+    """A JSON array answer is an unexpected payload at the outermost level."""
+
+    with (
+        patch.object(
+            registry,
+            "http_request",
+            return_value=_http_response(json.dumps(["1.0.0"])),
+        ),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_list_system_is_named() -> None:
+    """A system field that is neither missing, str, nor list is an
+    unexpected payload, not a TypeError from the ``in`` test."""
+    with (
+        _registry_response([{"system": 5, "checksum": {"sha256": "abc"}, "size": 1}]),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def _resolve_for(sizes: dict[str, int | None]):
+    def resolve(name: str, version: str):
+        size = sizes[name]
+        if size == -1:
+            raise EsphomeError("registry down")
+        return (f"http://x/{name}.tar.gz", "abc123", size)
+
+    return resolve
+
+
+def test_prefetch_packages_downloads_pending_in_parallel(tmp_path: Path) -> None:
+    """Two uninstalled packages download together under one combined bar,
+    with the registry's sha256 and size and a batch progress tracker."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert mock_download.call_count == 2
+    # Locking makes worker completion order nondeterministic
+    calls = sorted(mock_download.call_args_list, key=lambda c: c[0][0])
+    for call, (name, version, size) in zip(
+        calls, [("a", "1.0", 10), ("b", "2.0", 20)], strict=True
+    ):
+        assert call[0][0] == f"http://x/{name}.tar.gz"
+        assert call[0][1] == tmp_path / "dl" / f"{name}-{version}"
+        assert call[1]["sha256"] == "abc123"
+        assert call[1]["size"] == size
+        assert callable(call[1]["progress"])
+
+
+def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
+    """A dest whose marker appeared while the worker waited on the lock is
+    already installed; re-downloading would orphan an archive copy."""
+    dest = tmp_path / "a"
+    dest.mkdir()
+
+    from contextlib import contextmanager
+
+    @contextmanager
+    def marker_appears_under_lock(path, **kwargs):
+        # Simulates the concurrent build finishing while we waited
+        (dest / ".esphome_extracted").touch()
+        yield
+
+    with (
+        patch("filelock.FileLock", side_effect=marker_appears_under_lock),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10})
+        ),
+    ):
+        registry.prefetch_packages([("a", "1.0", dest, [])], tmp_path / "dl")
+    mock_download.assert_not_called()
+
+
+def test_already_installed_probe(tmp_path: Path) -> None:
+    """Both arms of the marker probe the prefetch worker keys on."""
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    assert registry._already_installed(dest) is False
+    (dest / ".esphome_extracted").touch()
+    assert registry._already_installed(dest) is True
+
+
+def test_prefetch_packages_dedupes_duplicate_entries(tmp_path: Path) -> None:
+    """Duplicate (name, version) entries would race each other between two
+    workers; only one survives (and one is too few to parallelize)."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("a", "1.0", tmp_path / "a", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_single_pending_skips(tmp_path: Path) -> None:
+    """One pending package has nothing to parallelize; the sequential
+    install keeps its own bar."""
+    marker_dest = tmp_path / "a"
+    marker_dest.mkdir()
+    (marker_dest / ".esphome_extracted").touch()
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", marker_dest, []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_mirror_and_sizeless_stay_sequential(
+    tmp_path: Path,
+) -> None:
+    """Mirror overrides and size-less registry entries are left to the
+    sequential path so its per-file bars stay trustworthy."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry,
+            "registry_download",
+            side_effect=_resolve_for({"b": None, "c": 30}),
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", ["http://mirror/{VERSION}"]),
+                ("b", "2.0", tmp_path / "b", []),
+                ("c", "3.0", tmp_path / "c", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_resolve_failure_defers_to_install(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A registry failure only skips the prefetch; install_package reports
+    the real error with context."""
+    caplog.set_level("DEBUG")
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": -1, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+    assert "Prefetch resolve for a failed" in caplog.text
+
+
+def test_prefetch_packages_complete_archive_skipped(tmp_path: Path) -> None:
+    """An archive already fully downloaded is not re-fetched."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x" * 10)
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            dl,
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_download_failure_is_debug(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed prefetch download is logged and left for install_package."""
+    caplog.set_level("DEBUG")
+    with (
+        patch.object(
+            registry, "download_with_resume", side_effect=OSError("boom")
+        ) as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert mock_download.call_count == 2
+    assert "Prefetch of a failed" in caplog.text
+    assert "Prefetch of b failed" in caplog.text
+
+
+def test_prefetch_packages_unexpected_failure_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A programming error (not a download failure) surfaces at WARNING
+    instead of becoming a permanent silent no-op."""
+    with (
+        patch.object(
+            registry, "download_with_resume", side_effect=TypeError("bad call")
+        ),
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert "TypeError" in caplog.text

tests/unit_tests/test_platformio_toolchain.py

@@ -431,8 +431,8 @@ def test_ccache_env_enabled_by_default(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -469,7 +469,7 @@ def test_ccache_env_disabled_without_binary(
 
     with (
         patch.dict(os.environ, env_vars, clear=True),
-        patch.object(toolchain.shutil, "which", return_value=None),
+        patch("shutil.which", return_value=None),
         caplog.at_level("WARNING"),
     ):
         env = toolchain._ccache_env()
@@ -494,8 +494,8 @@ def test_ccache_env_disabled_when_probe_fails(
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run", side_effect=probe_error),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=probe_error),
     ):
         env = toolchain._ccache_env()
 
@@ -508,8 +508,8 @@ def test_ccache_env_forced_on_skips_probe(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -537,9 +537,9 @@ def test_ccache_env_strips_win_long_path_prefix(setup_core: Path) -> None:
         patch.dict(os.environ, {}, clear=True),
         # shutil.which is patched, so the win32 code path of the real
         # implementation (which crashes on a POSIX host) is never reached.
-        patch("esphome.platformio.toolchain.sys.platform", "win32"),
-        patch.object(toolchain.shutil, "which", return_value=prefixed),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("esphome.framework_helpers.sys.platform", "win32"),
+        patch("shutil.which", return_value=prefixed),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -555,7 +555,7 @@ def test_ccache_env_opt_out(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -568,7 +568,7 @@ def test_ccache_env_normalizes_enable_value(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "yes"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -587,8 +587,8 @@ def test_ccache_env_respects_user_values_and_refreshes_basedir(
 
     with (
         patch.dict(os.environ, user_env, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -606,8 +606,8 @@ def test_run_platformio_cli_passes_ccache_env_to_subprocess_only(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         mock_run_external_process.return_value = 0
@@ -628,8 +628,8 @@ def test_ccache_env_requires_build_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
         pytest.raises(ValueError, match="CORE.build_path must be set"),
     ):
         toolchain._ccache_env()
@@ -642,8 +642,8 @@ def test_run_platformio_cli_merges_caller_env(
     CORE.build_path = str(setup_core / "build" / "test")
 
     with (
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         mock_run_external_process.return_value = 0
         toolchain.run_platformio_cli(
@@ -800,9 +800,7 @@ def test_ccache_env_real_probe_runs_stripped_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(
-            toolchain.shutil, "which", return_value="\\\\?\\" + sys.executable
-        ),
+        patch("shutil.which", return_value="\\\\?\\" + sys.executable),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         env = toolchain._ccache_env()
@@ -843,40 +841,6 @@ def test_ccache_wrapper_through_cmd_exe(
         assert marker.read_text() == "compiled"
 
 
-@pytest.mark.parametrize(
-    ("platform", "input_path", "expected"),
-    [
-        # win32: drive-letter extended-length prefix is stripped
-        (
-            "win32",
-            "\\\\?\\C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-        ),
-        # win32: UNC extended-length prefix is translated to a regular UNC path
-        (
-            "win32",
-            "\\\\?\\UNC\\server\\share\\python.exe",
-            "\\\\server\\share\\python.exe",
-        ),
-        # win32: paths without the prefix are returned unchanged
-        (
-            "win32",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-        ),
-        # non-win32: prefix is left alone (no-op)
-        ("linux", "\\\\?\\C:\\python.exe", "\\\\?\\C:\\python.exe"),
-        ("darwin", "/usr/bin/python3", "/usr/bin/python3"),
-    ],
-)
-def test_strip_win_long_path_prefix(
-    platform: str, input_path: str, expected: str
-) -> None:
-    r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
-    with patch("esphome.platformio.toolchain.sys.platform", platform):
-        assert toolchain._strip_win_long_path_prefix(input_path) == expected
-
-
 def test_run_platformio_cli_strips_win_long_path_prefix(
     setup_core: Path, mock_run_external_process: Mock
 ) -> None:
@@ -900,7 +864,7 @@ def test_run_platformio_cli_strips_win_long_path_prefix(
         # so the stdlib sees it too) would send shutil.which down the Windows
         # code path, which crashes on a POSIX host.
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=False),
-        patch("esphome.platformio.toolchain.sys.platform", "win32"),
+        patch("esphome.framework_helpers.sys.platform", "win32"),
         patch("esphome.platformio.toolchain.sys.executable", prefixed_exe),
     ):
         # Pop any pre-existing PYTHONEXEPATH so the assertion below reflects
@@ -932,7 +896,7 @@ def test_run_platformio_cli_does_not_set_pythonexepath_without_strip(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio.toolchain.sys.platform", "linux"),
+        patch("esphome.framework_helpers.sys.platform", "linux"),
         patch("esphome.platformio.toolchain.sys.executable", plain_exe),
     ):
         os.environ.pop("PYTHONEXEPATH", None)
@@ -1977,10 +1941,3 @@ def test_run_platformio_cli_invokes_heal(
     with patch.object(toolchain, "heal_platformio_python_env") as mock_heal:
         toolchain.run_platformio_cli("test")
     mock_heal.assert_called_once()
-
-
-def test_ccache_probe_spawns_with_close_fds_false() -> None:
-    """The probe follows the repo-wide posix_spawn convention."""
-    with patch("subprocess.run") as mock_run:
-        assert toolchain._ccache_runs("/usr/bin/ccache") is True
-    assert mock_run.call_args.kwargs["close_fds"] is False

tests/unit_tests/test_writer.py

@@ -14,6 +14,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from esphome.build_helpers.tools_cache import TOOLS_CACHE_SPECS
 from esphome.const import (
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
@@ -68,15 +69,12 @@ def _isolate_platformio_paths(tmp_path_factory: pytest.TempPathFactory) -> Any:
     test_clean_all_partial_exists) install their own inner patch which
     stacks on top of this one and wins for the duration of their block.
 
-    Also pin ``ESPHOME_ESP_IDF_PREFIX`` and ``ESPHOME_SDK_NRF_PREFIX`` to
-    nonexistent tmp dirs, and patch ``platformdirs.user_cache_dir``, for the
-    same reason: ``clean_all`` removes the machine-global toolchain installs
+    Also pin every ``TOOLS_CACHE_SPECS`` env override to a nonexistent tmp
+    dir, and patch ``platformdirs.user_cache_dir``, for the same reason: ``clean_all`` removes the machine-global toolchain installs
     and their default cache root, which otherwise resolve to the real
     ``~/.cache/esphome``.
     """
     pio_root = tmp_path_factory.mktemp("isolated_pio") / "nonexistent"
-    idf_root = tmp_path_factory.mktemp("isolated_idf") / "nonexistent"
-    sdk_nrf_root = tmp_path_factory.mktemp("isolated_sdk_nrf") / "nonexistent"
     cache_root = tmp_path_factory.mktemp("isolated_cache") / "nonexistent"
     mock_cfg = MagicMock()
     mock_cfg.get.side_effect = lambda section, option: (
@@ -90,8 +88,12 @@ def _isolate_platformio_paths(tmp_path_factory: pytest.TempPathFactory) -> Any:
         patch.dict(
             "os.environ",
             {
-                "ESPHOME_ESP_IDF_PREFIX": str(idf_root),
-                "ESPHOME_SDK_NRF_PREFIX": str(sdk_nrf_root),
+                # Derived from the registry so a new backend's cache can
+                # never drift out of the sandbox and hit a real toolchain
+                env_var: str(
+                    tmp_path_factory.mktemp(f"isolated_{subdir}") / "nonexistent"
+                )
+                for env_var, subdir in TOOLS_CACHE_SPECS
             },
         ),
         patch("platformdirs.user_cache_dir", return_value=str(cache_root)),
@@ -1057,6 +1059,28 @@ def test_clean_all_removes_global_sdk_nrf_install(
     assert str(sdk_nrf_install.resolve()) in caplog.text
 
 
+@patch("esphome.writer.CORE")
+def test_clean_all_removes_global_arduino8266_install(
+    mock_core: MagicMock,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """clean_all removes the machine-global native arduino8266 install dir."""
+    arduino8266_install = tmp_path / "arduino8266_install"
+    (arduino8266_install / "frameworks").mkdir(parents=True)
+    monkeypatch.setenv("ESPHOME_ARDUINO8266_PREFIX", str(arduino8266_install))
+
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+
+    with caplog.at_level("INFO"):
+        clean_all([str(config_dir)])
+
+    assert not arduino8266_install.exists()
+    assert str(arduino8266_install.resolve()) in caplog.text
+
+
 @patch("esphome.writer.CORE")
 def test_clean_all_removes_default_cache_root(
     mock_core: MagicMock,