Merge remote-tracking branch 'upstream/dev' into core-prebuilt-dir-upload

# Conflicts:
#	esphome/__main__.py

tests/components/nrf52/test.nrf52-adafruit.yaml

@@ -20,4 +20,4 @@ nrf52:
     voltage: 2.1V
     uicr_erase: true
   framework:
-    version: "2.6.1-a"
+    version: "2.6.1-b"

tests/components/zigbee/common_esp32.yaml

@@ -1,10 +1,10 @@
+packages:
+  - !include common.yaml
+
 binary_sensor:
   - platform: template
     name: "Garage Door Open 10"
     report: "enable"
-  - platform: template
-    name: "Garage Door Open 11"
-    report: "coordinator"
   - platform: template
     name: "Garage Door Open 12"
     report: "force"

tests/dashboard/test_web_server.py

@@ -128,8 +128,8 @@ def mock_storage_json() -> Generator[MagicMock]:
 
 @pytest.fixture
 def mock_idedata() -> Generator[MagicMock]:
-    """Fixture to mock platformio_api.IDEData."""
-    with patch("esphome.dashboard.web_server.platformio_api.IDEData") as mock:
+    """Fixture to mock platformio toolchain.IDEData."""
+    with patch("esphome.dashboard.web_server.toolchain.IDEData") as mock:
         yield mock
 
 

tests/integration/conftest.py

@@ -23,7 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
-from esphome.platformio_api import get_idedata
+from esphome.platformio.toolchain import get_idedata
 
 from .const import (
     API_CONNECTION_TIMEOUT,

tests/integration/fixtures/host_ota_rejects_garbage.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG

tests/integration/fixtures/host_ota_self_update.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+logger:
+  level: DEBUG

tests/integration/test_host_ota.py

@@ -0,0 +1,152 @@
+"""End-to-end OTA tests on the host platform.
+
+Exercises the native OTA protocol against a real host binary, then asserts
+pid is preserved across the post-OTA execv. A second OTA on the post-exec
+instance covers the FD_CLOEXEC path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Generator
+from contextlib import contextmanager
+import socket
+
+import pytest
+
+from esphome import espota2
+
+from .conftest import run_binary, wait_and_connect_api_client
+from .const import LOCALHOST, PORT_POLL_INTERVAL, PORT_WAIT_TIMEOUT
+from .types import CompileFunction, ConfigWriter
+
+DEVICE_NAME = "host-ota-test"
+
+
+@contextmanager
+def _reserve_port() -> Generator[tuple[int, socket.socket]]:
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind(("", 0))
+    try:
+        yield s.getsockname()[1], s
+    finally:
+        s.close()
+
+
+async def _wait_for_port(host: str, port: int, timeout: float) -> None:
+    """Poll until a TCP port accepts connections, or raise TimeoutError."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        try:
+            _, writer = await asyncio.open_connection(host, port)
+        except (ConnectionRefusedError, OSError):
+            await asyncio.sleep(PORT_POLL_INTERVAL)
+            continue
+        writer.close()
+        await writer.wait_closed()
+        return
+    raise TimeoutError(f"Port {port} on {host} did not open within {timeout}s")
+
+
+@pytest.mark.asyncio
+async def test_host_ota_self_update(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """Self-OTA: upload the running binary back to itself, expect re-exec."""
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+
+        loop = asyncio.get_running_loop()
+        ota_staged = loop.create_future()
+        rebooted = loop.create_future()
+
+        def on_log(line: str) -> None:
+            if not ota_staged.done() and "OTA staged at" in line:
+                ota_staged.set_result(True)
+            if not rebooted.done() and "Rebooting safely" in line:
+                rebooted.set_result(True)
+
+        async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info_before = await client.device_info()
+                assert info_before.name == DEVICE_NAME
+
+            # espota2 is blocking; run in executor.
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 0, "espota2 reported failure"
+
+            await asyncio.wait_for(ota_staged, timeout=10.0)
+            await asyncio.wait_for(rebooted, timeout=10.0)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+
+            # execv preserves pid; mismatch means external respawn.
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info_after = await client.device_info()
+                assert info_after.name == DEVICE_NAME
+                assert info_after.name == info_before.name
+
+            # Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 0, "second OTA failed -- listener leaked across execv"
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.pid == pid_before
+
+
+@pytest.mark.asyncio
+async def test_host_ota_rejects_garbage(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+    integration_test_dir,
+) -> None:
+    """Bogus payload is rejected and the device keeps running."""
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+
+        # 192 bytes that are neither ELF nor Mach-O.
+        bogus_path = integration_test_dir / "bogus.bin"
+        bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
+
+        api_socket.close()
+        ota_socket.close()
+
+        async with run_binary(binary_path) as (proc, _lines):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            loop = asyncio.get_running_loop()
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, bogus_path
+            )
+            assert rc == 1
+
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected OTA"
+            assert proc.pid == pid_before
+
+            async with wait_and_connect_api_client(port=api_port) as client:
+                info = await client.device_info()
+                assert info.name == DEVICE_NAME

tests/unit_tests/components/test_wifi.py

@@ -10,27 +10,44 @@ from esphome.const import Platform
 @pytest.mark.parametrize(
     "variant",
     [
+        # Upstream's canonical uppercase form.
         const.VARIANT_ESP32,
         const.VARIANT_ESP32S2,
         const.VARIANT_ESP32S3,
         const.VARIANT_ESP32C3,
         const.VARIANT_ESP32C6,
+        # Lowercase form external callers (e.g. device-builder's
+        # ``Esp32Variant`` StrEnum) surface.
+        "esp32",
+        "esp32s3",
+        "esp32c3",
+        # Mixed-case — defence in depth against future callers that
+        # pull the value off some other serialisation.
+        "Esp32",
     ],
 )
 def test_variant_has_wifi_for_native_phy_variants(variant: str) -> None:
-    """Variants with a native WiFi PHY → True."""
+    """Variants with a native WiFi PHY → True, case-insensitive."""
     assert variant_has_wifi(variant) is True
 
 
 @pytest.mark.parametrize(
     "variant",
     [
+        # Upstream's canonical uppercase form.
         const.VARIANT_ESP32H2,
         const.VARIANT_ESP32P4,
+        # Lowercase form external callers (e.g. device-builder's
+        # ``Esp32Variant`` StrEnum) surface.
+        "esp32h2",
+        "esp32p4",
+        # Mixed-case — defence in depth against future callers that
+        # pull the value off some other serialisation.
+        "Esp32H2",
     ],
 )
 def test_variant_has_wifi_for_no_phy_variants(variant: str) -> None:
-    """Variants that need ``esp32_hosted`` → False."""
+    """Variants that need ``esp32_hosted`` → False, case-insensitive."""
     assert variant_has_wifi(variant) is False
 
 
@@ -44,6 +61,18 @@ def test_has_native_wifi_dispatches_esp32_to_variant_check() -> None:
     )
 
 
