[cli] Add --ota-platform flag to pick web_server or native API OTA (#16207)

esphome/__main__.py

@@ -28,6 +28,7 @@ from esphome.const import (
     ALLOWED_NAME_CHARS,
     ARGUMENT_HELP_DEVICE,
     CONF_API,
+    CONF_AUTH,
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DEASSERT_RTS_DTR,
@@ -47,6 +48,8 @@ from esphome.const import (
     CONF_PORT,
     CONF_SUBSTITUTIONS,
     CONF_TOPIC,
+    CONF_USERNAME,
+    CONF_WEB_SERVER,
     ENV_NOGITIGNORE,
     KEY_CORE,
     KEY_NATIVE_IDF,
@@ -349,6 +352,17 @@ def choose_upload_log_host(
         elif bootsel.permission_error:
             bootsel_permission_error = True
 
+    # Annotate the OTA chooser entry only in the non-default case: when the
+    # config has web_server OTA but no native API OTA, the upload will fall
+    # through to the HTTP path and the user benefits from seeing that
+    # explicitly. The native-API path is the default and gets a plain label
+    # to avoid noise on the most common scenario. For LOGGING the OTA
+    # transport doesn't apply, so always leave the label plain.
+    if purpose == Purpose.UPLOADING and not has_native_ota() and has_web_server_ota():
+        ota_suffix = " via web_server"
+    else:
+        ota_suffix = ""
+
     def add_ota_options() -> None:
         """Add OTA options, using mDNS discovery if name_add_mac_suffix is enabled."""
         if (discovered := _discover_mac_suffix_devices()) is not None:
@@ -356,11 +370,11 @@ def choose_upload_log_host(
             # intentionally skip the base-name fallback since with
             # name_add_mac_suffix on, the base name doesn't exist on the net.
             for host in discovered:
-                options.append((f"Over The Air ({host})", host))
+                options.append((f"Over The Air{ota_suffix} ({host})", host))
         elif has_resolvable_address():
-            options.append((f"Over The Air ({CORE.address})", CORE.address))
+            options.append((f"Over The Air{ota_suffix} ({CORE.address})", CORE.address))
         if has_mqtt_ip_lookup():
-            options.append(("Over The Air (MQTT IP lookup)", "MQTTIP"))
+            options.append((f"Over The Air{ota_suffix} (MQTT IP lookup)", "MQTTIP"))
 
     if purpose == Purpose.LOGGING:
         if has_mqtt_logging():
@@ -429,7 +443,19 @@ def has_api() -> bool:
 
 
 def has_ota() -> bool:
-    """Check if OTA upload is available (requires platform: esphome)."""
+    """Check if any network OTA upload is available.
+
+    True if the config exposes either ``platform: esphome`` (native API
+    OTA) or ``platform: web_server`` (HTTP OTA). Both reach the device
+    over the same network stack, so the OTA discovery path treats them
+    interchangeably; ``upload_program`` picks the actual transport based
+    on ``--ota-platform`` and what's configured.
+    """
+    return has_native_ota() or has_web_server_ota()
+
+
+def has_native_ota() -> bool:
+    """Check if native API OTA upload is available (``platform: esphome``)."""
     if CONF_OTA not in CORE.config:
         return False
     return any(
@@ -438,6 +464,16 @@ def has_ota() -> bool:
     )
 
 
+def has_web_server_ota() -> bool:
+    """Check if web_server OTA upload is available (``platform: web_server``)."""
+    if CONF_OTA not in CORE.config:
+        return False
+    return any(
+        ota_item.get(CONF_PLATFORM) == CONF_WEB_SERVER
+        for ota_item in CORE.config[CONF_OTA]
+    )
+
+
 def has_mqtt_ip_lookup() -> bool:
     """Check if MQTT is available and IP lookup is supported."""
     from esphome.components.mqtt import CONF_DISCOVER_IP
@@ -1115,25 +1151,83 @@ def upload_program(
 
         return exit_code, host if exit_code == 0 else None
 
-    ota_conf = {}
+    requested_platform = getattr(args, "ota_platform", None)
+    chosen_platform = _choose_ota_platform(config, requested_platform)
+
+    # Resolve MQTT magic strings to actual IP addresses
+    network_devices = _resolve_network_devices(devices, config, args)
+
+    if chosen_platform == CONF_WEB_SERVER:
+        if getattr(args, "partition_table", False):
+            raise EsphomeError(
+                "--partition-table is only supported with the esphome OTA platform; "
+                "the web_server OTA path can only update the firmware image."
+            )
+        binary = CORE.firmware_bin
+        if getattr(args, "file", None) is not None:
+            binary = Path(args.file)
+        return _upload_via_web_server(config, network_devices, binary)
+
+    return _upload_via_native_api(config, network_devices, args)
+
+
+def _choose_ota_platform(config: ConfigType, requested: str | None) -> str:
+    """Pick the OTA platform to use, optionally honoring ``--ota-platform``.
+
+    Default behavior prefers ``esphome`` (native API) when it is configured.
+    The native API uses challenge-response auth with MD5/SHA256 hashing of a
+    server-issued nonce, so the password is never sent over the wire; the
+    ``web_server`` path uses HTTP Basic auth which transmits credentials in
+    cleartext over the LAN. (The native path also supports gzip compression
+    on ESP8266, where flash space is tight; on ESP32/RP2040/LibreTiny the
+    backend reports ``supports_compression() == false`` and the firmware is
+    sent uncompressed regardless of which platform is used.) Falls back to
+    ``web_server`` only when that is the only available platform.
+    """
+    # Use a dict (insertion-ordered) instead of a list so error messages and
+    # membership checks see one entry per platform even if the user has
+    # multiple ``ota:`` items of the same platform; the web_server OTA
+    # platform's final-validate hook merges duplicates anyway.
+    available: dict[str, None] = {}
     for ota_item in config.get(CONF_OTA, []):
-        if ota_item[CONF_PLATFORM] == CONF_ESPHOME:
+        platform = ota_item.get(CONF_PLATFORM)
+        if platform in (CONF_ESPHOME, CONF_WEB_SERVER):
+            available[platform] = None
+
+    if not available:
+        raise EsphomeError(
+            f"Cannot upload Over the Air as the {CONF_OTA} configuration is not "
+            f"present or does not include {CONF_PLATFORM}: {CONF_ESPHOME} or "
+            f"{CONF_PLATFORM}: {CONF_WEB_SERVER}"
+        )
+
+    if requested is not None:
+        if requested not in available:
+            raise EsphomeError(
+                f"--ota-platform {requested} was requested but the configuration "
+                f"only provides: {', '.join(available)}"
+            )
+        return requested
+
+    if CONF_ESPHOME in available:
+        return CONF_ESPHOME
+    return CONF_WEB_SERVER
+
+
+def _upload_via_native_api(
+    config: ConfigType, network_devices: list[str], args: ArgsProtocol
+) -> tuple[int, str | None]:
+    ota_conf: ConfigType = {}
+    for ota_item in config.get(CONF_OTA, []):
+        if ota_item.get(CONF_PLATFORM) == CONF_ESPHOME:
             ota_conf = ota_item
             break
 
-    if not ota_conf:
-        raise EsphomeError(
-            f"Cannot upload Over the Air as the {CONF_OTA} configuration is not present or does not include {CONF_PLATFORM}: {CONF_ESPHOME}"
-        )
-
     from esphome import espota2
 
     remote_port = int(ota_conf[CONF_PORT])
     password = ota_conf.get(CONF_PASSWORD)
 
-    # Resolve MQTT magic strings to actual IP addresses
-    network_devices = _resolve_network_devices(devices, config, args)
-
     binary = CORE.firmware_bin
     ota_type = espota2.OTA_TYPE_UPDATE_APP
     if getattr(args, "partition_table", False):
@@ -1157,6 +1251,28 @@ def upload_program(
     return espota2.run_ota(network_devices, remote_port, password, binary, ota_type)
 
 
+def _upload_via_web_server(
+    config: ConfigType, network_devices: list[str], binary: Path
+) -> tuple[int, str | None]:
+    web_conf = config.get(CONF_WEB_SERVER)
+    if not web_conf:
+        raise EsphomeError(
+            f"Cannot upload via web_server OTA: the {CONF_WEB_SERVER} component "
+            f"is not configured."
+        )
+
+    remote_port = int(web_conf[CONF_PORT])
+    auth = web_conf.get(CONF_AUTH) or {}
+    username = auth.get(CONF_USERNAME)
+    password = auth.get(CONF_PASSWORD)
+
+    from esphome import web_server_ota
+
+    return web_server_ota.run_ota(
+        network_devices, remote_port, username, password, binary
+    )
+
+
 # Layout of esp_partition_info_t on flash. Each entry is 32 bytes, leading with a
 # 16-bit little-endian magic. ESP-IDF defines ESP_PARTITION_MAGIC = 0x50AA (stored as
 # bytes 0xAA, 0x50) for partition entries and ESP_PARTITION_MAGIC_MD5 = 0xEBEB for the
@@ -1877,6 +1993,17 @@ def parse_args(argv):
         "--file",
         help="Manually specify the binary file to upload.",
     )
+    parser_upload.add_argument(
+        "--ota-platform",
+        choices=[CONF_ESPHOME, CONF_WEB_SERVER],
+        help=(
+            "OTA platform to use for network uploads. Defaults to "
+            f"'{CONF_ESPHOME}' (native API) when configured because it uses "
+            "challenge-response auth so the password is never sent in "
+            f"cleartext on the wire. Falls back to '{CONF_WEB_SERVER}' "
+            "(HTTP Basic auth) when that is the only configured platform."
+        ),
+    )
     parser_upload.add_argument(
         "--partition-table",
         help="Upload as partition table (OTA).",
@@ -1951,6 +2078,17 @@ def parse_args(argv):
         help="Build with native ESP-IDF instead of PlatformIO (ESP32 esp-idf framework only).",
         action="store_true",
     )
+    parser_run.add_argument(
+        "--ota-platform",
+        choices=[CONF_ESPHOME, CONF_WEB_SERVER],
+        help=(
+            "OTA platform to use for network uploads. Defaults to "
+            f"'{CONF_ESPHOME}' (native API) when configured because it uses "
+            "challenge-response auth so the password is never sent in "
+            f"cleartext on the wire. Falls back to '{CONF_WEB_SERVER}' "
+            "(HTTP Basic auth) when that is the only configured platform."
+        ),
+    )
 
     parser_clean = subparsers.add_parser(
         "clean-mqtt",

esphome/web_server_ota.py

@@ -0,0 +1,202 @@
+"""HTTP-based OTA upload via the ``web_server`` component's ``/update`` endpoint.
+
+This is the alternative to ``espota2`` (the native API OTA path). Useful when
+a device only has ``platform: web_server`` configured under ``ota:``, or when
+the user has lost the native OTA password but still has ``web_server`` basic
+auth credentials.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+from pathlib import Path
+import secrets
+import socket
+from typing import BinaryIO
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+from esphome.core import EsphomeError
+from esphome.helpers import ProgressBar, resolve_ip_address
+
+_LOGGER = logging.getLogger(__name__)
+
+OTA_PATH = "/update"
+FORM_FIELD = "update"
+# (connect_timeout, read_timeout). The device reboots after a successful
+# upload so the read side must allow for a slow flash + response.
+TIMEOUT = (20.0, 120.0)
+
+
+class WebServerOTAError(EsphomeError):
+    pass
+
+
+class _MultipartStreamer:
+    """Stream a single-file multipart/form-data body during transmission.
+
+    ``requests.post(files=...)`` materializes the entire body in memory before
+    sending, so a progress callback wired into the file-like fires during
+    encoding instead of during the network send. Pass this via ``data=``
+    (with ``__len__`` so urllib3 sets ``Content-Length`` instead of using
+    chunked transfer encoding); urllib3 then calls ``read(blocksize)``
+    repeatedly during the POST and the progress bar tracks bytes leaving the
+    host.
+    """
+
+    def __init__(self, file: BinaryIO, file_size: int, filename: str) -> None:
+        self.boundary = f"esphomeOTA{secrets.token_hex(16)}"
+        prefix = (
+            f"--{self.boundary}\r\n"
+            f'Content-Disposition: form-data; name="{FORM_FIELD}"; '
+            f'filename="{filename}"\r\n'
+            f"Content-Type: application/octet-stream\r\n\r\n"
+        ).encode()
+        suffix = f"\r\n--{self.boundary}--\r\n".encode()
+        # Walked in order; ``read()`` advances to the next source on EOF.
+        self._sources: list[BinaryIO] = [io.BytesIO(prefix), file, io.BytesIO(suffix)]
+        self._idx = 0
+        self._total = len(prefix) + file_size + len(suffix)
+        self._sent = 0
+        self.progress = ProgressBar()
+
+    def __len__(self) -> int:
+        return self._total
+
+    @property
+    def content_type(self) -> str:
+        return f"multipart/form-data; boundary={self.boundary}"
+
+    def read(self, size: int = -1) -> bytes:
+        remaining = self._total if size is None or size < 0 else size
+        out = bytearray()
+        while remaining > 0 and self._idx < len(self._sources):
+            chunk = self._sources[self._idx].read(remaining)
+            if not chunk:
+                self._idx += 1
+                continue
+            out += chunk
+            remaining -= len(chunk)
+        if out:
+            self._sent += len(out)
+            self.progress.update(self._sent / self._total)
+        return bytes(out)
+
+
+def _try_upload(
+    host: str,
+    port: int,
+    username: str | None,
+    password: str | None,
+    filename: Path,
+) -> tuple[int, str | None]:
+    from esphome.core import CORE
+
+    try:
+        addr_infos = resolve_ip_address(host, port, address_cache=CORE.address_cache)
+    except EsphomeError as err:
+        _LOGGER.error(
+            "Error resolving IP address of %s. Is it connected to WiFi?", host
+        )
+        if not CORE.dashboard:
+            _LOGGER.error("(If you know the IP, try --device <IP>)")
+        raise WebServerOTAError(err) from err
+
+    if not addr_infos:
+        _LOGGER.error("Could not resolve %s", host)
+        return 1, None
+
+    file_size = filename.stat().st_size
+    _LOGGER.info("Uploading %s (%s bytes) via web_server OTA", filename, file_size)
+    auth = HTTPBasicAuth(username, password) if username and password else None
+
+    # Iterate resolved IPs (IPv4 + IPv6 candidates) just like espota2 does.
+    for af, _socktype, _, _, sa in addr_infos:
+        ip = sa[0]
+        # IPv6 literals must be wrapped in brackets in URLs; link-local
+        # addresses need a percent-encoded zone index per RFC 6874.
+        if af == socket.AF_INET6:
+            scope = sa[3] if len(sa) >= 4 else 0
+            host_part = f"[{ip}%25{scope}]" if scope else f"[{ip}]"
+        else:
+            host_part = ip
+        url = f"http://{host_part}:{port}{OTA_PATH}"
+        _LOGGER.info("Connecting to %s port %s...", ip, port)
+
+        try:
+            with open(filename, "rb") as fh:
+                streamer = _MultipartStreamer(fh, file_size, filename.name)
+                try:
+                    response = requests.post(
+                        url,
+                        data=streamer,
+                        auth=auth,
+                        timeout=TIMEOUT,
+                        headers={
+                            "Content-Type": streamer.content_type,
+                            "Connection": "close",
+                        },
+                    )
+                finally:
+                    streamer.progress.done()
+        except requests.RequestException as err:
+            _LOGGER.error("OTA upload to %s port %s failed: %s", ip, port, err)
+            continue
+
+        if response.status_code == 401:
+            raise WebServerOTAError(
+                "Authentication failed (HTTP 401). Check the 'web_server' "
+                "'auth' username and password."
+            )
+        if response.status_code != 200:
+            detail = response.text.strip() or response.reason or "no response body"
+            raise WebServerOTAError(
+                f"Unexpected HTTP {response.status_code} response from device: {detail}"
+            )
+
+        # The endpoint returns HTTP 200 for both success and failure; the
+        # body is what tells us which (see ota_web_server.cpp handleRequest).
+        body = response.text.strip()
+        if "Successful" in body:
+            _LOGGER.info("Device response: %s", body)
+            _LOGGER.info("OTA successful")
+            return 0, ip
+
+        raise WebServerOTAError(
+            f"Device reported OTA failure: {body or 'no response body'}"
+        )
+
+    return 1, None
+
+
+def run_ota(
+    remote_hosts: str | list[str],
+    remote_port: int,
+    username: str | None,
+    password: str | None,
+    filename: Path,
+) -> tuple[int, str | None]:
+    """Upload ``filename`` to the first reachable host via ``web_server`` OTA.
+
+    Mirrors :func:`esphome.espota2.run_ota` so callers can swap between the
+    two paths with the same return contract: ``(0, host)`` on success or
+    ``(1, None)`` on failure.
+    """
+    hosts = [remote_hosts] if isinstance(remote_hosts, str) else list(remote_hosts)
+    for host in hosts:
+        try:
+            exit_code, used_host = _try_upload(
+                host, remote_port, username, password, filename
+            )
+        except WebServerOTAError as err:
+            _LOGGER.error("%s", err)
+            continue
+        if exit_code == 0:
+            return 0, used_host
+    # Reached only when every attempt failed; per-attempt errors were
+    # already logged. This summary line gives the user an unambiguous
+    # "stop reading, nothing worked" marker.
+    _LOGGER.error("OTA upload failed.")
+    return 1, None

tests/unit_tests/test_main.py

@@ -43,6 +43,7 @@ from esphome.__main__ import (
     has_non_ip_address,
     has_ota,
     has_resolvable_address,
+    has_web_server_ota,
     mqtt_get_ip,
     run_esphome,
     run_miniterm,
@@ -58,6 +59,7 @@ from esphome.components import esp32
 from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
 from esphome.const import (
     CONF_API,
+    CONF_AUTH,
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DISABLED,
@@ -76,6 +78,8 @@ from esphome.const import (
     CONF_SUBSTITUTIONS,
     CONF_TOPIC,
     CONF_USE_ADDRESS,
+    CONF_USERNAME,
+    CONF_WEB_SERVER,
     CONF_WIFI,
     KEY_CORE,
     KEY_TARGET_PLATFORM,
@@ -213,6 +217,13 @@ def mock_run_ota() -> Generator[Mock]:
         yield mock
 
 
+@pytest.fixture
+def mock_run_web_server_ota() -> Generator[Mock]:
+    """Mock web_server_ota.run_ota for testing."""
+    with patch("esphome.web_server_ota.run_ota") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_is_ip_address() -> Generator[Mock]:
     """Mock is_ip_address for testing."""
@@ -1114,6 +1125,7 @@ class MockArgs:
     reset: bool = False
     list_only: bool = False
     output: str | None = None
+    ota_platform: str | None = None
     partition_table: bool = False
 
 
@@ -1878,6 +1890,277 @@ def test_upload_program_ota_no_config(
         upload_program(config, args, devices)
 
 
+def test_has_web_server_ota_detects_platform() -> None:
+    """has_web_server_ota returns True when web_server OTA platform is configured."""
+    setup_core(
+        config={
+            CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        }
+    )
+    assert has_web_server_ota() is True
+    assert has_ota() is True
+
+
+def test_has_web_server_ota_returns_false_without_config() -> None:
+    """has_web_server_ota returns False when only native OTA is configured."""
+    setup_core(
+        config={
+            CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
+        }
+    )
+    assert has_web_server_ota() is False
+    assert has_ota() is True
+
+
+def test_upload_program_web_server_only_auto_dispatches(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """When only web_server OTA is configured, upload_program picks it automatically."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_web_server_ota.assert_called_once_with(
+        ["192.168.1.100"], 80, "admin", "pw", expected_firmware
+    )
+    mock_run_ota.assert_not_called()
+
+
+def test_upload_program_web_server_no_auth(
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """web_server OTA works without an auth block (passes None for credentials)."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        CONF_WEB_SERVER: {CONF_PORT: 8080},
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_web_server_ota.assert_called_once_with(
+        ["192.168.1.100"], 8080, None, None, expected_firmware
+    )
+
+
+def test_upload_program_both_platforms_default_prefers_native(
+    mock_run_ota: Mock,
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """When both OTA platforms are configured, default selection is native API."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {CONF_PORT: 80},
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    mock_run_ota.assert_called_once()
+    mock_run_web_server_ota.assert_not_called()
+
+
+def test_upload_program_ota_platform_override_to_web_server(
+    mock_run_ota: Mock,
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """--ota-platform web_server forces web_server OTA even when native is configured."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {CONF_PORT: 80},
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    mock_run_ota.assert_not_called()
+    mock_run_web_server_ota.assert_called_once()
+
+
+def test_upload_program_ota_platform_unavailable(
+    mock_get_port_type: Mock,
+) -> None:
+    """--ota-platform must reference a platform that is actually configured."""
+    setup_core(platform=PLATFORM_ESP32)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            }
+        ],
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="--ota-platform web_server"):
+        upload_program(config, args, devices)
+
+
+def test_upload_program_web_server_missing_component(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """web_server OTA without a web_server component fails with a clear error."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+        # No CONF_WEB_SERVER
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="web_server.*not configured"):
+        upload_program(config, args, devices)
+
+
+def test_upload_program_unrelated_ota_platform_ignored(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """OTA list entries that are neither esphome nor web_server are ignored.
+
+    Covers the false branch in _choose_ota_platform's filter loop and the
+    no-match branch in _upload_via_native_api's lookup loop.
+    """
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {CONF_PLATFORM: "http_request"},  # unrelated platform; ignored
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_PASSWORD: "secret",
+            },
+        ],
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    exit_code, host = upload_program(config, args, devices)
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    mock_run_ota.assert_called_once()
+
+
+def test_upload_program_duplicate_platform_dedup_in_error(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Duplicate same-platform OTA entries don't repeat in --ota-platform errors."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [
+            {CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232},
+            {CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3233},
+        ],
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError) as excinfo:
+        upload_program(config, args, devices)
+
+    # Error mentions esphome once in the platform list, not "esphome, esphome".
+    msg = str(excinfo.value)
+    assert "esphome, esphome" not in msg
+    assert msg.endswith(": esphome")
+
+
+def test_upload_program_only_unrelated_ota_platforms(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Only unrelated OTA platforms configured -> raises like missing OTA."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+
+    config = {
+        CONF_OTA: [{CONF_PLATFORM: "http_request"}],
+    }
+    args = MockArgs()
+    devices = ["192.168.1.100"]
+
+    with pytest.raises(EsphomeError, match="Cannot upload Over the Air"):
+        upload_program(config, args, devices)
+
+
 def test_upload_program_ota_with_mqtt_resolution(
     mock_mqtt_get_ip: Mock,
     mock_is_ip_address: Mock,