Merge branch 'dev' into 20260218-zigbee-proxy

tests/unit_tests/components/test_espnow.py

@@ -0,0 +1,48 @@
+"""Tests for the espnow component's final validation."""
+
+import pytest
+
+from esphome.components.esp32.const import (
+    VARIANT_ESP32C3,
+    VARIANT_ESP32H2,
+    VARIANT_ESP32P4,
+)
+from esphome.components.espnow import _validate_variant
+import esphome.config_validation as cv
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+
+def _run(
+    monkeypatch, variant: str, full_config: dict, config: ConfigType
+) -> ConfigType:
+    monkeypatch.setattr("esphome.components.espnow.get_esp32_variant", lambda: variant)
+    token = fv.full_config.set(full_config)
+    try:
+        return _validate_variant(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_variant_with_native_wifi_passes(monkeypatch) -> None:
+    """A variant with a native Wi-Fi PHY needs no shim; config passes through."""
+    config = {"id": "espnow"}
+    assert _run(monkeypatch, VARIANT_ESP32C3, {}, config) is config
+
+
+def test_radioless_non_p4_variant_rejected(monkeypatch) -> None:
+    """Radio-less variants without any ESP-NOW path are rejected outright."""
+    with pytest.raises(cv.Invalid, match="not supported"):
+        _run(monkeypatch, VARIANT_ESP32H2, {}, {})
+
+
+def test_p4_without_esp32_hosted_rejected(monkeypatch) -> None:
+    """The P4 needs the esp32_hosted shim to supply the esp_now_* symbols."""
+    with pytest.raises(cv.Invalid, match="esp32_hosted"):
+        _run(monkeypatch, VARIANT_ESP32P4, {}, {})
+
+
+def test_p4_with_esp32_hosted_passes(monkeypatch) -> None:
+    """The P4 with esp32_hosted present validates; config passes through."""
+    config = {"id": "espnow"}
+    assert _run(monkeypatch, VARIANT_ESP32P4, {"esp32_hosted": {}}, config) is config

tests/unit_tests/components/udp/test_init.py

@@ -0,0 +1,37 @@
+"""Tests for the udp component configuration schema."""
+
+from __future__ import annotations
+
+import pytest
+
+from esphome.components import udp
+from esphome.components.packet_transport import (
+    CONF_BINARY_SENSORS,
+    CONF_ENCRYPTION,
+    CONF_PING_PONG_ENABLE,
+    CONF_PROVIDERS,
+    CONF_ROLLING_CODE_ENABLE,
+    CONF_SENSORS,
+)
+import esphome.config_validation as cv
+
+
+@pytest.mark.parametrize(
+    "option",
+    [
+        CONF_PROVIDERS,
+        CONF_ENCRYPTION,
+        CONF_PING_PONG_ENABLE,
+        CONF_ROLLING_CODE_ENABLE,
+        CONF_SENSORS,
+        CONF_BINARY_SENSORS,
+    ],
+)
+def test_relocated_option_rejected(option: str) -> None:
+    """Options that moved to packet_transport raise a pointing error."""
+    with pytest.raises(cv.Invalid) as exc_info:
+        udp.CONFIG_SCHEMA({option: True})
+    assert (
+        f"The '{option}' option should now be configured in the 'packet_transport' component"
+        in str(exc_info.value)
+    )

tests/unit_tests/conftest.py

@@ -9,7 +9,7 @@ not be part of a unit test suite.
 
 """
 
-from collections.abc import Generator
+from collections.abc import Callable, Generator
 import os
 from pathlib import Path
 import sys
@@ -137,3 +137,40 @@ def mock_get_component() -> Generator[Mock, None, None]:
     """Mock get_component for config module."""
     with patch("esphome.config.get_component") as mock:
         yield mock
+
+
+@pytest.fixture
+def held_lock() -> Callable[..., Callable[..., None]]:
+    """Factory for a ``FileLock.acquire`` fake held by another downloader.
+
+    Each poll writes the next chunk to ``part`` (or runs it, for a callable)
+    and raises ``Timeout``; when the chunks run out the part is removed,
+    ``land()`` runs, and the acquire succeeds (also for any later job, so
+    ``land`` must be idempotent).
+    """
+    from filelock import Timeout
+
+    def make(
+        part: Path,
+        chunks: list[bytes | Callable[[], None]],
+        land: Callable[[], None],
+    ) -> Callable[..., None]:
+        polls = iter(chunks)
+
+        def acquire(*args, **kwargs) -> None:
+            try:
+                chunk = next(polls)
+            except StopIteration:
+                part.unlink(missing_ok=True)
+                land()
+                return
+            if callable(chunk):
+                chunk()
+            else:
+                part.parent.mkdir(parents=True, exist_ok=True)
+                part.write_bytes(chunk)
+            raise Timeout("held")
+
+        return acquire
+
+    return make

tests/unit_tests/test_config_validation.py

@@ -1394,6 +1394,35 @@ def test_entity_metadata_visibility_hints() -> None:
     assert web["web_server"].visibility is advanced
 
 
+def test_with_visibility_remarks_keys() -> None:
+    """``with_visibility`` re-marks the named keys, preserving each field's
+    default and validator, without touching the other keys or the input schema.
+    """
+    base = cv.Schema(
+        {
+            cv.Optional("a", default=7): cv.int_,
+            cv.Optional("b", visibility=cv.Visibility.ADVANCED): cv.string,
+        }
+    )
+    promoted = cv.with_visibility(base, cv.Visibility.UI, "a")
+
+    pm = {str(k): k for k in promoted.schema}
+    assert pm["a"].visibility is cv.Visibility.UI  # re-marked
+    assert pm["a"].default() == 7  # default preserved
+    assert pm["b"].visibility is cv.Visibility.ADVANCED  # sibling untouched
+    assert promoted({}) == {"a": 7}  # validator/default still applied
+
+    # The input schema is left untouched (no shared-marker mutation).
+    assert {str(k): k for k in base.schema}["a"].visibility is None
+
+
+def test_with_visibility_unknown_key_raises() -> None:
+    """A key not present in the schema is a typo — fail at build time."""
+    base = cv.Schema({cv.Optional("a"): cv.int_})
+    with pytest.raises(ValueError, match="not in schema"):
+        cv.with_visibility(base, cv.Visibility.UI, "nope")
+
+
 def _wrap_str(value: str) -> ESPHomeDataBase:
     """Wrap a raw string as an ESPHomeDataBase, mimicking a YAML-loaded value."""
     return make_data_base(value)

tests/unit_tests/test_espota2.py

@@ -416,6 +416,9 @@ def test_perform_ota_no_auth(
         "Update took 14.00 seconds (prepare 2.00, upload 5.00, commit 7.00)"
         in caplog.text
     )
+    # The data phase timeout must outlast the device's 105 s data timeout
+    mock_socket.settimeout.assert_any_call(espota2.DATA_PHASE_TIMEOUT)
+    assert espota2.DATA_PHASE_TIMEOUT > 105.0
 
 
 @pytest.mark.usefixtures("mock_time")

tests/unit_tests/test_espota2_noise.py

@@ -10,12 +10,15 @@ when the installed aioesphomeapi predates the noise module.
 from __future__ import annotations
 
 import base64
+from collections.abc import Callable
 import hashlib
 import io
+import logging
 from pathlib import Path
 import socket
 import sys
 import threading
+from typing import Any
 from unittest.mock import Mock, patch
 
 import pytest
@@ -65,8 +68,12 @@ class FakeEncryptedDevice(threading.Thread):
         offer_noise: bool = True,
         require_noise: bool = True,
         prologue_features_override: int | None = None,
+        connections: int = 1,
+        drop_handshakes: int = 0,
     ) -> None:
         super().__init__(daemon=True)
+        self.connections = connections
+        self.drop_handshakes = drop_handshakes  # hang up mid-handshake this many times
         self.psk = psk
         self.version = version
         self.offer_noise = offer_noise
@@ -81,10 +88,11 @@ class FakeEncryptedDevice(threading.Thread):
 
     def run(self) -> None:
         try:
-            sock, _ = self.listener.accept()
-            sock.settimeout(10)
-            with sock:
-                self._serve(sock)
+            for _ in range(self.connections):
+                sock, _ = self.listener.accept()
+                sock.settimeout(10)
+                with sock:
+                    self._serve(sock)
         except Exception as err:  # noqa: BLE001 - surfaced via join_and_check
             self.error = err
         finally:
@@ -109,8 +117,23 @@ class FakeEncryptedDevice(threading.Thread):
             return
         server_flags = espota2.SERVER_FEATURE_SUPPORTS_NOISE if self.offer_noise else 0
         sock.sendall(bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags]))
-        if not (self.offer_noise and noise_negotiated):
-            return  # the client fails closed; nothing further arrives
+        if not (noise_negotiated and self.offer_noise):
+            # A device that does not require encryption continues in
+            # plaintext whatever the client asked for, like older firmware
+            try:
+                self._transfer(
+                    lambda byte: sock.sendall(bytes([byte])),
+                    lambda length: _recv_exact(sock, length),
+                    lambda remaining: _recv_exact(
+                        sock, min(remaining, espota2.UPLOAD_BLOCK_SIZE)
+                    ),
+                )
+            except ConnectionError:
+                # A keyed client without fallback fails closed and hangs up
+                if noise_negotiated and not self.offer_noise:
+                    return
+                raise
+            return
 
         from cryptography.exceptions import InvalidTag
         from noise.connection import NoiseConnection
@@ -134,6 +157,9 @@ class FakeEncryptedDevice(threading.Thread):
 
         msg1 = _recv_frame(sock)
         assert msg1[0] == 0x00
+        if self.drop_handshakes > 0:
+            self.drop_handshakes -= 1
+            return  # a transport fault: the socket closes with no reply
         try:
             proto.read_message(msg1[1:])
         except InvalidTag:
@@ -149,6 +175,20 @@ class FakeEncryptedDevice(threading.Thread):
             assert len(plaintext) == length, "control units must be one per frame"
             return plaintext
 
+        def recv_data(_remaining: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            return plaintext
+
+        self._transfer(send_byte, recv_unit, recv_data)
+
+    def _transfer(
+        self,
+        send_byte: Callable[[int], None],
+        recv_unit: Callable[[int], bytes],
+        recv_data: Callable[[int], bytes],
+    ) -> None:
+        """The post-handshake exchange, identical over both transports."""
         send_byte(espota2.RESPONSE_AUTH_OK)
         recv_unit(1)  # ota type
         size = int.from_bytes(recv_unit(4), "big")
@@ -159,9 +199,7 @@ class FakeEncryptedDevice(threading.Thread):
         received = b""
         acked = 0
         while len(received) < size:
-            plaintext = proto.decrypt(_recv_frame(sock))
-            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
-            received += plaintext
+            received += recv_data(size - len(received))
             if self.version >= espota2.OTA_VERSION_2_0:
                 while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
                     len(received) == size and acked < size
@@ -176,7 +214,10 @@ class FakeEncryptedDevice(threading.Thread):
 
 
 def _upload(
-    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+    device: FakeEncryptedDevice,
+    firmware: bytes,
+    noise_psk: str | None,
+    plaintext_fallback: bool = False,
 ) -> None:
     device.start()
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -184,12 +225,35 @@ def _upload(
     sock.connect(("127.0.0.1", device.port))
     try:
         espota2.perform_ota(
-            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+            sock,
+            None,
+            io.BytesIO(firmware),
+            Path("firmware.bin"),
+            noise_psk=noise_psk,
+            plaintext_fallback=plaintext_fallback,
         )
     finally:
         sock.close()
 
 
+def _run_ota(
+    device: FakeEncryptedDevice, firmware: bytes, tmp_path: Path, noise_psk: str
+) -> int:
+    """Drive the retry loop, which is where the plaintext fallback reconnects."""
+    path = tmp_path / "firmware.bin"
+    path.write_bytes(firmware)
+    device.start()
+    rc, _ = espota2.run_ota(
+        "127.0.0.1",
+        device.port,
+        None,
+        path,
+        noise_psk=noise_psk,
+        plaintext_fallback=True,
+    )
+    return rc
+
+
 def test_encrypted_upload_success() -> None:
     """A full encrypted v2 upload spanning several 8192-byte blocks."""
     pytest.importorskip("aioesphomeapi.noise")
@@ -240,6 +304,56 @@ def test_client_fails_closed_when_device_lacks_encryption() -> None:
     device.join_and_check()
 
 
+# Remove before 2027.3.0
+def test_fallback_when_device_does_not_offer(caplog: pytest.LogCaptureFixture) -> None:
+    """The api key is tried opportunistically; an older device that cannot
+    encrypt still gets its update, with a warning."""
+    firmware = b"firmware"
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
+    with patch("time.sleep"), caplog.at_level(logging.WARNING):
+        _upload(device, firmware, PSK, plaintext_fallback=True)
+    device.join_and_check()
+    assert device.received == firmware
+    assert any("fallback is removed in 2027.3.0" in r.message for r in caplog.records)
+
+
+# Remove before 2027.3.0
+@pytest.mark.parametrize(
+    ("device_kwargs", "expected_rc", "fell_back"),
+    [
+        # A wrong key against an offering device reconnects in plaintext
+        ({"psk": OTHER_PSK, "require_noise": False, "connections": 2}, 0, True),
+        # The plaintext retry is refused by a device that requires encryption
+        ({"psk": OTHER_PSK, "require_noise": True, "connections": 2}, 1, True),
+        # A dropped connection inside the handshake is retried encrypted
+        ({"require_noise": False, "connections": 2, "drop_handshakes": 1}, 0, False),
+        # A second transport fault inside the handshake falls back
+        ({"require_noise": False, "connections": 3, "drop_handshakes": 2}, 0, True),
+    ],
+    ids=["wrong_key", "wrong_key_required", "one_fault", "two_faults"],
+)
+def test_fallback_through_the_retry_loop(
+    caplog: pytest.LogCaptureFixture,
+    tmp_path: Path,
+    device_kwargs: dict[str, Any],
+    expected_rc: int,
+    fell_back: bool,
+) -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"firmware"
+    device = FakeEncryptedDevice(**device_kwargs)
+    with patch("time.sleep"), caplog.at_level(logging.WARNING):
+        rc = _run_ota(device, firmware, tmp_path, PSK)
+    device.join_and_check()
+    assert rc == expected_rc
+    assert (device.received == firmware) is (expected_rc == 0)
+    assert (
+        any("Retrying in plaintext" in r.message for r in caplog.records) is fell_back
+    )
+    if expected_rc == 1:
+        assert any("requires an encrypted OTA" in r.message for r in caplog.records)
+
+
 def test_plaintext_client_gets_encryption_required_error() -> None:
     """A client without a key gets the device's 0x94 error message."""
     device = FakeEncryptedDevice()

tests/unit_tests/test_framework_helpers.py

@@ -2353,3 +2353,20 @@ def test_discard_partial_download_logs_undeletable(
     ):
         framework_helpers.discard_partial_download(dest)
     assert "Could not remove" in caplog.text
+
+
+def test_downloaded_bytes_reports_what_is_on_disk(tmp_path: Path) -> None:
+    """Part file first, then the landed file, both capped at size; else 0."""
+    dest = tmp_path / "archive"
+    assert framework_helpers.downloaded_bytes(dest, 4) == 0
+    part = tmp_path / "archive.part"
+    part.write_bytes(b"ab")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 2
+    part.write_bytes(b"abcdef")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 4
+    part.unlink()
+    dest.write_bytes(b"abc")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 3
+    assert framework_helpers.downloaded_bytes(dest) == 3
+    dest.write_bytes(b"abcdef")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 4

tests/unit_tests/test_main.py

@@ -2108,7 +2108,13 @@ def test_upload_program_ota_success(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        "secret",
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -2140,10 +2146,77 @@ def test_upload_program_ota_encryption_key(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, key
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        key,
+        plaintext_fallback=False,
     )
 
 
+def test_upload_program_ota_api_key_opportunistic(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Without an ota encryption block the api key is tried with a plaintext
+    fallback (removed in 2027.3.0)."""
+    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")
+
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    config = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: key}},
+        CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
+    }
+    exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        key,
+        plaintext_fallback=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "api_conf",
+    [{}, {CONF_ENCRYPTION: {}}],
+    ids=["no_encryption", "runtime_key"],
+)
+def test_upload_program_ota_no_usable_api_key_stays_plaintext(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    api_conf: dict[str, Any],
+) -> None:
+    """A missing or runtime provisioned api key gives the uploader nothing
+    to try."""
+    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_API: api_conf,
+        CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
+    }
+    exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert mock_run_ota.call_args.args[5] is None
+    assert mock_run_ota.call_args.kwargs == {"plaintext_fallback": False}
+
+
 def test_upload_program_ota_encryption_without_key_fails_closed(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2194,7 +2267,13 @@ def test_upload_program_ota_with_file_arg(
     assert exit_code == 0
     assert host == "192.168.1.100"
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        Path("custom.bin"),
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -2250,6 +2329,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2312,6 +2392,7 @@ def test_upload_program_ota_partition_table_mqttip(
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2500,6 +2581,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -2988,7 +3070,13 @@ def test_upload_program_ota_with_mqtt_resolution(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 
@@ -3038,7 +3126,13 @@ def test_upload_program_ota_with_mqtt_empty_broker(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.50"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5211,6 +5305,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5261,6 +5356,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
         None,
+        plaintext_fallback=False,
     )
 
 
@@ -5438,7 +5534,13 @@ def test_upload_program_ota_mqtt_timeout_fallback(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
+        ["192.168.1.100"],
+        3232,
+        None,
+        expected_firmware,
+        OTA_TYPE_UPDATE_APP,
+        None,
+        plaintext_fallback=False,
     )
 
 

tests/unit_tests/test_platformio_library.py

@@ -7,6 +7,7 @@ exercised in their own test modules)."""
 import json
 import logging
 from pathlib import Path
+from unittest.mock import Mock
 
 import pytest
 
@@ -228,6 +229,24 @@ def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
         _resolve_registry_version("owner", "pkg", set())
 
 
+def test_make_registry_client_skips_private_package_probe(monkeypatch):
+    """Our client answers the probe locally without patching PlatformIO's class."""
+    from platformio.account.client import AccountClient
+    from platformio.registry.client import RegistryClient
+
+    pio_probe = RegistryClient.__dict__["allowed_private_packages"]
+    monkeypatch.setattr(
+        AccountClient,
+        "get_account_info",
+        Mock(side_effect=AssertionError("account probe must not run")),
+    )
+
+    client = lib._make_registry_client().get_registry_client_instance()
+
+    assert client.allowed_private_packages() is False
+    assert RegistryClient.__dict__["allowed_private_packages"] is pio_probe
+
+
 def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
     """Stub the registry lookup so tests never touch the network."""
     monkeypatch.setattr(

tests/unit_tests/test_platformio_prefetch.py

@@ -454,23 +454,96 @@ def test_uri_fetch_job_waits_out_a_briefly_held_lock(tmp_path: Path) -> None:
     assert dl_path.read_bytes() == b"data"
 
 
-def test_lock_deadline_leaves_download_to_the_holder(tmp_path: Path) -> None:
-    """A lock held past the deadline means another process is fetching the
-    same file; skipping cleanly beats a misleading failure warning. The
-    tracker is still polled so a parked worker observes cancellation."""
+@pytest.mark.parametrize("staged", [b"", b"ab"])
+def test_lock_deadline_leaves_download_to_the_holder(
+    tmp_path: Path, staged: bytes
+) -> None:
+    """A lock held past the deadline is another process's download; skip
+    cleanly, polling the tracker with what the holder has staged so far."""
     dl_path = tmp_path / "archive"
+    (tmp_path / "archive.prefetch.part").write_bytes(staged)
     ticks: list[int] = []
     with (
         patch("esphome.framework_helpers.download_with_resume") as mock_download,
         patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
-        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
     ):
         pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
     mock_download.assert_not_called()
-    assert ticks == [0]
+    assert ticks == [len(staged)]
     assert not dl_path.exists()
 
 
+@pytest.mark.parametrize(
+    ("job", "part_name", "chunks", "expected"),
+    [
+        (
+            lambda dl_path: pf._registry_fetch_job(
+                MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4
+            ),
+            "archive.part",
+            [b"a", b"abc"],
+            [1, 3, 4],
+        ),
+        (
+            lambda dl_path: pf._uri_fetch_job(
+                MagicMock(), "https://x/a.zip", dl_path, 4
+            ),
+            "archive.prefetch.part",
+            [b"ab"],
+            [2, 4],
+        ),
+    ],
+    ids=["registry", "uri"],
+)
+def test_lock_wait_reports_the_holders_progress(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    held_lock,
+    job,
+    part_name: str,
+    chunks: list[bytes],
+    expected: list[int],
+) -> None:
+    """A waiting job reports the holder's part file (the staging one for a
+    URL job), then the full size once the holder lands the archive."""
+    dl_path = tmp_path / "archive"
+    ticks: list[int] = []
+    acquire = held_lock(
+        tmp_path / part_name, chunks, lambda: dl_path.write_bytes(b"abcd")
+    )
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+        caplog.at_level(logging.INFO),
+    ):
+        job(dl_path)(ticks.append)
+    mock_download.assert_not_called()
+    assert ticks == expected
+    assert caplog.text.count("Waiting for another process downloading archive") == 1
+
+
+def test_uri_lock_wait_prefers_the_landed_archive(tmp_path: Path, held_lock) -> None:
+    """Between the holder's promotion rename and its release the staging
+    part is gone; the landed cache file is credited instead of 0."""
+    dl_path = tmp_path / "archive"
+    ticks: list[int] = []
+    acquire = held_lock(
+        tmp_path / "archive.prefetch.part",
+        [b"ab", lambda: dl_path.write_bytes(b"abcd")],
+        lambda: None,
+    )
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
+    mock_download.assert_not_called()
+    assert ticks == [2, 4, 4]
+
+
 def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
     """A registry job that lost the download race to another process
     must not stamp a nonexistent archive into pio's usage.db."""
@@ -479,7 +552,7 @@ def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
     with (
         patch("esphome.framework_helpers.download_with_resume") as mock_download,
         patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
-        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
     ):
         pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)(
             lambda done: None
@@ -1152,6 +1225,20 @@ def test_main_runs_prefetch(tmp_path: Path) -> None:
     mock_prefetch.assert_called_once_with(tmp_path, "testenv")
 
 
+def test_main_skips_private_package_probe_before_prefetch(tmp_path: Path) -> None:
+    """The registry probe patch is applied before any package manager runs."""
+    order: list[str] = []
+    with (
+        patch.object(pf, "_prefetch", side_effect=lambda *_: order.append("prefetch")),
+        patch(
+            "esphome.platformio.runner.patch_registry_private_packages",
+            side_effect=lambda: order.append("patch"),
+        ),
+    ):
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    assert order == ["patch", "prefetch"]
+
+
 def test_main_bad_argv_is_a_distinct_exit(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -1576,7 +1663,7 @@ def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None:
         {"name": "SPI"},
     ]
     m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"])
-    pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
+    pf._preinstall(m, [("noise-c@0.1.26", _FakeSpec(name="noise-c"))])
     assert installed == ["noise-c", "libsodium"]  # dep deduped, SPI left out
     # The dep wave carries its compatibility so _install searches qualified
     dep_call = m._install.call_args_list[-1]
@@ -1596,7 +1683,7 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
     m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: (
         installed.append(getattr(spec, "name", str(spec)))
     )
-    pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
+    pf._preinstall(m, [("noise-c@0.1.26", _FakeSpec(name="noise-c"))])
     assert installed == ["noise-c"]
 
 

tests/unit_tests/test_platformio_registry.py

@@ -8,6 +8,7 @@ import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+from filelock import Timeout
 import pytest
 
 from esphome.core import EsphomeError
@@ -540,16 +541,13 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
     dest = tmp_path / "a"
     dest.mkdir()
 
-    from contextlib import contextmanager
-
-    @contextmanager
-    def marker_appears_under_lock(path, **kwargs):
+    def marker_appears_under_lock(*args, **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("filelock.FileLock.acquire", side_effect=marker_appears_under_lock),
+        patch("filelock.FileLock.release"),
         patch.object(registry, "download_with_resume") as mock_download,
         patch.object(
             registry, "registry_download", side_effect=_resolve_for({"a": 10})
@@ -559,6 +557,69 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
     mock_download.assert_not_called()
 
 
+def test_prefetch_packages_waits_with_the_holders_progress(
+    tmp_path: Path, held_lock
+) -> None:
+    """A worker parked on another build's lock reports that build's part
+    file, then the full size once the marker appears."""
+    dest = tmp_path / "a"
+    dest.mkdir()
+    ticks: list[int] = []
+    part = tmp_path / "dl" / "a-1.0.part"
+
+    def installed_and_pruned() -> None:
+        # install_package touches the marker, then unlinks the archive
+        (dest / ".esphome_extracted").touch()
+        part.unlink()
+
+    acquire = held_lock(
+        part,
+        [lambda: None, b"abc", installed_and_pruned],
+        (dest / ".esphome_extracted").touch,
+    )
+
+    def fake_batch(header, jobs):
+        for _name, _size, fetch in jobs:
+            fetch(ticks.append)
+        return []
+
+    with (
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+        patch.object(registry, "run_batch_downloads", side_effect=fake_batch),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
+        ),
+    ):
+        registry.prefetch_packages(
+            [("a", "1.0", dest, []), ("b", "2.0", tmp_path / "b", [])],
+            tmp_path / "dl",
+        )
+    assert ticks == [0, 3, 10, 10]
+    mock_download.assert_called_once()
+
+
+def test_prefetch_packages_leaves_a_long_held_lock_to_its_holder(
+    tmp_path: Path,
+) -> None:
+    """Past the deadline the worker skips; install_package waits on the same
+    lock later and verifies whatever the holder produced."""
+    with (
+        patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
+        ),
+    ):
+        registry.prefetch_packages(
+            [("a", "1.0", tmp_path / "a", []), ("b", "2.0", tmp_path / "b", [])],
+            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"

tests/unit_tests/test_platformio_runner.py

@@ -6,7 +6,9 @@ from collections.abc import Callable
 import io
 import sys
 from types import ModuleType
+from unittest.mock import Mock
 
+from platformio.registry.client import RegistryClient
 import pytest
 
 from esphome.platformio import runner
@@ -30,6 +32,7 @@ def _prepare_main(
     monkeypatch.setattr(sys, "stderr", stream)
     monkeypatch.setattr(runner, "patch_structhash", lambda: None)
     monkeypatch.setattr(runner, "patch_file_downloader", lambda: None)
+    monkeypatch.setattr(runner, "patch_registry_private_packages", lambda: None)
 
     platformio = ModuleType("platformio")
     platformio_main = ModuleType("platformio.__main__")
@@ -91,3 +94,40 @@ def test_main_still_filters_a_drained_partial_line(
 
     assert runner.main() == 0
     assert buf.getvalue() == b""
+
+
+def test_main_applies_registry_private_packages_patch(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The probe is patched before PlatformIO runs."""
+    order: list[str] = []
+    _prepare_main(monkeypatch, lambda: order.append("pio") or 0)
+    monkeypatch.setattr(
+        runner, "patch_registry_private_packages", lambda: order.append("patch")
+    )
+
+    assert runner.main() == 0
+    assert order == ["patch", "pio"]
+
+
+# Snapshot PlatformIO's own probe at import, before any test can patch it
+_PIO_PROBE = RegistryClient.__dict__["allowed_private_packages"]
+
+
+def test_patch_registry_private_packages_skips_account_probe(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Answers False without touching the account client."""
+    from platformio.account.client import AccountClient
+
+    monkeypatch.setattr(RegistryClient, "allowed_private_packages", _PIO_PROBE)
+    monkeypatch.setattr(
+        AccountClient,
+        "get_account_info",
+        Mock(side_effect=AssertionError("account probe must not run")),
+    )
+
+    runner.patch_registry_private_packages()
+
+    assert RegistryClient.allowed_private_packages() is False
+    assert RegistryClient().allowed_private_packages() is False

tests/unit_tests/test_wizard.py

@@ -37,7 +37,6 @@ def wizard_answers() -> list[str]:
         "nodemcuv2",  # board
         "SSID",  # ssid
         "psk",  # wifi password
-        "",  # ota password (empty for no password)
     ]
 
 
@@ -101,6 +100,25 @@ def test_config_file_should_include_ota(default_config: dict[str, Any]):
     assert "ota:" in config
 
 
+def test_config_file_should_use_encryption_when_api_key_set(
+    default_config: dict[str, Any],
+):
+    """
+    With an API encryption key and no OTA password the OTA block reuses the key
+    """
+    # Given
+    default_config["api_encryption_key"] = (
+        "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    )
+
+    # When
+    config = wz.wizard_file(**default_config)
+
+    # Then
+    assert "ota:\n  - platform: esphome\n    encryption:" in config
+    assert "password" not in config.split("ota:")[1].split("wifi:")[0]
+
+
 def test_config_file_should_include_ota_when_password_set(
     default_config: dict[str, Any],
 ):
@@ -630,15 +648,15 @@ def test_wizard_write_protects_existing_config(
     assert config_file.read_text() == original_content
 
 
-def test_wizard_accepts_ota_password(
+def test_wizard_uses_the_api_key_for_ota(
     tmp_path: Path, monkeypatch: MonkeyPatch, wizard_answers: list[str]
 ):
     """
-    The wizard should pass ota_password to wizard_write when the user provides one
+    The wizard generates an api key and does not ask for an OTA password;
+    the key secures OTA updates
     """
 
     # Given
-    wizard_answers[5] = "my_ota_password"  # Set OTA password
     config_file = tmp_path / "test.yaml"
     input_mock = MagicMock(side_effect=wizard_answers)
     monkeypatch.setattr("builtins.input", input_mock)
@@ -653,8 +671,9 @@ def test_wizard_accepts_ota_password(
     # Then
     assert retval == 0
     call_kwargs = wizard_write_mock.call_args.kwargs
-    assert "ota_password" in call_kwargs
-    assert call_kwargs["ota_password"] == "my_ota_password"
+    assert "api_encryption_key" in call_kwargs
+    assert "ota_password" not in call_kwargs
+    assert input_mock.call_count == len(wizard_answers)
 
 
 def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch):