Merge branch 'dev' into partition-table-ota

This commit is contained in:
Mat931
2026-04-24 19:59:32 +00:00
committed by GitHub
213 changed files with 9113 additions and 1895 deletions
@@ -46,7 +46,7 @@ from esphome.const import (
)
from esphome.core import CORE
from esphome.util import OrderedDict
from esphome.yaml_util import IncludeFile, add_context, load_yaml
from esphome.yaml_util import DocumentPath, IncludeFile, add_context, load_yaml
# Test strings
TEST_DEVICE_NAME = "test_device_name"
@@ -1113,7 +1113,7 @@ def test_packages_include_file_resolves_to_list(mock_resolve_include) -> None:
"""When packages: is an IncludeFile that resolves to a list, it is processed correctly."""
include_file = MagicMock(spec=IncludeFile)
package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
mock_resolve_include.return_value = ([package_content], None)
mock_resolve_include.return_value = [package_content]
config = {CONF_PACKAGES: include_file}
result = do_packages_pass(config)
@@ -1127,7 +1127,7 @@ def test_packages_include_file_resolves_to_dict(mock_resolve_include) -> None:
"""When packages: is an IncludeFile that resolves to a dict, it is processed correctly."""
include_file = MagicMock(spec=IncludeFile)
package_content = {CONF_WIFI: {CONF_SSID: TEST_PACKAGE_WIFI_SSID}}
mock_resolve_include.return_value = ({"network": package_content}, None)
mock_resolve_include.return_value = {"network": package_content}
config = {CONF_PACKAGES: include_file}
result = do_packages_pass(config)
@@ -1142,7 +1142,7 @@ def test_packages_include_file_resolves_to_invalid_type_raises(
) -> None:
"""When packages: is an IncludeFile that resolves to an invalid type, cv.Invalid is raised."""
include_file = MagicMock(spec=IncludeFile)
mock_resolve_include.return_value = ("not_a_dict_or_list", None)
mock_resolve_include.return_value = "not_a_dict_or_list"
config = {CONF_PACKAGES: include_file}
with pytest.raises(
@@ -1215,7 +1215,9 @@ def test_named_dict_with_include_files_no_false_deprecation_warning(
call_count = 0
def failing_callback(package_config: dict, context: object) -> dict:
def failing_callback(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
if call_count == 1:
@@ -1251,7 +1253,9 @@ def test_validate_deprecated_false_raises_directly(
call_count = 0
def failing_callback(package_config: dict, context: object) -> dict:
def failing_callback(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
if call_count == 1:
@@ -1283,7 +1287,9 @@ def test_error_on_first_declared_package_still_detected() -> None:
call_count = 0
def fail_on_last(package_config: dict, context: object) -> dict:
def fail_on_last(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal call_count
call_count += 1
# Reverse iteration: third_pkg (1), second_pkg (2), first_pkg (3)
@@ -1312,7 +1318,9 @@ def test_deprecated_single_package_fallback_still_works(
attempt = 0
def fail_then_succeed(package_config: dict, context: object) -> dict:
def fail_then_succeed(
package_config: dict, context: object, path: DocumentPath | None = None
) -> dict:
nonlocal attempt
attempt += 1
if attempt == 1:
+1
View File
@@ -3,3 +3,4 @@ output:
id: ac_dimmer_1
gate_pin: ${gate_pin}
zero_cross_pin: ${zero_cross_pin}
zero_cross_interrupt_type: ANY
+7
View File
@@ -0,0 +1,7 @@
psram:
media_source:
- platform: audio_http
id: audio_http_source
buffer_size: 100000
task_stack_in_psram: true
@@ -0,0 +1 @@
<<: !include common.yaml
+96
View File
@@ -117,4 +117,100 @@ TEST(FormatHexChar, UppercaseDigits) {
EXPECT_EQ(format_hex_pretty_char(15), 'F');
}
// --- small_pow10() ---
TEST(SmallPow10, Zero) { EXPECT_EQ(small_pow10(0), 1u); }
TEST(SmallPow10, One) { EXPECT_EQ(small_pow10(1), 10u); }
TEST(SmallPow10, Two) { EXPECT_EQ(small_pow10(2), 100u); }
TEST(SmallPow10, Three) { EXPECT_EQ(small_pow10(3), 1000u); }
// --- frac_to_str_unchecked() ---
TEST(FracToStr, OneDigit) {
char buf[8];
char *end = frac_to_str_unchecked(buf, 5, 1);
*end = '\0';
EXPECT_STREQ(buf, "5");
EXPECT_EQ(end - buf, 1);
}
TEST(FracToStr, TwoDigits) {
char buf[8];
char *end = frac_to_str_unchecked(buf, 46, 10);
*end = '\0';
EXPECT_STREQ(buf, "46");
}
TEST(FracToStr, ThreeDigits) {
char buf[8];
char *end = frac_to_str_unchecked(buf, 456, 100);
*end = '\0';
EXPECT_STREQ(buf, "456");
EXPECT_EQ(end - buf, 3);
}
TEST(FracToStr, LeadingZeros) {
char buf[8];
char *end = frac_to_str_unchecked(buf, 1, 100);
*end = '\0';
EXPECT_STREQ(buf, "001");
end = frac_to_str_unchecked(buf, 5, 10);
*end = '\0';
EXPECT_STREQ(buf, "05");
}
TEST(FracToStr, AllZeros) {
char buf[8];
char *end = frac_to_str_unchecked(buf, 0, 100);
*end = '\0';
EXPECT_STREQ(buf, "000");
end = frac_to_str_unchecked(buf, 0, 1);
*end = '\0';
EXPECT_STREQ(buf, "0");
}
TEST(FracToStr, ZeroDivisor) {
char buf[8];
buf[0] = 'X';
char *end = frac_to_str_unchecked(buf, 0, 0);
EXPECT_EQ(end, buf); // writes nothing
}
// --- buf_append_sep_str() ---
TEST(BufAppendSepStr, Basic) {
char buf[32] = "23.46";
char *start = buf + 5;
char *end = buf_append_sep_str(start, sizeof(buf) - 5, ' ', "°C", 3);
EXPECT_STREQ(buf, "23.46 °C");
EXPECT_EQ(end - buf, 9); // "°C" is 3 bytes (UTF-8)
}
TEST(BufAppendSepStr, EmptyString) {
char buf[32] = "100";
char *start = buf + 3;
char *end = buf_append_sep_str(start, sizeof(buf) - 3, ' ', "", 0);
EXPECT_STREQ(buf, "100 ");
EXPECT_EQ(end - start, 1); // just the separator
}
TEST(BufAppendSepStr, NoRoom) {
char buf[8] = "1234567";
char *start = buf + 7;
char *end = buf_append_sep_str(start, 1, ' ', "unit", 4);
EXPECT_EQ(end, start); // nothing written
}
TEST(BufAppendSepStr, Truncation) {
char buf[8] = "val";
char *start = buf + 3;
// remaining = 5, separator takes 1, so 3 chars of string fit + null
char *end = buf_append_sep_str(start, 5, ' ', "longunit", 8);
*end = '\0';
EXPECT_STREQ(buf, "val lon");
EXPECT_EQ(end - buf, 7);
}
} // namespace esphome::core::testing
@@ -0,0 +1,237 @@
#include <gtest/gtest.h>
#include <cmath>
#include <cstdio>
#include <cstring>
#include <span>
#include <string>
#include "esphome/core/helpers.h"
#include "esphome/core/string_ref.h"
namespace esphome::core::testing {
// Helper to call value_accuracy_to_buf and return as string
static std::string va_to_string(float value, int8_t accuracy_decimals) {
char buf[VALUE_ACCURACY_MAX_LEN];
std::span<char, VALUE_ACCURACY_MAX_LEN> sp(buf);
size_t len = value_accuracy_to_buf(sp, value, accuracy_decimals);
return std::string(buf, len);
}
// Helper: reference implementation using snprintf for comparison
static std::string va_reference(float value, int8_t accuracy_decimals) {
// Replicate normalize_accuracy_decimals logic
if (accuracy_decimals < 0) {
float divisor;
if (accuracy_decimals == -1) {
divisor = 10.0f;
} else if (accuracy_decimals == -2) {
divisor = 100.0f;
} else {
divisor = pow10_int(-accuracy_decimals);
}
value = roundf(value / divisor) * divisor;
accuracy_decimals = 0;
}
char buf[VALUE_ACCURACY_MAX_LEN];
snprintf(buf, sizeof(buf), "%.*f", accuracy_decimals, value);
return std::string(buf);
}
// --- Basic formatting ---
TEST(ValueAccuracyToBuf, ZeroDecimals) {
EXPECT_EQ(va_to_string(23.456f, 0), "23");
EXPECT_EQ(va_to_string(0.0f, 0), "0");
EXPECT_EQ(va_to_string(100.0f, 0), "100");
EXPECT_EQ(va_to_string(1.0f, 0), "1");
}
TEST(ValueAccuracyToBuf, OneDecimal) {
EXPECT_EQ(va_to_string(23.456f, 1), "23.5");
EXPECT_EQ(va_to_string(0.0f, 1), "0.0");
EXPECT_EQ(va_to_string(1.05f, 1), va_reference(1.05f, 1));
}
TEST(ValueAccuracyToBuf, TwoDecimals) {
EXPECT_EQ(va_to_string(23.456f, 2), "23.46");
EXPECT_EQ(va_to_string(0.0f, 2), "0.00");
EXPECT_EQ(va_to_string(1.005f, 2), va_reference(1.005f, 2));
}
TEST(ValueAccuracyToBuf, ThreeDecimals) {
EXPECT_EQ(va_to_string(23.456f, 3), "23.456");
EXPECT_EQ(va_to_string(0.0f, 3), "0.000");
}
// --- Negative values ---
TEST(ValueAccuracyToBuf, NegativeValues) {
EXPECT_EQ(va_to_string(-23.456f, 2), "-23.46");
EXPECT_EQ(va_to_string(-0.5f, 1), "-0.5");
EXPECT_EQ(va_to_string(-100.0f, 0), "-100");
}
// --- Negative accuracy_decimals (rounding to tens/hundreds) ---
TEST(ValueAccuracyToBuf, NegativeAccuracy) {
EXPECT_EQ(va_to_string(1234.0f, -1), va_reference(1234.0f, -1));
EXPECT_EQ(va_to_string(1234.0f, -2), va_reference(1234.0f, -2));
EXPECT_EQ(va_to_string(56.0f, -1), va_reference(56.0f, -1));
}
// --- Special float values ---
TEST(ValueAccuracyToBuf, NaN) {
std::string result = va_to_string(NAN, 2);
EXPECT_EQ(result, va_reference(NAN, 2));
}
TEST(ValueAccuracyToBuf, Infinity) {
std::string result = va_to_string(INFINITY, 2);
EXPECT_EQ(result, va_reference(INFINITY, 2));
}
TEST(ValueAccuracyToBuf, NegativeInfinity) {
std::string result = va_to_string(-INFINITY, 2);
EXPECT_EQ(result, va_reference(-INFINITY, 2));
}
// --- Edge cases ---
TEST(ValueAccuracyToBuf, VerySmallValues) {
EXPECT_EQ(va_to_string(0.001f, 3), "0.001");
EXPECT_EQ(va_to_string(0.001f, 2), "0.00");
EXPECT_EQ(va_to_string(0.009f, 2), "0.01");
}
TEST(ValueAccuracyToBuf, LargeValues) {
EXPECT_EQ(va_to_string(999999.0f, 0), va_reference(999999.0f, 0));
EXPECT_EQ(va_to_string(1013.25f, 2), "1013.25");
}
TEST(ValueAccuracyToBuf, Rounding) {
// 0.5 rounds up
EXPECT_EQ(va_to_string(23.5f, 0), "24");
EXPECT_EQ(va_to_string(23.45f, 1), "23.5"); // float: 23.45 -> 23.4 or 23.5
EXPECT_EQ(va_to_string(23.45f, 1), va_reference(23.45f, 1));
}
// --- Match snprintf for a range of typical sensor values ---
TEST(ValueAccuracyToBuf, MatchesSnprintf) {
float test_values[] = {0.0f, 1.0f, -1.0f, 23.456f, -23.456f, 100.0f, 0.1f, 0.01f, 99.99f, 1013.25f, -40.0f};
int8_t test_accuracies[] = {0, 1, 2, 3};
for (float value : test_values) {
for (int8_t acc : test_accuracies) {
EXPECT_EQ(va_to_string(value, acc), va_reference(value, acc))
<< "Mismatch for value=" << value << " accuracy=" << static_cast<int>(acc);
}
}
}
// --- Return value (length) ---
TEST(ValueAccuracyToBuf, ReturnsCorrectLength) {
char buf[VALUE_ACCURACY_MAX_LEN];
std::span<char, VALUE_ACCURACY_MAX_LEN> sp(buf);
size_t len = value_accuracy_to_buf(sp, 23.456f, 2);
EXPECT_EQ(len, 5u); // "23.46"
EXPECT_EQ(strlen(buf), len);
len = value_accuracy_to_buf(sp, 0.0f, 0);
EXPECT_EQ(len, 1u); // "0"
EXPECT_EQ(strlen(buf), len);
len = value_accuracy_to_buf(sp, -100.0f, 1);
EXPECT_EQ(len, 6u); // "-100.0"
EXPECT_EQ(strlen(buf), len);
}
TEST(ValueAccuracyToBuf, NegativeZero) {
// Hand-rolled formatter must preserve snprintf's sign-of-zero behavior.
EXPECT_EQ(va_to_string(-0.0f, 2), va_reference(-0.0f, 2));
EXPECT_EQ(va_to_string(-0.0f, 0), va_reference(-0.0f, 0));
// Tiny negative that rounds to zero at this precision must still render as "-0.00".
EXPECT_EQ(va_to_string(-0.001f, 2), va_reference(-0.001f, 2));
}
TEST(ValueAccuracyToBuf, OverflowFallsBackToSnprintf) {
// |value| * 10^acc must exceed UINT32_MAX to exercise the snprintf fallback path.
EXPECT_EQ(va_to_string(1.0e7f, 3), va_reference(1.0e7f, 3));
EXPECT_EQ(va_to_string(-1.0e7f, 3), va_reference(-1.0e7f, 3));
EXPECT_EQ(va_to_string(5.0e9f, 0), va_reference(5.0e9f, 0));
}
// --- value_accuracy_with_uom_to_buf ---
static std::string va_uom_to_string(float value, int8_t accuracy_decimals, const char *uom) {
char buf[VALUE_ACCURACY_MAX_LEN];
std::span<char, VALUE_ACCURACY_MAX_LEN> sp(buf);
StringRef ref(uom);
size_t len = value_accuracy_with_uom_to_buf(sp, value, accuracy_decimals, ref);
return std::string(buf, len);
}
static std::string va_uom_reference(float value, int8_t accuracy_decimals, const char *uom) {
char buf[VALUE_ACCURACY_MAX_LEN];
if (!uom || *uom == '\0') {
snprintf(buf, sizeof(buf), "%.*f", accuracy_decimals, value);
} else {
snprintf(buf, sizeof(buf), "%.*f %s", accuracy_decimals, value, uom);
}
return std::string(buf);
}
TEST(ValueAccuracyWithUomToBuf, BasicWithUnit) {
EXPECT_EQ(va_uom_to_string(23.456f, 2, "°C"), va_uom_reference(23.456f, 2, "°C"));
EXPECT_EQ(va_uom_to_string(1013.25f, 2, "hPa"), va_uom_reference(1013.25f, 2, "hPa"));
EXPECT_EQ(va_uom_to_string(-40.0f, 1, "°F"), va_uom_reference(-40.0f, 1, "°F"));
EXPECT_EQ(va_uom_to_string(100.0f, 0, "%"), va_uom_reference(100.0f, 0, "%"));
}
TEST(ValueAccuracyWithUomToBuf, EmptyUnit) {
EXPECT_EQ(va_uom_to_string(23.456f, 2, ""), "23.46");
EXPECT_EQ(va_uom_to_string(0.0f, 1, ""), "0.0");
}
TEST(ValueAccuracyWithUomToBuf, ReturnsCorrectLength) {
char buf[VALUE_ACCURACY_MAX_LEN];
std::span<char, VALUE_ACCURACY_MAX_LEN> sp(buf);
StringRef ref("°C");
size_t len = value_accuracy_with_uom_to_buf(sp, 23.46f, 2, ref);
EXPECT_EQ(strlen(buf), len);
EXPECT_EQ(len, strlen("23.46 °C"));
}
TEST(ValueAccuracyWithUomToBuf, NearBufferLimitTruncates) {
// Build a unit long enough that value + " " + unit exceeds VALUE_ACCURACY_MAX_LEN.
// "23.46" (5) + " " (1) + unit -> must cap at buf.size()-1 and stay null-terminated.
std::string long_unit(VALUE_ACCURACY_MAX_LEN, 'U');
char buf[VALUE_ACCURACY_MAX_LEN];
std::span<char, VALUE_ACCURACY_MAX_LEN> sp(buf);
StringRef ref(long_unit.c_str());
size_t len = value_accuracy_with_uom_to_buf(sp, 23.46f, 2, ref);
EXPECT_LT(len, VALUE_ACCURACY_MAX_LEN);
EXPECT_EQ(strlen(buf), len);
// Should begin with the formatted value and a separator.
EXPECT_EQ(std::string(buf, 6), "23.46 ");
}
TEST(ValueAccuracyWithUomToBuf, MatchesSnprintf) {
const char *units[] = {"°C", "hPa", "%", "W", "kWh", "m/s"};
float values[] = {0.0f, 23.456f, -40.0f, 1013.25f, 100.0f};
int8_t accs[] = {0, 1, 2, 3};
for (const char *u : units) {
for (float v : values) {
for (int8_t a : accs) {
EXPECT_EQ(va_uom_to_string(v, a, u), va_uom_reference(v, a, u))
<< "value=" << v << " acc=" << static_cast<int>(a) << " uom=" << u;
}
}
}
}
} // namespace esphome::core::testing
+6
View File
@@ -4,3 +4,9 @@ esphome:
- deep_sleep.prevent
- delay: 1s
- deep_sleep.allow
- if:
condition:
lambda: 'return false;'
then:
- deep_sleep.enter:
sleep_duration: 60min
@@ -0,0 +1,12 @@
deep_sleep:
run_duration: 10s
sleep_duration: 50s
<<: !include common.yaml
zigbee:
sensor:
- platform: template
name: "Temperature"
id: temperature_sensor
@@ -5,3 +5,10 @@ packages:
uart: !include ../../test_build_components/common/uart/esp32-ard.yaml
<<: !include common.yaml
sensor:
- platform: dsmr
energy_delivered_lux:
name: "Energy Consumed Luxembourg. OBIS: 1-0:1.8.0"
energy_delivered_tariff1:
name: "Energy Consumed Tariff 1. OBIS: 1-0:1.8.1"
+14
View File
@@ -5,3 +5,17 @@ packages:
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
<<: !include common.yaml
sensor:
- platform: dsmr
energy_delivered_lux:
name: "Energy Consumed Luxembourg. OBIS: 1-0:1.8.0"
energy_delivered_tariff1:
name: "Energy Consumed Tariff 1. OBIS: 1-0:1.8.1"
text_sensor:
- platform: dsmr
identification:
name: "DSMR Identification"
p1_version:
name: "DSMR Version. OBIS: 1-3:0.2.8"
@@ -5,3 +5,10 @@ packages:
uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
<<: !include common.yaml
text_sensor:
- platform: dsmr
identification:
name: "DSMR Identification"
p1_version:
name: "DSMR Version. OBIS: 1-3:0.2.8"
@@ -0,0 +1,7 @@
*** DO NOT USE THIS KEY...EVER ***
-----BEGIN EC PRIVATE KEY-----
MHcCAQEEIEZIp96p7Z7QN6vxOFE5FdRNm535vW81Ax07KnGxVjiMoAoGCCqGSM49
AwEHoUQDQgAEK+fBQDn1Q+r5lGwcDoMUgeg2Aq16LLrLUz7xWI6mS0PUClzolDIo
eaV/Pfjl7zAvkbQQsZq3rTNnr1eGAk5P+A==
-----END EC PRIVATE KEY-----
*** DO NOT USE THIS KEY...EVER ***
@@ -0,0 +1,10 @@
esp32:
variant: esp32
framework:
type: esp-idf
advanced:
signed_ota_verification:
signing_key: ../../components/esp32/dummy_signing_key_v1_ecdsa.pem
signing_scheme: ecdsa_v1
<<: !include common.yaml
@@ -20,6 +20,7 @@ esp32:
disable_regi2c_in_iram: true
disable_fatfs: true
sram1_as_iram: true
watchdog_timeout: 7s
wifi:
ssid: MySSID
@@ -0,0 +1,14 @@
wifi:
ssid: MySSID
password: password1
ota:
- platform: esphome
id: my_ota
port: 3287
password: ""
esphome:
on_boot:
then:
- lambda: id(my_ota).set_auth_password("runtime_password");
@@ -0,0 +1,8 @@
# `sendspin.switch` action enables the controller role, so we use a standalone test
packages:
base: !include common.yaml
wifi:
on_connect:
then:
- sendspin.switch:
@@ -0,0 +1,5 @@
<<: !include common.yaml
media_player:
- platform: sendspin
id: media_player_id
@@ -0,0 +1,9 @@
<<: !include common.yaml
media_source:
- platform: sendspin
id: media_source_id
buffer_size: 500000
initial_static_delay: 5ms
static_delay_adjustable: true
fixed_delay: 480us
@@ -0,0 +1,15 @@
<<: !include common.yaml
sensor:
- platform: sendspin
name: "Sendspin Track Progress"
type: track_progress
- platform: sendspin
name: "Sendspin Track Duration"
type: track_duration
- platform: sendspin
name: "Sendspin Year"
type: year
- platform: sendspin
name: "Sendspin Track"
type: track
@@ -0,0 +1,15 @@
<<: !include common.yaml
text_sensor:
- platform: sendspin
name: "Title"
type: title
- platform: sendspin
name: "Artist"
type: artist
- platform: sendspin
name: "Album"
type: album
- platform: sendspin
name: "Album Artist"
type: album_artist
+9
View File
@@ -0,0 +1,9 @@
wifi:
ap:
psram:
mode: quad
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
@@ -0,0 +1 @@
<<: !include common-action.yaml
@@ -0,0 +1 @@
<<: !include common-media_player.yaml
@@ -0,0 +1 @@
<<: !include common-media_source.yaml
@@ -0,0 +1 @@
<<: !include common-sensor.yaml
@@ -0,0 +1 @@
<<: !include common-text_sensor.yaml
@@ -0,0 +1 @@
<<: !include common.yaml
@@ -11,9 +11,9 @@ media_player:
id: speaker_media_player_id
announcement_pipeline:
speaker: speaker_id
format: NONE
buffer_size: 1000000
volume_increment: 0.02
volume_max: 0.95
volume_min: 0.0
task_stack_in_psram: true
codec_support_enabled: all
+1
View File
@@ -38,3 +38,4 @@ event:
update:
water_heater:
infrared:
radio_frequency:
-10
View File
@@ -1,4 +1,3 @@
---
binary_sensor:
- platform: template
name: "Garage Door Open 1"
@@ -22,12 +21,6 @@ sensor:
lambda: return 12.0;
internal: True
zigbee:
wipe_on_boot: true
on_join:
then:
- logger.log: "Joined network"
output:
- platform: template
id: output_factory
@@ -35,9 +28,6 @@ output:
write_action:
- zigbee.factory_reset
time:
- platform: zigbee
switch:
- platform: template
name: "Template Switch"
+14
View File
@@ -0,0 +1,14 @@
binary_sensor:
- platform: template
name: "Garage Door Open 10"
report: "enable"
- platform: template
name: "Garage Door Open 11"
report: "coordinator"
- platform: template
name: "Garage Door Open 12"
report: "force"
zigbee:
model: zigbee_test
router: true
+12
View File
@@ -0,0 +1,12 @@
packages:
- !include common.yaml
zigbee:
model: zigbee_test
wipe_on_boot: true
on_join:
then:
- logger.log: "Joined network"
time:
- platform: zigbee
@@ -0,0 +1 @@
<<: !include common_esp32.yaml
@@ -1 +1 @@
<<: !include common.yaml
<<: !include common_nrf52.yaml
@@ -1 +1 @@
<<: !include common.yaml
<<: !include common_nrf52.yaml
@@ -1,6 +1,7 @@
<<: !include common.yaml
<<: !include common_nrf52.yaml
zigbee:
wipe_on_boot: once
power_source: battery
ieee802154_vendor_oui: 0x231
sleepy: true
+66 -1
View File
@@ -1,6 +1,11 @@
"""Tests for time component cron expression parsing."""
from esphome.components.time import _parse_cron_part
import errno
from unittest.mock import MagicMock, patch
import pytest
from esphome.components.time import _load_tzdata, _parse_cron_part, validate_tz
def test_star_slash_seconds() -> None:
@@ -78,3 +83,63 @@ def test_range() -> None:
def test_single_value() -> None:
assert _parse_cron_part("30", 0, 59, {}) == {30}
def _mock_resources_with_error(error: Exception) -> MagicMock:
"""Return a mock of importlib.resources.files where read_bytes raises error."""
leaf = MagicMock()
leaf.read_bytes.side_effect = error
package = MagicMock()
package.__truediv__.return_value = leaf
return MagicMock(return_value=package)
def test_load_tzdata_returns_none_on_windows_einval() -> None:
"""On Windows, opening a tzdata path with NTFS-illegal chars raises OSError(EINVAL).
Regression test for crash when the system TZ resolves to a POSIX string like
"<+08>-8" (Asia/Shanghai, IST, etc.) and is fed back into _load_tzdata by
validate_tz to check whether it is also a valid IANA key.
"""
err = OSError(errno.EINVAL, "Invalid argument")
with patch(
"esphome.components.time.resources.files",
_mock_resources_with_error(err),
):
assert _load_tzdata("<+08>-8") is None
def test_load_tzdata_propagates_unexpected_oserror() -> None:
"""Unrelated OSErrors (e.g. PermissionError) must not be swallowed."""
with (
patch(
"esphome.components.time.resources.files",
_mock_resources_with_error(
PermissionError(errno.EACCES, "Permission denied")
),
),
pytest.raises(PermissionError),
):
_load_tzdata("Some/Zone")
def test_load_tzdata_returns_none_on_file_not_found() -> None:
"""Existing behavior: missing tz file returns None rather than raising."""
with patch(
"esphome.components.time.resources.files",
_mock_resources_with_error(FileNotFoundError()),
):
assert _load_tzdata("Not/A/Zone") is None
def test_validate_tz_accepts_posix_string_when_read_bytes_raises_einval() -> None:
"""validate_tz must not crash when _load_tzdata hits the Windows EINVAL path.
Simulates the Windows case where the auto-detected POSIX TZ string is fed
back through _load_tzdata and the underlying read_bytes raises errno 22.
"""
with patch(
"esphome.components.time.resources.files",
_mock_resources_with_error(OSError(errno.EINVAL, "Invalid argument")),
):
assert validate_tz("<+08>-8") == "<+08>-8"
+20
View File
@@ -121,6 +121,26 @@ def test_get_addresses_auto_detection() -> None:
assert cache.get_addresses("unknown.com") is None
def test_add_mdns_addresses_stores_and_normalizes() -> None:
"""add_mdns_addresses inserts entries under the normalized hostname."""
cache = AddressCache()
cache.add_mdns_addresses("Device.Local.", ["192.168.1.10", "192.168.1.11"])
assert cache.mdns_cache == {
normalize_hostname("Device.Local."): ["192.168.1.10", "192.168.1.11"]
}
# Overwrites on subsequent calls for the same host
cache.add_mdns_addresses("device.local", ["10.0.0.1"])
assert cache.mdns_cache[normalize_hostname("device.local")] == ["10.0.0.1"]
def test_add_mdns_addresses_empty_is_noop() -> None:
"""Passing an empty address list must not create an entry."""
cache = AddressCache()
cache.add_mdns_addresses("device.local", [])
assert cache.mdns_cache == {}
def test_has_cache() -> None:
"""Test checking if cache has entries."""
# Empty cache
+511 -2
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Generator
from collections.abc import Callable, Generator
from dataclasses import dataclass
import json
import logging
@@ -12,16 +12,18 @@ import re
import sys
import time
from typing import Any
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from pytest import CaptureFixture
from zeroconf import ServiceStateChange
from esphome import platformio_api
from esphome.__main__ import (
Purpose,
_get_configured_xtal_freq,
_make_crystal_freq_callback,
_resolve_network_devices,
choose_upload_log_host,
command_analyze_memory,
command_bundle,
@@ -36,6 +38,7 @@ from esphome.__main__ import (
has_mqtt,
has_mqtt_ip_lookup,
has_mqtt_logging,
has_name_add_mac_suffix,
has_non_ip_address,
has_ota,
has_resolvable_address,
@@ -48,6 +51,7 @@ from esphome.__main__ import (
upload_using_picotool,
upload_using_platformio,
)
from esphome.address_cache import AddressCache
from esphome.bundle import BUNDLE_EXTENSION, BundleFile, BundleResult
from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
from esphome.const import (
@@ -62,6 +66,7 @@ from esphome.const import (
CONF_MDNS,
CONF_MQTT,
CONF_NAME,
CONF_NAME_ADD_MAC_SUFFIX,
CONF_OTA,
CONF_PASSWORD,
CONF_PLATFORM,
@@ -79,6 +84,7 @@ from esphome.const import (
)
from esphome.core import CORE, EsphomeError
from esphome.util import BootselResult
from esphome.zeroconf import _await_discovery, discover_mdns_devices
def strip_ansi_codes(text: str) -> str:
@@ -2218,6 +2224,509 @@ def test_has_resolvable_address() -> None:
assert has_resolvable_address() is False
def test_has_name_add_mac_suffix() -> None:
"""Test has_name_add_mac_suffix function."""
# Test with name_add_mac_suffix enabled
setup_core(config={CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True}})
assert has_name_add_mac_suffix() is True
# Test with name_add_mac_suffix disabled
setup_core(config={CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: False}})
assert has_name_add_mac_suffix() is False
# Test with name_add_mac_suffix not set (defaults to False)
setup_core(config={CONF_ESPHOME: {}})
assert has_name_add_mac_suffix() is False
# Test with no esphome config
setup_core(config={})
assert has_name_add_mac_suffix() is False
# Test with no config at all
CORE.config = None
assert has_name_add_mac_suffix() is False
@pytest.fixture
def mock_mdns_discovery() -> Generator[MagicMock]:
"""Fixture to mock the async mDNS discovery infrastructure.
Patches ``AsyncEsphomeZeroconf``, ``AsyncServiceBrowser`` and
``AddressResolver`` in ``esphome.zeroconf`` and exposes hooks for tests to
stage browser events and control resolution results. The default
``AddressResolver`` stub simulates a cache hit returning no addresses, so
matched hosts appear in the discovery output with empty address lists
unless the test overrides ``_resolver_setup``.
"""
with (
patch("esphome.zeroconf.AsyncEsphomeZeroconf") as mock_aiozc_class,
patch("esphome.zeroconf.AsyncServiceBrowser") as mock_browser_class,
patch("esphome.zeroconf.AddressResolver") as mock_resolver_class,
):
mock_aiozc = MagicMock()
mock_aiozc.zeroconf = MagicMock()
mock_aiozc.async_close = AsyncMock(return_value=None)
mock_aiozc_class.return_value = mock_aiozc
mock_browser = MagicMock()
mock_browser.async_cancel = AsyncMock(return_value=None)
# Default: each host gets a fresh resolver that hits the cache and
# returns no addresses. Tests can override via ``_resolver_setup``.
def default_resolver_factory(name: str) -> MagicMock:
resolver = MagicMock()
resolver._name = name
resolver.load_from_cache.return_value = True
resolver.async_request = AsyncMock(return_value=True)
resolver.parsed_scoped_addresses.return_value = []
return resolver
mock_resolver_class.side_effect = default_resolver_factory
# Store references for test access
mock_aiozc._mock_browser_class = mock_browser_class
mock_aiozc._mock_browser = mock_browser
mock_aiozc._mock_class = mock_aiozc_class
mock_aiozc._mock_resolver_class = mock_resolver_class
yield mock_aiozc
@pytest.mark.parametrize(
("discovered_services", "base_name", "expected_hosts"),
[
# Matching devices; different-prefix device is filtered out
(
[
("mydevice-abc123._esphomelib._tcp.local.", ServiceStateChange.Added),
("mydevice-def456._esphomelib._tcp.local.", ServiceStateChange.Added),
(
"otherdevice-abcdef._esphomelib._tcp.local.",
ServiceStateChange.Added,
),
],
"mydevice",
["mydevice-abc123.local", "mydevice-def456.local"],
),
# No matches at all
(
[
(
"otherdevice-abcdef._esphomelib._tcp.local.",
ServiceStateChange.Added,
),
],
"mydevice",
[],
),
# Deduplication (same device Added then Updated)
(
[
("mydevice-abc123._esphomelib._tcp.local.", ServiceStateChange.Added),
("mydevice-abc123._esphomelib._tcp.local.", ServiceStateChange.Updated),
],
"mydevice",
["mydevice-abc123.local"],
),
# Suffix must be exactly 6 hex chars: wrong length and non-hex are rejected
(
[
# too short
("mydevice-abcd._esphomelib._tcp.local.", ServiceStateChange.Added),
# too long
(
"mydevice-abcdef1._esphomelib._tcp.local.",
ServiceStateChange.Added,
),
# non-hex
("mydevice-xyz123._esphomelib._tcp.local.", ServiceStateChange.Added),
# valid
("mydevice-012345._esphomelib._tcp.local.", ServiceStateChange.Added),
],
"mydevice",
["mydevice-012345.local"],
),
# Prefix-collision: base "foo" must not match "foo-bar-abc123"
(
[
("foo-abcdef._esphomelib._tcp.local.", ServiceStateChange.Added),
("foo-bar-abcdef._esphomelib._tcp.local.", ServiceStateChange.Added),
],
"foo",
["foo-abcdef.local"],
),
],
ids=[
"matching_with_filter",
"no_matches",
"deduplication",
"hex_suffix_filter",
"prefix_collision",
],
)
def test_discover_mdns_devices(
mock_mdns_discovery: MagicMock,
discovered_services: list[tuple[str, ServiceStateChange]],
base_name: str,
expected_hosts: list[str],
) -> None:
"""Test discover_mdns_devices filtering and deduplication."""
mock_browser = mock_mdns_discovery._mock_browser
def capture_callback(
zc: MagicMock,
service_type: str,
handlers: list[Callable[..., None]],
) -> MagicMock:
callback = handlers[0]
for service_name, state_change in discovered_services:
callback(
mock_mdns_discovery.zeroconf, service_type, service_name, state_change
)
return mock_browser
mock_mdns_discovery._mock_browser_class.side_effect = capture_callback
# Each discovered host gets a resolver that returns a unique IP string
# derived from its server name so we can assert per-host.
def resolver_factory(name: str) -> MagicMock:
resolver = MagicMock()
resolver._name = name
resolver.load_from_cache.return_value = True
resolver.async_request = AsyncMock(return_value=True)
resolver.parsed_scoped_addresses.return_value = [f"10.0.0.1#{name}"]
return resolver
mock_mdns_discovery._mock_resolver_class.side_effect = resolver_factory
result = discover_mdns_devices(base_name, timeout=0)
assert sorted(result) == expected_hosts
# Resolved addresses should be stored for matched hosts. AddressResolver
# receives the fully-qualified name (``<device>.local.``).
for host in expected_hosts:
short = host.partition(".")[0]
assert result[host] == [f"10.0.0.1#{short}.local."]
mock_browser.async_cancel.assert_awaited_once()
mock_mdns_discovery.async_close.assert_awaited_once()
def test_discover_mdns_devices_init_failure(caplog: pytest.LogCaptureFixture) -> None:
"""If AsyncEsphomeZeroconf fails to init, return empty dict and log warning."""
with (
patch(
"esphome.zeroconf.AsyncEsphomeZeroconf",
side_effect=OSError("no network"),
),
caplog.at_level(logging.WARNING, logger="esphome.zeroconf"),
):
result = discover_mdns_devices("mydevice", timeout=0)
assert result == {}
assert "mDNS discovery failed to initialize" in caplog.text
def test_discover_mdns_devices_resolution_failure(
mock_mdns_discovery: MagicMock,
) -> None:
"""If resolution raises, the host is still listed with an empty address list."""
mock_browser = mock_mdns_discovery._mock_browser
def capture_callback(
zc: MagicMock,
service_type: str,
handlers: list[Callable[..., None]],
) -> MagicMock:
handlers[0](
mock_mdns_discovery.zeroconf,
service_type,
"mydevice-abc123._esphomelib._tcp.local.",
ServiceStateChange.Added,
)
return mock_browser
mock_mdns_discovery._mock_browser_class.side_effect = capture_callback
# Resolver misses the cache, then async_request raises.
def failing_resolver_factory(name: str) -> MagicMock:
resolver = MagicMock()
resolver.load_from_cache.return_value = False
resolver.async_request = AsyncMock(side_effect=OSError("boom"))
resolver.parsed_scoped_addresses.return_value = []
return resolver
mock_mdns_discovery._mock_resolver_class.side_effect = failing_resolver_factory
result = discover_mdns_devices("mydevice", timeout=0)
assert result == {"mydevice-abc123.local": []}
def test_discover_mdns_devices_ignores_removed_state(
mock_mdns_discovery: MagicMock,
) -> None:
"""``Removed`` state changes are ignored and do not appear in the result."""
mock_browser = mock_mdns_discovery._mock_browser
def capture_callback(
zc: MagicMock,
service_type: str,
handlers: list[Callable[..., None]],
) -> MagicMock:
handlers[0](
mock_mdns_discovery.zeroconf,
service_type,
"mydevice-abc123._esphomelib._tcp.local.",
ServiceStateChange.Removed,
)
return mock_browser
mock_mdns_discovery._mock_browser_class.side_effect = capture_callback
result = discover_mdns_devices("mydevice", timeout=0)
assert result == {}
# No AddressResolver should have been constructed since no host matched.
mock_mdns_discovery._mock_resolver_class.assert_not_called()
def test_discover_mdns_devices_empty_resolution(
mock_mdns_discovery: MagicMock,
) -> None:
"""Host is listed with empty addresses when resolver returns no addresses."""
mock_browser = mock_mdns_discovery._mock_browser
def capture_callback(
zc: MagicMock,
service_type: str,
handlers: list[Callable[..., None]],
) -> MagicMock:
handlers[0](
mock_mdns_discovery.zeroconf,
service_type,
"mydevice-abc123._esphomelib._tcp.local.",
ServiceStateChange.Added,
)
return mock_browser
mock_mdns_discovery._mock_browser_class.side_effect = capture_callback
# Default fixture resolver is a cache-hit with no addresses — simulates
# the "browse found it but no A/AAAA records are available" case.
result = discover_mdns_devices("mydevice", timeout=0)
assert result == {"mydevice-abc123.local": []}
def test_resolve_network_devices_expands_cached_mdns_hosts(tmp_path: Path) -> None:
"""Hostnames in ``CORE.address_cache`` are expanded to their cached IPs."""
setup_core(tmp_path=tmp_path)
CORE.address_cache = AddressCache(
mdns_cache={
"device-abc123.local": ["10.0.0.1", "10.0.0.2"],
}
)
result = _resolve_network_devices(
["device-abc123.local", "192.168.1.50", "device-abc123.local"],
CORE.config,
MockArgs(),
)
# Cached hostname is replaced with its IPs (deduplicated across repeats)
# and the literal IP is preserved after.
assert result == ["10.0.0.1", "10.0.0.2", "192.168.1.50"]
def test_resolve_network_devices_keeps_uncached_hosts(tmp_path: Path) -> None:
"""Hostnames not in the cache pass through unchanged."""
setup_core(tmp_path=tmp_path)
CORE.address_cache = AddressCache()
result = _resolve_network_devices(
["unknown.local", "192.168.1.50"],
CORE.config,
MockArgs(),
)
assert result == ["unknown.local", "192.168.1.50"]
def test_await_discovery_timeout_returns_empty(
caplog: pytest.LogCaptureFixture,
) -> None:
"""If the discovery runner never sets its event, return {} and warn."""
stub = MagicMock()
stub.event.wait.return_value = False
stub.exception = None
stub.result = {"should_not_be_read": ["1.2.3.4"]}
with caplog.at_level(logging.WARNING, logger="esphome.zeroconf"):
result = _await_discovery(stub, timeout=0.01)
assert result == {}
assert "mDNS discovery timed out after 0.0s" in caplog.text
stub.event.wait.assert_called_once_with(timeout=pytest.approx(2.01))
def test_await_discovery_propagates_exception_as_empty(
caplog: pytest.LogCaptureFixture,
) -> None:
"""If the coroutine raised, log and return {} rather than re-raise."""
stub = MagicMock()
stub.event.wait.return_value = True
stub.exception = RuntimeError("boom")
stub.result = None
with caplog.at_level(logging.WARNING, logger="esphome.zeroconf"):
result = _await_discovery(stub, timeout=5.0)
assert result == {}
assert "mDNS discovery failed: boom" in caplog.text
@pytest.mark.usefixtures("mock_no_serial_ports")
def test_choose_upload_log_host_discovers_mac_suffix_devices(tmp_path: Path) -> None:
"""Interactive mode discovers MAC-suffixed devices and populates the cache."""
setup_core(
config={
CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True},
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
},
address="mydevice.local",
tmp_path=tmp_path,
name="mydevice",
)
CORE.address_cache = None
discovered = {
"mydevice-abc123.local": ["10.0.0.1"],
"mydevice-def456.local": ["10.0.0.2"],
}
with (
patch(
"esphome.__main__.discover_mdns_devices", return_value=discovered
) as mock_discover,
patch(
"esphome.__main__.choose_prompt", return_value="mydevice-abc123.local"
) as mock_prompt,
):
result = choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
assert result == ["mydevice-abc123.local"]
mock_discover.assert_called_once_with("mydevice")
mock_prompt.assert_called_once_with(
[
("Over The Air (mydevice-abc123.local)", "mydevice-abc123.local"),
("Over The Air (mydevice-def456.local)", "mydevice-def456.local"),
],
purpose=Purpose.UPLOADING,
)
# Resolved IPs should be cached so downstream resolution skips a second
# Zeroconf lookup.
assert CORE.address_cache is not None
assert CORE.address_cache.get_mdns_addresses("mydevice-abc123.local") == [
"10.0.0.1"
]
assert CORE.address_cache.get_mdns_addresses("mydevice-def456.local") == [
"10.0.0.2"
]
@pytest.mark.usefixtures("mock_no_serial_ports")
def test_choose_upload_log_host_mac_suffix_no_devices_found(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""When discovery finds nothing, no OTA option is offered and a warning logs."""
setup_core(
config={
CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True},
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
},
address="mydevice.local",
tmp_path=tmp_path,
name="mydevice",
)
with (
patch("esphome.__main__.discover_mdns_devices", return_value={}),
caplog.at_level(logging.WARNING, logger="esphome.__main__"),
pytest.raises(EsphomeError),
):
choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
assert "No devices matching 'mydevice-<mac>.local'" in caplog.text
def test_choose_upload_log_host_default_ota_discovers_mac_suffix(
tmp_path: Path,
) -> None:
"""``--device OTA`` also runs mDNS discovery when name_add_mac_suffix is on."""
setup_core(
config={
CONF_ESPHOME: {CONF_NAME_ADD_MAC_SUFFIX: True},
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
},
address="mydevice.local",
tmp_path=tmp_path,
name="mydevice",
)
CORE.address_cache = None
discovered = {
"mydevice-abc123.local": ["10.0.0.1"],
"mydevice-def456.local": ["10.0.0.2"],
}
with patch(
"esphome.__main__.discover_mdns_devices", return_value=discovered
) as mock_discover:
result = choose_upload_log_host(
default="OTA",
check_default=None,
purpose=Purpose.UPLOADING,
)
# Both discovered hostnames are returned so aioesphomeapi / espota2 can
# try each in turn with the cached IPs.
assert result == ["mydevice-abc123.local", "mydevice-def456.local"]
mock_discover.assert_called_once_with("mydevice")
assert CORE.address_cache is not None
assert CORE.address_cache.get_mdns_addresses("mydevice-abc123.local") == [
"10.0.0.1"
]
def test_choose_upload_log_host_default_ota_no_suffix_discovery(
tmp_path: Path,
) -> None:
"""``--device OTA`` without name_add_mac_suffix uses CORE.address as-is."""
setup_core(
config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]},
address="192.168.1.100",
tmp_path=tmp_path,
name="mydevice",
)
with patch("esphome.__main__.discover_mdns_devices") as mock_discover:
result = choose_upload_log_host(
default="OTA",
check_default=None,
purpose=Purpose.UPLOADING,
)
assert result == ["192.168.1.100"]
# Discovery must NOT run when name_add_mac_suffix is disabled.
mock_discover.assert_not_called()
def test_command_wizard(tmp_path: Path) -> None:
"""Test command_wizard function."""
config_file = tmp_path / "test.yaml"
+15 -18
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import re
import socket
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError
from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
@@ -115,24 +115,21 @@ def test_async_resolver_generic_exception() -> None:
def test_async_resolver_thread_timeout() -> None:
"""Test timeout when thread doesn't complete in time."""
# Mock the start method to prevent actual thread execution
with (
patch.object(AsyncResolver, "start"),
patch("esphome.resolver.hr.async_resolve_host"),
):
resolver = AsyncResolver(["test.local"], 6053)
# Override event.wait to simulate timeout (return False = timeout occurred)
with (
patch.object(resolver.event, "wait", return_value=False),
pytest.raises(
EsphomeError, match=re.escape("Timeout resolving IP address")
),
):
resolver.resolve()
"""Test timeout when the runner thread doesn't complete in time."""
# Patch AsyncThreadRunner inside esphome.resolver so we never actually
# start a thread and can control the wait return value directly.
fake_runner = MagicMock()
fake_runner.start = MagicMock()
fake_runner.event.wait.return_value = False # simulate timeout
# Verify thread start was called
resolver.start.assert_called_once()
with (
patch("esphome.resolver.AsyncThreadRunner", return_value=fake_runner),
patch("esphome.resolver.hr.async_resolve_host"),
pytest.raises(EsphomeError, match=re.escape("Timeout resolving IP address")),
):
AsyncResolver(["test.local"], 6053).resolve()
fake_runner.start.assert_called_once()
def test_async_resolver_ip_addresses(mock_addr_info_ipv4: AddrInfo) -> None:
+43 -4
View File
@@ -659,7 +659,7 @@ def test_resolve_package_max_depth_exceeded(tmp_path: Path) -> None:
cv.Invalid,
match=f"Maximum include nesting depth \\({MAX_INCLUDE_DEPTH}\\) exceeded",
):
processor.resolve_package(package_config, substitutions.ContextVars())
processor.resolve_package(package_config, substitutions.ContextVars(), [])
def test_include_filename_substitution_undefined_var(tmp_path: Path) -> None:
@@ -690,7 +690,7 @@ def test_raise_first_undefined_logs_extras_at_debug(
caplog.at_level(logging.DEBUG, logger="esphome.components.substitutions"),
pytest.raises(cv.Invalid) as exc_info,
):
substitutions.raise_first_undefined(errors, None, "package definition")
substitutions.raise_first_undefined(errors, "package definition")
# First error is surfaced as the cv.Invalid message.
raised = str(exc_info.value)
@@ -706,7 +706,7 @@ def test_raise_first_undefined_logs_extras_at_debug(
def test_raise_first_undefined_noop_on_empty() -> None:
"""An empty errors list is a no-op — no exception, no log."""
substitutions.raise_first_undefined([], None, "package definition")
substitutions.raise_first_undefined([], "package definition")
def test_do_substitution_pass_included_substitutions_must_be_mapping(
@@ -778,4 +778,43 @@ def test_resolve_package_undefined_var_in_include_filename(tmp_path: Path) -> No
)
processor = _PackageProcessor({}, None, False)
with pytest.raises(cv.Invalid, match="unresolved substitutions"):
processor.resolve_package(package_config, substitutions.ContextVars())
processor.resolve_package(package_config, substitutions.ContextVars(), [])
def test_resolve_include_error_shows_expanded_from_when_substituted(
tmp_path: Path,
) -> None:
"""When a substituted filename fails to load, the error includes '(expanded from ...)'."""
parent = tmp_path / "main.yaml"
parent.write_text("")
def failing_loader(_path: Path) -> None:
raise EsphomeError("File not found")
include = yaml_util.IncludeFile(parent, "${device}.yaml", None, failing_loader)
context = substitutions.ContextVars({"device": "my_device"})
with pytest.raises(cv.Invalid) as exc_info:
substitutions.resolve_include(include, [], context)
msg = str(exc_info.value)
assert "my_device.yaml" in msg
assert "expanded from '${device}.yaml'" in msg
def test_resolve_include_error_no_expanded_from_for_literal_filename(
tmp_path: Path,
) -> None:
"""When a literal filename fails to load, the error has no 'expanded from' clause."""
parent = tmp_path / "main.yaml"
parent.write_text("")
def failing_loader(_path: Path) -> None:
raise EsphomeError("File not found")
include = yaml_util.IncludeFile(parent, "literal.yaml", None, failing_loader)
with pytest.raises(cv.Invalid) as exc_info:
substitutions.resolve_include(include, [], substitutions.ContextVars())
assert "expanded from" not in str(exc_info.value)
+180 -1
View File
@@ -9,8 +9,9 @@ from esphome import core, yaml_util
from esphome.components import substitutions
from esphome.config_helpers import Extend, Remove
import esphome.config_validation as cv
from esphome.core import EsphomeError
from esphome.core import DocumentLocation, DocumentRange, EsphomeError
from esphome.util import OrderedDict
from esphome.yaml_util import ESPHomeDataBase, format_path, make_data_base
@pytest.fixture(autouse=True)
@@ -712,3 +713,181 @@ def test_yaml_merge_chain_include_depth_exceeded() -> None:
yaml_text = "base:\n <<: !include loop.yaml\n"
with pytest.raises(EsphomeError, match="Maximum include chain depth"):
yaml_util.parse_yaml(parent, io.StringIO(yaml_text), self_referencing_loader)
def _located(value, doc: str, line: int, col: int):
"""Return *value* wrapped with a fake ESPHomeDataBase source location."""
loc = DocumentLocation(doc, line, col)
obj = make_data_base(value)
if isinstance(obj, ESPHomeDataBase):
obj._esp_range = DocumentRange(loc, loc)
return obj
def test_format_path_no_location_info_returns_flat_path():
"""Plain path items with no esp_range produce a simple flat 'In:' line."""
result = format_path(["wifi", "ssid"], None)
assert result == "In: wifi->ssid"
def test_format_path_no_location_info_current_obj_adds_file():
"""When path has no location but current_obj does, its location is shown."""
obj = _located("${var}", "main.yaml", 5, 10)
result = format_path(["wifi", "ssid"], obj)
assert result == "In: wifi->ssid in main.yaml 6:11"
def test_format_path_single_frame_no_include_boundary():
"""All located keys from the same document → single 'In:' line, no 'Included from'."""
path = ["packages", _located("pkg1", "root.yaml", 5, 2)]
result = format_path(path, None)
assert result.startswith("In: packages->pkg1 in root.yaml 6:3")
assert "Included from" not in result
def test_format_path_two_frames_shows_included_from():
"""Keys from two different documents produce 'In:' + one 'Included from' line."""
path = [
"packages",
_located("device", "root.yaml", 10, 2),
"packages",
_located("inner", "hardware.yaml", 3, 2),
]
result = format_path(path, None)
assert "In: packages->inner in hardware.yaml 4:3" in result
assert "Included from packages->device in root.yaml 11:3" in result
def test_format_path_three_frames_full_include_stack():
"""Three document levels produce two 'Included from' lines in correct order."""
path = [
"packages",
_located("device", "root.yaml", 10, 2),
"packages",
_located("_wifi_", "hardware.yaml", 43, 2),
"packages",
_located("_roam_", "wifi.yaml", 25, 2),
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0].startswith("In: packages->_roam_ in wifi.yaml")
assert lines[1].startswith(" Included from packages->_wifi_ in hardware.yaml")
assert lines[2].startswith(" Included from packages->device in root.yaml")
def test_format_path_current_obj_overrides_innermost_location():
"""current_obj's esp_range replaces the key's column for the 'In:' line."""
path = ["packages", _located("pkg1", "root.yaml", 5, 2)]
# Value (the expression) sits at column 10, not column 2 like the key
value = _located("${undefined}", "root.yaml", 5, 10)
result = format_path(path, value)
assert "6:11" in result
assert "6:3" not in result
def test_format_path_empty_path_with_no_location():
"""Empty path with no location info returns 'In: '."""
result = format_path([], None)
assert result == "In: "
def test_format_path_integer_path_items_formatted_as_subscript():
"""Integer indices are rendered as [n] subscripts in the flat fallback."""
result = format_path(["packages", 0], None)
assert result == "In: packages[0]"
def test_format_path_integer_list_index_attached_to_previous_frame():
"""A list index between two include boundaries attaches to the outer frame."""
path = [
"packages",
_located("packages", "main.yaml", 5, 0),
0,
_located("packages", "level1.yaml", 2, 0),
0,
_located("esphome", "level2.yaml", 0, 0),
_located("name", "level2.yaml", 1, 8),
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0].startswith("In: esphome->name in level2.yaml")
assert "packages[0]" in lines[1] and "level1.yaml" in lines[1]
assert "packages[0]" in lines[2] and "main.yaml" in lines[2]
def test_format_path_trailing_unlocated_string_after_located_key():
"""Plain string keys after the last located key must still appear in output."""
path = [_located("packages", "main.yaml", 5, 0), "sub", "key"]
result = format_path(path, None)
assert result == "In: packages->sub->key in main.yaml 6:1"
def test_format_path_trailing_unlocated_int_attaches_to_current_frame():
"""Trailing ints attach to the open frame's last key (subscript), strings
buffer until end-of-path and then flush behind."""
path = [_located("packages", "main.yaml", 5, 0), 0, "sub"]
result = format_path(path, None)
# Int attaches to 'packages' as [0] subscript; trailing 'sub' is flushed
# at end and appears after.
assert result == "In: packages[0]->sub in main.yaml 6:1"
def test_format_path_only_trailing_unlocated_strings_are_preserved():
"""Trailing pending items must not be silently dropped after the last frame."""
path = [
_located("packages", "main.yaml", 5, 0),
_located("inner", "hardware.yaml", 3, 0),
"tail1",
"tail2",
]
result = format_path(path, None)
lines = result.splitlines()
assert lines[0] == "In: inner->tail1->tail2 in hardware.yaml 4:1"
assert lines[1] == " Included from packages in main.yaml 6:1"
def test_format_path_leading_int_with_no_current_doc_goes_to_pending():
"""An int before any located key is buffered and shown in the first frame."""
path = [0, _located("name", "main.yaml", 1, 0)]
result = format_path(path, None)
# Leading ints have no preceding name to subscript onto, so they render
# as bare [n] in the formatted segment.
assert result == "In: [0]->name in main.yaml 2:1"
def test_format_path_only_unlocated_int_returns_flat_fallback():
"""Path with only an int and no location info renders via the flat fallback."""
result = format_path([0], None)
assert result == "In: [0]"
def test_format_path_current_obj_in_different_doc_than_innermost_frame():
"""current_obj's location is preferred even when its document differs from the frame's."""
path = [_located("packages", "root.yaml", 1, 0)]
value = _located("${var}", "other.yaml", 9, 4)
result = format_path(path, value)
# Innermost line uses current_obj's mark (other.yaml 10:5), not the key's.
assert result == "In: packages in other.yaml 10:5"
def test_format_path_current_obj_without_location_falls_back_to_key():
"""An ESPHomeDataBase current_obj with no esp_range falls back to the key's location."""
class _NoRange(ESPHomeDataBase, str):
pass
obj = _NoRange.__new__(_NoRange, "value")
str.__init__(obj)
# No _esp_range set on this instance.
assert obj.esp_range is None
path = [_located("packages", "main.yaml", 5, 2)]
result = format_path(path, obj)
assert result == "In: packages in main.yaml 6:3"
def test_format_path_empty_path_with_located_current_obj():
"""An empty path with a located current_obj still surfaces the location."""
obj = _located("${var}", "main.yaml", 0, 0)
result = format_path([], obj)
assert result == "In: in main.yaml 1:1"