+def test_has_native_wifi_esp32_variant_case_insensitive() -> None:
+    """has_native_wifi accepts lowercase variant input.
+
+    External callers (device-builder's wizard, etc.) may surface
+    variant strings from their own enums that don't match upstream's
+    uppercase convention. The dispatcher should classify them
+    identically.
+    """
+    assert has_native_wifi(platform=Platform.ESP32, variant="esp32h2") is False
+    assert has_native_wifi(platform=Platform.ESP32, variant="esp32c3") is True
+
+
 def test_has_native_wifi_dispatches_rp2040_to_board_check() -> None:
     """RP2040 platform routes through ``rp2040.board_id_has_wifi``."""
     assert has_native_wifi(platform=Platform.RP2040, board="rpipicow") is True

tests/unit_tests/conftest.py

@@ -64,15 +64,15 @@ def mock_copy_file_if_changed() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_run_platformio_cli() -> Generator[Mock, None, None]:
-    """Mock run_platformio_cli for platformio_api."""
-    with patch("esphome.platformio_api.run_platformio_cli") as mock:
+    """Mock run_platformio_cli for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_platformio_cli") as mock:
         yield mock
 
 
 @pytest.fixture
 def mock_run_platformio_cli_run() -> Generator[Mock, None, None]:
-    """Mock run_platformio_cli_run for platformio_api."""
-    with patch("esphome.platformio_api.run_platformio_cli_run") as mock:
+    """Mock run_platformio_cli_run for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_platformio_cli_run") as mock:
         yield mock
 
 
