Merge branch 'esp8266-native-parallel-extract' into espidf-parallel-tool-extract

This commit is contained in:
J. Nick Koston
2026-09-08 18:06:04 +02:00
155 changed files with 6466 additions and 930 deletions
@@ -0,0 +1,13 @@
esphome:
name: test
esp32:
variant: esp32
wifi:
ssid: MySSID
password: password1
# esp32_ble_server is only auto-loaded here, so it has no services of its own.
esp32_improv:
authorizer: none
@@ -0,0 +1,9 @@
esphome:
name: test
esp32:
variant: esp32
esp32_ble_server:
id: ble_server
manufacturer_data: [0x72, 0x04, 0x00, 0x23]
@@ -0,0 +1,14 @@
esphome:
name: test
esp32:
variant: esp32
esp32_ble_server:
id: ble_server
services:
- uuid: 2a24b789-7aab-4535-af3e-ee76a35cc12d
characteristics:
- uuid: cad48e28-7fbe-41cf-bae9-d77a6c233423
read: true
value: [1, 2, 3, 4]
@@ -1,5 +1,10 @@
"""Tests for esp32_ble_server configuration helpers."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components.esp32_ble_server import (
@@ -45,3 +50,26 @@ def test_uuid_is_matches_descriptor_short_strings(uuid16) -> None:
assert uuid_is(uuid16, uuid16)
assert uuid_is(f"{uuid16:04X}", uuid16)
assert uuid_is(f"{uuid16:08X}", uuid16)
@pytest.mark.parametrize(
("config_file", "required"),
[
# Auto-loaded by esp32_improv only: nothing to find until Improv asks for it
("improv_only.yaml", False),
# The configuration defines a service clients are meant to connect to
("own_service.yaml", True),
# Manufacturer data is only useful if it is actually broadcast
("manufacturer_data_only.yaml", True),
],
)
def test_advertising_required(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
config_file: str,
required: bool,
) -> None:
"""The server only requests advertising when the configuration needs it."""
main_cpp = generate_main(component_config_path(config_file))
assert f"set_advertising_required({str(required).lower()})" in main_cpp
@@ -0,0 +1,32 @@
esphome:
name: test-keyboard-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- keyboard:
id: keyboard_widget
@@ -0,0 +1,34 @@
esphome:
name: test-qrcode-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- qrcode:
id: qr_widget
size: 100
text: "esphome.io"
@@ -0,0 +1,35 @@
esphome:
name: test-tabview-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- tabview:
id: tabview_widget
tabs:
- name: "Tab 1"
id: tab_1
@@ -0,0 +1,32 @@
"""Widgets whose LVGL C implementation creates or references labels
internally (tab titles, key legends, the QR canvas fallback) must declare
the label dependency in ``get_uses()``. Otherwise a config that contains
no ``label`` widget of its own compiles LVGL without ``LV_USE_LABEL`` and
fails at C compile time with undefined ``lv_label_*`` symbols.
"""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components.lvgl import defines as df
@pytest.mark.parametrize(
"yaml_file",
[
"qrcode_no_label.yaml",
"keyboard_no_label.yaml",
"tabview_no_label.yaml",
],
)
def test_label_less_config_enables_lv_use_label(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
yaml_file: str,
) -> None:
generate_main(component_config_path(yaml_file))
assert "LV_USE_LABEL" in df.get_defines()
@@ -5,11 +5,7 @@ from __future__ import annotations
import pytest
from esphome import config_validation as cv
from esphome.components.noise import (
decode_encryption_key,
is_reserved_key,
validate_encryption_key,
)
from esphome.components.noise import decode_encryption_key, validate_encryption_key
KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
@@ -41,6 +37,8 @@ def test_decode_encryption_key_rejects_short_decode() -> None:
decode_encryption_key("AAECAw==")
def test_is_reserved_key() -> None:
assert is_reserved_key("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")
assert not is_reserved_key(KEY)
def test_validate_encryption_key_rejects_all_zeros() -> None:
"""The all-zeros key is the provisioning sentinel the device treats as no
key, so it never reaches a build."""
with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
validate_encryption_key("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")
+189 -53
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any
@@ -14,6 +15,7 @@ from esphome.components.esphome.ota import (
_validate_no_password_with_encryption,
ota_esphome_final_validate,
)
from esphome.components.noise import static_encryption_key
from esphome.const import (
CONF_API,
CONF_ENCRYPTION,
@@ -115,7 +117,6 @@ def test_non_esphome_ota_unaffected() -> None:
API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
OTHER_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
ZEROS_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
def test_encryption_key_inherited_from_api() -> None:
@@ -197,36 +198,6 @@ def test_encryption_without_any_key_rejected() -> None:
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 = {
@@ -295,13 +266,14 @@ def test_encryption_explicit_key_with_runtime_provisioned_api_accepted() -> None
fv.full_config.reset(token)
@pytest.mark.parametrize("component", ["web_server", "prometheus"])
def test_encryption_with_web_server_ota_warns(
caplog: pytest.LogCaptureFixture,
caplog: pytest.LogCaptureFixture, component: str
) -> None:
"""With the web_server component the plaintext /update endpoint is always
on; the combination validates with a warning."""
"""web_server and prometheus keep the shared listener up, so the
plaintext /update endpoint is always on and the combination warns."""
full_conf = {
"web_server": {},
component: {},
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)},
@@ -316,12 +288,12 @@ def test_encryption_with_web_server_ota_warns(
fv.full_config.reset(token)
def test_encryption_with_captive_portal_web_server_ota_warns(
def test_encryption_with_captive_portal_does_not_warn(
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."""
web_server component; its endpoint only exists while the fallback AP is
active and is the intended recovery path, so there is no warning."""
full_conf = {
"captive_portal": {},
CONF_OTA: [
@@ -333,7 +305,10 @@ def test_encryption_with_captive_portal_web_server_ota_warns(
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
assert any("captive_portal" in record.message for record in caplog.records)
assert not any(
"OTA encryption does not cover" in record.message
for record in caplog.records
)
esphome_conf = next(
conf
for conf in fv.full_config.get()[CONF_OTA]
@@ -344,6 +319,100 @@ def test_encryption_with_captive_portal_web_server_ota_warns(
fv.full_config.reset(token)
def test_password_with_api_key_warns(caplog: pytest.LogCaptureFixture) -> None:
"""A static api key makes the device offer encryption and the CLI take
it, so the password is dead weight; the config validates with a warning."""
full_conf = {
CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
assert any("wastes significant flash" in r.message for r in caplog.records)
finally:
fv.full_config.reset(token)
def test_password_with_runtime_api_key_warns_differently(
caplog: pytest.LogCaptureFixture,
) -> None:
"""The CLI still needs the password, but the provisioned key also
authenticates uploads; the warning says so without the flash advice."""
full_conf = {
CONF_API: {CONF_ENCRYPTION: {}},
CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
messages = [r.message for r in caplog.records]
assert any("provisioned at runtime also authenticates" in m for m in messages)
assert not any("wastes significant flash" in m for m in messages)
finally:
fv.full_config.reset(token)
def test_password_without_api_key_no_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Without an api key there is no offer, so nothing to warn about."""
full_conf = {
CONF_API: {},
CONF_OTA: [_make_ota_config(port=3232, **{CONF_PASSWORD: "pw"})],
}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
assert not any("authenticates" in r.message for r in caplog.records)
finally:
fv.full_config.reset(token)
def test_web_server_component_without_ota_platform_does_not_warn(
caplog: pytest.LogCaptureFixture,
) -> None:
"""The web_server component alone has no /update endpoint."""
full_conf = {
"web_server": {},
CONF_OTA: [
_make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
],
}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
ota_esphome_final_validate({})
assert not any(
"OTA encryption does not cover" in r.message for r in caplog.records
)
finally:
fv.full_config.reset(token)
def test_web_server_ota_platform_alone_does_not_warn(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Only the web_server component starts the shared listener, so the ota
platform on its own never exposes /update."""
full_conf = {
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 not any("plaintext /update" in r.message for r in caplog.records)
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 = {
@@ -370,20 +439,87 @@ def test_auto_load_pulls_noise_only_for_encryption() -> 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_static_encryption_key() -> None:
"""Only a build-time key counts; a runtime provisioned one does not."""
assert static_encryption_key({}) is None
assert static_encryption_key({CONF_ENCRYPTION: {}}) is None
assert static_encryption_key({CONF_ENCRYPTION: {CONF_KEY: API_KEY}}) == API_KEY
@pytest.mark.parametrize(
("yaml_name", "defines_present", "defines_absent"),
[
# An api key alone compiles the transport in without requiring it;
# the device uses the api server's key, not a copy
(
"api_key_offer",
{"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_FROM_API"},
{"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_PROVISIONED"},
),
# A password still guards plaintext uploads on an offering device
(
"api_key_offer_password",
{"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_FROM_API", "USE_OTA_PASSWORD"},
{"USE_OTA_ENCRYPTION_REQUIRED", "USE_OTA_ENCRYPTION_PROVISIONED"},
),
# The ota encryption block is what makes the device refuse plaintext
(
"encryption_required",
{
"USE_OTA_ENCRYPTION",
"USE_OTA_ENCRYPTION_REQUIRED",
"USE_OTA_ENCRYPTION_FROM_API",
},
{"USE_OTA_ENCRYPTION_PROVISIONED"},
),
# Without api encryption the ota key is the device's own
(
"own_key",
{"USE_OTA_ENCRYPTION", "USE_OTA_ENCRYPTION_REQUIRED"},
{"USE_OTA_ENCRYPTION_FROM_API", "USE_OTA_ENCRYPTION_PROVISIONED"},
),
# A key provisioned at runtime lives in the api server; the device
# offers with it once provisioned and never requires it
(
"runtime_api_key",
{
"USE_OTA_ENCRYPTION",
"USE_OTA_ENCRYPTION_FROM_API",
"USE_OTA_ENCRYPTION_PROVISIONED",
},
{"USE_OTA_ENCRYPTION_REQUIRED"},
),
# No api encryption at all keeps the noise glue out of the build
(
"plain",
set(),
{
"USE_OTA_ENCRYPTION",
"USE_OTA_ENCRYPTION_REQUIRED",
"USE_OTA_ENCRYPTION_FROM_API",
"USE_OTA_ENCRYPTION_PROVISIONED",
},
),
],
)
def test_encryption_offer_codegen(
generate_main: Callable[[str], str],
yaml_name: str,
defines_present: set[str],
defines_absent: set[str],
) -> None:
main_cpp = generate_main(
f"tests/component_tests/ota/test_esphome_ota_{yaml_name}.yaml"
)
defines = {define.name for define in CORE.defines}
assert defines_present <= defines
assert not (defines_absent & defines)
encrypted = "USE_OTA_ENCRYPTION" in defines_present
own_key = encrypted and "USE_OTA_ENCRYPTION_FROM_API" not in defines_present
assert ("esphome_esphomeotacomponent_id->set_noise_psk(" in main_cpp) is own_key
assert ("set_auth_password(" in main_cpp) is ("USE_OTA_PASSWORD" in defines_present)
# The noise transport source compiles only when the define is set
assert FILTER_SOURCE_FILES() == ([] if encrypted else ["ota_esphome_noise.cpp"])
def test_password_with_encryption_rejected() -> None:
@@ -0,0 +1,11 @@
esphome:
name: ota-offer
host:
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
@@ -0,0 +1,12 @@
esphome:
name: ota-offer-password
host:
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
password: "superlongpasswordthatnoonewillknow"
@@ -0,0 +1,12 @@
esphome:
name: ota-encryption-required
host:
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
encryption:
@@ -0,0 +1,11 @@
esphome:
name: ota-own-key
host:
api:
ota:
- platform: esphome
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
@@ -0,0 +1,9 @@
esphome:
name: ota-plain
host:
api:
ota:
- platform: esphome
@@ -0,0 +1,10 @@
esphome:
name: ota-runtime-key
host:
api:
encryption:
ota:
- platform: esphome
@@ -0,0 +1,145 @@
"""Tests for template climate config validation."""
import pytest
from esphome import config_validation as cv
from esphome.components.template.climate import (
CONF_SET_TARGET_HUMIDITY_ACTION,
CONF_SET_TARGET_TEMPERATURE_ACTION,
CONF_SET_TARGET_TEMPERATURE_HIGH_ACTION,
CONF_SET_TARGET_TEMPERATURE_LOW_ACTION,
CONF_SUPPORTS_CURRENT_HUMIDITY,
CONF_SUPPORTS_CURRENT_TEMPERATURE,
CONF_SUPPORTS_TARGET_HUMIDITY,
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE,
CONF_TARGET_HUMIDITY,
_resolve_supports,
_validate_initial_state,
_validate_set_actions,
)
from esphome.const import (
CONF_HUMIDITY_SENSOR,
CONF_INITIAL_STATE,
CONF_SENSOR,
CONF_TARGET_TEMPERATURE,
CONF_TARGET_TEMPERATURE_HIGH,
CONF_TARGET_TEMPERATURE_LOW,
)
from esphome.types import ConfigType
def test_supports_current_temperature_derived_from_sensor() -> None:
config: ConfigType = {CONF_SENSOR: "some_sensor"}
assert _resolve_supports(config)[CONF_SUPPORTS_CURRENT_TEMPERATURE] is True
def test_supports_current_temperature_false_without_sensor() -> None:
assert _resolve_supports({})[CONF_SUPPORTS_CURRENT_TEMPERATURE] is False
def test_supports_current_temperature_explicit_true_without_sensor_allowed() -> None:
# The value can still be reported with climate.template.publish.
config: ConfigType = {CONF_SUPPORTS_CURRENT_TEMPERATURE: True}
assert _resolve_supports(config)[CONF_SUPPORTS_CURRENT_TEMPERATURE] is True
def test_supports_current_temperature_false_with_sensor_rejected() -> None:
config: ConfigType = {
CONF_SENSOR: "some_sensor",
CONF_SUPPORTS_CURRENT_TEMPERATURE: False,
}
with pytest.raises(cv.Invalid, match="cannot be false"):
_resolve_supports(config)
def test_supports_current_humidity_false_with_sensor_rejected() -> None:
config: ConfigType = {
CONF_HUMIDITY_SENSOR: "some_sensor",
CONF_SUPPORTS_CURRENT_HUMIDITY: False,
}
with pytest.raises(cv.Invalid, match="cannot be false"):
_resolve_supports(config)
def test_two_point_derived_from_set_actions() -> None:
config: ConfigType = {
CONF_SET_TARGET_TEMPERATURE_LOW_ACTION: [{}],
CONF_SET_TARGET_TEMPERATURE_HIGH_ACTION: [{}],
}
assert _resolve_supports(config)[CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE] is True
def test_two_point_false_with_set_action_rejected() -> None:
config: ConfigType = {
CONF_SET_TARGET_TEMPERATURE_LOW_ACTION: [{}],
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE: False,
}
with pytest.raises(cv.Invalid, match="cannot be false"):
_resolve_supports(config)
def test_target_humidity_derived_from_set_action() -> None:
config: ConfigType = {CONF_SET_TARGET_HUMIDITY_ACTION: [{}]}
assert _resolve_supports(config)[CONF_SUPPORTS_TARGET_HUMIDITY] is True
def test_set_target_temperature_low_requires_high() -> None:
config: ConfigType = {CONF_SET_TARGET_TEMPERATURE_LOW_ACTION: [{}]}
with pytest.raises(cv.Invalid, match="must be used together"):
_validate_set_actions(config)
def test_set_target_temperature_conflicts_with_two_point_actions() -> None:
config: ConfigType = {
CONF_SET_TARGET_TEMPERATURE_ACTION: [{}],
CONF_SET_TARGET_TEMPERATURE_LOW_ACTION: [{}],
CONF_SET_TARGET_TEMPERATURE_HIGH_ACTION: [{}],
}
with pytest.raises(cv.Invalid, match="cannot be used together"):
_validate_set_actions(config)
def test_initial_state_target_temperature_rejected_with_two_point() -> None:
config: ConfigType = {
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE: True,
CONF_SUPPORTS_TARGET_HUMIDITY: False,
CONF_INITIAL_STATE: {CONF_TARGET_TEMPERATURE: 21.0},
}
with pytest.raises(cv.Invalid, match="is not available"):
_validate_initial_state(config)
def test_initial_state_two_point_values_rejected_without_two_point() -> None:
config: ConfigType = {
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE: False,
CONF_SUPPORTS_TARGET_HUMIDITY: False,
CONF_INITIAL_STATE: {
CONF_TARGET_TEMPERATURE_LOW: 18.0,
CONF_TARGET_TEMPERATURE_HIGH: 24.0,
},
}
with pytest.raises(cv.Invalid, match="requires"):
_validate_initial_state(config)
def test_initial_state_target_humidity_rejected_without_support() -> None:
config: ConfigType = {
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE: False,
CONF_SUPPORTS_TARGET_HUMIDITY: False,
CONF_INITIAL_STATE: {CONF_TARGET_HUMIDITY: 50},
}
with pytest.raises(cv.Invalid, match="requires"):
_validate_initial_state(config)
def test_initial_state_matching_two_point_accepted() -> None:
config: ConfigType = {
CONF_SUPPORTS_TWO_POINT_TARGET_TEMPERATURE: True,
CONF_SUPPORTS_TARGET_HUMIDITY: True,
CONF_INITIAL_STATE: {
CONF_TARGET_TEMPERATURE_LOW: 18.0,
CONF_TARGET_TEMPERATURE_HIGH: 24.0,
CONF_TARGET_HUMIDITY: 50,
},
}
assert _validate_initial_state(config) is config
+5
View File
@@ -0,0 +1,5 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.dependencies = manifest.dependencies + ["sensor", "spi"]
@@ -0,0 +1,62 @@
#include <gtest/gtest.h>
#include "esphome/components/atm90e32/atm90e32.h"
namespace esphome::atm90e32::testing {
TEST(ATM90E32OffsetRegisterVerification, AcceptsExactSignedReadback) {
EXPECT_TRUE(offset_register_value_matches(0x007B, 123));
EXPECT_TRUE(offset_register_value_matches(0xFF85, -123));
}
TEST(ATM90E32OffsetRegisterVerification, RejectsMismatchedReadback) {
EXPECT_FALSE(offset_register_value_matches(0x007C, 123));
EXPECT_FALSE(offset_register_value_matches(0xFF84, -123));
}
TEST(ATM90E32OffsetRestoreState, ReportsVerifiedStoredValuesAsRestored) {
const auto state = resolve_offset_restore_state(true, true, false);
EXPECT_TRUE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsVerifiedConfigFallbackAsNotRestored) {
const auto state = resolve_offset_restore_state(true, false, true);
EXPECT_FALSE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsFailedConfigFallbackAsUnverified) {
const auto state = resolve_offset_restore_state(true, false, false);
EXPECT_FALSE(state.restored);
EXPECT_FALSE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsConfigWithoutStoredValuesAsNotRestored) {
const auto state = resolve_offset_restore_state(false, true, false);
EXPECT_FALSE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetPersistence, RollsBackStoredValuesOrZeroSentinel) {
const OffsetCalibration previous[3]{{1, -1}, {2, -2}, {3, -3}};
OffsetCalibration rollback[3]{};
prepare_offset_rollback(previous, true, rollback);
for (uint8_t phase = 0; phase < 3; phase++) {
EXPECT_EQ(rollback[phase].first_offset, previous[phase].first_offset);
EXPECT_EQ(rollback[phase].second_offset, previous[phase].second_offset);
}
prepare_offset_rollback(previous, false, rollback);
for (const auto &phase : rollback) {
EXPECT_EQ(phase.first_offset, 0);
EXPECT_EQ(phase.second_offset, 0);
}
}
} // namespace esphome::atm90e32::testing
+1 -2
View File
@@ -30,8 +30,7 @@ climate:
- switch.turn_on: climate_heater_switch
- switch.turn_off: climate_cooler_switch
# Thermostat-based climate so climate.control: action variants get build
# coverage (bang_bang doesn't support fan modes, presets, etc.). Climate
# has no template platform, so thermostat is the right vehicle.
# coverage (bang_bang doesn't support fan modes, presets, etc.).
- platform: thermostat
id: climate_test_thermostat
name: Test Thermostat
@@ -0,0 +1,5 @@
# Exercises the ESP-NOW-over-hosted shim: on the ESP32-P4 host, esp32_hosted
# supplies the esp_now_* symbols that the espnow component links against.
packages:
esp32_hosted: !include common.yaml
espnow: !include ../espnow/common.yaml
@@ -68,6 +68,14 @@ class Initiator {
static const uint8_t PROLOGUE[] = {'t', 'e', 's', 't', 'p', 'r', 'o', 'l', 'o', 'g', 'u', 'e'};
// The context only points at the key and init() copies it before returning,
// so a temporary context over a temporary key is safe within one call
static NoiseContext ctx_for(const psk_t &psk) {
NoiseContext ctx;
ctx.set_psk(psk.data());
return ctx;
}
static psk_t make_psk(uint8_t seed) {
psk_t psk;
for (size_t i = 0; i < psk.size(); i++) {
@@ -102,7 +110,7 @@ TEST(NoiseResponderHandshakeTest, MessageMethodsErrorBeforeInit) {
TEST(NoiseResponderHandshakeTest, FullHandshakeAndTransportRoundTrip) {
const psk_t psk = make_psk(7);
NoiseResponderHandshake responder;
ASSERT_EQ(responder.init(psk, PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(ctx_for(psk), PROLOGUE, sizeof(PROLOGUE)), 0);
EXPECT_EQ(responder.action(), Action::ACTION_READ);
Initiator initiator(psk, PROLOGUE, sizeof(PROLOGUE));
@@ -155,8 +163,8 @@ TEST(NoiseResponderHandshakeTest, ReInitRestartsHandshake) {
// proves the restart took effect; the old state surviving would fail the
// MAC here.
NoiseResponderHandshake responder;
ASSERT_EQ(responder.init(make_psk(7), PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(make_psk(9), PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(ctx_for(make_psk(7)), PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(ctx_for(make_psk(9)), PROLOGUE, sizeof(PROLOGUE)), 0);
EXPECT_EQ(responder.action(), Action::ACTION_READ);
Initiator initiator(make_psk(9), PROLOGUE, sizeof(PROLOGUE));
@@ -168,7 +176,7 @@ TEST(NoiseResponderHandshakeTest, ReInitRestartsHandshake) {
TEST(NoiseResponderHandshakeTest, WrongPskFailsWithMacFailure) {
NoiseResponderHandshake responder;
ASSERT_EQ(responder.init(make_psk(7), PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(ctx_for(make_psk(7)), PROLOGUE, sizeof(PROLOGUE)), 0);
Initiator initiator(make_psk(200), PROLOGUE, sizeof(PROLOGUE));
uint8_t msg[MAX_HANDSHAKE_SIZE];
@@ -185,7 +193,7 @@ TEST(NoiseResponderHandshakeTest, MismatchedPrologueFailsWithMacFailure) {
// tampered preamble must fail even with the right key.
const psk_t psk = make_psk(7);
NoiseResponderHandshake responder;
ASSERT_EQ(responder.init(psk, PROLOGUE, sizeof(PROLOGUE)), 0);
ASSERT_EQ(responder.init(ctx_for(psk), PROLOGUE, sizeof(PROLOGUE)), 0);
static const uint8_t TAMPERED[] = {'x'};
Initiator initiator(psk, TAMPERED, sizeof(TAMPERED));
@@ -17,12 +17,17 @@ TEST(NoiseContextTest, AllZerosPskIsReserved) {
EXPECT_FALSE(NoiseContext::is_all_zeros(psk));
NoiseContext ctx;
psk_t loaded;
EXPECT_FALSE(ctx.has_psk());
ctx.set_psk(zeros);
EXPECT_FALSE(ctx.has_psk());
ctx.set_psk(psk);
ctx.load_psk(loaded);
EXPECT_EQ(loaded, zeros);
ctx.set_psk(psk.data());
EXPECT_TRUE(ctx.has_psk());
EXPECT_EQ(ctx.get_psk(), psk);
ctx.load_psk(loaded);
EXPECT_EQ(loaded, psk);
// Callers map the reserved key to nullptr; the context just stores what it is given
ctx.set_psk(nullptr);
EXPECT_FALSE(ctx.has_psk());
}
TEST(WireFormatTest, FrameHeaderIsIndicatorPlusBigEndianLength) {
+12
View File
@@ -0,0 +1,12 @@
wifi:
ssid: MySSID
password: password1
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
port: 3290
password: "superlongpasswordthatnoonewillknow"
+10
View File
@@ -0,0 +1,10 @@
wifi:
ssid: MySSID
password: password1
api:
encryption:
ota:
- platform: esphome
port: 3291
@@ -0,0 +1,2 @@
packages:
ota: !include api_key_offer.yaml
@@ -0,0 +1,2 @@
packages:
ota: !include api_key_offer.yaml
@@ -0,0 +1,2 @@
packages:
ota: !include api_runtime_key.yaml
@@ -0,0 +1,2 @@
packages:
ota: !include api_runtime_key.yaml
+113
View File
@@ -25,6 +25,27 @@ esphome:
away: !lambda "return true;"
is_on: !lambda "return false;"
- climate.template.publish:
id: template_climate
current_temperature: 21.0
mode: HEAT
fan_mode: AUTO
swing_mode: "OFF"
preset: NONE
target_temperature: 22.0
# Templated
- climate.template.publish:
id: template_climate
current_temperature: !lambda "return 21.5f;"
mode: !lambda "return climate::CLIMATE_MODE_COOL;"
target_temperature: !lambda "return 23.0f;"
- climate.template.publish:
id: template_climate_custom_modes
custom_fan_mode: "turbo"
custom_preset: "eco_plus"
# Test C++ API: set_template() with stateless lambda (no captures)
# NOTE: set_template() is not intended to be a public API, but we test it to ensure it doesn't break.
- lambda: |-
@@ -513,6 +534,98 @@ alarm_control_panel:
codes:
- "1234"
climate:
- platform: template
id: template_climate
name: "Template Climate"
optimistic: true
sensor: template_template_sens
supports_action: true
supports_current_humidity: true
restore_mode: NO_RESTORE
initial_state:
mode: HEAT
target_temperature: 21.0
fan_mode: LOW
supported_modes:
- "OFF"
- HEAT
- COOL
supported_fan_modes:
- AUTO
- LOW
- HIGH
supported_swing_modes:
- "OFF"
- VERTICAL
supported_presets:
- NONE
- ECO
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
set_mode_action:
- logger.log:
format: "set_mode_action %d"
args: ["(int) x"]
set_target_temperature_action:
- logger.log:
format: "set_target_temperature_action %.1f"
args: ["x"]
set_target_humidity_action:
- logger.log:
format: "set_target_humidity_action %.1f"
args: ["x"]
set_fan_mode_action:
- logger.log:
format: "set_fan_mode_action %d"
args: ["(int) x"]
set_swing_mode_action:
- logger.log:
format: "set_swing_mode_action %d"
args: ["(int) x"]
set_preset_action:
- logger.log:
format: "set_preset_action %d"
args: ["(int) x"]
on_control:
- logger.log: "on_control fired"
on_state:
- logger.log: "on_state fired"
- platform: template
id: template_climate_custom_modes
name: "Template Climate Custom Modes"
optimistic: true
sensor: template_template_sens
supported_modes:
- "OFF"
- HEAT
custom_fan_modes:
- turbo
- silent
- eco
custom_presets:
- eco_plus
- power_save
- max
set_custom_fan_mode_action:
- logger.log:
format: "set_custom_fan_mode_action %s"
args: ["x.c_str()"]
set_custom_preset_action:
- logger.log:
format: "set_custom_preset_action %s"
args: ["x.c_str()"]
initial_state:
custom_fan_mode: eco
custom_preset: max
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
water_heater:
- platform: template
id: template_water_heater
@@ -0,0 +1,29 @@
# Tuya without any network component (no wifi/ethernet/api), as used on
# serial-only or BLE-only Tuya MCU boards. Regression test for
# https://github.com/esphome/esphome/issues/18942
substitutions:
status_pin: P6
packages:
uart: !include ../../test_build_components/common/uart/bk72xx-ard.yaml
tuya:
status_pin: ${status_pin}
binary_sensor:
- platform: tuya
id: tuya_presence
sensor_datapoint: 101
sensor:
- platform: tuya
id: tuya_light_intensity
sensor_datapoint: 103
number:
- platform: tuya
id: tuya_far_detection
number_datapoint: 109
min_value: 0
max_value: 600
step: 1
+1
View File
@@ -14,6 +14,7 @@ esphome:
condition: wifi.ap_active
then:
- logger.log: "WiFi AP is active!"
- wifi.roam
wifi:
networks:
+7
View File
@@ -162,6 +162,13 @@ def integration_test_dir() -> Generator[Path]:
yield Path(tmpdir)
@pytest.fixture
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Host preferences persist per device name; give the test its own so a
provisioned key never leaks into another run."""
monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
@pytest.fixture
def reserved_tcp_port() -> Generator[tuple[int, socket.socket]]:
"""Reserve an unused TCP port by holding the socket open."""
+7
View File
@@ -9,6 +9,13 @@ API_CONNECTION_TIMEOUT = 30.0 # seconds
PORT_WAIT_TIMEOUT = 30.0 # seconds
PORT_POLL_INTERVAL = 0.1 # seconds
# The well-known all-zeros provisioning PSK, a key to provision over it, and
# the time the device takes to activate a newly saved key (100 ms timer plus
# margin)
ZERO_PSK = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
PROVISIONING_PSK = b"bm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm5ubm4="
KEY_ACTIVATION_DELAY = 0.5 # seconds
# Process shutdown timeouts
SIGINT_TIMEOUT = 5.0 # seconds
SIGTERM_TIMEOUT = 2.0 # seconds
@@ -0,0 +1,12 @@
esphome:
name: host-ota-test
host:
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
port: __OTA_PORT__
password: "hunter2"
logger:
level: DEBUG
@@ -0,0 +1,10 @@
esphome:
name: host-ota-test
host:
api:
encryption:
ota:
- platform: esphome
port: __OTA_PORT__
logger:
level: DEBUG
@@ -0,0 +1,72 @@
esphome:
name: tmpl-clim-basic
on_boot:
- climate.template.publish:
id: test_climate
action: IDLE
host:
api:
logger:
climate:
- platform: template
id: test_climate
name: Test Basic Climate
optimistic: true
sensor: test_climate_current_temperature
humidity_sensor: test_climate_current_humidity
supports_action: true
supported_modes:
- "OFF"
- HEAT
- COOL
supported_fan_modes:
- AUTO
- LOW
- HIGH
supported_swing_modes:
- "OFF"
- VERTICAL
supported_presets:
- NONE
- ECO
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
on_control:
- lambda: |-
if (x.get_mode().has_value())
ESP_LOGD("test", "on_control mode=%d", (int) *x.get_mode());
if (x.get_target_temperature().has_value())
ESP_LOGD("test", "on_control target_temperature=%.1f", *x.get_target_temperature());
if (x.get_fan_mode().has_value())
ESP_LOGD("test", "on_control fan_mode=%d", (int) *x.get_fan_mode());
if (x.get_swing_mode().has_value())
ESP_LOGD("test", "on_control swing_mode=%d", (int) *x.get_swing_mode());
if (x.get_preset().has_value())
ESP_LOGD("test", "on_control preset=%d", (int) *x.get_preset());
sensor:
- platform: template
id: test_climate_current_temperature
name: Test Climate Current Temperature
lambda: "return 22.5f;"
update_interval: 10ms
- platform: template
id: test_climate_current_humidity
name: Test Climate Current Humidity
lambda: "return 55.0f;"
update_interval: 10ms
button:
- platform: template
id: simulate_device_report
name: Simulate Device Report
on_press:
- climate.template.publish:
id: test_climate
mode: "OFF"
fan_mode: AUTO
swing_mode: "OFF"
preset: NONE
@@ -0,0 +1,47 @@
esphome:
name: tmpl-clim-custom
host:
api:
logger:
climate:
- platform: template
id: test_climate
name: Test Custom Mode Climate
optimistic: true
sensor: test_climate_current_temperature
supported_modes:
- "OFF"
- HEAT
- COOL
custom_fan_modes:
- turbo
- silent
- eco
custom_presets:
- eco_plus
- power_save
- max
on_control:
- lambda: |-
if (x.has_custom_fan_mode())
ESP_LOGD("test", "on_control custom_fan_mode=%s", x.get_custom_fan_mode().c_str());
if (x.has_custom_preset())
ESP_LOGD("test", "on_control custom_preset=%s", x.get_custom_preset().c_str());
sensor:
- platform: template
id: test_climate_current_temperature
name: Test Climate Current Temperature
lambda: "return 22.5f;"
update_interval: 10ms
button:
- platform: template
id: simulate_device_report
name: Simulate Device Report
on_press:
- climate.template.publish:
id: test_climate
custom_fan_mode: "eco"
custom_preset: "max"
@@ -0,0 +1,56 @@
esphome:
name: tmpl-clim-nonopt
host:
api:
logger:
climate:
- platform: template
id: test_climate
name: Test Template Climate Nonoptimistic
optimistic: false
supported_modes:
- "OFF"
- HEAT
- COOL
- FAN_ONLY
supported_fan_modes:
- AUTO
- LOW
- HIGH
supported_swing_modes:
- "OFF"
- VERTICAL
supported_presets:
- NONE
- ECO
- AWAY
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
on_control:
- lambda: |-
if (x.get_mode().has_value())
ESP_LOGD("test", "on_control mode=%d", (int) *x.get_mode());
if (x.get_target_temperature().has_value())
ESP_LOGD("test", "on_control target_temperature=%.1f", *x.get_target_temperature());
if (x.get_fan_mode().has_value())
ESP_LOGD("test", "on_control fan_mode=%d", (int) *x.get_fan_mode());
if (x.get_swing_mode().has_value())
ESP_LOGD("test", "on_control swing_mode=%d", (int) *x.get_swing_mode());
if (x.get_preset().has_value())
ESP_LOGD("test", "on_control preset=%d", (int) *x.get_preset());
button:
- platform: template
id: simulate_device_confirmation
name: Simulate Device Confirmation
on_press:
- climate.template.publish:
id: test_climate
mode: HEAT
target_temperature: 22.5
fan_mode: HIGH
swing_mode: VERTICAL
preset: AWAY
@@ -0,0 +1,26 @@
esphome:
name: tmpl-clim-oc-order
host:
api:
logger:
# on_control fires with the full ClimateCall (arg `x`) from the base Climate component's
# ClimateCall::perform(), before validate_()/control() run -- so when the lambda action below
# runs, the entity's own .mode is still the OLD value, even though x.get_mode() already reports
# the NEW requested value. on_state fires afterward, once control() has applied it.
climate:
- platform: template
id: test_climate
name: Test On Control Ordering
optimistic: true
supported_modes:
- "OFF"
- HEAT
on_control:
- lambda: |-
ESP_LOGD("test", "on_control requested_mode=%d current_mode_before_apply=%d",
x.get_mode().has_value() ? (int) *x.get_mode() : -1,
(int) id(test_climate).mode);
on_state:
- lambda: |-
ESP_LOGD("test", "on_state mode=%d", (int) x.mode);
@@ -0,0 +1,63 @@
esphome:
name: tmpl-clim-publish-all
host:
api:
logger:
climate:
- platform: template
id: test_climate
name: Test Publish All Fields
optimistic: true
# current_temperature/current_humidity/action are only sent over the API at all if their
# trait is advertised: current_temperature/current_humidity because a sensor/humidity_sensor
# is referenced below, action because supports_action is set. The sensors' fixed readings
# match what climate.template.publish pushes, so the sensor callback (guarded to only publish
# on an actual change) doesn't produce an extra, unexpected state update of its own.
sensor: test_climate_current_temperature
humidity_sensor: test_climate_current_humidity
supports_action: true
supported_modes:
- "OFF"
- HEAT
supported_fan_modes:
- AUTO
- HIGH
supported_swing_modes:
- "OFF"
- VERTICAL
supported_presets:
- NONE
- ECO
on_control:
# Should never fire in this test: climate.template.publish is a pure bypass and must not
# re-trigger on_control as if the entity were freshly commanded.
- logger.log: "on_control fired"
sensor:
- platform: template
id: test_climate_current_temperature
name: Test Climate Current Temperature
lambda: "return 20.0f;"
update_interval: 10ms
- platform: template
id: test_climate_current_humidity
name: Test Climate Current Humidity
lambda: "return 60.0f;"
update_interval: 10ms
button:
- platform: template
id: publish_all
name: Publish All
on_press:
- climate.template.publish:
id: test_climate
current_temperature: 20.0
current_humidity: 60.0
target_temperature: 23.0
mode: HEAT
action: HEATING
fan_mode: HIGH
swing_mode: VERTICAL
preset: ECO
@@ -0,0 +1,49 @@
esphome:
name: tmpl-clim-sensor-push
host:
api:
logger:
# No lambda/update_interval: these sensors only ever report a value when a button below
# publishes one (standing in for e.g. a BLE scan callback in a real config).
sensor:
- platform: template
id: room_temperature
name: Room Temperature
- platform: template
id: room_humidity
name: Room Humidity
climate:
- platform: template
id: test_climate
name: Test Sensor Push Climate
optimistic: true
sensor: room_temperature
humidity_sensor: room_humidity
supported_modes:
- "OFF"
- HEAT
button:
- platform: template
id: publish_temperature
name: Publish Temperature
on_press:
- sensor.template.publish:
id: room_temperature
state: 24.0
- platform: template
id: publish_temperature_same
name: Publish Temperature Same Value
on_press:
- sensor.template.publish:
id: room_temperature
state: 24.0
- platform: template
id: publish_humidity
name: Publish Humidity
on_press:
- sensor.template.publish:
id: room_humidity
state: 65.0
@@ -0,0 +1,89 @@
esphome:
name: tmpl-clim-set-act
host:
api:
logger:
# Every settable field forwards its requested value to a set_*_action. supports_two_point and
# supports_target_humidity are not declared here: they are derived from the low/high and humidity
# set actions being present.
climate:
- platform: template
id: test_climate
name: Test Set Actions
optimistic: false
restore_mode: NO_RESTORE
supported_modes:
- "OFF"
- HEAT
- COOL
supported_fan_modes:
- AUTO
- LOW
supported_swing_modes:
- "OFF"
- VERTICAL
supported_presets:
- NONE
- ECO
custom_fan_modes:
- turbo
custom_presets:
- eco_plus
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
set_mode_action:
- logger.log:
format: "set_mode_action %d"
args: ["(int) x"]
set_target_temperature_low_action:
- logger.log:
format: "set_target_temperature_low_action %.1f"
args: ["x"]
set_target_temperature_high_action:
- logger.log:
format: "set_target_temperature_high_action %.1f"
args: ["x"]
set_target_humidity_action:
- logger.log:
format: "set_target_humidity_action %.0f"
args: ["x"]
set_fan_mode_action:
- logger.log:
format: "set_fan_mode_action %d"
args: ["(int) x"]
set_custom_fan_mode_action:
- logger.log:
format: "set_custom_fan_mode_action %s"
args: ["x.c_str()"]
set_swing_mode_action:
- logger.log:
format: "set_swing_mode_action %d"
args: ["(int) x"]
set_preset_action:
- logger.log:
format: "set_preset_action %d"
args: ["(int) x"]
set_custom_preset_action:
- logger.log:
format: "set_custom_preset_action %s"
args: ["x.c_str()"]
button:
- platform: template
id: report_device_state
name: Report Device State
on_press:
- climate.template.publish:
id: test_climate
mode: HEAT
- platform: template
id: report_unsupported_mode
name: Report Unsupported Mode
on_press:
- climate.template.publish:
id: test_climate
mode: DRY
@@ -0,0 +1,52 @@
esphome:
name: tmpl-clim-two-point
host:
api:
logger:
climate:
- platform: template
id: test_climate
name: Test Two-Point Heatpump
optimistic: true
sensor: test_climate_current_temperature
supports_two_point_target_temperature: true
supports_target_humidity: true
supported_modes:
- "OFF"
- HEAT_COOL
- HEAT
- COOL
visual:
min_temperature: 16.0
max_temperature: 30.0
temperature_step: 0.5
on_control:
- lambda: |-
if (x.get_mode().has_value())
ESP_LOGD("test", "on_control mode=%d", (int) *x.get_mode());
if (x.get_target_temperature_low().has_value())
ESP_LOGD("test", "on_control target_temperature_low=%.1f", *x.get_target_temperature_low());
if (x.get_target_temperature_high().has_value())
ESP_LOGD("test", "on_control target_temperature_high=%.1f", *x.get_target_temperature_high());
if (x.get_target_humidity().has_value())
ESP_LOGD("test", "on_control target_humidity=%.1f", *x.get_target_humidity());
sensor:
- platform: template
id: test_climate_current_temperature
name: Test Climate Current Temperature
lambda: "return 21.0f;"
update_interval: 10ms
button:
- platform: template
id: simulate_device_report
name: Simulate Device Report
on_press:
- climate.template.publish:
id: test_climate
mode: HEAT_COOL
target_temperature_low: 18.0
target_temperature_high: 24.0
target_humidity: 50.0
@@ -10,34 +10,40 @@ from __future__ import annotations
import asyncio
import base64
import socket
from aioesphomeapi import InvalidEncryptionKeyAPIError, RequiresEncryptionAPIError
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
from .conftest import run_binary_and_wait_for_port
from .const import KEY_ACTIVATION_DELAY, LOCALHOST, PROVISIONING_PSK, ZERO_PSK
from .types import (
APIClientConnectedFactory,
CompileFunction,
ConfigWriter,
RunCompiledFunction,
)
# The well-known provisioning PSK: base64 of 32 zero bytes
ZERO_PSK = base64.b64encode(bytes(32)).decode()
# A real key to provision
NEW_KEY = base64.b64encode(b"n" * 32)
# Time for the device to activate a newly saved key (100ms timer plus margin)
KEY_ACTIVATION_DELAY = 0.5
@pytest.fixture(autouse=True)
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
"""Keep host preferences per-test so every run starts unprovisioned."""
monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
pytestmark = pytest.mark.usefixtures("isolated_preferences")
NEW_KEY = PROVISIONING_PSK
@pytest.mark.asyncio
async def test_api_zero_psk_provisioning(
yaml_config: str,
run_compiled: RunCompiledFunction,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
reserved_tcp_port: tuple[int, socket.socket],
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Exercise the reject paths, then provision a key over the zero-PSK channel."""
async with run_compiled(yaml_config):
"""Exercise the reject paths, provision a key over the zero-PSK channel,
and check the key comes back from preferences on the next boot."""
port, port_socket = reserved_tcp_port
config_path = await write_yaml_config(yaml_config)
binary_path = await compile_esphome(config_path)
port_socket.close()
async with run_binary_and_wait_for_port(binary_path, LOCALHOST, port):
# --- Pre-provisioning reject paths (device state is unchanged) ---
# A wrong (non-zero) PSK fails against the zero provisioning PSK
@@ -97,6 +103,19 @@ async def test_api_zero_psk_provisioning(
async with api_client_connected(timeout=5) as client:
await client.device_info()
# The key is loaded from preferences on the next boot
lines: list[str] = []
async with run_binary_and_wait_for_port(
binary_path, LOCALHOST, port, line_callback=lines.append
):
async with api_client_connected(noise_psk=NEW_KEY.decode()) as client:
device_info = await client.device_info()
assert device_info.api_encryption_provisionable is False
with pytest.raises(InvalidEncryptionKeyAPIError):
async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client:
await client.device_info()
assert any("Loaded saved Noise PSK" in line for line in lines)
@pytest.mark.asyncio
async def test_api_zero_psk_provisioning_plaintext(
+258 -115
View File
@@ -8,9 +8,12 @@ instance covers the FD_CLOEXEC path.
from __future__ import annotations
import asyncio
import base64
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass
import functools
from pathlib import Path
import socket
import pytest
@@ -18,10 +21,18 @@ import pytest
from esphome import espota2
from .conftest import run_binary, wait_and_connect_api_client
from .const import LOCALHOST, PORT_POLL_INTERVAL, PORT_WAIT_TIMEOUT
from .types import CompileFunction, ConfigWriter
from .const import (
KEY_ACTIVATION_DELAY,
LOCALHOST,
PORT_POLL_INTERVAL,
PORT_WAIT_TIMEOUT,
PROVISIONING_PSK,
ZERO_PSK,
)
from .types import APIClientConnectedFactory, CompileFunction, ConfigWriter
DEVICE_NAME = "host-ota-test"
API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
@contextmanager
@@ -35,6 +46,14 @@ def _reserve_port() -> Generator[tuple[int, socket.socket]]:
s.close()
async def _wait_for_line(lines: list[str], needle: str, timeout: float = 5.0) -> None:
"""The config dump prints after every setup, a little after the api port
opens, so wait for it rather than assert on the lines seen so far."""
async with asyncio.timeout(timeout):
while not any(needle in line for line in lines):
await asyncio.sleep(PORT_POLL_INTERVAL)
async def _wait_for_port(host: str, port: int, timeout: float) -> None:
"""Poll until a TCP port accepts connections, or raise TimeoutError."""
loop = asyncio.get_running_loop()
@@ -51,6 +70,102 @@ async def _wait_for_port(host: str, port: int, timeout: float) -> None:
raise TimeoutError(f"Port {port} on {host} did not open within {timeout}s")
async def _build(
yaml_config: str,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
reserved_tcp_port: tuple[int, socket.socket],
) -> tuple[int, int, Path]:
"""Reserve an OTA port, compile the fixture with it, and release both
ports right before the binary is started."""
api_port, api_socket = reserved_tcp_port
with _reserve_port() as (ota_port, ota_socket):
config_path = await write_yaml_config(
yaml_config.replace("__OTA_PORT__", str(ota_port))
)
binary_path = await compile_esphome(config_path)
api_socket.close()
ota_socket.close()
return api_port, ota_port, binary_path
async def _run_ota(
ota_port: int,
password: str | None,
binary_path: Path,
noise_psk: str | None,
plaintext_fallback: bool = False,
) -> int:
"""espota2 is blocking; run it in the executor and return its exit code."""
rc, _ = await asyncio.get_running_loop().run_in_executor(
None,
functools.partial(
espota2.run_ota,
LOCALHOST,
ota_port,
password,
binary_path,
noise_psk=noise_psk,
plaintext_fallback=plaintext_fallback,
),
)
return rc
@dataclass
class _Device:
"""A running host binary and the checks every successful OTA repeats:
a safe reboot, the api port back up, and the pid preserved by execv."""
api_port: int
ota_port: int
binary_path: Path
proc: asyncio.subprocess.Process | None = None
reboots: int = 0
def __post_init__(self) -> None:
self._rebooted = asyncio.Event()
def on_log(self, line: str) -> None:
if "Rebooting safely" in line:
self.reboots += 1
self._rebooted.set()
async def wait_reboot(self, count: int, timeout: float = 10.0) -> None:
async with asyncio.timeout(timeout):
while self.reboots < count:
self._rebooted.clear()
await self._rebooted.wait()
async def ota(
self,
password: str | None,
noise_psk: str | None,
msg: str,
plaintext_fallback: bool = False,
) -> None:
"""Upload, then expect the re-exec with the pid preserved."""
pid_before = self.proc.pid
expected_reboots = self.reboots + 1
rc = await _run_ota(
self.ota_port, password, self.binary_path, noise_psk, plaintext_fallback
)
assert rc == 0, msg
await self.wait_reboot(expected_reboots)
await _wait_for_port(LOCALHOST, self.api_port, PORT_WAIT_TIMEOUT)
assert self.proc.returncode is None, "process exited instead of execing"
assert self.proc.pid == pid_before
async def refused_ota(
self, password: str | None, noise_psk: str | None, msg: str
) -> None:
"""Upload must fail and the device must keep running."""
rc = await _run_ota(self.ota_port, password, self.binary_path, noise_psk)
assert rc == 1, msg
await asyncio.sleep(0.5)
assert self.proc.returncode is None, "process died on rejected OTA"
@pytest.mark.asyncio
async def test_host_ota_self_update(
yaml_config: str,
@@ -59,57 +174,34 @@ async def test_host_ota_self_update(
reserved_tcp_port: tuple[int, socket.socket],
) -> None:
"""Self-OTA: upload the running binary back to itself, expect re-exec."""
api_port, api_socket = reserved_tcp_port
with _reserve_port() as (ota_port, ota_socket):
yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
config_path = await write_yaml_config(yaml_config)
binary_path = await compile_esphome(config_path)
api_socket.close()
ota_socket.close()
dev = _Device(
*await _build(
yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
)
)
staged = asyncio.Event()
loop = asyncio.get_running_loop()
ota_staged = loop.create_future()
rebooted = loop.create_future()
def on_log(line: str) -> None:
if "OTA staged at" in line:
staged.set()
dev.on_log(line)
def on_log(line: str) -> None:
if not ota_staged.done() and "OTA staged at" in line:
ota_staged.set_result(True)
if not rebooted.done() and "Rebooting safely" in line:
rebooted.set_result(True)
async with run_binary(dev.binary_path, line_callback=on_log) as (proc, _lines):
dev.proc = proc
await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
async with wait_and_connect_api_client(port=dev.api_port) as client:
info_before = await client.device_info()
assert info_before.name == DEVICE_NAME
async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
pid_before = proc.pid
async with wait_and_connect_api_client(port=api_port) as client:
info_before = await client.device_info()
assert info_before.name == DEVICE_NAME
await dev.ota(None, None, "espota2 reported failure")
assert staged.is_set()
# espota2 is blocking; run in executor.
rc, _ = await loop.run_in_executor(
None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
)
assert rc == 0, "espota2 reported failure"
async with wait_and_connect_api_client(port=dev.api_port) as client:
info_after = await client.device_info()
assert info_after.name == info_before.name
await asyncio.wait_for(ota_staged, timeout=10.0)
await asyncio.wait_for(rebooted, timeout=10.0)
await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
# execv preserves pid; mismatch means external respawn.
assert proc.returncode is None, "process exited instead of execing"
assert proc.pid == pid_before
async with wait_and_connect_api_client(port=api_port) as client:
info_after = await client.device_info()
assert info_after.name == DEVICE_NAME
assert info_after.name == info_before.name
# Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
rc, _ = await loop.run_in_executor(
None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
)
assert rc == 0, "second OTA failed -- listener leaked across execv"
await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
assert proc.pid == pid_before
# Second OTA: catches FD_CLOEXEC regressions (EADDRINUSE on rebind).
await dev.ota(None, None, "second OTA failed -- listener leaked across execv")
@pytest.mark.asyncio
@@ -121,51 +213,110 @@ async def test_host_ota_encrypted(
) -> 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()
dev = _Device(
*await _build(
yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
)
)
async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, _lines):
dev.proc = proc
await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
await dev.refused_ota(
None, None, "plaintext upload to an encrypted device must fail"
)
await dev.ota(None, API_KEY, "encrypted OTA reported failure")
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)
@pytest.mark.asyncio
async def test_host_ota_api_key_offer_with_password(
yaml_config: str,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
reserved_tcp_port: tuple[int, socket.socket],
caplog: pytest.LogCaptureFixture,
) -> None:
"""With only an api key the device offers encryption without requiring
it: the password still guards plaintext uploads, the key alone
authenticates an encrypted one, and until 2027.3.0 a failed encrypted
attempt falls back to plaintext."""
pytest.importorskip("aioesphomeapi.noise")
wrong_key = base64.b64encode(b"w" * 32).decode()
dev = _Device(
*await _build(
yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
)
)
async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, lines):
dev.proc = proc
await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
await _wait_for_line(lines, "Encryption: offered")
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
await dev.refused_ota(
None, None, "plaintext upload without the password must fail"
)
await dev.ota(
"hunter2", None, "plaintext upload with the password must succeed"
)
await dev.ota(None, API_KEY, "encrypted upload with the api key must succeed")
# 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
# Remove before 2027.3.0: a wrong key falls back to plaintext, which
# the password still guards
with caplog.at_level("WARNING", logger="esphome.espota2"):
await dev.ota(
"hunter2",
wrong_key,
"the plaintext retry with the password must succeed",
plaintext_fallback=True,
)
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"
assert any("Retrying in plaintext" in r.message for r in caplog.records)
await dev.ota(
None,
API_KEY,
"the right api key encrypts without touching the fallback",
plaintext_fallback=True,
)
# 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
@pytest.mark.usefixtures("isolated_preferences")
async def test_host_ota_provisioned_api_key(
yaml_config: str,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
reserved_tcp_port: tuple[int, socket.socket],
api_client_connected: APIClientConnectedFactory,
) -> None:
"""A key provisioned over the api feeds the OTA offer: plaintext works
while unprovisioned, the provisioned key encrypts, the key loaded from
preferences on the next boot keeps encrypting, and plaintext stays
accepted because only the ota block requires encryption."""
pytest.importorskip("aioesphomeapi.noise")
dev = _Device(
*await _build(
yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
)
)
async with run_binary(dev.binary_path, line_callback=dev.on_log) as (proc, lines):
dev.proc = proc
await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
await _wait_for_line(lines, "once the api key is provisioned")
await dev.ota(
None, None, "plaintext upload to an unprovisioned device must succeed"
)
async with api_client_connected(
port=dev.api_port, noise_psk=ZERO_PSK
) as client:
assert await client.noise_encryption_set_key(PROVISIONING_PSK) is True
await asyncio.sleep(KEY_ACTIVATION_DELAY)
key = PROVISIONING_PSK.decode()
await dev.ota(
None, key, "encrypted upload with the provisioned key must succeed"
)
await dev.ota(None, key, "the key loaded at boot must feed the OTA offer")
await dev.ota(None, None, "plaintext must stay accepted on an offering device")
@pytest.mark.asyncio
@@ -177,33 +328,25 @@ async def test_host_ota_rejects_garbage(
integration_test_dir,
) -> None:
"""Bogus payload is rejected and the device keeps running."""
api_port, api_socket = reserved_tcp_port
with _reserve_port() as (ota_port, ota_socket):
yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
config_path = await write_yaml_config(yaml_config)
binary_path = await compile_esphome(config_path)
dev = _Device(
*await _build(
yaml_config, write_yaml_config, compile_esphome, reserved_tcp_port
)
)
# 192 bytes that are neither ELF nor Mach-O.
bogus_path = integration_test_dir / "bogus.bin"
bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
# 192 bytes that are neither ELF nor Mach-O.
bogus_path = integration_test_dir / "bogus.bin"
bogus_path.write_bytes(b"NOT-AN-EXECUTABLE-AT-ALL" * 8)
async with run_binary(dev.binary_path) as (proc, _lines):
dev.proc = proc
await _wait_for_port(LOCALHOST, dev.api_port, PORT_WAIT_TIMEOUT)
pid_before = proc.pid
rc = await _run_ota(dev.ota_port, None, bogus_path, None)
assert rc == 1
await asyncio.sleep(0.5)
assert proc.returncode is None, "process died on rejected OTA"
assert proc.pid == pid_before
api_socket.close()
ota_socket.close()
async with run_binary(binary_path) as (proc, _lines):
await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
pid_before = proc.pid
loop = asyncio.get_running_loop()
rc, _ = await loop.run_in_executor(
None, espota2.run_ota, LOCALHOST, ota_port, None, bogus_path
)
assert rc == 1
await asyncio.sleep(0.5)
assert proc.returncode is None, "process died on rejected OTA"
assert proc.pid == pid_before
async with wait_and_connect_api_client(port=api_port) as client:
info = await client.device_info()
assert info.name == DEVICE_NAME
async with wait_and_connect_api_client(port=dev.api_port) as client:
info = await client.device_info()
assert info.name == DEVICE_NAME
@@ -0,0 +1,146 @@
"""Integration test for template climate: sensor-pushed measured values, on_control + publish
for the settable ones.
current_temperature/current_humidity are pushed by a referenced sensor/humidity_sensor (no
polling); action is set once at boot via climate.template.publish, since it has no sensor
equivalent. mode/target_temperature/fan_mode/swing_mode/preset are plain internal state:
on_control fires exactly once per command (never before the first one), and
climate.template.publish simulates the device reporting its own state independent of any prior
command -- that report is authoritative, overriding whatever was optimistically applied earlier.
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import (
ButtonInfo,
ClimateAction,
ClimateFanMode,
ClimateInfo,
ClimateMode,
ClimatePreset,
ClimateSwingMode,
)
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-basic"
@pytest.mark.asyncio
async def test_template_climate_basic(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Sensor-pushed measured values, on_control + publish for settable ones."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
def on_log_line(line: str) -> None:
if "on_control " in line:
log_lines.append(line)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
async def wait_for_climate_state(
timeout: float = 5.0,
) -> aioesphomeapi.ClimateState:
return await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState), timeout
)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
# Advertised capabilities come straight from the supported_*/custom_* config lists.
assert ClimateMode.OFF in test_climate.supported_modes
assert ClimateMode.HEAT in test_climate.supported_modes
assert ClimateMode.COOL in test_climate.supported_modes
assert ClimateFanMode.AUTO in test_climate.supported_fan_modes
assert ClimateFanMode.LOW in test_climate.supported_fan_modes
assert ClimateFanMode.HIGH in test_climate.supported_fan_modes
assert ClimateSwingMode.OFF in test_climate.supported_swing_modes
assert ClimateSwingMode.VERTICAL in test_climate.supported_swing_modes
assert ClimatePreset.NONE in test_climate.supported_presets
assert ClimatePreset.ECO in test_climate.supported_presets
report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
client.subscribe_states(
initial_state_helper.on_state_wrapper(lambda state: None)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
initial = initial_state_helper.initial_states.get(test_climate.key)
assert initial is not None, "No initial climate state received"
assert isinstance(initial, aioesphomeapi.ClimateState)
assert initial.current_temperature == pytest.approx(22.5, abs=0.1)
assert initial.current_humidity == pytest.approx(55.0, abs=0.1)
assert initial.action == ClimateAction.IDLE
assert initial.mode == ClimateMode.OFF
# Nothing was commanded yet: on_control must not have fired.
assert not log_lines
# Commands apply optimistically and on_control fires with the same values.
client.climate_command(test_climate.key, mode=ClimateMode.HEAT)
state = await wait_for_climate_state()
assert state.mode == ClimateMode.HEAT
client.climate_command(test_climate.key, target_temperature=22.5)
state = await wait_for_climate_state()
assert state.target_temperature == pytest.approx(22.5, abs=0.1)
client.climate_command(test_climate.key, fan_mode=ClimateFanMode.HIGH)
state = await wait_for_climate_state()
assert state.fan_mode == ClimateFanMode.HIGH
client.climate_command(test_climate.key, swing_mode=ClimateSwingMode.VERTICAL)
state = await wait_for_climate_state()
assert state.swing_mode == ClimateSwingMode.VERTICAL
client.climate_command(test_climate.key, preset=ClimatePreset.ECO)
state = await wait_for_climate_state()
assert state.preset == ClimatePreset.ECO
await asyncio.sleep(0.2)
assert any(
"on_control mode=3" in line for line in log_lines
) # CLIMATE_MODE_HEAT
assert any("on_control target_temperature=22.5" in line for line in log_lines)
assert any("on_control fan_mode=" in line for line in log_lines)
assert any("on_control swing_mode=" in line for line in log_lines)
assert any("on_control preset=" in line for line in log_lines)
# Exactly one on_control log line per command, none extra (e.g. from a stray republish).
assert len(log_lines) == 5
# measured values are untouched by any of the above (no set action exists for them).
assert state.current_temperature == pytest.approx(22.5, abs=0.1)
assert state.current_humidity == pytest.approx(55.0, abs=0.1)
assert state.action == ClimateAction.IDLE
# The device's report is authoritative and overrides everything commanded above.
client.button_command(report_button.key)
state = await wait_for_climate_state()
assert state.mode == ClimateMode.OFF
assert state.fan_mode == ClimateFanMode.AUTO
assert state.swing_mode == ClimateSwingMode.OFF
assert state.preset == ClimatePreset.NONE
@@ -0,0 +1,98 @@
"""Integration test for template climate: custom fan modes and presets.
Same on_control (forward) + climate.template.publish (device report, authoritative) pattern as
the enum-based mode/preset fields, but for the custom string variants.
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import ButtonInfo, ClimateInfo
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-custom"
@pytest.mark.asyncio
async def test_template_climate_custom_modes(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Custom fan mode/preset: traits, on_control forwarding, and publish precedence."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
def on_log_line(line: str) -> None:
if "on_control " in line:
log_lines.append(line)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
async def wait_for_climate_state(
timeout: float = 5.0,
) -> aioesphomeapi.ClimateState:
return await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState), timeout
)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
assert set(test_climate.supported_custom_fan_modes) == {
"turbo",
"silent",
"eco",
}
assert set(test_climate.supported_custom_presets) == {
"eco_plus",
"power_save",
"max",
}
report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
client.subscribe_states(
initial_state_helper.on_state_wrapper(lambda state: None)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
initial = initial_state_helper.initial_states.get(test_climate.key)
assert initial is not None, "No initial climate state received"
assert isinstance(initial, aioesphomeapi.ClimateState)
assert initial.custom_fan_mode == ""
assert initial.custom_preset == ""
client.climate_command(test_climate.key, custom_fan_mode="turbo")
state = await wait_for_climate_state()
assert state.custom_fan_mode == "turbo"
client.climate_command(test_climate.key, custom_preset="power_save")
state = await wait_for_climate_state()
assert state.custom_preset == "power_save"
await asyncio.sleep(0.2)
assert any("on_control custom_fan_mode=turbo" in line for line in log_lines)
assert any("on_control custom_preset=power_save" in line for line in log_lines)
# The device's report is authoritative and overrides what was commanded above.
client.button_command(report_button.key)
state = await wait_for_climate_state()
assert state.custom_fan_mode == "eco"
assert state.custom_preset == "max"
@@ -0,0 +1,107 @@
"""Integration test for template climate: optimistic: false.
A command still fires on_control (so a real device-backed config can forward it out), but must
NOT change the entity's own state -- only an explicit climate.template.publish call (standing in
for the device confirming the command actually took effect) does that.
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import (
ButtonInfo,
ClimateFanMode,
ClimateInfo,
ClimateMode,
ClimatePreset,
ClimateSwingMode,
)
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-nonopt"
@pytest.mark.asyncio
async def test_template_climate_nonoptimistic(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Nonoptimistic: a command doesn't change state until explicitly published."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
state_updates: list[aioesphomeapi.ClimateState] = []
def on_log_line(line: str) -> None:
if "on_control " in line:
log_lines.append(line)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
def on_state(state: aioesphomeapi.EntityState) -> None:
if isinstance(state, aioesphomeapi.ClimateState):
state_updates.append(state)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
confirm_button = require_entity(
entities, "simulate_device_confirmation", ButtonInfo
)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
initial = initial_state_helper.initial_states.get(test_climate.key)
assert initial is not None, "No initial climate state received"
assert isinstance(initial, aioesphomeapi.ClimateState)
assert initial.mode == ClimateMode.OFF
# Send every settable field in one command. on_control must fire with all of them, but
# nothing may be applied to the entity's own state -- no ClimateState update at all.
client.climate_command(
test_climate.key,
mode=ClimateMode.HEAT,
target_temperature=22.5,
fan_mode=ClimateFanMode.HIGH,
swing_mode=ClimateSwingMode.VERTICAL,
preset=ClimatePreset.AWAY,
)
await asyncio.sleep(0.3)
assert any(
"on_control mode=3" in line for line in log_lines
) # CLIMATE_MODE_HEAT
assert any("on_control target_temperature=22.5" in line for line in log_lines)
assert any("on_control fan_mode=" in line for line in log_lines)
assert any("on_control swing_mode=" in line for line in log_lines)
assert any("on_control preset=" in line for line in log_lines)
assert not state_updates, (
"optimistic: false must not publish a state until climate.template.publish reports it"
)
# The device confirms the command actually took effect.
client.button_command(confirm_button.key)
state = await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
)
assert state.mode == ClimateMode.HEAT
assert state.target_temperature == pytest.approx(22.5, abs=0.1)
assert state.fan_mode == ClimateFanMode.HIGH
assert state.swing_mode == ClimateSwingMode.VERTICAL
assert state.preset == ClimatePreset.AWAY
@@ -0,0 +1,83 @@
"""Integration test: on_control fires before control()/on_state, with the full ClimateCall.
on_control's lambda argument exposes get_mode()/etc. on the *requested* ClimateCall, while the
entity's own .mode field still reflects the state *before* control() applies the change --
proving the firing order is on_control, then control(), then on_state.
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import ClimateInfo, ClimateMode
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-oc-order"
@pytest.mark.asyncio
async def test_template_climate_on_control_ordering(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""on_control sees the requested value while the entity's own state is still the old one."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
def on_log_line(line: str) -> None:
if "on_control " in line or "on_state " in line:
log_lines.append(line)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
async def wait_for_climate_state(
timeout: float = 5.0,
) -> aioesphomeapi.ClimateState:
return await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState), timeout
)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
client.subscribe_states(
initial_state_helper.on_state_wrapper(lambda state: None)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
client.climate_command(test_climate.key, mode=ClimateMode.HEAT)
state = await wait_for_climate_state()
assert state.mode == ClimateMode.HEAT
await asyncio.sleep(0.2)
# on_control saw the new requested mode (3 == CLIMATE_MODE_HEAT) while the entity's own
# state was still the old one (0 == CLIMATE_MODE_OFF) -- proving it fired before control().
assert any(
"on_control requested_mode=3 current_mode_before_apply=0" in line
for line in log_lines
)
# on_state fired afterward, reporting the now-applied mode.
assert any("on_state mode=3" in line for line in log_lines)
control_index = next(
i for i, line in enumerate(log_lines) if "on_control " in line
)
state_index = next(i for i, line in enumerate(log_lines) if "on_state " in line)
assert control_index < state_index, "on_control must fire before on_state"
@@ -0,0 +1,96 @@
"""Integration test for template climate: climate.template.publish covering every field at once.
A single climate.template.publish call resolves into exactly one ClimateState update, and never
triggers on_control (which would misrepresent a device state report as a fresh command). This also
exercises that a sensor/humidity_sensor whose reading matches what's about to be published doesn't
sneak in an extra state update of its own (the sensor callback only re-publishes on an actual
change).
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import (
ButtonInfo,
ClimateAction,
ClimateFanMode,
ClimateInfo,
ClimateMode,
ClimatePreset,
ClimateSwingMode,
)
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-publish-all"
@pytest.mark.asyncio
async def test_template_climate_publish_all_fields(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""One climate.template.publish call setting every field resolves to one state update."""
clear_host_prefs(DEVICE_NAME)
state_updates: list[aioesphomeapi.ClimateState] = []
on_control_count = 0
def on_log_line(line: str) -> None:
nonlocal on_control_count
if "on_control fired" in line:
on_control_count += 1
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
def on_state(state: aioesphomeapi.EntityState) -> None:
if isinstance(state, aioesphomeapi.ClimateState):
state_updates.append(state)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
publish_button = require_entity(entities, "publish_all", ButtonInfo)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
client.button_command(publish_button.key)
try:
state = await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
)
except TimeoutError:
pytest.fail("Timeout waiting for the published climate state")
assert state.current_temperature == pytest.approx(20.0, abs=0.1)
assert state.current_humidity == pytest.approx(60.0, abs=0.1)
assert state.target_temperature == pytest.approx(23.0, abs=0.1)
assert state.mode == ClimateMode.HEAT
assert state.action == ClimateAction.HEATING
assert state.fan_mode == ClimateFanMode.HIGH
assert state.swing_mode == ClimateSwingMode.VERTICAL
assert state.preset == ClimatePreset.ECO
# Give any stray extra update (there shouldn't be one) a moment to arrive.
await asyncio.sleep(0.2)
assert len(state_updates) == 1, (
f"Expected exactly one ClimateState update, got {len(state_updates)}"
)
assert on_control_count == 0, (
"climate.template.publish must not trigger on_control"
)
@@ -0,0 +1,88 @@
"""Integration test for template climate: current_temperature/current_humidity live sensor push.
A *later* change to a backing sensor's value -- not just its initial reading at boot -- propagates
into a new climate state via add_on_state_callback. Re-publishing the same sensor value again must
not cause a redundant climate state update.
"""
from __future__ import annotations
import asyncio
import math
import aioesphomeapi
from aioesphomeapi import ButtonInfo, ClimateInfo
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-sensor-push"
@pytest.mark.asyncio
async def test_template_climate_sensor_push(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""A later change to the backing sensor pushes a new climate state; an unchanged republish does not."""
clear_host_prefs(DEVICE_NAME)
state_updates: list[aioesphomeapi.ClimateState] = []
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
def on_state(state: aioesphomeapi.EntityState) -> None:
if isinstance(state, aioesphomeapi.ClimateState):
state_updates.append(state)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
publish_temp = require_entity(entities, "publish_temperature", ButtonInfo)
publish_temp_same = require_entity(
entities, "publish_temperature_same", ButtonInfo
)
publish_humidity = require_entity(entities, "publish_humidity", ButtonInfo)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
initial = initial_state_helper.initial_states.get(test_climate.key)
assert initial is not None, "No initial climate state received"
assert isinstance(initial, aioesphomeapi.ClimateState)
# Neither backing sensor has published anything yet.
assert math.isnan(initial.current_temperature)
assert math.isnan(initial.current_humidity)
# A later sensor reading -- not the initial one -- pushes a new climate state.
client.button_command(publish_temp.key)
state = await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
)
assert state.current_temperature == pytest.approx(24.0, abs=0.1)
client.button_command(publish_humidity.key)
state = await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
)
assert state.current_humidity == pytest.approx(65.0, abs=0.1)
# Re-publishing the same temperature must not cause a redundant climate state update.
updates_before = len(state_updates)
client.button_command(publish_temp_same.key)
await asyncio.sleep(0.3)
assert len(state_updates) == updates_before, (
"Re-publishing an unchanged sensor reading must not republish the climate state"
)
@@ -0,0 +1,114 @@
"""Integration test: each settable field forwards its value to the matching set_*_action.
With optimistic: false the entity state stays put until climate.template.publish reports the
device's actual state back, so the actions are the only thing that reacts to a command.
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import (
ButtonInfo,
ClimateFanMode,
ClimateInfo,
ClimateMode,
ClimatePreset,
ClimateSwingMode,
)
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-set-act"
@pytest.mark.asyncio
async def test_template_climate_set_actions(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Every set_*_action fires with the requested value; state waits for a publish."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
def on_log_line(line: str) -> None:
if "_action " in line or "Unsupported" in line:
log_lines.append(line)
def logged(fragment: str) -> bool:
return any(fragment in line for line in log_lines)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
report_button = require_entity(entities, "report_device_state", ButtonInfo)
unsupported_button = require_entity(
entities, "report_unsupported_mode", ButtonInfo
)
client.subscribe_states(
initial_state_helper.on_state_wrapper(lambda state: None)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
# Both traits are derived from the low/high and humidity set actions, not declared.
assert test_climate.supports_two_point_target_temperature
assert test_climate.supports_target_humidity
client.climate_command(test_climate.key, mode=ClimateMode.HEAT)
client.climate_command(
test_climate.key, target_temperature_low=18.0, target_temperature_high=24.0
)
client.climate_command(test_climate.key, target_humidity=55)
client.climate_command(test_climate.key, fan_mode=ClimateFanMode.LOW)
client.climate_command(test_climate.key, custom_fan_mode="turbo")
client.climate_command(test_climate.key, swing_mode=ClimateSwingMode.VERTICAL)
client.climate_command(test_climate.key, preset=ClimatePreset.ECO)
client.climate_command(test_climate.key, custom_preset="eco_plus")
for _ in range(50):
await asyncio.sleep(0.1)
if logged("set_custom_preset_action eco_plus"):
break
assert logged("set_mode_action 3") # CLIMATE_MODE_HEAT
assert logged("set_target_temperature_low_action 18.0")
assert logged("set_target_temperature_high_action 24.0")
assert logged("set_target_humidity_action 55")
assert logged("set_fan_mode_action 3") # CLIMATE_FAN_LOW
assert logged("set_custom_fan_mode_action turbo")
assert logged("set_swing_mode_action 2") # CLIMATE_SWING_VERTICAL
assert logged("set_preset_action 5") # CLIMATE_PRESET_ECO
assert logged("set_custom_preset_action eco_plus")
# optimistic: false, so none of the commands above touched the entity's own state --
# a device report is what actually moves it.
client.button_command(report_button.key)
state = await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState)
)
assert state.mode == ClimateMode.HEAT
# A publish naming a mode outside supported_modes warns instead of publishing it.
client.button_command(unsupported_button.key)
for _ in range(50):
await asyncio.sleep(0.1)
if logged("Unsupported mode"):
break
assert logged("Unsupported mode")
@@ -0,0 +1,118 @@
"""Integration tests for template climate: two-point target temperature + humidity.
Covers the supports_two_point_target_temperature/supports_target_humidity boolean flags plus
on_control (forwarding commands out) and climate.template.publish (the device reporting its own
authoritative state, independent of any prior command -- e.g. a device that owns its own setpoint,
changed via a physical remote).
"""
from __future__ import annotations
import asyncio
import aioesphomeapi
from aioesphomeapi import ButtonInfo, ClimateInfo, ClimateMode
import pytest
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity, wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
DEVICE_NAME = "tmpl-clim-two-point"
@pytest.mark.asyncio
async def test_template_climate_two_point_temperature(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Two-point target temperature + humidity: booleans, on_control, and publish precedence."""
clear_host_prefs(DEVICE_NAME)
log_lines: list[str] = []
def on_log_line(line: str) -> None:
if "on_control " in line:
log_lines.append(line)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
async def wait_for_climate_state(
timeout: float = 5.0,
) -> aioesphomeapi.ClimateState:
return await wait_for_state(
client, lambda s: isinstance(s, aioesphomeapi.ClimateState), timeout
)
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) == 1, "Expected exactly 1 climate entity"
test_climate = climate_infos[0]
assert test_climate.name == "Test Two-Point Heatpump"
assert test_climate.supports_two_point_target_temperature
assert test_climate.supports_target_humidity
report_button = require_entity(entities, "simulate_device_report", ButtonInfo)
client.subscribe_states(
initial_state_helper.on_state_wrapper(lambda state: None)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
initial = initial_state_helper.initial_states.get(test_climate.key)
assert initial is not None, "No initial climate state received"
assert isinstance(initial, aioesphomeapi.ClimateState)
# Nothing has been published yet: settable fields have no sensor to seed them from, so
# the entity starts at ESPHome's plain defaults. current_temperature is pushed by the
# referenced sensor, which has already settled by the time we get here.
assert initial.mode == ClimateMode.OFF
assert initial.current_temperature == pytest.approx(21.0, abs=0.1)
# The device reports its actual state for the first time.
client.button_command(report_button.key)
state = await wait_for_climate_state()
assert state.mode == ClimateMode.HEAT_COOL
assert state.target_temperature_low == pytest.approx(18.0, abs=0.1)
assert state.target_temperature_high == pytest.approx(24.0, abs=0.1)
assert state.target_humidity == pytest.approx(50.0, abs=0.1)
# Commands apply optimistically (settable fields are plain internal state), and on_control
# fires with the same values so a real config could forward them to the device.
client.climate_command(
test_climate.key, target_temperature_low=19.0, target_temperature_high=25.0
)
state = await wait_for_climate_state()
assert state.target_temperature_low == pytest.approx(19.0, abs=0.1)
assert state.target_temperature_high == pytest.approx(25.0, abs=0.1)
await asyncio.sleep(0.2)
assert any(
"on_control target_temperature_low=19.0" in line for line in log_lines
)
assert any(
"on_control target_temperature_high=25.0" in line for line in log_lines
)
client.climate_command(test_climate.key, target_humidity=45.0)
state = await wait_for_climate_state()
assert state.target_humidity == pytest.approx(45.0, abs=0.1)
await asyncio.sleep(0.2)
assert any("on_control target_humidity=45.0" in line for line in log_lines)
# The device's next report is authoritative and overrides whatever was optimistically
# applied above -- this is the whole point of climate.template.publish: a device that owns
# its own state (e.g. changed by a physical remote) always wins.
client.button_command(report_button.key)
state = await wait_for_climate_state()
assert state.target_temperature_low == pytest.approx(18.0, abs=0.1)
assert state.target_temperature_high == pytest.approx(24.0, abs=0.1)
assert state.target_humidity == pytest.approx(50.0, abs=0.1)
+17 -17
View File
@@ -35,8 +35,8 @@ def _load_script():
def test_spec_key_collapses_destinations() -> None:
"""Two specs delivering one package share a directory and one key."""
mod = _load_script()
assert mod.spec_key("esphome/noise-c @ 0.1.21") == "noise-c"
assert mod.spec_key("esphome/noise-c@0.1.21") == "noise-c"
assert mod.spec_key("esphome/noise-c @ 0.1.24") == "noise-c"
assert mod.spec_key("esphome/noise-c@0.1.24") == "noise-c"
assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
"esp32async/asynctcp @ 3.5.0"
)
@@ -54,23 +54,23 @@ def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
"[env:a]\n"
"platform = fake/platform@1\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.21\n"
" esphome/noise-c @ 0.1.24\n"
" ${common.lib_deps}\n"
" internal_lib\n"
"[env:b]\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.21\n"
" esphome/noise-c @ 0.1.24\n"
)
mod = _load_script()
args = Namespace(libraries=True, platforms=True, tools=False)
libs, platforms, tools = mod.parse_specs(str(ini), args)
# exact-string duplicates collapse; distinct version pins survive
assert libs == ["esphome/noise-c @ 0.1.21"]
assert libs == ["esphome/noise-c @ 0.1.24"]
assert platforms == ["fake/platform@1"]
assert tools == []
assert mod.build_cli_args(libs, platforms, tools) == [
"-l",
"esphome/noise-c @ 0.1.21",
"esphome/noise-c @ 0.1.24",
"-p",
"fake/platform@1",
]
@@ -162,13 +162,13 @@ def test_parallel_install_behavior(tmp_path: Path) -> None:
mod.parallel_install(
cls,
[
"esphome/noise-c @ 0.1.21",
"esphome/noise-c @ 0.1.21",
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 0.1.24",
"esphome/already @ 1.0",
"https://x/framework.tar.xz",
],
)
assert cls.calls == ["esphome/noise-c @ 0.1.21"]
assert cls.calls == ["esphome/noise-c @ 0.1.24"]
assert cls.lock_events == ["lock", "unlock"]
@@ -205,7 +205,7 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.21": [
"esphome/noise-c @ 0.1.24": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
{"name": "SPI"},
],
@@ -213,12 +213,12 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21", "esphome/wg @ 1.0"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24", "esphome/wg @ 1.0"])
assert len(cls.calls) == 3 # the shared dep installs exactly once
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
# Wave-1 strings carry no compatibility; the dependency wave does
compats = dict(cls.compat_calls)
assert compats["esphome/noise-c @ 0.1.21"] is None
assert compats["esphome/noise-c @ 0.1.24"] is None
dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
assert dep_compat is not None # mirrors pio's install_dependency
@@ -229,11 +229,11 @@ def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.21": [
"esphome/noise-c @ 0.1.24": [
{"name": "vendored", "version": "https://github.com/x/y.git"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
@@ -348,13 +348,13 @@ def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
"""Already-installed top-level packages still feed the dependency
wave; a warm store can be missing a transitive dep."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.21"})
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.24"})
cls.deps = {
"esphome/noise-c @ 0.1.21": [
"esphome/noise-c @ 0.1.24": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
@@ -0,0 +1,219 @@
"""Unit tests for script/sync_dependency_versions.py."""
from pathlib import Path
import subprocess
import sys
import pytest
import yamlrocks
sys.path.insert(0, str((Path(__file__).parent / ".." / ".." / "script").resolve()))
import sync_dependency_versions as sync_mod # noqa: E402
PRECOMMIT = """\
# See https://pre-commit.com for more information
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
# Ruff version.
rev: v0.1.0
hooks:
- id: ruff
- repo: https://github.com/PyCQA/flake8
rev: 7.0.0
hooks:
- id: flake8
- repo: https://github.com/asottile/pyupgrade
rev: v3.0.0
hooks:
- id: pyupgrade
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: v13.0.1
hooks:
- id: clang-format
- repo: https://github.com/adrienverge/yamllint.git
rev: v1.0.0
hooks:
- id: yamllint
- repo: local
hooks:
- id: pylint
"""
REQ_TEST = """\
pylint==4.0.8
flake8==7.1.0
ruff==0.2.0 # comment
pyupgrade==3.0.0
"""
REQ_DEV = """\
clang-format==13.0.1
yamllint==1.0.0
"""
RUFF_REPO = "https://github.com/astral-sh/ruff-pre-commit"
DUPLICATE_RUFF_BLOCK = f" - repo: {RUFF_REPO}\n rev: v0.3.0\n hooks: []\n"
EXPECTED_DRIFT = ["ruff: 0.1.0 -> 0.2.0", "flake8: 7.0.0 -> 7.1.0"]
EXPECTED_PRECOMMIT = PRECOMMIT.replace("rev: v0.1.0", "rev: v0.2.0").replace(
"rev: 7.0.0", "rev: 7.1.0"
)
@pytest.fixture
def root(tmp_path: Path) -> Path:
"""A fake checkout where ruff (v-prefixed) and flake8 (bare) have drifted."""
(tmp_path / ".pre-commit-config.yaml").write_text(PRECOMMIT)
(tmp_path / "requirements_test.txt").write_text(REQ_TEST)
(tmp_path / "requirements_dev.txt").write_text(REQ_DEV)
return tmp_path
def _load(text: str) -> object:
return yamlrocks.loads(text.encode(), option=yamlrocks.OPT_ROUND_TRIP)
@pytest.mark.parametrize(
("requirements", "expected"),
[
("prek==0.5.1 # comment\n", "0.5.1"),
("Prek==0.5.1\n", "0.5.1"),
("other==1.0\nprek==0.5.1\n", "0.5.1"),
("prek>=0.5.1\n", None),
("prek-extra==0.5.1\n", None),
("", None),
],
)
def test_read_requirement_version(requirements: str, expected: str | None) -> None:
assert sync_mod.read_requirement_version(requirements, "prek") == expected
def test_find_repo_entry() -> None:
entry = sync_mod.find_repo_entry(_load(PRECOMMIT), RUFF_REPO)
assert entry["rev"] == "v0.1.0"
@pytest.mark.parametrize(
("text", "message"),
[
("hooks: []\n", "missing key 'repos'"),
("repos:\n - rev: 1.0.0\n", "missing key 'repo'"),
(PRECOMMIT + DUPLICATE_RUFF_BLOCK, "found 2"),
("repos:\n - repo: other\n rev: 1.0.0\n", "found 0"),
],
)
def test_find_repo_entry_errors(text: str, message: str) -> None:
with pytest.raises(sync_mod.SyncError, match=message):
sync_mod.find_repo_entry(_load(text), RUFF_REPO)
@pytest.mark.parametrize(
("rev", "expected"),
[("v0.1.0", ("v", "0.1.0")), ("7.0.0", ("", "7.0.0")), ("'1.0'", ("", "1.0"))],
)
def test_current_rev(rev: str, expected: tuple[str, str]) -> None:
doc = _load(f"repos:\n - repo: {RUFF_REPO}\n rev: {rev}\n")
assert sync_mod.current_rev(doc["repos"][0], RUFF_REPO) == expected
@pytest.mark.parametrize(
("block", "message"),
[(" hooks: []\n", "has no rev"), (" rev: 1.0\n", "not a string: 1.0")],
)
def test_current_rev_errors(block: str, message: str) -> None:
doc = _load(f"repos:\n - repo: {RUFF_REPO}\n{block}")
with pytest.raises(sync_mod.SyncError, match=message):
sync_mod.current_rev(doc["repos"][0], RUFF_REPO)
def test_sync_reports_without_writing(root: Path) -> None:
assert sync_mod.sync(root, write=False) == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
def test_sync_writes_keeps_layout_and_is_idempotent(root: Path) -> None:
assert sync_mod.sync(root, write=True) == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == EXPECTED_PRECOMMIT
assert sync_mod.sync(root, write=True) == []
def test_sync_does_not_touch_a_config_that_matches(root: Path) -> None:
(root / ".pre-commit-config.yaml").write_text(EXPECTED_PRECOMMIT)
before = (root / ".pre-commit-config.yaml").stat().st_mtime_ns
assert sync_mod.sync(root, write=True) == []
assert (root / ".pre-commit-config.yaml").stat().st_mtime_ns == before
def test_sync_missing_requirement_pin(root: Path) -> None:
(root / "requirements_dev.txt").write_text("")
with pytest.raises(sync_mod.SyncError, match="no 'clang-format==' pin"):
sync_mod.sync(root, write=True)
def test_sync_propagates_config_errors(root: Path) -> None:
(root / ".pre-commit-config.yaml").write_text(PRECOMMIT + DUPLICATE_RUFF_BLOCK)
with pytest.raises(sync_mod.SyncError, match="found 2"):
sync_mod.sync(root, write=True)
def test_main_check_reports_drift(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert sync_mod.main(["--check", "--root", str(root)]) == 1
assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
def test_main_writes_then_check_is_clean(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert sync_mod.main(["--root", str(root)]) == 0
assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
assert sync_mod.main(["--check", "--root", str(root)]) == 0
assert capsys.readouterr().out == ""
def test_main_reports_sync_error(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
(root / "requirements_dev.txt").write_text("")
assert sync_mod.main(["--root", str(root)]) == 1
assert (
"error: requirements_dev.txt: no 'clang-format==' pin"
in capsys.readouterr().err
)
def test_main_defaults_to_repo_root(monkeypatch: pytest.MonkeyPatch) -> None:
seen: dict[str, object] = {}
def fake_sync(root: Path, *, write: bool) -> list[str]:
seen["root"] = root
seen["write"] = write
return []
monkeypatch.setattr(sync_mod, "sync", fake_sync)
assert sync_mod.main([]) == 0
assert seen == {"root": sync_mod.REPO_ROOT, "write": True}
def test_repository_is_in_sync() -> None:
"""The real checkout must match; a failure here means a rev has drifted.
Also proves every SYNC_TARGETS entry still resolves in the real files.
"""
assert sync_mod.sync(sync_mod.REPO_ROOT, write=False) == []
def test_cli_entry_point(root: Path) -> None:
"""Run the script the way the workflow does, as a subprocess."""
script = Path(sync_mod.__file__)
result = subprocess.run(
[sys.executable, str(script), "--check", "--root", str(root)],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 1
assert result.stdout.splitlines() == EXPECTED_DRIFT
@@ -0,0 +1,48 @@
"""Tests for the espnow component's final validation."""
import pytest
from esphome.components.esp32.const import (
VARIANT_ESP32C3,
VARIANT_ESP32H2,
VARIANT_ESP32P4,
)
from esphome.components.espnow import _validate_variant
import esphome.config_validation as cv
import esphome.final_validate as fv
from esphome.types import ConfigType
def _run(
monkeypatch, variant: str, full_config: dict, config: ConfigType
) -> ConfigType:
monkeypatch.setattr("esphome.components.espnow.get_esp32_variant", lambda: variant)
token = fv.full_config.set(full_config)
try:
return _validate_variant(config)
finally:
fv.full_config.reset(token)
def test_variant_with_native_wifi_passes(monkeypatch) -> None:
"""A variant with a native Wi-Fi PHY needs no shim; config passes through."""
config = {"id": "espnow"}
assert _run(monkeypatch, VARIANT_ESP32C3, {}, config) is config
def test_radioless_non_p4_variant_rejected(monkeypatch) -> None:
"""Radio-less variants without any ESP-NOW path are rejected outright."""
with pytest.raises(cv.Invalid, match="not supported"):
_run(monkeypatch, VARIANT_ESP32H2, {}, {})
def test_p4_without_esp32_hosted_rejected(monkeypatch) -> None:
"""The P4 needs the esp32_hosted shim to supply the esp_now_* symbols."""
with pytest.raises(cv.Invalid, match="esp32_hosted"):
_run(monkeypatch, VARIANT_ESP32P4, {}, {})
def test_p4_with_esp32_hosted_passes(monkeypatch) -> None:
"""The P4 with esp32_hosted present validates; config passes through."""
config = {"id": "espnow"}
assert _run(monkeypatch, VARIANT_ESP32P4, {"esp32_hosted": {}}, config) is config
@@ -0,0 +1,37 @@
"""Tests for the udp component configuration schema."""
from __future__ import annotations
import pytest
from esphome.components import udp
from esphome.components.packet_transport import (
CONF_BINARY_SENSORS,
CONF_ENCRYPTION,
CONF_PING_PONG_ENABLE,
CONF_PROVIDERS,
CONF_ROLLING_CODE_ENABLE,
CONF_SENSORS,
)
import esphome.config_validation as cv
@pytest.mark.parametrize(
"option",
[
CONF_PROVIDERS,
CONF_ENCRYPTION,
CONF_PING_PONG_ENABLE,
CONF_ROLLING_CODE_ENABLE,
CONF_SENSORS,
CONF_BINARY_SENSORS,
],
)
def test_relocated_option_rejected(option: str) -> None:
"""Options that moved to packet_transport raise a pointing error."""
with pytest.raises(cv.Invalid) as exc_info:
udp.CONFIG_SCHEMA({option: True})
assert (
f"The '{option}' option should now be configured in the 'packet_transport' component"
in str(exc_info.value)
)
+38 -1
View File
@@ -9,7 +9,7 @@ not be part of a unit test suite.
"""
from collections.abc import Generator
from collections.abc import Callable, Generator
import os
from pathlib import Path
import sys
@@ -137,3 +137,40 @@ def mock_get_component() -> Generator[Mock, None, None]:
"""Mock get_component for config module."""
with patch("esphome.config.get_component") as mock:
yield mock
@pytest.fixture
def held_lock() -> Callable[..., Callable[..., None]]:
"""Factory for a ``FileLock.acquire`` fake held by another downloader.
Each poll writes the next chunk to ``part`` (or runs it, for a callable)
and raises ``Timeout``; when the chunks run out the part is removed,
``land()`` runs, and the acquire succeeds (also for any later job, so
``land`` must be idempotent).
"""
from filelock import Timeout
def make(
part: Path,
chunks: list[bytes | Callable[[], None]],
land: Callable[[], None],
) -> Callable[..., None]:
polls = iter(chunks)
def acquire(*args, **kwargs) -> None:
try:
chunk = next(polls)
except StopIteration:
part.unlink(missing_ok=True)
land()
return
if callable(chunk):
chunk()
else:
part.parent.mkdir(parents=True, exist_ok=True)
part.write_bytes(chunk)
raise Timeout("held")
return acquire
return make
+125 -11
View File
@@ -10,12 +10,15 @@ when the installed aioesphomeapi predates the noise module.
from __future__ import annotations
import base64
from collections.abc import Callable
import hashlib
import io
import logging
from pathlib import Path
import socket
import sys
import threading
from typing import Any
from unittest.mock import Mock, patch
import pytest
@@ -65,8 +68,12 @@ class FakeEncryptedDevice(threading.Thread):
offer_noise: bool = True,
require_noise: bool = True,
prologue_features_override: int | None = None,
connections: int = 1,
drop_handshakes: int = 0,
) -> None:
super().__init__(daemon=True)
self.connections = connections
self.drop_handshakes = drop_handshakes # hang up mid-handshake this many times
self.psk = psk
self.version = version
self.offer_noise = offer_noise
@@ -81,10 +88,11 @@ class FakeEncryptedDevice(threading.Thread):
def run(self) -> None:
try:
sock, _ = self.listener.accept()
sock.settimeout(10)
with sock:
self._serve(sock)
for _ in range(self.connections):
sock, _ = self.listener.accept()
sock.settimeout(10)
with sock:
self._serve(sock)
except Exception as err: # noqa: BLE001 - surfaced via join_and_check
self.error = err
finally:
@@ -109,8 +117,23 @@ class FakeEncryptedDevice(threading.Thread):
return
server_flags = espota2.SERVER_FEATURE_SUPPORTS_NOISE if self.offer_noise else 0
sock.sendall(bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags]))
if not (self.offer_noise and noise_negotiated):
return # the client fails closed; nothing further arrives
if not (noise_negotiated and self.offer_noise):
# A device that does not require encryption continues in
# plaintext whatever the client asked for, like older firmware
try:
self._transfer(
lambda byte: sock.sendall(bytes([byte])),
lambda length: _recv_exact(sock, length),
lambda remaining: _recv_exact(
sock, min(remaining, espota2.UPLOAD_BLOCK_SIZE)
),
)
except ConnectionError:
# A keyed client without fallback fails closed and hangs up
if noise_negotiated and not self.offer_noise:
return
raise
return
from cryptography.exceptions import InvalidTag
from noise.connection import NoiseConnection
@@ -134,6 +157,9 @@ class FakeEncryptedDevice(threading.Thread):
msg1 = _recv_frame(sock)
assert msg1[0] == 0x00
if self.drop_handshakes > 0:
self.drop_handshakes -= 1
return # a transport fault: the socket closes with no reply
try:
proto.read_message(msg1[1:])
except InvalidTag:
@@ -149,6 +175,20 @@ class FakeEncryptedDevice(threading.Thread):
assert len(plaintext) == length, "control units must be one per frame"
return plaintext
def recv_data(_remaining: int) -> bytes:
plaintext = proto.decrypt(_recv_frame(sock))
assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
return plaintext
self._transfer(send_byte, recv_unit, recv_data)
def _transfer(
self,
send_byte: Callable[[int], None],
recv_unit: Callable[[int], bytes],
recv_data: Callable[[int], bytes],
) -> None:
"""The post-handshake exchange, identical over both transports."""
send_byte(espota2.RESPONSE_AUTH_OK)
recv_unit(1) # ota type
size = int.from_bytes(recv_unit(4), "big")
@@ -159,9 +199,7 @@ class FakeEncryptedDevice(threading.Thread):
received = b""
acked = 0
while len(received) < size:
plaintext = proto.decrypt(_recv_frame(sock))
assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
received += plaintext
received += recv_data(size - len(received))
if self.version >= espota2.OTA_VERSION_2_0:
while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
len(received) == size and acked < size
@@ -176,7 +214,10 @@ class FakeEncryptedDevice(threading.Thread):
def _upload(
device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
device: FakeEncryptedDevice,
firmware: bytes,
noise_psk: str | None,
plaintext_fallback: bool = False,
) -> None:
device.start()
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -184,12 +225,35 @@ def _upload(
sock.connect(("127.0.0.1", device.port))
try:
espota2.perform_ota(
sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
sock,
None,
io.BytesIO(firmware),
Path("firmware.bin"),
noise_psk=noise_psk,
plaintext_fallback=plaintext_fallback,
)
finally:
sock.close()
def _run_ota(
device: FakeEncryptedDevice, firmware: bytes, tmp_path: Path, noise_psk: str
) -> int:
"""Drive the retry loop, which is where the plaintext fallback reconnects."""
path = tmp_path / "firmware.bin"
path.write_bytes(firmware)
device.start()
rc, _ = espota2.run_ota(
"127.0.0.1",
device.port,
None,
path,
noise_psk=noise_psk,
plaintext_fallback=True,
)
return rc
def test_encrypted_upload_success() -> None:
"""A full encrypted v2 upload spanning several 8192-byte blocks."""
pytest.importorskip("aioesphomeapi.noise")
@@ -240,6 +304,56 @@ def test_client_fails_closed_when_device_lacks_encryption() -> None:
device.join_and_check()
# Remove before 2027.3.0
def test_fallback_when_device_does_not_offer(caplog: pytest.LogCaptureFixture) -> None:
"""The api key is tried opportunistically; an older device that cannot
encrypt still gets its update, with a warning."""
firmware = b"firmware"
device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
with patch("time.sleep"), caplog.at_level(logging.WARNING):
_upload(device, firmware, PSK, plaintext_fallback=True)
device.join_and_check()
assert device.received == firmware
assert any("fallback is removed in 2027.3.0" in r.message for r in caplog.records)
# Remove before 2027.3.0
@pytest.mark.parametrize(
("device_kwargs", "expected_rc", "fell_back"),
[
# A wrong key against an offering device reconnects in plaintext
({"psk": OTHER_PSK, "require_noise": False, "connections": 2}, 0, True),
# The plaintext retry is refused by a device that requires encryption
({"psk": OTHER_PSK, "require_noise": True, "connections": 2}, 1, True),
# A dropped connection inside the handshake is retried encrypted
({"require_noise": False, "connections": 2, "drop_handshakes": 1}, 0, False),
# A second transport fault inside the handshake falls back
({"require_noise": False, "connections": 3, "drop_handshakes": 2}, 0, True),
],
ids=["wrong_key", "wrong_key_required", "one_fault", "two_faults"],
)
def test_fallback_through_the_retry_loop(
caplog: pytest.LogCaptureFixture,
tmp_path: Path,
device_kwargs: dict[str, Any],
expected_rc: int,
fell_back: bool,
) -> None:
pytest.importorskip("aioesphomeapi.noise")
firmware = b"firmware"
device = FakeEncryptedDevice(**device_kwargs)
with patch("time.sleep"), caplog.at_level(logging.WARNING):
rc = _run_ota(device, firmware, tmp_path, PSK)
device.join_and_check()
assert rc == expected_rc
assert (device.received == firmware) is (expected_rc == 0)
assert (
any("Retrying in plaintext" in r.message for r in caplog.records) is fell_back
)
if expected_rc == 1:
assert any("requires an encrypted OTA" in r.message for r in caplog.records)
def test_plaintext_client_gets_encryption_required_error() -> None:
"""A client without a key gets the device's 0x94 error message."""
device = FakeEncryptedDevice()
@@ -2422,3 +2422,20 @@ def test_discard_partial_download_logs_undeletable(
):
framework_helpers.discard_partial_download(dest)
assert "Could not remove" in caplog.text
def test_downloaded_bytes_reports_what_is_on_disk(tmp_path: Path) -> None:
"""Part file first, then the landed file, both capped at size; else 0."""
dest = tmp_path / "archive"
assert framework_helpers.downloaded_bytes(dest, 4) == 0
part = tmp_path / "archive.part"
part.write_bytes(b"ab")
assert framework_helpers.downloaded_bytes(dest, 4) == 2
part.write_bytes(b"abcdef")
assert framework_helpers.downloaded_bytes(dest, 4) == 4
part.unlink()
dest.write_bytes(b"abc")
assert framework_helpers.downloaded_bytes(dest, 4) == 3
assert framework_helpers.downloaded_bytes(dest) == 3
dest.write_bytes(b"abcdef")
assert framework_helpers.downloaded_bytes(dest, 4) == 4
+108 -6
View File
@@ -2109,7 +2109,13 @@ def test_upload_program_ota_success(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
["192.168.1.100"],
3232,
"secret",
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -2141,10 +2147,77 @@ def test_upload_program_ota_encryption_key(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, key
["192.168.1.100"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
key,
plaintext_fallback=False,
)
def test_upload_program_ota_api_key_opportunistic(
mock_run_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""Without an ota encryption block the api key is tried with a plaintext
fallback (removed in 2027.3.0)."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_ota.return_value = (0, "192.168.1.100")
key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
config = {
CONF_API: {CONF_ENCRYPTION: {CONF_KEY: key}},
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
}
exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
assert exit_code == 0
expected_firmware = (
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
key,
plaintext_fallback=True,
)
@pytest.mark.parametrize(
"api_conf",
[{}, {CONF_ENCRYPTION: {}}],
ids=["no_encryption", "runtime_key"],
)
def test_upload_program_ota_no_usable_api_key_stays_plaintext(
mock_run_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
api_conf: dict[str, Any],
) -> None:
"""A missing or runtime provisioned api key gives the uploader nothing
to try."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_ota.return_value = (0, "192.168.1.100")
config = {
CONF_API: api_conf,
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}],
}
exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
assert exit_code == 0
assert mock_run_ota.call_args.args[5] is None
assert mock_run_ota.call_args.kwargs == {"plaintext_fallback": False}
def test_upload_program_ota_encryption_without_key_fails_closed(
mock_run_ota: Mock,
mock_get_port_type: Mock,
@@ -2195,7 +2268,13 @@ def test_upload_program_ota_with_file_arg(
assert exit_code == 0
assert host == "192.168.1.100"
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
["192.168.1.100"],
3232,
None,
Path("custom.bin"),
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -2251,6 +2330,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
partition_file,
OTA_TYPE_UPDATE_PARTITION_TABLE,
None,
plaintext_fallback=False,
)
@@ -2313,6 +2393,7 @@ def test_upload_program_ota_partition_table_mqttip(
partition_file,
OTA_TYPE_UPDATE_PARTITION_TABLE,
None,
plaintext_fallback=False,
)
@@ -2501,6 +2582,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
bootloader_file,
OTA_TYPE_UPDATE_BOOTLOADER,
None,
plaintext_fallback=False,
)
@@ -2989,7 +3071,13 @@ def test_upload_program_ota_with_mqtt_resolution(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
["192.168.1.100"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -3039,7 +3127,13 @@ def test_upload_program_ota_with_mqtt_empty_broker(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
["192.168.1.50"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
# Verify warning was logged
assert "MQTT IP discovery failed" in caplog.text
@@ -5212,6 +5306,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -5262,6 +5357,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -5439,7 +5535,13 @@ def test_upload_program_ota_mqtt_timeout_fallback(
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_ota.assert_called_once_with(
["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
["192.168.1.100"],
3232,
None,
expected_firmware,
OTA_TYPE_UPDATE_APP,
None,
plaintext_fallback=False,
)
@@ -7,6 +7,7 @@ exercised in their own test modules)."""
import json
import logging
from pathlib import Path
from unittest.mock import Mock
import pytest
@@ -228,6 +229,24 @@ def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
_resolve_registry_version("owner", "pkg", set())
def test_make_registry_client_skips_private_package_probe(monkeypatch):
"""Our client answers the probe locally without patching PlatformIO's class."""
from platformio.account.client import AccountClient
from platformio.registry.client import RegistryClient
pio_probe = RegistryClient.__dict__["allowed_private_packages"]
monkeypatch.setattr(
AccountClient,
"get_account_info",
Mock(side_effect=AssertionError("account probe must not run")),
)
client = lib._make_registry_client().get_registry_client_instance()
assert client.allowed_private_packages() is False
assert RegistryClient.__dict__["allowed_private_packages"] is pio_probe
def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
"""Stub the registry lookup so tests never touch the network."""
monkeypatch.setattr(
+96 -9
View File
@@ -454,23 +454,96 @@ def test_uri_fetch_job_waits_out_a_briefly_held_lock(tmp_path: Path) -> None:
assert dl_path.read_bytes() == b"data"
def test_lock_deadline_leaves_download_to_the_holder(tmp_path: Path) -> None:
"""A lock held past the deadline means another process is fetching the
same file; skipping cleanly beats a misleading failure warning. The
tracker is still polled so a parked worker observes cancellation."""
@pytest.mark.parametrize("staged", [b"", b"ab"])
def test_lock_deadline_leaves_download_to_the_holder(
tmp_path: Path, staged: bytes
) -> None:
"""A lock held past the deadline is another process's download; skip
cleanly, polling the tracker with what the holder has staged so far."""
dl_path = tmp_path / "archive"
(tmp_path / "archive.prefetch.part").write_bytes(staged)
ticks: list[int] = []
with (
patch("esphome.framework_helpers.download_with_resume") as mock_download,
patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
):
pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
mock_download.assert_not_called()
assert ticks == [0]
assert ticks == [len(staged)]
assert not dl_path.exists()
@pytest.mark.parametrize(
("job", "part_name", "chunks", "expected"),
[
(
lambda dl_path: pf._registry_fetch_job(
MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4
),
"archive.part",
[b"a", b"abc"],
[1, 3, 4],
),
(
lambda dl_path: pf._uri_fetch_job(
MagicMock(), "https://x/a.zip", dl_path, 4
),
"archive.prefetch.part",
[b"ab"],
[2, 4],
),
],
ids=["registry", "uri"],
)
def test_lock_wait_reports_the_holders_progress(
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
held_lock,
job,
part_name: str,
chunks: list[bytes],
expected: list[int],
) -> None:
"""A waiting job reports the holder's part file (the staging one for a
URL job), then the full size once the holder lands the archive."""
dl_path = tmp_path / "archive"
ticks: list[int] = []
acquire = held_lock(
tmp_path / part_name, chunks, lambda: dl_path.write_bytes(b"abcd")
)
with (
patch("esphome.framework_helpers.download_with_resume") as mock_download,
patch("filelock.FileLock.acquire", side_effect=acquire),
patch("filelock.FileLock.release"),
caplog.at_level(logging.INFO),
):
job(dl_path)(ticks.append)
mock_download.assert_not_called()
assert ticks == expected
assert caplog.text.count("Waiting for another process downloading archive") == 1
def test_uri_lock_wait_prefers_the_landed_archive(tmp_path: Path, held_lock) -> None:
"""Between the holder's promotion rename and its release the staging
part is gone; the landed cache file is credited instead of 0."""
dl_path = tmp_path / "archive"
ticks: list[int] = []
acquire = held_lock(
tmp_path / "archive.prefetch.part",
[b"ab", lambda: dl_path.write_bytes(b"abcd")],
lambda: None,
)
with (
patch("esphome.framework_helpers.download_with_resume") as mock_download,
patch("filelock.FileLock.acquire", side_effect=acquire),
patch("filelock.FileLock.release"),
):
pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
mock_download.assert_not_called()
assert ticks == [2, 4, 4]
def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
"""A registry job that lost the download race to another process
must not stamp a nonexistent archive into pio's usage.db."""
@@ -479,7 +552,7 @@ def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
with (
patch("esphome.framework_helpers.download_with_resume") as mock_download,
patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
):
pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)(
lambda done: None
@@ -1152,6 +1225,20 @@ def test_main_runs_prefetch(tmp_path: Path) -> None:
mock_prefetch.assert_called_once_with(tmp_path, "testenv")
def test_main_skips_private_package_probe_before_prefetch(tmp_path: Path) -> None:
"""The registry probe patch is applied before any package manager runs."""
order: list[str] = []
with (
patch.object(pf, "_prefetch", side_effect=lambda *_: order.append("prefetch")),
patch(
"esphome.platformio.runner.patch_registry_private_packages",
side_effect=lambda: order.append("patch"),
),
):
assert pf.main([str(tmp_path), "testenv"]) == 0
assert order == ["patch", "prefetch"]
def test_main_bad_argv_is_a_distinct_exit(
caplog: pytest.LogCaptureFixture,
) -> None:
@@ -1576,7 +1663,7 @@ def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None:
{"name": "SPI"},
]
m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"])
pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c", "libsodium"] # dep deduped, SPI left out
# The dep wave carries its compatibility so _install searches qualified
dep_call = m._install.call_args_list[-1]
@@ -1596,7 +1683,7 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: (
installed.append(getattr(spec, "name", str(spec)))
)
pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c"]
+67 -6
View File
@@ -9,6 +9,7 @@ import os
from pathlib import Path
from unittest.mock import MagicMock, patch
from filelock import Timeout
import pytest
from esphome.core import EsphomeError
@@ -545,16 +546,13 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
dest = tmp_path / "a"
dest.mkdir()
from contextlib import contextmanager
@contextmanager
def marker_appears_under_lock(path, **kwargs):
def marker_appears_under_lock(*args, **kwargs):
# Simulates the concurrent build finishing while we waited
(dest / ".esphome_extracted").touch()
yield
with (
patch("filelock.FileLock", side_effect=marker_appears_under_lock),
patch("filelock.FileLock.acquire", side_effect=marker_appears_under_lock),
patch("filelock.FileLock.release"),
patch.object(registry, "download_with_resume") as mock_download,
patch.object(
registry, "registry_download", side_effect=_resolve_for({"a": 10})
@@ -564,6 +562,69 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
mock_download.assert_not_called()
def test_prefetch_packages_waits_with_the_holders_progress(
tmp_path: Path, held_lock
) -> None:
"""A worker parked on another build's lock reports that build's part
file, then the full size once the marker appears."""
dest = tmp_path / "a"
dest.mkdir()
ticks: list[int] = []
part = tmp_path / "dl" / "a-1.0.part"
def installed_and_pruned() -> None:
# install_package touches the marker, then unlinks the archive
(dest / ".esphome_extracted").touch()
part.unlink()
acquire = held_lock(
part,
[lambda: None, b"abc", installed_and_pruned],
(dest / ".esphome_extracted").touch,
)
def fake_batch(header, jobs):
for _name, _size, fetch in jobs:
fetch(ticks.append)
return []
with (
patch("filelock.FileLock.acquire", side_effect=acquire),
patch("filelock.FileLock.release"),
patch.object(registry, "run_batch_downloads", side_effect=fake_batch),
patch.object(registry, "download_with_resume") as mock_download,
patch.object(
registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
),
):
registry.prefetch_packages(
[("a", "1.0", dest, []), ("b", "2.0", tmp_path / "b", [])],
tmp_path / "dl",
)
assert ticks == [0, 3, 10, 10]
mock_download.assert_called_once()
def test_prefetch_packages_leaves_a_long_held_lock_to_its_holder(
tmp_path: Path,
) -> None:
"""Past the deadline the worker skips; install_package waits on the same
lock later and verifies whatever the holder produced."""
with (
patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
patch.object(registry, "download_with_resume") as mock_download,
patch.object(
registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
),
):
registry.prefetch_packages(
[("a", "1.0", tmp_path / "a", []), ("b", "2.0", tmp_path / "b", [])],
tmp_path / "dl",
)
mock_download.assert_not_called()
def test_already_installed_probe(tmp_path: Path) -> None:
"""Both arms of the marker probe the prefetch worker keys on."""
dest = tmp_path / "pkg"
@@ -6,7 +6,9 @@ from collections.abc import Callable
import io
import sys
from types import ModuleType
from unittest.mock import Mock
from platformio.registry.client import RegistryClient
import pytest
from esphome.platformio import runner
@@ -30,6 +32,7 @@ def _prepare_main(
monkeypatch.setattr(sys, "stderr", stream)
monkeypatch.setattr(runner, "patch_structhash", lambda: None)
monkeypatch.setattr(runner, "patch_file_downloader", lambda: None)
monkeypatch.setattr(runner, "patch_registry_private_packages", lambda: None)
platformio = ModuleType("platformio")
platformio_main = ModuleType("platformio.__main__")
@@ -91,3 +94,40 @@ def test_main_still_filters_a_drained_partial_line(
assert runner.main() == 0
assert buf.getvalue() == b""
def test_main_applies_registry_private_packages_patch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The probe is patched before PlatformIO runs."""
order: list[str] = []
_prepare_main(monkeypatch, lambda: order.append("pio") or 0)
monkeypatch.setattr(
runner, "patch_registry_private_packages", lambda: order.append("patch")
)
assert runner.main() == 0
assert order == ["patch", "pio"]
# Snapshot PlatformIO's own probe at import, before any test can patch it
_PIO_PROBE = RegistryClient.__dict__["allowed_private_packages"]
def test_patch_registry_private_packages_skips_account_probe(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Answers False without touching the account client."""
from platformio.account.client import AccountClient
monkeypatch.setattr(RegistryClient, "allowed_private_packages", _PIO_PROBE)
monkeypatch.setattr(
AccountClient,
"get_account_info",
Mock(side_effect=AssertionError("account probe must not run")),
)
runner.patch_registry_private_packages()
assert RegistryClient.allowed_private_packages() is False
assert RegistryClient().allowed_private_packages() is False
+25 -6
View File
@@ -37,7 +37,6 @@ def wizard_answers() -> list[str]:
"nodemcuv2", # board
"SSID", # ssid
"psk", # wifi password
"", # ota password (empty for no password)
]
@@ -101,6 +100,25 @@ def test_config_file_should_include_ota(default_config: dict[str, Any]):
assert "ota:" in config
def test_config_file_should_use_encryption_when_api_key_set(
default_config: dict[str, Any],
):
"""
With an API encryption key and no OTA password the OTA block reuses the key
"""
# Given
default_config["api_encryption_key"] = (
"AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
)
# When
config = wz.wizard_file(**default_config)
# Then
assert "ota:\n - platform: esphome\n encryption:" in config
assert "password" not in config.split("ota:")[1].split("wifi:")[0]
def test_config_file_should_include_ota_when_password_set(
default_config: dict[str, Any],
):
@@ -630,15 +648,15 @@ def test_wizard_write_protects_existing_config(
assert config_file.read_text() == original_content
def test_wizard_accepts_ota_password(
def test_wizard_uses_the_api_key_for_ota(
tmp_path: Path, monkeypatch: MonkeyPatch, wizard_answers: list[str]
):
"""
The wizard should pass ota_password to wizard_write when the user provides one
The wizard generates an api key and does not ask for an OTA password;
the key secures OTA updates
"""
# Given
wizard_answers[5] = "my_ota_password" # Set OTA password
config_file = tmp_path / "test.yaml"
input_mock = MagicMock(side_effect=wizard_answers)
monkeypatch.setattr("builtins.input", input_mock)
@@ -653,8 +671,9 @@ def test_wizard_accepts_ota_password(
# Then
assert retval == 0
call_kwargs = wizard_write_mock.call_args.kwargs
assert "ota_password" in call_kwargs
assert call_kwargs["ota_password"] == "my_ota_password"
assert "api_encryption_key" in call_kwargs
assert "ota_password" not in call_kwargs
assert input_mock.call_count == len(wizard_answers)
def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch):