Merge branch 'esp8266-arduino-toolchain' into esp8266-native-pch

tests/component_tests/epaper_spi/config/uc8179_e1001_test.yaml

@@ -0,0 +1,15 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32-s3-devkitc-1
+  variant: esp32s3
+
+spi:
+  clk_pin: GPIO7
+  mosi_pin: GPIO9
+
+display:
+  - platform: epaper_spi
+    id: epaper_display
+    model: seeed-reterminal-e1001

tests/component_tests/epaper_spi/test_init.py

@@ -439,6 +439,23 @@ def test_enable_pin_multiple(
     assert all(pin["mode"]["output"] is True for pin in enable_pins)
 
 
+def test_uc8179_e1001_code_generation(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Test that the reTerminal E1001 model generates the UC8179 driver and init sequence."""
+    main_cpp = generate_main(component_config_path("uc8179_e1001_test.yaml"))
+
+    # The model must instantiate the UC8179 driver class with the panel dimensions
+    assert "epaper_spi::EPaperUC8179" in main_cpp
+    assert re.search(r'"SEEED-RETERMINAL-E1001",\s*800,\s*480', main_cpp)
+
+    # The generated init sequence must contain the UC8179 resolution setting
+    # for 800x480: command 0x61, 4 data bytes 0x03 0x20 0x01 0xE0
+    # (rendered as decimal in the generated array)
+    assert "97, 4, 3, 32, 1, 224" in main_cpp
+
+
 def test_enable_pin_code_generation(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],

tests/component_tests/esp32/config/execute_from_psram_s31.yaml

@@ -0,0 +1,13 @@
+esphome:
+  name: test
+
+esp32:
+  variant: esp32s31
+  board: esp32-s31-devkitc
+  framework:
+    type: esp-idf
+    advanced:
+      execute_from_psram: true
+
+psram:
+  mode: octal

tests/component_tests/esp32/test_esp32.py

@@ -203,6 +203,18 @@ def test_esp32_rejects_unsupported_cli_toolchain(
             r"'execute_from_psram' requires PSRAM to be configured @ data\['framework'\]\['advanced'\]\['execute_from_psram'\]",
             id="execute_from_psram_requires_psram_p4_config",
         ),
+        pytest.param(
+            {
+                "variant": "esp32s31",
+                "board": "esp32-s31-devkitc",
+                "framework": {
+                    "type": "esp-idf",
+                    "advanced": {"execute_from_psram": True},
+                },
+            },
+            r"'execute_from_psram' requires PSRAM to be configured @ data\['framework'\]\['advanced'\]\['execute_from_psram'\]",
+            id="execute_from_psram_requires_psram_s31_config",
+        ),
         pytest.param(
             {
                 "variant": "esp32s3",
@@ -422,12 +434,12 @@ def test_execute_from_psram_s3_sdkconfig(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Test that execute_from_psram on ESP32-S3 sets the correct sdkconfig options."""
+    """Test that execute_from_psram on ESP32-S3 sets the correct sdkconfig option."""
     generate_main(component_config_path("execute_from_psram_s3.yaml"))
     sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
-    assert sdkconfig.get("CONFIG_SPIRAM_FETCH_INSTRUCTIONS") is True
-    assert sdkconfig.get("CONFIG_SPIRAM_RODATA") is True
-    assert "CONFIG_SPIRAM_XIP_FROM_PSRAM" not in sdkconfig
+    assert sdkconfig.get("CONFIG_SPIRAM_XIP_FROM_PSRAM") is True
+    assert "CONFIG_SPIRAM_FETCH_INSTRUCTIONS" not in sdkconfig
+    assert "CONFIG_SPIRAM_RODATA" not in sdkconfig
 
 
 def test_execute_from_psram_p4_sdkconfig(
@@ -442,6 +454,18 @@ def test_execute_from_psram_p4_sdkconfig(
     assert "CONFIG_SPIRAM_RODATA" not in sdkconfig
 
 
+def test_execute_from_psram_s31_sdkconfig(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Test that execute_from_psram on ESP32-S31 sets the correct sdkconfig option."""
+    generate_main(component_config_path("execute_from_psram_s31.yaml"))
+    sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
+    assert sdkconfig.get("CONFIG_SPIRAM_XIP_FROM_PSRAM") is True
+    assert "CONFIG_SPIRAM_FETCH_INSTRUCTIONS" not in sdkconfig
+    assert "CONFIG_SPIRAM_RODATA" not in sdkconfig
+
+
 def test_nvs_encryption_sdkconfig(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
@@ -1268,3 +1292,50 @@ def test_parse_pio_platform_version(value: str, expected: str) -> None:
     from esphome.components.esp32 import _parse_pio_platform_version
 
     assert _parse_pio_platform_version(value) == expected
+
+
+def test_esp32_s31_gpio_validation(
+    set_core_config: SetCoreConfigCallable,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """S31: GPIO26-28/30-32 are reserved for the SPI flash interface, GPIO29
+    and GPIO41 do not exist, GPIO33 is a normal pin, and GPIO36 is a
+    strapping pin."""
+    from esphome.components.esp32.const import VARIANT_ESP32S31
+    from esphome.components.esp32.gpio import validate_supports
+    from esphome.const import CONF_INPUT, CONF_MODE, CONF_OPEN_DRAIN, CONF_OUTPUT
+
+    set_core_config(
+        PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32S31}
+    )
+
+    input_mode = {CONF_INPUT: True, CONF_OUTPUT: False, CONF_OPEN_DRAIN: False}
+
+    # Not reserved; a normal GPIO
+    pin = {CONF_NUMBER: 33, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
+    assert validate_gpio_pin(pin)[CONF_NUMBER] == 33
+
+    # Reserved for the SPI flash interface, but can be bypassed with
+    # ignore_pin_validation_error
+    for num in (26, 27, 28, 30, 31, 32):
+        with pytest.raises(cv.Invalid, match=f"GPIO{num} is reserved"):
+            validate_gpio_pin(
+                {CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
+            )
+        pin = {CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: True}
+        assert validate_gpio_pin(pin)[CONF_NUMBER] == num
+
+    for num in (29, 41):
+        with pytest.raises(cv.Invalid, match=f"GPIO{num} does not exist"):
+            validate_gpio_pin(
+                {CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
+            )
+        # Also rejected in validate_supports so ignore_pin_validation_error
+        # cannot bypass it
+        with pytest.raises(cv.Invalid, match=f"GPIO{num} does not exist"):
+            validate_supports({CONF_NUMBER: num, CONF_MODE: input_mode})
+
+    pin = {CONF_NUMBER: 36, CONF_MODE: input_mode}
+    with caplog.at_level("WARNING"):
+        validate_supports(pin)
+    assert "GPIO36 is a strapping PIN" in caplog.text

tests/component_tests/mipi_rgb/test_mipi_rgb_config.py

@@ -10,7 +10,13 @@ from esphome import config_validation as cv
 # via ch422g) can be validated by the mipi_rgb CONFIG_SCHEMA in this test.
 import esphome.components.ch422g  # noqa: F401
 from esphome.components.display import get_display_metadata
-from esphome.components.esp32 import KEY_BOARD, VARIANT_ESP32S3
+from esphome.components.esp32 import (
+    KEY_BOARD,
+    VARIANT_ESP32C3,
+    VARIANT_ESP32P4,
+    VARIANT_ESP32S3,
+    VARIANT_ESP32S31,
+)
 import esphome.components.pca9554  # noqa: F401
 import esphome.components.xl9535  # noqa: F401
 from esphome.const import (
@@ -135,3 +141,63 @@ def test_metadata_records_rotation(
 
     config = CONFIG_SCHEMA({**base, "id": "unrotated"})
     assert get_display_metadata(config["id"]).rotation == 0
+
+
+@pytest.mark.parametrize(
+    ("variant", "board", "model"),
+    [
+        # ESP32-8048S070 is a real Sunton board wired for ESP32-S3 (e.g. its
+        # default de_pin is GPIO41, which doesn't exist on S31), so it is
+        # only meaningful as a config on that variant.
+        (VARIANT_ESP32S3, "esp32-s3-devkitc-1", "ESP32-8048S070"),
+        # P4 and S31 use the pin-agnostic CUSTOM model so this only checks
+        # that the chip itself is accepted, independent of board wiring.
+        (VARIANT_ESP32P4, "esp32-p4-evboard", "CUSTOM"),
+        # No dedicated board is registered for ESP32-S31 yet; an unknown board
+        # name simply skips per-board pin validation.
+        (VARIANT_ESP32S31, "esp32-s31-devkitc", "CUSTOM"),
+    ],
+)
+def test_configuration_succeeds_on_supported_variants(
+    variant: str, board: str, model: str, set_core_config: SetCoreConfigCallable
+) -> None:
+    """mipi_rgb requires a chip with an RGB LCD peripheral: S3, P4 or S31."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: board, KEY_VARIANT: variant},
+    )
+
+    from esphome.components.mipi_rgb.display import CONFIG_SCHEMA
+
+    config = {"model": model, "data_pins": DATA_PINS, "pclk_pin": 21}
+    if model == "CUSTOM":
+        config[CONF_INIT_SEQUENCE] = [[0xA0, 0x01]]
+        config[CONF_DIMENSIONS] = {CONF_WIDTH: 480, CONF_HEIGHT: 480}
+    CONFIG_SCHEMA(config)
+
+
+def test_only_on_variant_rejects_unsupported_variant(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """A variant without the RGB LCD peripheral (e.g. ESP32-C3) is rejected.
+
+    Exercises the exact ``only_on_variant`` call used by ``mipi_rgb.display``
+    directly, since building a full model config with GPIO numbers that are
+    also valid on an unsupported variant like ESP32-C3 is unrelated to what
+    this checks.
+    """
+    from esphome.components.esp32 import only_on_variant
+
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_VARIANT: VARIANT_ESP32C3},
+    )
+
+    validator = only_on_variant(
+        supported=[VARIANT_ESP32S3, VARIANT_ESP32P4, VARIANT_ESP32S31]
+    )
+    with pytest.raises(
+        cv.Invalid,
+        match=r"This feature is only available on ESP32S3, ESP32P4, ESP32S31",
+    ):
+        validator({})

tests/component_tests/noise/test_encryption_key.py

@@ -5,7 +5,11 @@ from __future__ import annotations
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.noise import decode_encryption_key, validate_encryption_key
+from esphome.components.noise import (
+    decode_encryption_key,
+    is_reserved_key,
+    validate_encryption_key,
+)
 
 KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
 
@@ -35,3 +39,8 @@ def test_decode_encryption_key_rejects_short_decode() -> None:
     a zero padded PSK on the device."""
     with pytest.raises(cv.Invalid, match="32 bytes"):
         decode_encryption_key("AAECAw==")
+
+
+def test_is_reserved_key() -> None:
+    assert is_reserved_key("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")
+    assert not is_reserved_key(KEY)

tests/component_tests/ota/test_esphome_ota.py

@@ -8,17 +8,25 @@ from typing import Any
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.esphome.ota import ota_esphome_final_validate
+from esphome.components.esphome.ota import (
+    AUTO_LOAD,
+    FILTER_SOURCE_FILES,
+    _validate_no_password_with_encryption,
+    ota_esphome_final_validate,
+)
 from esphome.const import (
+    CONF_API,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
     CONF_ID,
+    CONF_KEY,
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PLATFORM,
     CONF_PORT,
     CONF_VERSION,
 )
-from esphome.core import ID
+from esphome.core import CORE, ID
 import esphome.final_validate as fv
 
 
@@ -103,3 +111,305 @@ def test_non_esphome_ota_unaffected() -> None:
         assert len(updated[CONF_OTA]) == 3
     finally:
         fv.full_config.reset(token)
+
+
+API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+OTHER_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+ZEROS_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+
+
+def test_encryption_key_inherited_from_api() -> None:
+    """A bare encryption block resolves to the api encryption key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_matching_api_accepted() -> None:
+    """An explicit ota key equal to the api key validates."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_key_differing_from_api_rejected() -> None:
+    """There is one key per device; an ota key differing from the api key raises."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="must match the 'api' encryption key"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_without_api_encryption_accepted() -> None:
+    """An explicit ota key with a plaintext api has nothing to match; it stands."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_without_any_key_rejected() -> None:
+    """A bare encryption block with no api key to inherit raises."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="no 'api' encryption key to inherit"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_all_zeros_key_rejected() -> None:
+    """The all-zeros key is the provisioning sentinel; the device would treat
+    it as no PSK and accept plaintext, so it must fail validation."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_inherited_all_zeros_key_rejected() -> None:
+    """An all-zeros api key must not silently disable ota encryption either."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_key_mismatch_between_merged_configs_rejected() -> None:
+    """Same-port configs with different encryption keys raise."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="encryption is inconsistent"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize("keyed_first", [True, False])
+def test_encryption_bare_and_keyed_blocks_merge(keyed_first: bool) -> None:
+    """A bare encryption block (package/device split) is compatible with a
+    keyed one on the same port; the merge resolves to the keyed result."""
+    keyed = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+    bare = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})
+    full_conf = {
+        CONF_OTA: [keyed, bare] if keyed_first else [bare, keyed],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_runtime_provisioned_api_key_not_inheritable() -> None:
+    """A keyless api encryption block provisions its key at runtime; a bare
+    ota encryption block cannot inherit it and the message says so."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="provisioned at runtime"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_with_runtime_provisioned_api_accepted() -> None:
+    """The documented remedy for a runtime-provisioned api key: set an
+    explicit ota key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_web_server_ota_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """With the web_server component the plaintext /update endpoint is always
+    on; the combination validates with a warning."""
+    full_conf = {
+        "web_server": {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert any("plaintext /update" in record.message for record in caplog.records)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_captive_portal_web_server_ota_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """captive_portal auto-loads the web_server ota platform without the
+    web_server component; encryption stays usable and only warns, so the
+    fallback AP recovery path is not lost."""
+    full_conf = {
+        "captive_portal": {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        assert any("captive_portal" in record.message for record in caplog.records)
+        esphome_conf = next(
+            conf
+            for conf in fv.full_config.get()[CONF_OTA]
+            if conf.get(CONF_PLATFORM) == CONF_ESPHOME
+        )
+        assert esphome_conf[CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_web_server_ota_without_encryption_unaffected() -> None:
+    """web_server ota stays valid alongside an unencrypted esphome entry."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        assert len(fv.full_config.get()[CONF_OTA]) == 2
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_auto_load_pulls_noise_only_for_encryption() -> None:
+    """A plain ota entry must never pull noise-c into the build."""
+    assert AUTO_LOAD({CONF_PORT: 3232}) == ["sha256", "socket"]
+    assert "noise" in AUTO_LOAD({CONF_ENCRYPTION: {}})
+    # Tooling probes must get the maximal set: None from dependency
+    # resolution, {} from the components-graph platform probe
+    assert "noise" in AUTO_LOAD(None)
+    assert "noise" in AUTO_LOAD({})
+
+
+def test_filter_source_files_excludes_noise_without_encryption() -> None:
+    """The noise transport source compiles only for encrypted builds."""
+    old_config = CORE.config
+    try:
+        CORE.config = {CONF_OTA: [_make_ota_config(port=3232)]}
+        assert FILTER_SOURCE_FILES() == ["ota_esphome_noise.cpp"]
+        CORE.config = {
+            CONF_OTA: [
+                _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
+            ]
+        }
+        assert FILTER_SOURCE_FILES() == []
+    finally:
+        CORE.config = old_config
+
+
+def test_password_with_encryption_rejected() -> None:
+    """The password and encryption options are mutually exclusive."""
+    config = {CONF_PASSWORD: "pw", CONF_ENCRYPTION: {CONF_KEY: API_KEY}}
+    with pytest.raises(cv.Invalid, match="cannot be combined"):
+        _validate_no_password_with_encryption(config)
+
+
+def test_password_alone_accepted() -> None:
+    """A password without encryption still validates."""
+    config = {CONF_PASSWORD: "pw"}
+    assert _validate_no_password_with_encryption(config) is config
+
+
+def test_merged_password_and_encryption_rejected() -> None:
+    """A password block and an encryption block merged on one port raise."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_PASSWORD: "pw"}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="cannot be combined"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)

tests/components/d01/common.yaml

@@ -0,0 +1,3 @@
+sensor:
+  - platform: d01
+    name: D01 PM2.5 Concentration

tests/components/d01/test.esp32-idf.yaml

@@ -0,0 +1,7 @@
+substitutions:
+  tx_pin: GPIO4
+  rx_pin: GPIO5
+
+packages:
+  uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
+  d01: !include common.yaml

tests/components/d01/test.esp8266-ard.yaml

@@ -0,0 +1,7 @@
+substitutions:
+  tx_pin: GPIO0
+  rx_pin: GPIO2
+
+packages:
+  uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
+  d01: !include common.yaml

tests/components/d01/test.rp2040-ard.yaml

@@ -0,0 +1,7 @@
+substitutions:
+  tx_pin: GPIO4
+  rx_pin: GPIO5
+
+packages:
+  uart: !include ../../test_build_components/common/uart/rp2040-ard.yaml
+  d01: !include common.yaml

tests/components/ds1603l/common.yaml

@@ -0,0 +1,3 @@
+sensor:
+  - platform: ds1603l
+    name: ds1603l Distance

tests/components/ds1603l/test.esp32-idf.yaml

@@ -0,0 +1,7 @@
+substitutions:
+  tx_pin: GPIO1
+  rx_pin: GPIO3
+
+packages:
+  uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
+  ds1603l: !include common.yaml

tests/components/ds1603l/test.esp8266-ard.yaml

@@ -0,0 +1,7 @@
+substitutions:
+  tx_pin: GPIO0
+  rx_pin: GPIO2
+
+packages:
+  uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
+  ds1603l: !include common.yaml

tests/components/epaper_spi/test.esp32-s3-idf.yaml

@@ -255,3 +255,45 @@ display:
       it.filled_rectangle(0, 0, it.get_width(), it.get_height(), Color::WHITE);
       it.circle(it.get_width() / 2, it.get_height() / 2, 100, Color::BLACK);
       it.circle(it.get_width() / 2, it.get_height() / 2, 60, Color(255, 0, 0));
+
+  # Waveshare 7.5" V2 mono (800x480, UC8179 controller, EPD_7in5_V2)
+  # full_update_every > 1 exercises the fast/partial refresh paths
+  - platform: epaper_spi
+    spi_id: spi_bus
+    model: waveshare-7.5in-v2
+    full_update_every: 4
+    cs_pin:
+      allow_other_uses: true
+      number: GPIO5
+    dc_pin:
+      allow_other_uses: true
+      number: GPIO17
+    reset_pin:
+      allow_other_uses: true
+      number: GPIO16
+    busy_pin:
+      allow_other_uses: true
+      number: GPIO4
+      inverted: true
+    lambda: |-
+      it.filled_rectangle(0, 0, it.get_width(), it.get_height(), Color::WHITE);
+      it.circle(it.get_width() / 2, it.get_height() / 2, 100, Color::BLACK);
+
+  # Seeed reTerminal E1001 - 7.5" mono e-paper (800x480, UC8179)
+  # Pins overridden to avoid conflicts with the E1002 defaults above
+  - platform: epaper_spi
+    spi_id: spi_bus
+    model: seeed-reterminal-e1001
+    cs_pin:
+      allow_other_uses: true
+      number: GPIO5
+    dc_pin:
+      allow_other_uses: true
+      number: GPIO17
+    reset_pin:
+      allow_other_uses: true
+      number: GPIO16
+    busy_pin:
+      allow_other_uses: true
+      number: GPIO4
+      inverted: true

tests/components/improv_serial/common-ethernet.yaml

@@ -0,0 +1,17 @@
+ethernet:
+  type: W5500
+  clk_pin: 19
+  mosi_pin: 21
+  miso_pin: 17
+  cs_pin: 18
+  interrupt_pin: 36
+  reset_pin: 12
+  clock_speed: 10Mhz
+
+logger:
+  hardware_uart: UART0
+
+# Exercises the per-interface webserver URL collection at compile time
+web_server:
+
+improv_serial:

tests/components/improv_serial/test-ethernet.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  improv_serial: !include common-ethernet.yaml

tests/components/lvgl/lvgl-package.yaml

@@ -188,6 +188,8 @@ lvgl:
     dark_mode: true
     obj:
       border_width: 1
+      user_1:
+        bg_color: black
 
   gradients:
     - id: color_bar
@@ -717,6 +719,30 @@ lvgl:
                 id: button_with_text
                 text: Clicked
 
+        # Exercises the LV_STATE_USER_1..USER_4 states: setting them at creation
+        # (both literal and lambda), styling each of them individually, and
+        # setting/clearing them at runtime with lvgl.widget.update.
+        - button:
+            id: user_flags_button
+            text: User flags
+            state:
+              user_1: true
+              user_2: !lambda return true;
+            user_1:
+              bg_color: 0xFF00FF
+            user_2:
+              bg_color: 0x00FFFF
+            user_3:
+              bg_color: 0xFFFF00
+            user_4:
+              bg_color: 0x808080
+            on_click:
+              - lvgl.widget.update:
+                  id: user_flags_button
+                  state:
+                    user_3: true
+                    user_4: !lambda return !lv_obj_has_state(id(user_flags_button), LV_STATE_USER_4);
+
         - button:
             layout: 2x1
             id: button_button

tests/components/ota/encryption.yaml

@@ -0,0 +1,9 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+ota:
+  - platform: esphome
+    port: 3288
+    encryption:
+      key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="

tests/components/ota/encryption_inherit.yaml

@@ -0,0 +1,12 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+    key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+
+ota:
+  - platform: esphome
+    port: 3289
+    encryption:

tests/components/ota/test-encryption.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include encryption.yaml

tests/components/ota/test-encryption.esp8266-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include encryption.yaml

tests/components/ota/test-encryption.rp2040-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include encryption.yaml

tests/components/ota/test-encryption_inherit.esp8266-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  ota: !include encryption_inherit.yaml

tests/components/sen6x/common.yaml

@@ -28,8 +28,21 @@ sensor:
       accuracy_decimals: 1
     nox_index:
       name: NOx Index
+      algorithm_tuning:
+        index_offset: 8
+        learning_time_offset_hours: 6
+        learning_time_gain_hours: 24
+        gating_max_duration_minutes: 900
+        gain_factor: 180
     voc_index:
       name: VOC Index
+      algorithm_tuning:
+        index_offset: 120
+        learning_time_offset_hours: 6
+        learning_time_gain_hours: 24
+        gating_max_duration_minutes: 240
+        std_initial: 75
+        gain_factor: 180
     co2:
       name: Carbon Dioxide
     formaldehyde:

tests/components/sen6x/validate.esp32-idf.yaml

@@ -0,0 +1,18 @@
+# Config-only: partial algorithm_tuning blocks, so the schema defaults fill in the
+# keys that are left out.
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
+
+sensor:
+  - platform: sen6x
+    id: sen6x_partial_tuning
+    type: SEN65
+    i2c_id: i2c_bus
+    voc_index:
+      name: VOC Index
+      algorithm_tuning:
+        index_offset: 60
+    nox_index:
+      name: NOx Index
+      algorithm_tuning:
+        gain_factor: 45

tests/integration/conftest.py

@@ -23,6 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
+from esphome.helpers import get_usable_cpu_count
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -67,6 +68,14 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     env["PLATFORMIO_LIBDEPS_DIR"] = str(cache_dir / "libdeps" / worker)
     # Prevent cache cleaning during integration tests
     env["ESPHOME_SKIP_CLEAN_BUILD"] = "1"
+    # Cap each compile's -j so several xdist workers do not each spawn a
+    # full-width compiler fan-out on the same machine. An explicit env wins.
+    if "ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT" not in os.environ:
+        workers = int(os.environ.get("PYTEST_XDIST_WORKER_COUNT", "1"))
+        # Floor of 2 keeps a lone tail compile from running fully serial
+        env["ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT"] = str(
+            max(2, get_usable_cpu_count() // workers)
+        )
     # Compile with THIS tree's esphome sources, not wherever the venv's editable
     # install points (which may be a different git worktree or checkout).
     repo_root = str(Path(__file__).resolve().parent.parent.parent)
@@ -78,7 +87,8 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
 @pytest.fixture(scope="session")
 def shared_platformio_cache() -> Generator[Path]:
     """Initialize a shared PlatformIO cache for all integration tests."""
-    # Use a dedicated directory for integration tests to avoid conflicts
+    # Use a dedicated directory for integration tests to avoid conflicts.
+    # CI caches parts of this path; keep in sync with ci.yml integration-tests.
     test_cache_dir = Path.home() / ".esphome-integration-tests"
     cache_dir = test_cache_dir / "platformio"
 

tests/integration/fixtures/api_homeassistant_binary_sensor_initial_state.yaml

@@ -0,0 +1,59 @@
+esphome:
+  name: ha-bs-initial
+
+host:
+
+api:
+
+logger:
+  level: DEBUG
+
+binary_sensor:
+  # trigger_on_initial_state: true must fire on_press for the first state from HA
+  - platform: homeassistant
+    name: Initial On
+    entity_id: binary_sensor.initial_on
+    trigger_on_initial_state: true
+    on_press:
+      - logger.log: "initial_on on_press"
+    on_release:
+      - logger.log: "initial_on on_release"
+
+  # Default (false) must not fire on the first state, only on later changes
+  - platform: homeassistant
+    name: Default
+    entity_id: binary_sensor.default
+    on_press:
+      - logger.log: "default on_press"
+    on_release:
+      - logger.log: "default on_release"
+
+  # Real HA startup shape: 'unavailable' arrives before the first real state
+  - platform: homeassistant
+    name: Unavailable First
+    entity_id: binary_sensor.unavailable_first
+    trigger_on_initial_state: true
+    on_press:
+      - logger.log: "unavailable_first on_press"
+    on_release:
+      - logger.log: "unavailable_first on_release"
+
+  # Initial 'off' must fire on_release when trigger_on_initial_state is set
+  - platform: homeassistant
+    name: Initial Off
+    entity_id: binary_sensor.initial_off
+    trigger_on_initial_state: true
+    on_press:
+      - logger.log: "initial_off on_press"
+    on_release:
+      - logger.log: "initial_off on_release"
+
+  # Same 'unavailable' first shape without the flag; must stay quiet on the
+  # first real state and only fire on the later change
+  - platform: homeassistant
+    name: Default Unavailable First
+    entity_id: binary_sensor.default_unavail
+    on_press:
+      - logger.log: "default_unavail on_press"
+    on_release:
+      - logger.log: "default_unavail on_release"

tests/integration/fixtures/external_components/wifi/wifi_component.cpp

@@ -29,6 +29,8 @@ void WiFiComponent::set_sta(const WiFiAP &ap) { ESP_LOGI(TAG, "set_sta ssid=%s",
 
 void WiFiComponent::start_connecting(const WiFiAP &ap) {
   ESP_LOGI(TAG, "start_connecting ssid=%s", ap.get_ssid().c_str());
+  // Connecting succeeds immediately, so the requested network is the connected one
+  this->connected_ssid_ = ap.get_ssid().c_str();
 }
 
 void WiFiComponent::clear_sta() { ESP_LOGI(TAG, "clear_sta"); }

tests/integration/fixtures/external_components/wifi/wifi_component.h

@@ -13,11 +13,15 @@
 #include "esphome/core/component.h"
 #include "esphome/core/string_ref.h"
 
+#include <cstdio>
+#include <span>
 #include <string>
 #include <vector>
 
 namespace esphome::wifi {
 
+static constexpr size_t SSID_BUFFER_SIZE = 33;
+
 class WiFiAP {
  public:
   void set_ssid(const char *ssid) { this->ssid_ = ssid; }
@@ -58,6 +62,12 @@ class WiFiComponent : public Component {
   bool is_disabled() const { return false; }
   // Always connected so network::is_connected() keeps the API server accepting clients
   bool is_connected() const { return true; }
+  // Reports the network start_connecting() was last asked for, so a consumer checking that it
+  // joined the network it requested (rather than an earlier one) sees the connect succeed
+  const char *wifi_ssid_to(std::span<char, SSID_BUFFER_SIZE> buffer) {
+    snprintf(buffer.data(), buffer.size(), "%s", this->connected_ssid_.c_str());
+    return buffer.data();
+  }
   void start_scanning();
   const std::vector<WiFiScanResult> &get_scan_result() const { return this->scan_result_; }
   void set_sta(const WiFiAP &ap);
@@ -70,6 +80,7 @@ class WiFiComponent : public Component {
 
  protected:
   std::vector<WiFiScanResult> scan_result_;
+  std::string connected_ssid_;
 };
 
 extern WiFiComponent *global_wifi_component;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)

tests/integration/fixtures/host_ota_encrypted.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+    encryption:
+      key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+logger:
+  level: DEBUG

tests/integration/integration_test_durations.json

@@ -0,0 +1,142 @@
+{
+  "tests/integration/test_action_concurrent_reentry.py": 45.23,
+  "tests/integration/test_addressable_light_transition.py": 74.47,
+  "tests/integration/test_alarm_control_panel_state_transitions.py": 74.1,
+  "tests/integration/test_api_action_metadata.py": 62.1,
+  "tests/integration/test_api_action_responses.py": 71.08,
+  "tests/integration/test_api_action_timeout.py": 21.64,
+  "tests/integration/test_api_conditional_memory.py": 13.72,
+  "tests/integration/test_api_custom_services.py": 24.16,
+  "tests/integration/test_api_get_time_response_timezone.py": 23.48,
+  "tests/integration/test_api_homeassistant.py": 37.87,
+  "tests/integration/test_api_homeassistant_action_no_subscriber.py": 14.38,
+  "tests/integration/test_api_list_entities_backpressure.py": 26.85,
+  "tests/integration/test_api_message_size_batching.py": 33.36,
+  "tests/integration/test_api_reboot_timeout.py": 13.63,
+  "tests/integration/test_api_string_lambda.py": 25.04,
+  "tests/integration/test_api_vv_logging.py": 16.6,
+  "tests/integration/test_api_zero_psk_provisioning.py": 43.14,
+  "tests/integration/test_areas_and_devices.py": 25.98,
+  "tests/integration/test_automation_wait_actions.py": 21.91,
+  "tests/integration/test_automations.py": 42.43,
+  "tests/integration/test_batch_delay_zero_rapid_transitions.py": 16.65,
+  "tests/integration/test_binary_sensor_autorepeat_filter.py": 28.67,
+  "tests/integration/test_binary_sensor_invalidate_state.py": 23.69,
+  "tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 22.99,
+  "tests/integration/test_build_info.py": 24.96,
+  "tests/integration/test_camera_mock.py": 14.47,
+  "tests/integration/test_climate_control_action.py": 31.07,
+  "tests/integration/test_climate_custom_modes.py": 28.59,
+  "tests/integration/test_continuation_actions.py": 14.96,
+  "tests/integration/test_cover_control_action.py": 26.14,
+  "tests/integration/test_crc8_helper.py": 10.92,
+  "tests/integration/test_device_id_in_state.py": 64.97,
+  "tests/integration/test_duplicate_entities.py": 30.81,
+  "tests/integration/test_entity_icon.py": 32.85,
+  "tests/integration/test_fan_turn_on_action.py": 24.91,
+  "tests/integration/test_fnv1_hash_object_id.py": 12.54,
+  "tests/integration/test_fnv1a_hash.py": 21.8,
+  "tests/integration/test_gpio_expander_cache.py": 5.2,
+  "tests/integration/test_host_logger_thread_safety.py": 21.7,
+  "tests/integration/test_host_mode_basic.py": 13.62,
+  "tests/integration/test_host_mode_batch_delay.py": 14.56,
+  "tests/integration/test_host_mode_climate_basic_state.py": 30.95,
+  "tests/integration/test_host_mode_climate_control.py": 29.06,
+  "tests/integration/test_host_mode_empty_string_options.py": 27.22,
+  "tests/integration/test_host_mode_entity_fields.py": 30.95,
+  "tests/integration/test_host_mode_fan_preset.py": 14.44,
+  "tests/integration/test_host_mode_many_entities.py": 54.13,
+  "tests/integration/test_host_mode_many_entities_multiple_connections.py": 32.17,
+  "tests/integration/test_host_mode_noise_encryption.py": 42.77,
+  "tests/integration/test_host_mode_reconnect.py": 4.06,
+  "tests/integration/test_host_mode_sensor.py": 13.47,
+  "tests/integration/test_host_ota.py": 21.4,
+  "tests/integration/test_host_preferences.py": 25.43,
+  "tests/integration/test_host_preferences_suspend_resume.py": 19.2,
+  "tests/integration/test_improv_serial_uart.py": 31.52,
+  "tests/integration/test_large_message_batching.py": 15.64,
+  "tests/integration/test_legacy_area.py": 22.63,
+  "tests/integration/test_legacy_climate_compat.py": 26.13,
+  "tests/integration/test_legacy_fan_compat.py": 24.05,
+  "tests/integration/test_light_automations.py": 30.86,
+  "tests/integration/test_light_binary_effect_off_phase.py": 23.19,
+  "tests/integration/test_light_calls.py": 32.35,
+  "tests/integration/test_light_constant_brightness.py": 29.89,
+  "tests/integration/test_light_control_action.py": 29.06,
+  "tests/integration/test_light_dim_relative_action.py": 29.61,
+  "tests/integration/test_light_effect_zero_brightness.py": 18.68,
+  "tests/integration/test_light_initial_state.py": 24.49,
+  "tests/integration/test_light_toggle_action.py": 26.46,
+  "tests/integration/test_lock_automations.py": 23.28,
+  "tests/integration/test_logger_buffered_recursion_guard.py": 24.29,
+  "tests/integration/test_loop_disable_enable.py": 45.28,
+  "tests/integration/test_loop_interval_decoupling.py": 28.35,
+  "tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.97,
+  "tests/integration/test_micros_to_millis.py": 20.79,
+  "tests/integration/test_multi_click_trigger.py": 26.2,
+  "tests/integration/test_multi_device_preferences.py": 16.87,
+  "tests/integration/test_noise_encryption_key_protection.py": 77.05,
+  "tests/integration/test_object_id_api_verification.py": 73.51,
+  "tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 62.33,
+  "tests/integration/test_object_id_no_friendly_name.py": 43.47,
+  "tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 32.21,
+  "tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 56.86,
+  "tests/integration/test_online_image_bmp.py": 50.9,
+  "tests/integration/test_oversized_payloads.py": 53.2,
+  "tests/integration/test_preference_key_stability.py": 26.09,
+  "tests/integration/test_runtime_stats.py": 18.34,
+  "tests/integration/test_safe_mode_loop_runs.py": 10.07,
+  "tests/integration/test_scheduler_blocking_warning.py": 40.91,
+  "tests/integration/test_scheduler_bulk_cleanup.py": 23.14,
+  "tests/integration/test_scheduler_defer_cancel.py": 24.54,
+  "tests/integration/test_scheduler_defer_cancel_regular.py": 13.48,
+  "tests/integration/test_scheduler_defer_fifo_simple.py": 26.86,
+  "tests/integration/test_scheduler_defer_stress.py": 27.23,
+  "tests/integration/test_scheduler_heap_stress.py": 24.02,
+  "tests/integration/test_scheduler_internal_id_no_collision.py": 24.57,
+  "tests/integration/test_scheduler_interval_reschedule.py": 13.12,
+  "tests/integration/test_scheduler_interval_zero_coerced.py": 22.91,
+  "tests/integration/test_scheduler_null_name.py": 23.46,
+  "tests/integration/test_scheduler_numeric_id_test.py": 24.54,
+  "tests/integration/test_scheduler_pool.py": 25.0,
+  "tests/integration/test_scheduler_rapid_cancellation.py": 14.68,
+  "tests/integration/test_scheduler_recursive_timeout.py": 25.35,
+  "tests/integration/test_scheduler_removed_item_race.py": 26.19,
+  "tests/integration/test_scheduler_self_keyed.py": 23.43,
+  "tests/integration/test_scheduler_simultaneous_callbacks.py": 22.16,
+  "tests/integration/test_scheduler_string_test.py": 15.22,
+  "tests/integration/test_script_array_params.py": 14.67,
+  "tests/integration/test_script_delay_params.py": 15.65,
+  "tests/integration/test_script_queued.py": 24.93,
+  "tests/integration/test_script_queued_idle_loop.py": 5.04,
+  "tests/integration/test_script_wait_on_boot.py": 13.08,
+  "tests/integration/test_select_stringref_trigger.py": 29.6,
+  "tests/integration/test_sensor_filters_delta.py": 28.01,
+  "tests/integration/test_sensor_filters_ring_buffer.py": 25.04,
+  "tests/integration/test_sensor_filters_sliding_window.py": 71.5,
+  "tests/integration/test_sensor_filters_value_list.py": 16.94,
+  "tests/integration/test_sensor_timeout_filter.py": 29.48,
+  "tests/integration/test_socket_wake_gate_tcp.py": 20.36,
+  "tests/integration/test_status_flags.py": 37.42,
+  "tests/integration/test_strftime_to.py": 22.61,
+  "tests/integration/test_syslog.py": 16.34,
+  "tests/integration/test_template_alarm_control_panel_many_sensors.py": 29.81,
+  "tests/integration/test_template_text_save.py": 25.43,
+  "tests/integration/test_text_command.py": 23.34,
+  "tests/integration/test_text_sensor_raw_state.py": 69.57,
+  "tests/integration/test_uart_mock_ld2410.py": 37.95,
+  "tests/integration/test_uart_mock_ld2412.py": 93.22,
+  "tests/integration/test_uart_mock_ld2420.py": 43.24,
+  "tests/integration/test_uart_mock_ld2450.py": 31.75,
+  "tests/integration/test_uart_mock_modbus.py": 667.4,
+  "tests/integration/test_udp.py": 9.38,
+  "tests/integration/test_use_address_runtime.py": 37.05,
+  "tests/integration/test_valve_control_action.py": 24.47,
+  "tests/integration/test_varint_five_byte_device_id.py": 25.03,
+  "tests/integration/test_wait_until_mid_loop_timing.py": 23.73,
+  "tests/integration/test_wait_until_on_boot.py": 9.16,
+  "tests/integration/test_wait_until_ordering.py": 13.3,
+  "tests/integration/test_wait_until_reentrant_restart.py": 25.23,
+  "tests/integration/test_wake_loop_forces_phase_b.py": 23.34,
+  "tests/integration/test_water_heater_template.py": 17.67
+}

tests/integration/log_utils.py

@@ -28,6 +28,11 @@ class LineWaiter:
             self._future.set_result(line)
             self._future = None
 
+    async def wait_for_each(self, *texts: str, timeout: float = 10.0) -> None:
+        """Await each text in turn; a text may match a line already received."""
+        for text in texts:
+            await self.wait_for(text, timeout=timeout)
+
     async def wait_for(self, *needles: str, timeout: float = 10.0) -> str:
         """Return the first line, past or future, containing every needle."""
         for line in self.lines:

tests/integration/test_api_homeassistant_binary_sensor_initial_state.py

@@ -0,0 +1,98 @@
+"""Test on_press/on_release for homeassistant binary sensors on the first HA state."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .log_utils import LineWaiter
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+ENTITIES = (
+    "binary_sensor.initial_on",
+    "binary_sensor.default",
+    "binary_sensor.unavailable_first",
+    "binary_sensor.initial_off",
+    "binary_sensor.default_unavail",
+)
+
+
+@pytest.mark.asyncio
+async def test_api_homeassistant_binary_sensor_initial_state(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """The first state from HA fires on_press only with trigger_on_initial_state."""
+    loop = asyncio.get_running_loop()
+    waiter = LineWaiter()
+    subscribed: set[str] = set()
+    all_subscribed = loop.create_future()
+
+    def on_state_sub(entity_id: str, _attribute: str | None) -> None:
+        subscribed.add(entity_id)
+        if not all_subscribed.done() and subscribed.issuperset(ENTITIES):
+            all_subscribed.set_result(None)
+
+    async with (
+        run_compiled(yaml_config, line_callback=waiter.callback),
+        api_client_connected() as client,
+    ):
+        client.subscribe_home_assistant_states(on_state_sub)
+        try:
+            await asyncio.wait_for(all_subscribed, timeout=5.0)
+        except TimeoutError:
+            pytest.fail(f"never subscribed: {set(ENTITIES) - subscribed}")
+
+        # First state from HA
+        client.send_home_assistant_state("binary_sensor.initial_on", "", "on")
+        client.send_home_assistant_state("binary_sensor.default", "", "on")
+        client.send_home_assistant_state(
+            "binary_sensor.unavailable_first", "", "unavailable"
+        )
+        client.send_home_assistant_state("binary_sensor.unavailable_first", "", "on")
+        client.send_home_assistant_state(
+            "binary_sensor.default_unavail", "", "unavailable"
+        )
+        client.send_home_assistant_state("binary_sensor.default_unavail", "", "on")
+        client.send_home_assistant_state("binary_sensor.initial_off", "", "off")
+
+        await waiter.wait_for("initial_on on_press", timeout=5.0)
+        await waiter.wait_for("unavailable_first on_press", timeout=5.0)
+        # Pin that the 'unavailable' message actually arrived and was rejected
+        await waiter.wait_for("Can't convert 'unavailable'", timeout=5.0)
+        # initial_off is the last state sent, so this wait also proves the
+        # earlier 'default' initial state was already processed
+        await waiter.wait_for("initial_off on_release", timeout=5.0)
+        # Both 'unavailable' senders must have been seen and rejected
+        assert sum("Can't convert 'unavailable'" in line for line in waiter.lines) == 2
+        # Guard every phase 2 needle against being satisfied by a stale
+        # phase 1 line, and pin that the initial states fired nothing else
+        for absent in (
+            "initial_on on_release",
+            "default on_press",
+            "default on_release",
+            "default_unavail on_press",
+            "default_unavail on_release",
+            "unavailable_first on_release",
+            "initial_off on_press",
+        ):
+            assert not any(absent in line for line in waiter.lines), (
+                f"unexpected trigger before the second state change: {absent}"
+            )
+
+        # A later change fires for all of them
+        client.send_home_assistant_state("binary_sensor.initial_on", "", "off")
+        client.send_home_assistant_state("binary_sensor.default", "", "off")
+        client.send_home_assistant_state("binary_sensor.unavailable_first", "", "off")
+        client.send_home_assistant_state("binary_sensor.initial_off", "", "on")
+        client.send_home_assistant_state("binary_sensor.default_unavail", "", "off")
+        await waiter.wait_for_each(
+            "initial_on on_release",
+            "default on_release",
+            "default_unavail on_release",
+            "unavailable_first on_release",
+            "initial_off on_press",
+            timeout=5.0,
+        )

tests/integration/test_host_ota.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Generator
 from contextlib import contextmanager
+import functools
 import socket
 
 import pytest
@@ -111,6 +112,62 @@ async def test_host_ota_self_update(
             assert proc.pid == pid_before
 
 
+@pytest.mark.asyncio
+async def test_host_ota_encrypted(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """Encrypted self-OTA succeeds; a plaintext upload to the same device fails."""
+    pytest.importorskip("aioesphomeapi.noise")
+    noise_psk = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    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()
+        rebooted = loop.create_future()
+
+        def on_log(line: str) -> None:
+            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
+
+            # A plaintext upload must be refused with the device unharmed
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 1, "plaintext upload to an encrypted device must fail"
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected plaintext OTA"
+
+            # The encrypted upload goes through and the device re-execs
+            rc, _ = await loop.run_in_executor(
+                None,
+                functools.partial(
+                    espota2.run_ota,
+                    LOCALHOST,
+                    ota_port,
+                    None,
+                    binary_path,
+                    noise_psk=noise_psk,
+                ),
+            )
+            assert rc == 0, "encrypted OTA reported failure"
+            await asyncio.wait_for(rebooted, timeout=10.0)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+
+
 @pytest.mark.asyncio
 async def test_host_ota_rejects_garbage(
     yaml_config: str,

tests/script/test_ci_custom.py

@@ -0,0 +1,147 @@
+"""Unit tests for the ESP_LOG-needs-braces lint rule in script/ci-custom.py.
+
+The rule flags an if/else/for/while whose only body is an unbraced ESP_LOG*() call (which becomes an
+empty statement -- and a -Wempty-body warning -- once the log level compiles the macro out). These
+tests pin the comment/string/raw-string masker, the accepted control-statement shapes, and the
+NOLINT escape hatch at both placements a contributor would try.
+"""
+
+import importlib.util
+from pathlib import Path
+import sys
+
+SCRIPT_DIR = (Path(__file__).parent / ".." / ".." / "script").resolve()
+sys.path.insert(0, str(SCRIPT_DIR))
+_spec = importlib.util.spec_from_file_location("ci_custom", SCRIPT_DIR / "ci-custom.py")
+ci_custom = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(ci_custom)
+
+mask = ci_custom._mask_cpp_comments_strings
+
+
+def _lint(content: str) -> list:
+    return ci_custom.lint_esp_log_needs_braces("test.cpp", content)
+
+
+# --- masker ---
+
+
+def test_mask_preserves_length_newlines_and_real_parens() -> None:
+    src = 'foo("bar") + baz();\nqux();\n'
+    masked = mask(src)
+    assert len(masked) == len(src)
+    assert masked.count("\n") == src.count("\n")
+    assert masked.count("(") == src.count("(")  # real parens survive for balancing
+
+
+def test_mask_blanks_line_and_block_comments() -> None:
+    assert "ESP_LOGD" not in mask("a; // if (x) ESP_LOGD(t);\n")
+    assert "ESP_LOGD" not in mask("a; /* if (x) ESP_LOGD(t); */ b;\n")
+
+
+def test_mask_blanks_string_literals() -> None:
+    assert "if" not in mask('x = "if (y) ESP_LOGD";\n')
+
+
+def test_mask_handles_raw_string_without_desync() -> None:
+    # A raw string full of quotes/parens must be consumed as one unit; code after it stays intact.
+    src = 's.print(R"(<a href="x">)");\nreturn;\n'
+    masked = mask(src)
+    assert "href" not in masked
+    assert "return;" in masked  # not swallowed by a desynced string scan
+
+
+# --- rule: flags real violations ---
+
+
+def test_flags_unbraced_if_next_line() -> None:
+    assert _lint("if (x)\n  ESP_LOGD(t);\n")
+
+
+def test_flags_unbraced_same_line() -> None:
+    assert _lint("if (x) ESP_LOGW(t);\n")
+
+
+def test_flags_c_style_for() -> None:
+    assert _lint("for (int i = 0; i < n; i++)\n  ESP_LOGD(t, i);\n")
+
+
+def test_flags_range_for_and_else() -> None:
+    assert _lint("for (auto &x : v)\n  ESP_LOGCONFIG(t);\n")
+    assert _lint("else\n  ESP_LOGE(t);\n")
+
+
+def test_flags_for_header_with_nested_call() -> None:
+    assert _lint("for (auto it = v.begin(); it != v.end(); ++it)\n  ESP_LOGD(t);\n")
+
+
+def test_for_header_does_not_reach_into_a_later_statement() -> None:
+    # The 'for' header is bounded to its own statement, so it cannot swallow the loop body and latch
+    # onto a later ')'. Without that, the '#if' line below is reported as an unbraced body even though
+    # the '#' preprocessor check should skip it.
+    assert not _lint(
+        "for (int i = 0; i < n; i++)\n  arr[i] = 0;\n#if defined(USE_X)\n  ESP_LOGD(t);\n#endif\n"
+    )
+
+
+def test_violation_after_a_for_loop_is_reported_at_its_own_line() -> None:
+    errors = _lint(
+        "for (int i = 0; i < n; i++)\n  sum += a[i];\nif (verbose)\n  ESP_LOGD(t, sum);\n"
+    )
+    lines = [line for line, _col, _msg in errors]
+    assert lines == [3]  # the 'if', not the 'for' on line 1
+
+
+def test_flags_lowercase_esph_log_family() -> None:
+    # core/log.h defines esph_log_*() alongside ESP_LOG*(); both expand to nothing below their level.
+    assert _lint('if (x)\n  esph_log_config(t, "m");\n')
+    assert _lint('if (err != ESP_OK)\n  esph_log_e(t, "m");\n')
+
+
+def test_digit_separator_does_not_disable_the_rest_of_the_file() -> None:
+    # A "'" digit separator must not be read as a char-literal opener, which blanked everything after.
+    assert _lint("uint32_t x = 1'000;\nif (y)\n  ESP_LOGD(t);\n")
+
+
+def test_mask_still_blanks_real_char_literals() -> None:
+    assert "ESP_LOGD" not in mask("char c = '\"'; // if (x) ESP_LOGD(t);\n")
+    assert not _lint("char sep = ';';\nif (x) {\n  ESP_LOGD(t);\n}\n")
+
+
+def test_flags_multiline_log_body() -> None:
+    assert _lint('if (x)\n  ESP_LOGD(t, "%d %d",\n           a, b);\n')
+
+
+def test_raw_string_before_violation_still_caught() -> None:
+    # Regression for the masker desyncing on a raw string and disabling the check for the rest.
+    assert _lint('s.print(R"(<a href="x">)");\nif (y)\n  ESP_LOGD(t);\n')
+
+
+# --- rule: ignores non-violations ---
+
+
+def test_ignores_braced_body() -> None:
+    assert not _lint("if (x) {\n  ESP_LOGD(t);\n}\n")
+
+
+def test_ignores_commented_out_code() -> None:
+    assert not _lint("// if (x) ESP_LOGD(t);\n")
+
+
+def test_ignores_preprocessor_else() -> None:
+    assert not _lint("#else\n  ESP_LOGCONFIG(t);\n#endif\n")
+
+
+def test_ignores_non_log_body() -> None:
+    assert not _lint("if (x)\n  return false;\n")
+
+
+# --- NOLINT escape hatch, both placements ---
+
+
+def test_nolint_at_end_of_log_line_suppresses() -> None:
+    assert not _lint("if (x)\n  ESP_LOGD(t);  // NOLINT\n")
+
+
+def test_nolint_on_control_line_suppresses() -> None:
+    assert not _lint("if (x)  // NOLINT\n  ESP_LOGD(t);\n")

tests/script/test_determine_jobs.py

@@ -165,9 +165,14 @@ def test_main_all_tests_should_run(
         patch.object(determine_jobs, "_is_clang_tidy_full_scan", return_value=False),
         patch.object(
             determine_jobs,
-            "_all_integration_test_files",
+            "all_integration_test_files",
             return_value=fake_test_files,
         ),
+        patch.object(
+            determine_jobs,
+            "load_integration_durations",
+            return_value=dict.fromkeys(fake_test_files, 200.0),
+        ),
         patch.object(
             determine_jobs,
             "get_changed_components",
@@ -203,24 +208,12 @@ def test_main_all_tests_should_run(
     output = json.loads(captured.out)
 
     assert output["integration_tests"] is True
-    # run_all=True expands to the full glob and pre-buckets into 3 parts.
-    # Each bucket's `tests` is a JSON list of file paths.
+    assert output["integration_run_all"] is True
+    # run_all=True expands to the full glob; balance and naming are pinned
+    # by the unit tests, main() only needs to round-trip the structure
     assert isinstance(output["integration_test_buckets"], list)
-    assert len(output["integration_test_buckets"]) == 3
-    assert [b["name"] for b in output["integration_test_buckets"]] == [
-        "1/3",
-        "2/3",
-        "3/3",
-    ]
-    for bucket in output["integration_test_buckets"]:
-        assert isinstance(bucket["tests"], list)
-        for path in bucket["tests"]:
-            assert isinstance(path, str)
     bucket_files = [f for b in output["integration_test_buckets"] for f in b["tests"]]
-    assert bucket_files == fake_test_files
-    # Bucket sizes are balanced (max-min difference at most 1).
-    sizes = [len(b["tests"]) for b in output["integration_test_buckets"]]
-    assert max(sizes) - min(sizes) <= 1
+    assert sorted(bucket_files) == fake_test_files
     assert output["clang_tidy"] is True
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is True
@@ -529,14 +522,24 @@ def test_compute_integration_test_buckets_at_threshold_stays_single() -> None:
 
 
 def test_compute_integration_test_buckets_just_over_threshold_splits() -> None:
-    """One file over the threshold triggers the 3-bucket fan-out, balanced."""
+    """One file over the threshold fans out fully when the weights demand it."""
     n = determine_jobs.INTEGRATION_TESTS_SPLIT_THRESHOLD + 1
     files = [f"tests/integration/test_{i:02d}.py" for i in range(n)]
-    run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 200.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
     assert run is True
-    assert [b["name"] for b in buckets] == ["1/3", "2/3", "3/3"]
-    union = [path for b in buckets for path in b["tests"]]
+    # threshold+1 files x 200s caps at the maximum bucket count.
+    n_buckets = determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert [b["name"] for b in buckets] == [
+        f"{i + 1}/{n_buckets}" for i in range(n_buckets)
+    ]
+    union = sorted(path for b in buckets for path in b["tests"])
     assert union == sorted(files)
+    # Equal weights => bucket sizes are balanced (difference at most 1).
     sizes = [len(b["tests"]) for b in buckets]
     assert max(sizes) - min(sizes) <= 1
 
@@ -546,7 +549,7 @@ def test_compute_integration_test_buckets_run_all_with_empty_glob_disables_run()
 ):
     """run_all=True but glob returns no files => run suppressed (otherwise
     pytest would collect tests outside tests/integration/)."""
-    with patch.object(determine_jobs, "_all_integration_test_files", return_value=[]):
+    with patch.object(determine_jobs, "all_integration_test_files", return_value=[]):
         run, buckets = determine_jobs._compute_integration_test_buckets(True, [])
     assert run is False
     assert buckets == []
@@ -572,6 +575,13 @@ def test_determine_integration_tests(
         assert run_all is True
         assert test_files == []
 
+    # Dependency pins and the session init fixture trigger run_all
+    for trigger in sorted(determine_jobs.INTEGRATION_TESTS_TRIGGER_FILES):
+        with patch.object(determine_jobs, "changed_files", return_value=[trigger]):
+            run_all, test_files = determine_jobs.determine_integration_tests()
+            assert run_all is True
+            assert test_files == []
+
     # Python files directly in esphome/ do NOT trigger tests
     with patch.object(
         determine_jobs, "changed_files", return_value=["esphome/config.py"]
@@ -3231,3 +3241,86 @@ def test_esp8266_native_components_to_test_narrowing(
     ):
         result = determine_jobs.esp8266_native_components_to_test()
         assert result == expected
+
+
+def test_compute_integration_test_buckets_no_durations_full_fanout() -> None:
+    """Without recorded durations the fan-out stays at the maximum."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(determine_jobs, "load_integration_durations", return_value={}):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) == determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_compute_integration_test_buckets_adaptive_count() -> None:
+    """A small recorded total weight collapses to one bucket above the threshold."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 10.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    # 15 files x 10s recorded = 150s, under the per-bucket weight target.
+    assert [b["name"] for b in buckets] == ["1/1"]
+    assert buckets[0]["tests"] == files
+
+
+def test_compute_integration_test_buckets_duration_weighted() -> None:
+    """Heavy files spread across buckets instead of clustering by sorted name."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(12)]
+    durations = dict.fromkeys(files, 10.0)
+    durations[files[0]] = 600.0
+    durations[files[1]] = 600.0
+    with patch.object(
+        determine_jobs, "load_integration_durations", return_value=durations
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) >= 2
+    heavy_buckets = [b for b in buckets if set(files[:2]) & set(b["tests"])]
+    assert len(heavy_buckets) == 2, "heavy files should land in different buckets"
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_load_integration_durations_missing_or_corrupt(tmp_path: Path) -> None:
+    """Missing or unparsable durations data degrades to an empty mapping."""
+    with patch.object(helpers, "root_path", str(tmp_path)):
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+        durations_file.parent.mkdir(parents=True)
+        durations_file.write_text("not json")
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file.write_text('{"tests/integration/test_a.py": 12.5}')
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # Non-positive entries are dropped, valid ones survive
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": -1}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # One non-numeric entry cannot discard the whole recording
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": null}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # A non-dict top level degrades to empty
+        durations_file.write_text("[12.5]")
+        assert determine_jobs.load_integration_durations() == {}
+
+
+def test_committed_integration_durations_are_sane() -> None:
+    """The committed recording itself holds positive bounded floats."""
+    raw = json.loads(
+        (Path(helpers.root_path) / helpers.INTEGRATION_TEST_DURATIONS_FILE).read_text()
+    )
+    assert raw, "committed durations file missing or empty"
+    assert all(isinstance(v, (int, float)) and 0 < v < 86400 for v in raw.values())
+    assert all(k.startswith("tests/integration/test_") for k in raw)

tests/script/test_helpers.py

@@ -2120,3 +2120,30 @@ def test_get_cpp_changed_components_independent_of_cwd(
     assert helpers.get_cpp_changed_components(
         ["tests/components/time/__init__.py"]
     ) == ["time"]
+
+
+def test_lpt_partition_balances_skewed_weights() -> None:
+    """Heavy items spread across groups instead of clustering."""
+    items = [f"i{n}" for n in range(6)]
+    weights = {"i0": 100.0, "i1": 90.0, "i2": 10.0, "i3": 10.0, "i4": 5.0, "i5": 5.0}
+    groups = helpers.lpt_partition(items, weights, 2)
+    group_weights = sorted(sum(weights[i] for i in g) for g in groups)
+    # Contiguous split would give 200 vs 20; LPT lands at 110 vs 110
+    assert group_weights == [110.0, 110.0]
+    assert sorted(i for g in groups for i in g) == items
+
+
+def test_lpt_partition_more_groups_than_items() -> None:
+    """Surplus groups come back empty; every item still lands somewhere."""
+    items = ["a", "b"]
+    groups = helpers.lpt_partition(items, {"a": 1.0, "b": 1.0}, 4)
+    assert len(groups) == 4
+    assert sorted(i for g in groups for i in g) == items
+    assert sum(not g for g in groups) == 2
+
+
+def test_lpt_partition_tie_determinism() -> None:
+    """Equal weights assign in input order, so output is reproducible."""
+    items = [f"i{n}" for n in range(4)]
+    weights = dict.fromkeys(items, 1.0)
+    assert helpers.lpt_partition(items, weights, 2) == [["i0", "i2"], ["i1", "i3"]]

tests/script/test_update_integration_test_durations.py

@@ -0,0 +1,130 @@
+"""Unit tests for script/update_integration_test_durations.py."""
+
+import json
+from pathlib import Path
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Add the script directory to Python path so we can import the module
+script_dir = str((Path(__file__).parent / ".." / ".." / "script").resolve())
+sys.path.insert(0, script_dir)
+
+import helpers  # noqa: E402
+import update_integration_test_durations as uitd  # noqa: E402
+
+JUNIT_TEMPLATE = """<?xml version="1.0" encoding="utf-8"?>
+<testsuites><testsuite>{testcases}</testsuite></testsuites>
+"""
+
+KNOWN = {
+    "tests/integration/test_a.py",
+    "tests/integration/test_b.py",
+}
+
+
+def _write_junit(path: Path, testcases: str) -> None:
+    path.write_text(JUNIT_TEMPLATE.format(testcases=testcases), encoding="utf-8")
+
+
+def test_collect_durations_sums_per_file(tmp_path: Path) -> None:
+    """Testcases from the same module sum."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t1" time="1.5"/>'
+        '<testcase classname="tests.integration.test_a" name="t2" time="2.0"/>'
+        '<testcase classname="tests.integration.test_b" name="t1" time="4.0"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 3.5,
+        "tests/integration/test_b.py": 4.0,
+    }
+
+
+def test_collect_durations_class_based_testcase(tmp_path: Path) -> None:
+    """A class-based classname still maps to its module file."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a.TestFoo" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 2.5
+    }
+
+
+def test_collect_durations_unknown_module_skipped(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A classname that maps to no known file is skipped with a warning."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_gone" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+    assert "test_gone" in capsys.readouterr().err
+
+
+def test_collect_durations_skips_skipped_testcases(tmp_path: Path) -> None:
+    """Skipped testcases do not record a bogus zero duration."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="0">'
+        "<skipped/></testcase>",
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+
+
+def test_collect_durations_unexpected_classname_aborts(tmp_path: Path) -> None:
+    """A classname outside tests.integration means the junit layout changed."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.unit_tests.test_x" name="t" time="9.0"/>',
+    )
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_collect_durations_empty_dir_aborts(tmp_path: Path) -> None:
+    """No junit XML at all is a hard error, not an empty recording."""
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_main_merges_partial_run(tmp_path: Path) -> None:
+    """A partial run merges over the previous data instead of truncating it."""
+    tests_dir = tmp_path / "tests" / "integration"
+    tests_dir.mkdir(parents=True)
+    for name in ("test_a", "test_b", "test_c"):
+        (tests_dir / f"{name}.py").write_text("", encoding="utf-8")
+    durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+    durations_file.write_text(
+        json.dumps(
+            {
+                "tests/integration/test_a.py": 5.0,
+                "tests/integration/test_b.py": 7.0,
+                "tests/integration/test_gone.py": 9.0,
+            }
+        ),
+        encoding="utf-8",
+    )
+    junit_dir = tmp_path / "junit"
+    junit_dir.mkdir()
+    _write_junit(
+        junit_dir / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="6.0"/>',
+    )
+    with (
+        patch.object(helpers, "root_path", str(tmp_path)),
+        patch.object(uitd, "DURATIONS_FILE", durations_file),
+    ):
+        # 1 of 3 files covered: refused without --allow-partial
+        with patch.object(sys, "argv", ["uitd", str(junit_dir)]):
+            assert uitd.main() == uitd.EXIT_LOW_COVERAGE
+        with patch.object(sys, "argv", ["uitd", str(junit_dir), "--allow-partial"]):
+            assert uitd.main() == 0
+    # test_a updated, test_b kept, deleted test_gone dropped
+    assert json.loads(durations_file.read_text()) == {
+        "tests/integration/test_a.py": 6.0,
+        "tests/integration/test_b.py": 7.0,
+    }

tests/unit_tests/test_espota2_noise.py

@@ -0,0 +1,407 @@
+"""Unit tests for encrypted OTA uploads in esphome.espota2.
+
+A fake device implementing the responder side of the wire protocol (via
+noiseprotocol, which esphome already has through aioesphomeapi) serves a real
+TCP loopback connection, so these exercise the actual handshake, framing, and
+cipher interop of the client code. Tests that need the client-side crypto skip
+when the installed aioesphomeapi predates the noise module.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+from pathlib import Path
+import socket
+import sys
+import threading
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esphome import espota2
+
+PSK = base64.b64encode(bytes(range(32))).decode()
+OTHER_PSK = base64.b64encode(bytes(range(1, 33))).decode()
+
+MAGIC = bytes(espota2.MAGIC_BYTES)
+
+
+def _recv_exact(sock: socket.socket, amount: int) -> bytes:
+    data = b""
+    while len(data) < amount:
+        chunk = sock.recv(amount - len(data))
+        if not chunk:
+            raise ConnectionError("client closed")
+        data += chunk
+    return data
+
+
+def _frame(payload: bytes) -> bytes:
+    return (
+        bytes([espota2.NOISE_FRAME_INDICATOR, len(payload) >> 8, len(payload) & 0xFF])
+        + payload
+    )
+
+
+def _send_frame(sock: socket.socket, payload: bytes) -> None:
+    sock.sendall(_frame(payload))
+
+
+def _recv_frame(sock: socket.socket) -> bytes:
+    header = _recv_exact(sock, 3)
+    assert header[0] == 0x01
+    return _recv_exact(sock, (header[1] << 8) | header[2])
+
+
+class FakeEncryptedDevice(threading.Thread):
+    """Responder side of the encrypted OTA wire protocol."""
+
+    def __init__(
+        self,
+        psk: str = PSK,
+        version: int = 2,
+        offer_noise: bool = True,
+        require_noise: bool = True,
+        prologue_features_override: int | None = None,
+    ) -> None:
+        super().__init__(daemon=True)
+        self.psk = psk
+        self.version = version
+        self.offer_noise = offer_noise
+        self.require_noise = require_noise
+        self.prologue_features_override = prologue_features_override
+        self.received: bytes | None = None
+        self.error: Exception | None = None
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.port = self.listener.getsockname()[1]
+
+    def run(self) -> None:
+        try:
+            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:
+            self.listener.close()
+
+    def join_and_check(self) -> None:
+        self.join(timeout=10)
+        assert not self.is_alive(), "fake device did not finish"
+        if self.error is not None:
+            raise self.error
+
+    def _serve(self, sock: socket.socket) -> None:
+        assert _recv_exact(sock, 5) == MAGIC
+        sock.sendall(bytes([espota2.RESPONSE_OK, self.version]))
+        features = _recv_exact(sock, 1)[0]
+        noise_negotiated = bool(
+            features & espota2.CLIENT_FEATURE_SUPPORTS_NOISE
+            and features & espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+        )
+        if self.require_noise and not noise_negotiated:
+            sock.sendall(bytes([espota2.RESPONSE_ERROR_ENCRYPTION_REQUIRED]))
+            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
+
+        from cryptography.exceptions import InvalidTag
+        from noise.connection import NoiseConnection
+
+        prologue_features = (
+            features
+            if self.prologue_features_override is None
+            else self.prologue_features_override
+        )
+        prologue = (
+            espota2.NOISE_PROLOGUE_INIT
+            + MAGIC
+            + bytes([espota2.RESPONSE_OK, self.version, prologue_features])
+            + bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags])
+        )
+        proto = NoiseConnection.from_name(b"Noise_NNpsk0_25519_ChaChaPoly_SHA256")
+        proto.set_as_responder()
+        proto.set_psks(base64.b64decode(self.psk))
+        proto.set_prologue(prologue)
+        proto.start_handshake()
+
+        msg1 = _recv_frame(sock)
+        assert msg1[0] == 0x00
+        try:
+            proto.read_message(msg1[1:])
+        except InvalidTag:
+            _send_frame(sock, b"\x01" + espota2.NOISE_MAC_FAILURE_REASON.encode())
+            return
+        _send_frame(sock, b"\x00" + bytes(proto.write_message()))
+
+        def send_byte(byte: int) -> None:
+            _send_frame(sock, proto.encrypt(bytes([byte])))
+
+        def recv_unit(length: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert len(plaintext) == length, "control units must be one per frame"
+            return plaintext
+
+        send_byte(espota2.RESPONSE_AUTH_OK)
+        recv_unit(1)  # ota type
+        size = int.from_bytes(recv_unit(4), "big")
+        send_byte(espota2.RESPONSE_UPDATE_PREPARE_OK)
+        md5_hex = recv_unit(32)
+        send_byte(espota2.RESPONSE_BIN_MD5_OK)
+
+        received = b""
+        acked = 0
+        while len(received) < size:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            received += plaintext
+            if self.version >= espota2.OTA_VERSION_2_0:
+                while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
+                    len(received) == size and acked < size
+                ):
+                    send_byte(espota2.RESPONSE_CHUNK_OK)
+                    acked += espota2.UPLOAD_BLOCK_SIZE
+        assert hashlib.md5(received).hexdigest().encode() == md5_hex
+        send_byte(espota2.RESPONSE_RECEIVE_OK)
+        send_byte(espota2.RESPONSE_UPDATE_END_OK)
+        assert recv_unit(1) == bytes([espota2.RESPONSE_OK])
+        self.received = received
+
+
+def _upload(
+    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+) -> None:
+    device.start()
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(10)
+    sock.connect(("127.0.0.1", device.port))
+    try:
+        espota2.perform_ota(
+            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+        )
+    finally:
+        sock.close()
+
+
+def test_encrypted_upload_success() -> None:
+    """A full encrypted v2 upload spanning several 8192-byte blocks."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = bytes(range(256)) * 80  # 20480 bytes, crosses chunk-ack boundaries
+    device = FakeEncryptedDevice()
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_encrypted_upload_version_1() -> None:
+    """Version 1 protocol (no chunk acks) works through the noise transport."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"v1 firmware image" * 100
+    device = FakeEncryptedDevice(version=1)
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_wrong_key_fails_with_clear_error() -> None:
+    """A key mismatch surfaces the device's handshake reject readably."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(psk=OTHER_PSK)
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_tampered_negotiation_breaks_handshake() -> None:
+    """A negotiation byte differing between the sides breaks the prologue MAC."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(
+        prologue_features_override=espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+    )
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_client_fails_closed_when_device_lacks_encryption() -> None:
+    """With a key configured, a device not offering noise aborts the upload."""
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
+    with pytest.raises(espota2.OTAError, match="refusing to send the image"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_plaintext_client_gets_encryption_required_error() -> None:
+    """A client without a key gets the device's 0x94 error message."""
+    device = FakeEncryptedDevice()
+    with pytest.raises(espota2.OTAError, match="requires an encrypted OTA"):
+        _upload(device, b"firmware", None)
+    device.join_and_check()
+
+
+def test_missing_aioesphomeapi_noise_module_message() -> None:
+    """An aioesphomeapi without the noise module produces a clear error."""
+    with (
+        patch.dict(sys.modules, {"aioesphomeapi.noise": None}),
+        pytest.raises(espota2.OTAError, match="requires a newer aioesphomeapi"),
+    ):
+        espota2.NoiseSocketWrapper(Mock(), PSK, b"prologue")
+
+
+class ScriptedSocket:
+    """Serves scripted recv chunks; b"" means the peer closed."""
+
+    def __init__(self, *chunks: bytes | Exception) -> None:
+        self.chunks = list(chunks)
+        self.sent: list[bytes] = []
+
+    def sendall(self, data: bytes) -> None:
+        self.sent.append(data)
+
+    def settimeout(self, timeout: float) -> None:
+        pass
+
+    def recv(self, amount: int) -> bytes:
+        if not self.chunks:
+            return b""
+        chunk = self.chunks[0]
+        if isinstance(chunk, Exception):
+            self.chunks.pop(0)
+            raise chunk
+        take, rest = chunk[:amount], chunk[amount:]
+        if rest:
+            self.chunks[0] = rest
+        else:
+            self.chunks.pop(0)
+        return take
+
+
+def _wrapper(*chunks: bytes | Exception) -> espota2.NoiseSocketWrapper:
+    pytest.importorskip("aioesphomeapi.noise")
+    return espota2.NoiseSocketWrapper(ScriptedSocket(*chunks), PSK, b"prologue")
+
+
+def test_wrapper_rejects_malformed_psk() -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    with pytest.raises(espota2.OTAError, match="Invalid OTA encryption key"):
+        espota2.NoiseSocketWrapper(ScriptedSocket(), "not-base64!!!", b"prologue")
+
+
+def test_handshake_socket_error_is_network_error() -> None:
+    wrapper = _wrapper(OSError("boom"))
+    with pytest.raises(espota2.OTANetworkError, match="noise handshake"):
+        wrapper.do_handshake()
+
+
+def test_handshake_closed_at_frame_boundary() -> None:
+    wrapper = _wrapper()
+    with pytest.raises(espota2.OTANetworkError, match="closed connection during"):
+        wrapper.do_handshake()
+
+
+def test_handshake_reject_with_other_reason() -> None:
+    wrapper = _wrapper(_frame(b"\x01Handshake error"))
+    with pytest.raises(
+        espota2.OTAError, match="rejected the noise handshake: Handshake error"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_garbage_second_message() -> None:
+    """A valid-looking point with a garbage MAC fails cleanly."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(range(48))))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_invalid_curve_point() -> None:
+    """An all-zero x25519 point is rejected as a clean error, not a crash."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(48)))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_recv_closed_at_frame_boundary_returns_empty() -> None:
+    wrapper = _wrapper()
+    assert wrapper.recv(1) == b""
+
+
+def test_recv_corrupt_frame_is_retryable_network_error() -> None:
+    from cryptography.exceptions import InvalidTag
+
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(side_effect=InvalidTag()))
+    with pytest.raises(espota2.OTANetworkError, match="decryption failed"):
+        wrapper.recv(1)
+
+
+def test_wrapper_blocks_unencrypted_socket_methods() -> None:
+    """Byte-moving socket methods must not bypass the encrypted transport."""
+    wrapper = _wrapper()
+    # The harmless socket controls pass through to the wrapped socket
+    wrapper._sock = Mock()
+    wrapper.settimeout(1)
+    wrapper._sock.settimeout.assert_called_once_with(1)
+    wrapper.setsockopt(6, 1, 1)
+    wrapper._sock.setsockopt.assert_called_once_with(6, 1, 1)
+    wrapper.close()
+    wrapper._sock.close.assert_called_once_with()
+    with pytest.raises(AttributeError):
+        _ = wrapper.send
+    with pytest.raises(AttributeError):
+        _ = wrapper.recv_into
+
+
+def test_recv_empty_plaintext_frame_is_protocol_error() -> None:
+    """A MAC-only frame decrypts to nothing; b'' from recv must mean close."""
+    wrapper = _wrapper(_frame(bytes(16)))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b""))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper.recv(1)
+
+
+def test_recv_frame_bad_indicator_is_retryable() -> None:
+    wrapper = _wrapper(b"\x02\x00\x01x")
+    with pytest.raises(espota2.OTANetworkError, match="Bad noise frame indicator"):
+        wrapper._recv_frame()
+
+
+def test_recv_frame_zero_length_is_retryable() -> None:
+    wrapper = _wrapper(bytes([espota2.NOISE_FRAME_INDICATOR, 0, 0]))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper._recv_frame()
+
+
+def test_perform_ota_blank_key_refuses_plaintext() -> None:
+    with pytest.raises(espota2.OTAError, match="empty OTA encryption key"):
+        espota2.perform_ota(
+            ScriptedSocket(), None, io.BytesIO(b"x"), Path("f.bin"), noise_psk=""
+        )
+
+
+def test_recv_exact_closed_mid_frame() -> None:
+    wrapper = _wrapper(_frame(b"partial")[:5])
+    with pytest.raises(OSError, match="closed inside a noise frame"):
+        wrapper._recv_frame()
+
+
+def test_recv_serves_buffered_plaintext_without_new_frame() -> None:
+    """A second recv drains the decrypted buffer without reading another frame."""
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b"AB"))
+    assert wrapper.recv(1) == b"A"  # reads and decrypts one frame
+    assert wrapper.recv(1) == b"B"  # served from the buffer, no new frame
+    wrapper._decrypt.decrypt.assert_called_once()

tests/unit_tests/test_git.py

@@ -714,6 +714,25 @@ def test_run_git_command_without_git_dir_raises_error(
         git.run_git_command(["git", "clone", "https://invalid.url/repo.git"])
 
 
+def test_has_complete_clone(tmp_path: Path) -> None:
+    """The lock-free probe tracks the completion marker, subpath included."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    subpath = Path("lib")
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    repo_dir = _compute_repo_dir(url, "v1", "test_domain") / subpath
+    (repo_dir / ".git").mkdir(parents=True)
+    # A directory without the marker is an incomplete clone
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    _mark_clone_complete(repo_dir)
+    assert git.has_complete_clone(url, "v1", "test_domain", subpath)
+    # The ref is part of the cache key
+    assert not git.has_complete_clone(url, "v2", "test_domain", subpath)
+
+
 def test_clone_or_update_with_never_refresh(
     tmp_path: Path, mock_run_git_command: Mock
 ) -> None:

tests/unit_tests/test_main.py

@@ -87,7 +87,9 @@ from esphome.const import (
     CONF_BROKER,
     CONF_DISABLED,
     CONF_DISCOVER_IP,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
+    CONF_KEY,
     CONF_LEVEL,
     CONF_LOG,
     CONF_LOG_TOPIC,
@@ -113,6 +115,7 @@ from esphome.const import (
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
+    PLATFORM_HOST,
     PLATFORM_NRF52,
     PLATFORM_RP2,
     Toolchain,
@@ -2106,10 +2109,65 @@ 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
+        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
+def test_upload_program_ota_encryption_key(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """The resolved encryption key is passed through to run_ota."""
+    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_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: key},
+            }
+        ]
+    }
+    exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    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
+    )
+
+
+def test_upload_program_ota_encryption_without_key_fails_closed(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """An encryption block with no resolved key must never upload plaintext."""
+    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_ENCRYPTION: {},
+            }
+        ]
+    }
+    with pytest.raises(EsphomeError, match="no key was resolved"):
+        upload_program(config, MockArgs(), ["192.168.1.100"])
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_ota_with_file_arg(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2137,7 +2195,7 @@ 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
+        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2192,6 +2250,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2253,6 +2312,7 @@ def test_upload_program_ota_partition_table_mqttip(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2440,6 +2500,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         None,
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
+        None,
     )
 
 
@@ -2602,6 +2663,42 @@ def test_has_web_server_logging_respects_log_disabled() -> None:
     assert has_web_server_logging() is False
 
 
+def test_upload_program_web_server_warns_when_encryption_configured(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Explicitly picking web_server OTA on an encrypted config warns about
+    the plaintext upload path."""
+    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_ENCRYPTION: {CONF_KEY: "test_key"},
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    with caplog.at_level(logging.WARNING):
+        exit_code, _ = upload_program(config, args, ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert any("plaintext HTTP" in record.message for record in caplog.records)
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_web_server_only_auto_dispatches(
     mock_run_web_server_ota: Mock,
     mock_run_ota: Mock,
@@ -2892,7 +2989,7 @@ 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
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2942,7 +3039,7 @@ 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
+        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5114,6 +5211,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5163,6 +5261,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5340,7 +5439,7 @@ 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
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -7513,3 +7612,54 @@ async def test_wrap_to_code_comment_is_insertion_order_independent() -> None:
     assert first == second
     assert second.index("alpha") < second.index("beta")
     assert second.index("a: 2") < second.index("z: 1")
+
+
+def test_host_program_path_platformio_toolchain() -> None:
+    """Host + PlatformIO toolchain reads the memoized idedata path."""
+    setup_core(platform=PLATFORM_HOST)
+    idedata = SimpleNamespace(firmware_elf_path="/build/x/.pioenvs/x/program")
+    with patch(
+        "esphome.platformio.toolchain.get_idedata", return_value=idedata
+    ) as mock_get:
+        assert main._host_program_path({}) == "/build/x/.pioenvs/x/program"
+    mock_get.assert_called_once_with({})
+
+
+def test_host_program_path_esp_idf_toolchain() -> None:
+    """Host + native ESP-IDF toolchain asks the espidf toolchain for the ELF."""
+    setup_core(platform=PLATFORM_HOST)
+    CORE.toolchain = Toolchain.ESP_IDF
+    with patch(
+        "esphome.espidf.toolchain.get_elf_path", return_value=Path("/b/app.elf")
+    ):
+        assert main._host_program_path({}) == str(Path("/b/app.elf"))
+
+
+def test_command_compile_host_logs_program_path(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """command_compile on host logs the compiled program path."""
+    setup_core(platform=PLATFORM_HOST)
+    with (
+        patch.object(main, "write_cpp", return_value=0),
+        patch.object(main, "compile_program", return_value=0),
+        patch.object(main, "_host_program_path", return_value="/b/program"),
+        caplog.at_level(logging.INFO),
+    ):
+        assert main.command_compile(SimpleNamespace(only_generate=False), {}) == 0
+    assert "Successfully compiled program to path '/b/program'" in caplog.text
+
+
+def test_command_run_host_executes_program(caplog: pytest.LogCaptureFixture) -> None:
+    """command_run on host logs and executes the compiled program directly."""
+    setup_core(platform=PLATFORM_HOST)
+    with (
+        patch.object(main, "write_cpp", return_value=0),
+        patch.object(main, "compile_program", return_value=0),
+        patch.object(main, "_host_program_path", return_value="/b/program"),
+        patch.object(main, "run_external_process", return_value=0) as mock_run,
+        caplog.at_level(logging.INFO),
+    ):
+        assert main.command_run(SimpleNamespace(), {}) == 0
+    mock_run.assert_called_with("/b/program")
+    assert "Running program from path '/b/program'" in caplog.text

tests/unit_tests/test_platformio_library.py

@@ -638,7 +638,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     """Registry archives in one wave download concurrently, deduped by URL;
-    git/local sources and failures are left to the sequential call."""
+    local sources and failures are left to the sequential call."""
     calls: list[str] = []
 
     def fake_download(
@@ -658,7 +658,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
         # into the same cache directory)
         ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))),
         ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))),
-        ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
+        ("l", ConvertedLibrary("l", "*", LocalSource("/some/lib"))),
     ]
     lib._prefetch_wave(wave, "", "idf")
     assert sorted(calls) == [
@@ -670,6 +670,83 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     assert "Prefetch of c failed (retrying sequentially)" in caplog.text
 
 
+def test_prefetch_wave_clones_git_sources_in_parallel(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Git sources join the same prefetch batch as the archives, deduped by
+    clone target; a clone failure warns and is left to the sequential call."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+
+    def fake_clone(self, dir_suffix, force=False, salt="", namespace=""):
+        calls.append(f"{self}/{dir_suffix}")
+        if "boom" in self.url:
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(GitSource, "download", fake_clone)
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        # Same url@ref and target dir must clone once
+        ("g2", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        ("h", ConvertedLibrary("h", "*", GitSource("https://x/boom.git", None))),
+    ]
+    monkeypatch.setattr(
+        URLSource, "download", lambda self, dir_suffix, progress=None, **kw: None
+    )
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/boom.git/h", "https://x/g.git#v1/g"]
+    assert "Cloning 2 library repo(s): g, h" in caplog.text
+    assert "Prefetch of h failed (retrying sequentially)" in caplog.text
+
+
+def test_source_base_prefetch_defaults() -> None:
+    """The base Source is not prefetchable and reports cached (nothing to do)."""
+    source = Source()
+    assert source.prefetch_key("x") is None
+    assert source.is_cached("x") is True
+
+
+def test_prefetch_wave_single_clone_uses_the_batch(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A wave with only git sources still clones through the batch runner."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: False)
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, force=False, salt="", namespace="": calls.append(
+            self.url
+        ),
+    )
+    lib._prefetch_wave(
+        [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))],
+        "",
+        "idf",
+    )
+    assert calls == ["https://x/g.git"]
+    assert "Cloning 1 library repo(s): g" in caplog.text
+    assert "Downloading" not in caplog.text
+
+
+def test_prefetch_wave_warm_git_cache_is_silent(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An already-complete clone is neither re-fetched nor announced."""
+    caplog.set_level("INFO")
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, **kw: (_ for _ in ()).throw(AssertionError("cloned")),
+    )
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: True)
+    wave = [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))]
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Cloning" not in caplog.text
+
+
 def test_prefetch_wave_unknown_size_left_to_sequential(
     setup_core, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/unit_tests/test_platformio_prefetch.py

@@ -13,6 +13,7 @@ from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 from filelock import Timeout
+from platformio.dependencies import get_core_dependencies
 from platformio.package.manager._install import PackageManagerInstallMixin
 from platformio.package.manager.base import BasePackageManager
 from platformio.package.manager.library import LibraryPackageManager
@@ -1417,6 +1418,76 @@ def test_prefetch_installs_cached_archives_without_downloads(
     assert not (tmp_path / pf._SENTINEL_NAME).exists()
 
 
+@pytest.mark.parametrize(
+    ("platform_group", "lib_group", "expected"),
+    [
+        (
+            [("toolchain-x@1", _FakeSpec(name="toolchain-x"))],
+            [],
+            ["configure", "install", "configure"],
+        ),
+        ([], [("noise-c@1.0", _FakeSpec(name="noise-c"))], ["configure", "install"]),
+    ],
+)
+def test_prefetch_reconfigures_only_after_platform_installs(
+    tmp_path: Path, platform_group: list, lib_group: list, expected: list[str]
+) -> None:
+    """Installed platform packages get a second configure pass; libraries do not."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    order: list[str] = []
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    fake_platform.configure_project_packages.side_effect = lambda env, targets: (
+        order.append("configure")
+    )
+    config = _fake_config(
+        tmp_path, {"platform": "fake/p@1", "lib_deps": ["esphome/noise-c@1.0"]}
+    )
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(
+            pf,
+            "_registry_jobs",
+            side_effect=[([], 0, platform_group), ([], 0, lib_group)],
+        ),
+        patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
+        patch.object(pf, "_preinstall", side_effect=lambda *_: order.append("install")),
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    assert order == expected
+
+
+@pytest.mark.parametrize(
+    "err", [RuntimeError("idf_tools.py failed"), SystemExit("postinstall exited")]
+)
+def test_prefetch_settle_failure_warns_and_continues(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture, err: BaseException
+) -> None:
+    """A failing second configure pass only costs the speedup."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    fake_platform.configure_project_packages.side_effect = [None, err]
+    config = _fake_config(tmp_path, {"platform": "fake/p@1"})
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(
+            pf,
+            "_registry_jobs",
+            side_effect=[
+                ([], 0, [("toolchain-x@1", _FakeSpec(name="toolchain-x"))]),
+                ([], 0, []),
+            ],
+        ),
+        patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
+        patch.object(pf, "_preinstall"),
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    assert f"Could not settle platform packages: {err}" in caplog.text
+
+
 def test_preinstall_extracts_in_parallel_under_one_lock(tmp_path: Path) -> None:
     """The manager lock wraps the whole batch; per-thread managers share
     its package dir; one failing install leaves the rest alone."""
@@ -1658,30 +1729,31 @@ def test_preinstall_unlocks_even_when_pool_fails(tmp_path: Path) -> None:
     m.unlock.assert_called_once_with()
 
 
-def test_prefetch_skips_duplicate_tool_scons(tmp_path: Path) -> None:
-    """A platform that lists tool-scons itself does not get it appended."""
+def test_prefetch_replaces_platform_tool_scons_with_core_spec(tmp_path: Path) -> None:
+    """A platform's own tool-scons spec gives way to the core's registry spec."""
     _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
     fake_platform = MagicMock()
     fake_platform.packages = {"tool-scons": {"optional": False}}
     fake_platform.get_package_spec.side_effect = lambda name: _FakeSpec(
-        uri=None, name=name
+        uri="https://x/scons.zip", name=name, owner=None
     )
     config = _fake_config(tmp_path, {"platform": "fake/p@1"})
     modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
-    batches: list[list[str]] = []
+    batches: list[list] = []
     with (
         patch.dict("sys.modules", modules),
         patch.object(
             pf,
             "_registry_jobs",
             side_effect=lambda mgr, specs, seen: (
-                batches.append([s.name for s in specs]) or ([], 0, [])
+                batches.append(list(specs)) or ([], 0, [])
             ),
         ),
         patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
     ):
         pf._prefetch(tmp_path, "testenv")
-    assert batches[0] == ["tool-scons"]
+    (spec,) = batches[0]
+    assert (spec.name, spec.owner, spec.uri) == ("tool-scons", "platformio", None)
 
 
 def test_platformio_private_api_contract() -> None:
@@ -1714,6 +1786,8 @@ def test_platformio_private_api_contract() -> None:
         assert callable(getattr(BasePackageManager, name))
     # The dependency wave mirrors install_dependency's builtin skip
     assert callable(LibraryPackageManager.is_builtin_lib)
+    # The prefetch keys tool-scons on this core dependency
+    assert "tool-scons" in get_core_dependencies()
     # The pre-install passes these positionally / by keyword
     assert "compatibility" in inspect.signature(BasePackageManager.__init__).parameters
     lib_params = inspect.signature(LibraryPackageManager.__init__).parameters