@@ -92,8 +92,8 @@ def mock_esp8266_decode_pc() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_run_external_process() -> Generator[Mock, None, None]:
-    """Mock run_external_process for platformio_api."""
-    with patch("esphome.platformio_api.run_external_process") as mock:
+    """Mock run_external_process for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.run_external_process") as mock:
         yield mock
 
 
@@ -113,8 +113,8 @@ def mock_subprocess_run() -> Generator[Mock, None, None]:
 
 @pytest.fixture
 def mock_get_idedata() -> Generator[Mock, None, None]:
-    """Mock get_idedata for platformio_api."""
-    with patch("esphome.platformio_api.get_idedata") as mock:
+    """Mock get_idedata for platformio toolchain."""
+    with patch("esphome.platformio.toolchain.get_idedata") as mock:
         yield mock
 
 

tests/unit_tests/test_core.py

@@ -591,6 +591,30 @@ class TestEsphomeCore:
         assert target.is_esp32 is False
         assert target.is_esp8266 is True
 
+    def test_firmware_bin__default(self, target):
+        """Default platforms produce <pioenvs>/<name>/firmware.bin."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "esp32"}
+        assert target.firmware_bin == Path(
+            "foo/build/.pioenvs/test-device/firmware.bin"
+        )
+
+    def test_firmware_bin__libretiny(self, target):
+        """The libretiny platform produces firmware.uf2."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "bk72xx"}
+        assert target.firmware_bin == Path(
+            "foo/build/.pioenvs/test-device/firmware.uf2"
+        )
+
+    def test_firmware_bin__host(self, target):
+        """Host platform produces a native ELF/Mach-O named `program`,
+        not firmware.bin -- needed for `esphome upload` to find the
+        right artifact for the host OTA backend."""
+        target.name = "test-device"
+        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "host"}
+        assert target.firmware_bin == Path("foo/build/.pioenvs/test-device/program")
+
     @pytest.mark.skipif(os.name == "nt", reason="Unix-specific test")
     def test_data_dir_default_unix(self, target):
         """Test data_dir returns .esphome in config directory by default on Unix."""

tests/unit_tests/test_main.py

@@ -18,7 +18,6 @@ import pytest
 from pytest import CaptureFixture
 from zeroconf import ServiceStateChange
 
-from esphome import platformio_api
 from esphome.__main__ import (
     Purpose,
     _get_configured_xtal_freq,
@@ -97,6 +96,7 @@ from esphome.espota2 import (
     OTA_TYPE_UPDATE_BOOTLOADER,
     OTA_TYPE_UPDATE_PARTITION_TABLE,
 )
+from esphome.platformio import toolchain
 from esphome.util import BootselResult, FlashImage
 from esphome.zeroconf import _await_discovery, discover_mdns_devices
 
@@ -288,7 +288,7 @@ def mock_run_external_process() -> Generator[Mock]:
 
 @pytest.fixture
 def mock_run_external_command_main() -> Generator[Mock]:
-    """Mock run_external_command in __main__ module (different from platformio_api)."""
+    """Mock run_external_command in __main__ module (different from platformio toolchain)."""
     with patch("esphome.__main__.run_external_command") as mock:
         mock.return_value = 0  # Default to success
         yield mock
@@ -1201,7 +1201,7 @@ def test_upload_using_esptool_path_conversion(
     CORE.data[KEY_ESP32] = {KEY_VARIANT: VARIANT_ESP32}
 
     # Create mock IDEData with Path objects
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
         FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
@@ -1279,7 +1279,7 @@ def test_upload_using_esptool_skips_missing_extra_flash_images(
 
     missing_path = tmp_path / "variants" / "tasmota" / "tinyuf2.bin"
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = [
         FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
@@ -1547,8 +1547,8 @@ def test_upload_using_platformio_creates_signed_bin_for_rp2040(
     mock_idedata.firmware_elf_path = str(firmware_elf)
 
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
-        patch("esphome.platformio_api.run_platformio_cli_run", return_value=0),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.run_platformio_cli_run", return_value=0),
     ):
         result = upload_using_platformio({}, "/dev/ttyACM0")
 
@@ -1564,7 +1564,7 @@ def test_upload_using_platformio_skips_signed_bin_for_non_rp2040(
     """Test that upload_using_platformio doesn't create signed bin for non-RP2040."""
     setup_core(platform=PLATFORM_ESP32)
 
-    with patch("esphome.platformio_api.run_platformio_cli_run", return_value=0):
+    with patch("esphome.platformio.toolchain.run_platformio_cli_run", return_value=0):
         result = upload_using_platformio({}, "/dev/ttyUSB0")
 
     assert result == 0
@@ -1621,7 +1621,7 @@ def test_upload_program_prebuilt_dir_installs_libretiny_platform_if_missing(
         patch("esphome.__main__.get_ltchiptool_path", return_value=None) as mock_find,
         patch("esphome.__main__.write_cpp", return_value=0) as mock_write_cpp,
         patch(
-            "esphome.platformio_api.prepare_platform_for_upload", return_value=0
+            "esphome.platformio.toolchain.prepare_platform_for_upload", return_value=0
         ) as mock_prep,
         patch("esphome.__main__.upload_using_ltchiptool", return_value=0),
     ):
@@ -1655,7 +1655,7 @@ def test_upload_program_prebuilt_dir_skips_install_when_tool_present(
             return_value=tmp_path / "ltchiptool",
         ),
         patch("esphome.__main__.write_cpp") as mock_write_cpp,
-        patch("esphome.platformio_api.prepare_platform_for_upload") as mock_prep,
+        patch("esphome.platformio.toolchain.prepare_platform_for_upload") as mock_prep,
         patch("esphome.__main__.upload_using_ltchiptool", return_value=0),
     ):
         exit_code, _ = upload_program({}, args, ["/dev/ttyUSB0"])
@@ -1684,7 +1684,7 @@ def test_upload_program_prebuilt_dir_write_cpp_failure_aborts_upload(
     with (
         patch("esphome.__main__.get_ltchiptool_path", return_value=None),
         patch("esphome.__main__.write_cpp", return_value=3),
-        patch("esphome.platformio_api.prepare_platform_for_upload") as mock_prep,
+        patch("esphome.platformio.toolchain.prepare_platform_for_upload") as mock_prep,
         patch("esphome.__main__.upload_using_ltchiptool") as mock_upload,
     ):
         exit_code, host = upload_program({}, args, ["/dev/ttyUSB0"])
@@ -1713,7 +1713,9 @@ def test_upload_program_prebuilt_dir_pkg_install_failure_aborts_upload(
     with (
         patch("esphome.__main__._find_picotool", return_value=None),
         patch("esphome.__main__.write_cpp", return_value=0),
-        patch("esphome.platformio_api.prepare_platform_for_upload", return_value=2),
+        patch(
+            "esphome.platformio.toolchain.prepare_platform_for_upload", return_value=2
+        ),
         patch("esphome.__main__.upload_using_picotool") as mock_upload,
         patch("esphome.__main__._rp2040_serial_reset_to_bootsel") as mock_reset,
     ):
@@ -1773,7 +1775,7 @@ def test_upload_using_picotool_falls_back_to_firmware_bin_when_elf_missing(
 
     # Stub the picotool lookup to short-circuit the toolchain probe.
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
         patch(
             "esphome.__main__.get_picotool_path",
             return_value=tmp_path / "picotool",
@@ -2019,7 +2021,7 @@ def test_upload_using_picotool_success(tmp_path: Path) -> None:
 
     config = {}
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
         patch("subprocess.run", return_value=mock_result),
     ):
         exit_code = upload_using_picotool(config)
@@ -2039,7 +2041,7 @@ def test_upload_using_picotool_no_elf(tmp_path: Path) -> None:
     mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
+    with patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata):
         exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
@@ -2059,7 +2061,7 @@ def test_upload_using_picotool_not_found(tmp_path: Path) -> None:
     mock_idedata.cc_path = "/fake/path/gcc"
 
     config = {}
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
+    with patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata):
         exit_code = upload_using_picotool(config)
 
     assert exit_code == 1
@@ -2093,7 +2095,7 @@ def test_upload_using_picotool_permission_error(tmp_path: Path) -> None:
 
     config = {}
     with (
-        patch("esphome.platformio_api.get_idedata", return_value=mock_idedata),
+        patch("esphome.platformio.toolchain.get_idedata", return_value=mock_idedata),
         patch("subprocess.run", return_value=mock_result),
     ):
         exit_code = upload_using_picotool(config)
@@ -5211,7 +5213,7 @@ def test_command_analyze_memory_success(
     firmware_elf.write_text("mock elf file")
 
     # Mock idedata
-    mock_idedata_obj = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata_obj = MagicMock(spec=toolchain.IDEData)
     mock_idedata_obj.firmware_elf_path = str(firmware_elf)
     mock_idedata_obj.objdump_path = "/path/to/objdump"
     mock_idedata_obj.readelf_path = "/path/to/readelf"
@@ -5283,7 +5285,7 @@ def test_command_analyze_memory_with_external_components(
     firmware_elf.write_text("mock elf file")
 
     # Mock idedata
-    mock_idedata_obj = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata_obj = MagicMock(spec=toolchain.IDEData)
     mock_idedata_obj.firmware_elf_path = str(firmware_elf)
     mock_idedata_obj.objdump_path = "/path/to/objdump"
     mock_idedata_obj.readelf_path = "/path/to/readelf"
@@ -5374,16 +5376,18 @@ def test_command_analyze_memory_no_idedata(
 
 @pytest.fixture
 def mock_compile_build_info_run_compile() -> Generator[Mock]:
-    """Mock platformio_api.run_compile for build_info tests."""
-    with patch("esphome.platformio_api.run_compile", return_value=0) as mock:
+    """Mock toolchain.run_compile for build_info tests."""
+    with patch("esphome.platformio.toolchain.run_compile", return_value=0) as mock:
         yield mock
 
 
 @pytest.fixture
 def mock_compile_build_info_get_idedata() -> Generator[Mock]:
-    """Mock platformio_api.get_idedata for build_info tests."""
+    """Mock toolchain.get_idedata for build_info tests."""
     mock_idedata = MagicMock()
-    with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata) as mock:
+    with patch(
+        "esphome.platformio.toolchain.get_idedata", return_value=mock_idedata
+    ) as mock:
         yield mock
 
 
@@ -6293,7 +6297,7 @@ def test_upload_using_esptool_passes_crystal_callback(
     sdkconfig = build_dir / "sdkconfig.test"
     sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n")
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = []
     mock_get_idedata.return_value = mock_idedata
@@ -6323,7 +6327,7 @@ def test_upload_using_esptool_subprocess_passes_crystal_callback(
     sdkconfig = build_dir / "sdkconfig.test"
     sdkconfig.write_text("CONFIG_XTAL_FREQ=40\n")
 
-    mock_idedata = MagicMock(spec=platformio_api.IDEData)
+    mock_idedata = MagicMock(spec=toolchain.IDEData)
     mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
     mock_idedata.extra_flash_images = []
     mock_get_idedata.return_value = mock_idedata

tests/unit_tests/test_platformio_toolchain.py

@@ -1,4 +1,4 @@
-"""Tests for platformio_api.py path functions."""
+"""Tests for esphome.platformio.toolchain path functions."""
 
 # pylint: disable=protected-access
 
@@ -11,8 +11,8 @@ from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
 
-from esphome import platformio_api, platformio_runner
 from esphome.core import CORE, EsphomeError
+from esphome.platformio import runner, toolchain
 from esphome.util import FlashImage
 
 
@@ -21,7 +21,7 @@ def test_idedata_firmware_elf_path(setup_core: Path) -> None:
     CORE.build_path = setup_core / "build" / "test"
     CORE.name = "test"
     raw_data = {"prog_path": "/path/to/firmware.elf"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     assert idedata.firmware_elf_path == Path("/path/to/firmware.elf")
 
@@ -32,7 +32,7 @@ def test_idedata_firmware_bin_path(setup_core: Path) -> None:
     CORE.name = "test"
     prog_path = str(Path("/path/to/firmware.elf"))
     raw_data = {"prog_path": prog_path}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.firmware_bin_path
     assert isinstance(result, Path)
@@ -47,7 +47,7 @@ def test_idedata_firmware_bin_path_preserves_directory(setup_core: Path) -> None
     CORE.name = "test"
     prog_path = str(Path("/complex/path/to/build/firmware.elf"))
     raw_data = {"prog_path": prog_path}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.firmware_bin_path
     expected = Path("/complex/path/to/build/firmware.bin")
@@ -67,7 +67,7 @@ def test_idedata_extra_flash_images(setup_core: Path) -> None:
             ]
         },
     }
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     images = idedata.extra_flash_images
     assert len(images) == 2
@@ -83,7 +83,7 @@ def test_idedata_extra_flash_images_empty(setup_core: Path) -> None:
     CORE.build_path = setup_core / "build" / "test"
     CORE.name = "test"
     raw_data = {"prog_path": "/path/to/firmware.elf", "extra": {"flash_images": []}}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     images = idedata.extra_flash_images
     assert images == []
@@ -97,7 +97,7 @@ def test_idedata_cc_path(setup_core: Path) -> None:
         "prog_path": "/path/to/firmware.elf",
         "cc_path": "/Users/test/.platformio/packages/toolchain-xtensa32/bin/xtensa-esp32-elf-gcc",
     }
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     assert (
         idedata.cc_path
@@ -132,7 +132,7 @@ def test_load_idedata_returns_dict(
     mock_run_platformio_cli_run.return_value = '{"prog_path": "/test/firmware.elf"}'
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     assert result is not None
     assert isinstance(result, dict)
@@ -161,7 +161,7 @@ def test_load_idedata_uses_cache_when_valid(
     os.utime(idedata_path, (platformio_ini_mtime + 1, platformio_ini_mtime + 1))
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should not call _run_idedata since cache is valid
     mock_run_platformio_cli_run.assert_not_called()
@@ -194,7 +194,7 @@ def test_load_idedata_regenerates_when_platformio_ini_newer(
     mock_run_platformio_cli_run.return_value = json.dumps(new_data)
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should call _run_idedata since platformio.ini is newer
     mock_run_platformio_cli_run.assert_called_once()
@@ -228,7 +228,7 @@ def test_load_idedata_regenerates_on_corrupted_cache(
     mock_run_platformio_cli_run.return_value = json.dumps(new_data)
 
     config = {"name": "test"}
-    result = platformio_api._load_idedata(config)
+    result = toolchain._load_idedata(config)
 
     # Should call _run_idedata since cache is corrupted
     mock_run_platformio_cli_run.assert_called_once()
@@ -255,7 +255,7 @@ def test_load_idedata_uses_prebuilt_dir_when_set(
 
     CORE.prebuilt_dir = prebuilt_dir
 
-    result = platformio_api._load_idedata({"name": "test"})
+    result = toolchain._load_idedata({"name": "test"})
 
     assert result["prog_path"] == str(prebuilt_dir / "firmware.elf")
     # Never re-runs PlatformIO when prebuilt idedata is supplied: the dashboard
@@ -293,7 +293,7 @@ def test_load_idedata_resolves_basenames_against_prebuilt_dir(
     )
     CORE.prebuilt_dir = prebuilt_dir
 
-    result = platformio_api._load_idedata({"name": "test"})
+    result = toolchain._load_idedata({"name": "test"})
 
     assert result["prog_path"] == str(prebuilt_dir / "firmware.elf")
     images = result["extra"]["flash_images"]
@@ -330,7 +330,7 @@ def test_load_idedata_absolute_paths_in_prebuilt_pass_through(
     )
     CORE.prebuilt_dir = prebuilt_dir
 
-    result = platformio_api._load_idedata({"name": "test"})
+    result = toolchain._load_idedata({"name": "test"})
 
     assert result["prog_path"] == str(prebuilt_dir / "firmware.elf")
     assert result["extra"]["flash_images"][0]["path"] == abs_bootloader
@@ -358,7 +358,7 @@ def test_load_idedata_falls_back_when_prebuilt_idedata_missing(
         {"prog_path": "/local/firmware.elf"}
     )
 
-    result = platformio_api._load_idedata({"name": "test"})
+    result = toolchain._load_idedata({"name": "test"})
 
     assert result["prog_path"] == "/local/firmware.elf"
     mock_run_platformio_cli_run.assert_called_once()
@@ -381,7 +381,7 @@ def test_run_idedata_parses_json_from_output(
         f"Some preamble\n{json.dumps(expected_data)}\nSome postamble"
     )
 
-    result = platformio_api._run_idedata(config)
+    result = toolchain._run_idedata(config)
 
     assert result == expected_data
 
@@ -395,7 +395,7 @@ def test_run_idedata_raises_on_no_json(
     mock_run_platformio_cli_run.return_value = "No JSON in this output"
 
     with pytest.raises(EsphomeError):
-        platformio_api._run_idedata(config)
+        toolchain._run_idedata(config)
 
 
 def test_run_idedata_raises_on_invalid_json(
@@ -407,7 +407,7 @@ def test_run_idedata_raises_on_invalid_json(
 
     # The ValueError from json.loads is re-raised
     with pytest.raises(ValueError):
-        platformio_api._run_idedata(config)
+        toolchain._run_idedata(config)
 
 
 def test_run_platformio_cli_sets_environment_variables(
@@ -418,7 +418,7 @@ def test_run_platformio_cli_sets_environment_variables(
 
     with patch.dict(os.environ, {}, clear=False):
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         # Check environment variables were set
         assert os.environ["PLATFORMIO_FORCE_COLOR"] == "true"
@@ -431,11 +431,11 @@ def test_run_platformio_cli_sets_environment_variables(
         assert "PYTHONWARNINGS" in os.environ
 
         # Check command was called correctly — runs PlatformIO as a subprocess
-        # via the esphome.platformio_runner entry point.
+        # via the esphome.platformio.runner entry point.
         mock_run_external_process.assert_called_once()
         args = mock_run_external_process.call_args[0]
         assert "-m" in args
-        assert "esphome.platformio_runner" in args
+        assert "esphome.platformio.runner" in args
         assert "test" in args
         assert "arg" in args
 
@@ -470,8 +470,8 @@ 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_api.sys.platform", platform):
-        assert platformio_api._strip_win_long_path_prefix(input_path) == expected
+    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(
@@ -494,15 +494,15 @@ def test_run_platformio_cli_strips_win_long_path_prefix(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio_api.sys.platform", "win32"),
-        patch("esphome.platformio_api.sys.executable", prefixed_exe),
+        patch("esphome.platformio.toolchain.sys.platform", "win32"),
+        patch("esphome.platformio.toolchain.sys.executable", prefixed_exe),
     ):
         # Pop any pre-existing PYTHONEXEPATH so the assertion below reflects
         # what run_platformio_cli set, not whatever the test runner's
         # environment happened to contain.
         os.environ.pop("PYTHONEXEPATH", None)
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         # The subprocess is invoked with the stripped executable path.
         mock_run_external_process.assert_called_once()
@@ -526,12 +526,12 @@ def test_run_platformio_cli_does_not_set_pythonexepath_without_strip(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio_api.sys.platform", "linux"),
-        patch("esphome.platformio_api.sys.executable", plain_exe),
+        patch("esphome.platformio.toolchain.sys.platform", "linux"),
+        patch("esphome.platformio.toolchain.sys.executable", plain_exe),
     ):
         os.environ.pop("PYTHONEXEPATH", None)
         mock_run_external_process.return_value = 0
-        platformio_api.run_platformio_cli("test", "arg")
+        toolchain.run_platformio_cli("test", "arg")
 
         mock_run_external_process.assert_called_once()
         args = mock_run_external_process.call_args[0]
@@ -547,7 +547,7 @@ def test_run_platformio_cli_run_builds_command(
     mock_run_platformio_cli.return_value = 0
 
     config = {"name": "test"}
-    platformio_api.run_platformio_cli_run(config, True, "extra", "args")
+    toolchain.run_platformio_cli_run(config, True, "extra", "args")
 
     mock_run_platformio_cli.assert_called_once_with(
         "run", "-d", CORE.build_path, "-v", "extra", "args"
@@ -562,7 +562,7 @@ def test_run_compile(setup_core: Path, mock_run_platformio_cli_run: Mock) -> Non
     config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 4}}
     mock_run_platformio_cli_run.return_value = 0
 
-    platformio_api.run_compile(config, verbose=True)
+    toolchain.run_compile(config, verbose=True)
 
     mock_run_platformio_cli_run.assert_called_once_with(config, True, "-j4")
 
@@ -589,22 +589,22 @@ def test_get_idedata_caches_result(
     config = {"name": "test"}
 
     # First call should load and cache
-    result1 = platformio_api.get_idedata(config)
+    result1 = toolchain.get_idedata(config)
     mock_run_platformio_cli_run.assert_called_once()
 
     # Second call should use cache from CORE.data
-    result2 = platformio_api.get_idedata(config)
+    result2 = toolchain.get_idedata(config)
     mock_run_platformio_cli_run.assert_called_once()  # Still only called once
 
     assert result1 is result2
-    assert isinstance(result1, platformio_api.IDEData)
+    assert isinstance(result1, toolchain.IDEData)
     assert result1.firmware_elf_path == Path("/test/firmware.elf")
 
 
 def test_idedata_addr2line_path_windows(setup_core: Path) -> None:
     """Test IDEData.addr2line_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.addr2line_path
     assert result == "C:\\tools\\addr2line.exe"
@@ -613,7 +613,7 @@ def test_idedata_addr2line_path_windows(setup_core: Path) -> None:
 def test_idedata_addr2line_path_unix(setup_core: Path) -> None:
     """Test IDEData.addr2line_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.addr2line_path
     assert result == "/usr/bin/addr2line"
@@ -622,7 +622,7 @@ def test_idedata_addr2line_path_unix(setup_core: Path) -> None:
 def test_idedata_objdump_path_windows(setup_core: Path) -> None:
     """Test IDEData.objdump_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.objdump_path
     assert result == "C:\\tools\\objdump.exe"
@@ -631,7 +631,7 @@ def test_idedata_objdump_path_windows(setup_core: Path) -> None:
 def test_idedata_objdump_path_unix(setup_core: Path) -> None:
     """Test IDEData.objdump_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.objdump_path
     assert result == "/usr/bin/objdump"
@@ -640,7 +640,7 @@ def test_idedata_objdump_path_unix(setup_core: Path) -> None:
 def test_idedata_readelf_path_windows(setup_core: Path) -> None:
     """Test IDEData.readelf_path on Windows."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "C:\\tools\\gcc.exe"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.readelf_path
     assert result == "C:\\tools\\readelf.exe"
@@ -649,7 +649,7 @@ def test_idedata_readelf_path_windows(setup_core: Path) -> None:
 def test_idedata_readelf_path_unix(setup_core: Path) -> None:
     """Test IDEData.readelf_path on Unix."""
     raw_data = {"prog_path": "/path/to/firmware.elf", "cc_path": "/usr/bin/gcc"}
-    idedata = platformio_api.IDEData(raw_data)
+    idedata = toolchain.IDEData(raw_data)
 
     result = idedata.readelf_path
     assert result == "/usr/bin/readelf"
@@ -675,7 +675,7 @@ def test_patch_structhash(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Verify both modules had clean_build_dir patched
         # Check that clean_build_dir was set on both modules
@@ -727,7 +727,7 @@ def test_patched_clean_build_dir_removes_outdated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -777,7 +777,7 @@ def test_patched_clean_build_dir_keeps_updated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -825,7 +825,7 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_runner.patch_structhash()
+        runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -855,7 +855,7 @@ def test_patch_file_downloader_succeeds_first_try() -> None:
             ),
         },
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -894,7 +894,7 @@ def test_patch_file_downloader_retries_on_failure() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -935,7 +935,7 @@ def test_patch_file_downloader_raises_after_max_retries() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -983,7 +983,7 @@ def test_patch_file_downloader_closes_session_and_response_between_retries() ->
         ),
         patch("time.sleep"),
     ):
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -1018,9 +1018,9 @@ def test_patch_file_downloader_idempotent() -> None:
         },
     ):
         # Patch multiple times
-        platformio_runner.patch_file_downloader()
-        platformio_runner.patch_file_downloader()
-        platformio_runner.patch_file_downloader()
+        runner.patch_file_downloader()
+        runner.patch_file_downloader()
+        runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -1038,9 +1038,7 @@ def _filter_through_redirect(line: str) -> str:
     from esphome.util import RedirectText
 
     captured = io.StringIO()
-    redirect = RedirectText(
-        captured, filter_lines=platformio_runner.FILTER_PLATFORMIO_LINES
-    )
+    redirect = RedirectText(captured, filter_lines=runner.FILTER_PLATFORMIO_LINES)
     redirect.write(line + "\n")
     return captured.getvalue()
 

tests/unit_tests/test_upload_targets.py

@@ -0,0 +1,81 @@
+"""Tests for the stable upload-targets classification helpers."""
+
+import pytest
+
+from esphome.upload_targets import PortType, get_port_type
+
+
+@pytest.mark.parametrize(
+    "port",
+    [
+        "/dev/ttyUSB0",
+        "/dev/ttyACM0",
+        "/dev/cu.usbserial-1410",
+        "/dev/tty.usbmodem1101",
+        "COM1",
+        "COM23",
+    ],
+)
+def test_get_port_type_serial(port: str) -> None:
+    """Local serial devices classify as SERIAL."""
+    assert get_port_type(port) is PortType.SERIAL
+
+
+def test_get_port_type_bootsel() -> None:
+    """``BOOTSEL`` magic string classifies as BOOTSEL."""
+    assert get_port_type("BOOTSEL") is PortType.BOOTSEL
+
+
+def test_get_port_type_mqtt() -> None:
+    """``MQTT`` magic string classifies as MQTT."""
+    assert get_port_type("MQTT") is PortType.MQTT
+
+
+def test_get_port_type_mqttip() -> None:
+    """``MQTTIP`` magic string classifies as MQTTIP."""
+    assert get_port_type("MQTTIP") is PortType.MQTTIP
+
+
+@pytest.mark.parametrize(
+    "port",
+    [
+        "192.168.1.10",
+        "fe80::1",
+        "device.local",
+        "my-esp.example.com",
+    ],
+)
+def test_get_port_type_network(port: str) -> None:
+    """IP addresses, mDNS, and hostnames classify as NETWORK."""
+    assert get_port_type(port) is PortType.NETWORK
+
+
+def test_port_type_values_are_stable() -> None:
+    """Member values are part of the stable surface.
+
+    External tooling (device-builder, etc.) may compare against the
+    string values directly. Renaming or changing these breaks
+    downstream consumers — guard against accidental edits.
+    """
+    assert PortType.SERIAL.value == "SERIAL"
+    assert PortType.NETWORK.value == "NETWORK"
+    assert PortType.MQTT.value == "MQTT"
+    assert PortType.MQTTIP.value == "MQTTIP"
+    assert PortType.BOOTSEL.value == "BOOTSEL"
+
+
+def test_main_re_exports_for_backwards_compat() -> None:
+    """``esphome.__main__`` re-exports the stable surface.
+
+    The CLI entry point pre-dated the stable module and existing
+    internal callers (and any third-party code that snuck in via
+    ``__main__``) still import from there. The re-export must
+    resolve to the same objects.
+    """
+    from esphome.__main__ import (
+        PortType as MainPortType,
+        get_port_type as main_get_port_type,
+    )
+
+    assert MainPortType is PortType
+    assert main_get_port_type is get_port_type