Merge branch 'dev' into platformio-prefetch-git-clones

# Conflicts:
#	esphome/platformio/prefetch.py
This commit is contained in:
J. Nick Koston
2026-08-31 13:23:59 -05:00
175 changed files with 3960 additions and 1034 deletions
@@ -0,0 +1,155 @@
"""Tests for user-defined action field metadata (description / example)."""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components.api import (
_action_strings,
_action_strings_size,
_has_action_metadata,
_validate_esp8266_action_strings,
validate_variable,
)
from esphome.config_validation import Invalid
from esphome.const import PlatformFramework
from esphome.core import CORE
from esphome.cpp_generator import safe_exp
from esphome.helpers import fnv1_hash
from tests.component_tests.helpers import get_define_value
from tests.component_tests.types import SetCoreConfigCallable
CONFIG = "tests/component_tests/api/test_action_metadata.yaml"
CONFIG_ESP8266 = "tests/component_tests/api/test_action_metadata_esp8266.yaml"
CONFIG_SHORTHAND = "tests/component_tests/api/test_action_metadata_shorthand.yaml"
def test_metadata_is_emitted_as_progmem_table(
generate_main: Callable[[str | Path], str],
) -> None:
"""Every action string is a PROGMEM array referenced from one PROGMEM table."""
main_cpp = generate_main(CONFIG)
assert (
'static constexpr char api_action_str0[] PROGMEM = "play_buzzer";' in main_cpp
)
assert (
'static constexpr char api_action_str1[] PROGMEM = "Play an RTTTL melody on the buzzer";'
in main_cpp
)
assert (
'static constexpr char api_action_str4[] PROGMEM = "two_short:d=4,o=5,b=100:16e6,16e6";'
in main_cpp
)
assert (
"static constexpr const char * api_action0_strings[] PROGMEM = {"
"api_action_str0, api_action_str1, api_action_str2, api_action_str3, "
"api_action_str4, api_action_str5, nullptr, nullptr};" in main_cpp
)
# An action without metadata still carries the metadata slots (as nullptr)
assert (
"static constexpr const char * api_action1_strings[] PROGMEM = {"
"api_action_str6, nullptr, api_action_str7, nullptr, nullptr};" in main_cpp
)
assert f"(api_action0_strings, {safe_exp(fnv1_hash('play_buzzer'))});" in main_cpp
assert "USE_API_USER_DEFINED_ACTION_METADATA" in {d.name for d in CORE.defines}
assert get_define_value("API_USER_ACTION_STRINGS_SCRATCH_SIZE") is None
def test_esp8266_sizes_scratch_buffer_for_largest_action(
generate_main: Callable[[str | Path], str],
) -> None:
"""ESP8266 gets a scratch buffer define equal to the byte total of the largest action."""
generate_main(CONFIG_ESP8266)
# play_buzzer: name, description, two variable names, one description, one example,
# each with a terminator
assert get_define_value("API_USER_ACTION_STRINGS_SCRATCH_SIZE") == "117"
def test_shorthand_variables_emit_no_metadata(
generate_main: Callable[[str | Path], str],
) -> None:
"""The name: type shorthand emits a name-only table and no define."""
main_cpp = generate_main(CONFIG_SHORTHAND)
assert (
"static constexpr const char * api_action0_strings[] PROGMEM = "
"{api_action_str0, api_action_str1};" in main_cpp
)
assert "USE_API_USER_DEFINED_ACTION_METADATA" not in {d.name for d in CORE.defines}
def test_variable_shorthand_normalizes_to_mapping() -> None:
"""A bare type string validates to the mapping form."""
assert validate_variable("string") == {"type": "string"}
@pytest.mark.parametrize(
"value",
[
{"description": "no type given"},
{"type": "string", "selector": "text"},
"stringy",
{"type": "stringy"},
],
)
def test_variable_rejects_invalid(value: object) -> None:
"""Missing or unknown type and unknown keys raise in both forms."""
with pytest.raises(Invalid):
validate_variable(value)
def _oversized_action_config() -> dict:
return {
"actions": [
{
"action": "big",
"description": "x" * 300,
"variables": {"a": {"type": "string", "example": "y" * 300}},
}
]
}
def test_esp8266_rejects_actions_over_string_budget(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP8266_ARDUINO)
with pytest.raises(Invalid, match="ESP8266 allows at most 384 bytes"):
_validate_esp8266_action_strings(_oversized_action_config())
def test_other_platforms_have_no_string_budget(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF)
config = _oversized_action_config()
assert _validate_esp8266_action_strings(config) is config
def test_empty_metadata_is_unset_and_not_counted() -> None:
"""An empty description or example emits nullptr and takes no scratch space."""
conf = {
"action": "a",
"description": "",
"variables": {"b": {"type": "int", "description": "", "example": "ex"}},
}
strings = _action_strings(conf, has_metadata=True)
assert strings == ["a", None, "b", None, "ex"]
# Every emitted string counts its terminator: "a" + "b" + "ex"
assert _action_strings_size(strings) == 2 + 2 + 3
def test_empty_metadata_does_not_enable_the_define() -> None:
actions = [
{
"action": "a",
"description": "",
"variables": {"b": {"type": "int", "example": ""}},
}
]
assert not _has_action_metadata(actions)
actions[0]["variables"]["b"]["example"] = "1"
assert _has_action_metadata(actions)
@@ -0,0 +1,14 @@
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: MySSID
password: password1
logger:
packages:
api: !include test_action_metadata_common.yaml
@@ -0,0 +1,18 @@
api:
actions:
- action: play_buzzer
description: Play an RTTTL melody on the buzzer
variables:
song_str:
type: string
description: RTTTL melody string
example: "two_short:d=4,o=5,b=100:16e6,16e6"
volume:
type: int
then:
- logger.log: Action Called
- action: plain_action
variables:
value: int
then:
- logger.log: Action Called
@@ -0,0 +1,14 @@
esphome:
name: test
esp8266:
board: d1_mini
wifi:
ssid: MySSID
password: password1
logger:
packages:
api: !include test_action_metadata_common.yaml
@@ -0,0 +1,19 @@
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: MySSID
password: password1
logger:
api:
actions:
- action: plain_action
variables:
value: int
then:
- logger.log: Action Called
@@ -9,7 +9,7 @@ def test_synchronous_chain_keeps_zero_copy_args(generate_main):
assert (
"api::UserServiceTrigger<api::enums::SUPPORTS_RESPONSE_NONE, StringRef>"
'("zero_copy_args", {"message"})' in main_cpp
"(api_action0_strings," in main_cpp
)
@@ -22,7 +22,7 @@ def test_response_callback_args_are_owning(generate_main):
assert (
"api::UserServiceTrigger<api::enums::SUPPORTS_RESPONSE_NONE, std::string>"
'("response_args", {"message"})' in main_cpp
"(api_action1_strings," in main_cpp
)
assert "api::HomeAssistantServiceCallAction<std::string>" in main_cpp
assert "api::HomeAssistantServiceCallAction<StringRef>" not in main_cpp
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
sensor:
- platform: internal_temperature
name: Internal Temperature
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32-s3-devkitc-1
framework:
type: esp-idf
sensor:
- platform: internal_temperature
name: Internal Temperature
+62
View File
@@ -313,6 +313,12 @@ def test_esp32_configuration_errors(
("esp_wifi",),
id="espnow",
),
pytest.param(
# temprature_sens_read() on the original ESP32 lives in the esp_phy blob.
"exclusion_reincludes_internal_temperature.yaml",
("esp_phy",),
id="internal_temperature",
),
],
)
def test_default_exclusions_reincluded_by_owning_components(
@@ -337,6 +343,15 @@ def test_default_exclusions_reincluded_by_owning_components(
assert ("esp_http_server" in excluded) == ("esp_http_server" not in reincluded)
def test_esp_phy_stays_excluded_for_internal_temperature_on_newer_variants(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Only the original ESP32 reads the PHY blob; other variants use esp_driver_tsens."""
generate_main(component_config_path("exclusion_stays_internal_temperature_s3.yaml"))
assert "esp_phy" in CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS]
def test_nvs_sec_provider_stays_excluded_when_encryption_is_off(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
@@ -1253,3 +1268,50 @@ def test_parse_pio_platform_version(value: str, expected: str) -> None:
from esphome.components.esp32 import _parse_pio_platform_version
assert _parse_pio_platform_version(value) == expected
def test_esp32_s31_gpio_validation(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""S31: GPIO26-28/30-32 are reserved for the SPI flash interface, GPIO29
and GPIO41 do not exist, GPIO33 is a normal pin, and GPIO36 is a
strapping pin."""
from esphome.components.esp32.const import VARIANT_ESP32S31
from esphome.components.esp32.gpio import validate_supports
from esphome.const import CONF_INPUT, CONF_MODE, CONF_OPEN_DRAIN, CONF_OUTPUT
set_core_config(
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32S31}
)
input_mode = {CONF_INPUT: True, CONF_OUTPUT: False, CONF_OPEN_DRAIN: False}
# Not reserved; a normal GPIO
pin = {CONF_NUMBER: 33, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
assert validate_gpio_pin(pin)[CONF_NUMBER] == 33
# Reserved for the SPI flash interface, but can be bypassed with
# ignore_pin_validation_error
for num in (26, 27, 28, 30, 31, 32):
with pytest.raises(cv.Invalid, match=f"GPIO{num} is reserved"):
validate_gpio_pin(
{CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
)
pin = {CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: True}
assert validate_gpio_pin(pin)[CONF_NUMBER] == num
for num in (29, 41):
with pytest.raises(cv.Invalid, match=f"GPIO{num} does not exist"):
validate_gpio_pin(
{CONF_NUMBER: num, CONF_IGNORE_PIN_VALIDATION_ERROR: False}
)
# Also rejected in validate_supports so ignore_pin_validation_error
# cannot bypass it
with pytest.raises(cv.Invalid, match=f"GPIO{num} does not exist"):
validate_supports({CONF_NUMBER: num, CONF_MODE: input_mode})
pin = {CONF_NUMBER: 36, CONF_MODE: input_mode}
with caplog.at_level("WARNING"):
validate_supports(pin)
assert "GPIO36 is a strapping PIN" in caplog.text
@@ -10,7 +10,13 @@ from esphome import config_validation as cv
# via ch422g) can be validated by the mipi_rgb CONFIG_SCHEMA in this test.
import esphome.components.ch422g # noqa: F401
from esphome.components.display import get_display_metadata
from esphome.components.esp32 import KEY_BOARD, VARIANT_ESP32S3
from esphome.components.esp32 import (
KEY_BOARD,
VARIANT_ESP32C3,
VARIANT_ESP32P4,
VARIANT_ESP32S3,
VARIANT_ESP32S31,
)
import esphome.components.pca9554 # noqa: F401
import esphome.components.xl9535 # noqa: F401
from esphome.const import (
@@ -135,3 +141,63 @@ def test_metadata_records_rotation(
config = CONFIG_SCHEMA({**base, "id": "unrotated"})
assert get_display_metadata(config["id"]).rotation == 0
@pytest.mark.parametrize(
("variant", "board", "model"),
[
# ESP32-8048S070 is a real Sunton board wired for ESP32-S3 (e.g. its
# default de_pin is GPIO41, which doesn't exist on S31), so it is
# only meaningful as a config on that variant.
(VARIANT_ESP32S3, "esp32-s3-devkitc-1", "ESP32-8048S070"),
# P4 and S31 use the pin-agnostic CUSTOM model so this only checks
# that the chip itself is accepted, independent of board wiring.
(VARIANT_ESP32P4, "esp32-p4-evboard", "CUSTOM"),
# No dedicated board is registered for ESP32-S31 yet; an unknown board
# name simply skips per-board pin validation.
(VARIANT_ESP32S31, "esp32-s31-devkitc", "CUSTOM"),
],
)
def test_configuration_succeeds_on_supported_variants(
variant: str, board: str, model: str, set_core_config: SetCoreConfigCallable
) -> None:
"""mipi_rgb requires a chip with an RGB LCD peripheral: S3, P4 or S31."""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: board, KEY_VARIANT: variant},
)
from esphome.components.mipi_rgb.display import CONFIG_SCHEMA
config = {"model": model, "data_pins": DATA_PINS, "pclk_pin": 21}
if model == "CUSTOM":
config[CONF_INIT_SEQUENCE] = [[0xA0, 0x01]]
config[CONF_DIMENSIONS] = {CONF_WIDTH: 480, CONF_HEIGHT: 480}
CONFIG_SCHEMA(config)
def test_only_on_variant_rejects_unsupported_variant(
set_core_config: SetCoreConfigCallable,
) -> None:
"""A variant without the RGB LCD peripheral (e.g. ESP32-C3) is rejected.
Exercises the exact ``only_on_variant`` call used by ``mipi_rgb.display``
directly, since building a full model config with GPIO numbers that are
also valid on an unsupported variant like ESP32-C3 is unrelated to what
this checks.
"""
from esphome.components.esp32 import only_on_variant
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_VARIANT: VARIANT_ESP32C3},
)
validator = only_on_variant(
supported=[VARIANT_ESP32S3, VARIANT_ESP32P4, VARIANT_ESP32S31]
)
with pytest.raises(
cv.Invalid,
match=r"This feature is only available on ESP32S3, ESP32P4, ESP32S31",
):
validator({})
+5 -1
View File
@@ -61,8 +61,12 @@ api:
reboot_timeout: 0min
actions:
- action: hello_world
description: Log a greeting
variables:
name: string
name:
type: string
description: Name to greet
example: World
then:
- logger.log:
format: Hello World %s!
+2 -1
View File
@@ -1,4 +1,5 @@
<<: !include common-base.yaml
packages:
base: !include common-base.yaml
api:
encryption:
+3 -3
View File
@@ -328,10 +328,10 @@ TEST(StepToAccuracyDecimals, RoundsUpToWholeNumber) {
}
TEST(StepToAccuracyDecimals, OutsideFixedNotationRange) {
// %.5g prints these in exponent form, so the count comes from parsing "1e-05" or "1.2346e+05".
EXPECT_EQ(step_to_accuracy_decimals(0.00001f), 0);
// %.5g would print these in exponent form; the count is now the real one rather than a parse of "1e-05".
EXPECT_EQ(step_to_accuracy_decimals(0.00001f), 5);
EXPECT_EQ(step_to_accuracy_decimals(0.000125f), 6);
EXPECT_EQ(step_to_accuracy_decimals(123456.0f), 8);
EXPECT_EQ(step_to_accuracy_decimals(123456.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(1000000.0f), 0);
}
@@ -0,0 +1,17 @@
ethernet:
type: W5500
clk_pin: 19
mosi_pin: 21
miso_pin: 17
cs_pin: 18
interrupt_pin: 36
reset_pin: 12
clock_speed: 10Mhz
logger:
hardware_uart: UART0
# Exercises the per-interface webserver URL collection at compile time
web_server:
improv_serial:
@@ -0,0 +1,2 @@
packages:
improv_serial: !include common-ethernet.yaml
@@ -0,0 +1,92 @@
#include <gtest/gtest.h>
#include <algorithm>
#include <array>
#include <cmath>
#include <cstdint>
#include "esphome/components/light/esp_color_correction.h"
namespace esphome::light::testing {
namespace {
// A representative fixture for ESPColorCorrection/gamma_table_reverse_search tests below --
// not a spec for generate_gamma_table() itself, which the Python tests own.
std::array<uint16_t, 256> build_gamma_table(double gamma) {
std::array<uint16_t, 256> table{};
table[0] = 0;
for (int i = 1; i < 256; i++) {
double raw = std::round(std::pow(i / 255.0, gamma) * 65535.0);
table[i] = static_cast<uint16_t>(std::max(1.0, std::min(65535.0, raw)));
}
return table;
}
// Bundles a table with an ESPColorCorrection pointing at it, since the correction only holds
// a raw pointer into the table and doesn't own it.
struct GammaFixture {
explicit GammaFixture(double gamma) : table(build_gamma_table(gamma)) { correction.set_gamma_table(table.data()); }
std::array<uint16_t, 256> table;
ESPColorCorrection correction;
};
} // namespace
// Regression test for esphome/esphome#18842: ESPColorCorrection's own 16-bit -> 8-bit
// conversion must never round a non-zero table entry down to a zero 8-bit output.
TEST(GammaCorrection, NonZeroInputsSurviveConversion) {
for (double gamma : {1.0, 1.8, 2.0, 2.2, 2.8, 3.0, 4.0}) {
GammaFixture fixture(gamma);
for (int i = 1; i < 256; i++) {
EXPECT_GE(fixture.correction.color_correct_red(i), 1) << "gamma=" << gamma << " index=" << i;
}
}
}
TEST(GammaCorrection, ZeroInputStaysZero) {
for (double gamma : {1.0, 2.2, 2.8, 4.0}) {
GammaFixture fixture(gamma);
EXPECT_EQ(fixture.correction.color_correct_red(0), 0) << "gamma=" << gamma;
}
}
TEST(GammaCorrection, FullBrightnessStaysFull) {
for (double gamma : {1.0, 2.2, 2.8, 4.0}) {
GammaFixture fixture(gamma);
EXPECT_EQ(fixture.correction.color_correct_red(255), 255) << "gamma=" << gamma;
}
}
// Reproduces the reporter's own numbers from esphome/esphome#18842 at gamma=2.8: codes
// 1-27 previously collapsed to an 8-bit output of 0 and must now be non-zero.
TEST(GammaCorrection, DeadZoneFixedAtGamma28) {
GammaFixture fixture(2.8);
for (int i = 1; i < 28; i++) {
EXPECT_GE(fixture.correction.color_correct_red(i), 1) << "index=" << i << " still collapses to 0";
}
}
TEST(GammaCorrection, ReverseSearchFindsLargestIndexLessEqualTarget) {
auto table = build_gamma_table(2.8);
for (uint16_t target : {0, 128, 129, 135, 1000, 32768, 65535}) {
uint8_t lo = gamma_table_reverse_search(table.data(), target);
EXPECT_LE(table[lo], target) << "target=" << target;
if (lo < 255) {
EXPECT_GT(table[lo + 1], target) << "target=" << target;
}
}
}
// color_uncorrect_* binary-searches the table via gamma_table_reverse_search().
TEST(GammaCorrection, UncorrectStaysMonotonic) {
GammaFixture fixture(2.8);
uint8_t prev = 0;
for (int i = 1; i < 256; i++) {
uint8_t result = fixture.correction.color_uncorrect_red(i);
EXPECT_GE(result, prev) << "index=" << i;
prev = result;
}
}
} // namespace esphome::light::testing
+65
View File
@@ -209,6 +209,63 @@ lvgl:
position: 212
- color: 0xFF0000
position: 255
- id: linear_grad
direction: LINEAR
linear:
from_x: 0%
from_y: 0%
to_x: 100%
to_y: 0%
extend: REFLECT
stops:
- color: 0xFF0000
position: 0
- color: 0x0000FF
position: 255
- id: radial_grad
direction: RADIAL
radial:
center_x: 50%
center_y: 50%
to_x: 100%
to_y: 50%
extend: PAD
stops:
- color: 0xFFFFFF
position: 0
- color: 0x000000
position: 255
- id: radial_focal_grad
direction: RADIAL
radial:
center_x: 50%
center_y: 50%
to_x: 100%
to_y: 50%
focal_x: 40%
focal_y: 40%
focal_radius: 10
extend: REPEAT
stops:
- color: 0xFF0000
position: 0
- color: 0x0000FF
position: 255
- id: conical_grad
direction: CONICAL
conical:
center_x: 50%
center_y: 50%
start_angle: 0
end_angle: 360
extend: PAD
stops:
- color: 0xFF0000
position: 0
- color: 0x00FF00
position: 127
- color: 0xFF0000
position: 255
style_definitions:
- id: style_test
@@ -1070,6 +1127,14 @@ lvgl:
logger.log:
format: Slider released at %d/%d with value %.0f
args: ['(int) point.x', '(int) point.y', x]
# Exercises the style-application path for a complex gradient, not just its
# lv_grad_*_init() codegen: the other new gradients are only ever declared.
- obj:
bg_opa: cover
bg_grad: conical_grad
width: 40
height: 40
- button:
styles: spin_button
id: spin_up
+46
View File
@@ -0,0 +1,46 @@
mk2pvrouter:
id: test_mk2pvrouter
uart_id: uart_bus
sensor:
- platform: mk2pvrouter
name: Power
tag: P
mk2pvrouter_id: test_mk2pvrouter
unit_of_measurement: W
device_class: power
state_class: measurement
accuracy_decimals: 0
- platform: mk2pvrouter
name: Voltage
tag: V
mk2pvrouter_id: test_mk2pvrouter
unit_of_measurement: V
device_class: voltage
state_class: measurement
accuracy_decimals: 2
filters:
# Device sends voltage * 100
- multiply: 0.01
- platform: mk2pvrouter
name: Energy
tag: E
mk2pvrouter_id: test_mk2pvrouter
unit_of_measurement: Wh
device_class: energy
state_class: total_increasing
accuracy_decimals: 0
- platform: mk2pvrouter
name: Temperature
tag: T1
mk2pvrouter_id: test_mk2pvrouter
unit_of_measurement: "°C"
device_class: temperature
state_class: measurement
accuracy_decimals: 2
filters:
# Device sends temperature * 100
- multiply: 0.01
@@ -0,0 +1,3 @@
packages:
uart_9600_even_7bits: !include ../../test_build_components/common/uart_9600_even_7bits/esp32-idf.yaml
mk2pvrouter: !include common.yaml
@@ -0,0 +1,3 @@
packages:
uart_9600_even_7bits: !include ../../test_build_components/common/uart_9600_even_7bits/esp8266-ard.yaml
mk2pvrouter: !include common.yaml
@@ -0,0 +1,3 @@
packages:
uart_9600_even_7bits: !include ../../test_build_components/common/uart_9600_even_7bits/rp2040-ard.yaml
mk2pvrouter: !include common.yaml
+8 -1
View File
@@ -11,7 +11,14 @@ namespace esphome::modbus::testing {
// A UART that discards all writes, for tests that never inspect the wire.
class NullUART : public uart::UARTComponent {
public:
NullUART() { this->set_baud_rate(115200); }
// 8N1, matching what the uart schema emits for a real hub; the framing drives the modbus
// interframe timing, so leaving data/stop bits at their zero defaults would not be representative.
NullUART() {
this->set_baud_rate(115200);
this->set_data_bits(8);
this->set_stop_bits(1);
this->set_parity(uart::UART_CONFIG_PARITY_NONE);
}
void write_array(const uint8_t *data, size_t len) override {}
bool peek_byte(uint8_t *data) override { return false; }
bool read_array(uint8_t *data, size_t len) override { return false; }
@@ -0,0 +1,64 @@
#include <gtest/gtest.h>
#include <cstdint>
#include "common.h"
#include "esphome/components/modbus/modbus.h"
namespace esphome::modbus::testing {
namespace {
// Exposes the timing values setup() derives from the UART framing.
class FramingProbeHub : public ModbusClientHub {
public:
uint32_t bits_per_char() const { return this->bits_per_char_; }
uint32_t frame_delay_us() const { return this->frame_delay_us_; }
};
class FramedUART : public NullUART {
public:
FramedUART(uint32_t baud_rate, uint8_t data_bits, uint8_t stop_bits, uart::UARTParityOptions parity) {
this->set_baud_rate(baud_rate);
this->set_data_bits(data_bits);
this->set_stop_bits(stop_bits);
this->set_parity(parity);
}
};
} // namespace
// 8N1 is 10 bits on the wire, so t3.5 at 9600 baud is 3.5 * 10 / 9600 = 3645.8us.
TEST(ModbusFraming, EightNoneOneDerivesTenBits) {
FramedUART uart(9600, 8, 1, uart::UART_CONFIG_PARITY_NONE);
FramingProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
EXPECT_EQ(hub.bits_per_char(), 10u);
EXPECT_EQ(hub.frame_delay_us(), 3646u);
}
// Spec-conformant RTU framing is 11 bits, which lengthens the interframe gap to
// 3.5 * 11 / 9600 = 4010.4us, rounded up.
TEST(ModbusFraming, EightEvenOneDerivesElevenBits) {
FramedUART uart(9600, 8, 1, uart::UART_CONFIG_PARITY_EVEN);
FramingProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
EXPECT_EQ(hub.bits_per_char(), 11u);
EXPECT_EQ(hub.frame_delay_us(), 4011u);
}
// Above 19200 baud the spec's fixed 1750us floor governs instead of 3.5 characters.
TEST(ModbusFraming, FastBaudUsesSpecFloor) {
FramedUART uart(115200, 8, 1, uart::UART_CONFIG_PARITY_NONE);
FramingProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
EXPECT_EQ(hub.frame_delay_us(), 1750u);
}
} // namespace esphome::modbus::testing
@@ -427,11 +427,132 @@ TEST(ModbusHelpersTest, RegistersToNumberMatchesPayloadToNumber) {
}
}
TEST(ModbusHelpersTest, RegistersToNumberMatchesPayloadToNumberForQwords) {
// The word shuffle the QWORD_R decode replaces is the least obvious code in the byte path, so pin
// it against that path rather than against registers_to_value(). The top bit is set, which is where
// U_QWORD's unsigned value and this function's int64_t return deliberately diverge.
const uint16_t registers[] = {0xF123, 0x4567, 0x89AB, 0xCDEF};
const std::vector<uint8_t> bytes{0xF1, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF};
for (auto value_type :
{SensorValueType::U_QWORD, SensorValueType::S_QWORD, SensorValueType::U_QWORD_R, SensorValueType::S_QWORD_R}) {
EXPECT_EQ(registers_to_number(registers, 4, value_type),
payload_to_number(std::span<const uint8_t>(bytes), value_type, 0, 0xFFFFFFFF))
<< "value_type=" << static_cast<int>(value_type);
}
}
TEST(ModbusHelpersTest, RegistersToNumberTreatsRawAndBitAsNothingToDecode) {
// Both have no fixed-width number, so they decode to 0 whatever the span holds - including none.
const uint16_t registers[] = {0x1234};
EXPECT_EQ(registers_to_number(registers, 1, SensorValueType::RAW), std::optional<int64_t>(0));
EXPECT_EQ(registers_to_number(registers, 0, SensorValueType::RAW), std::optional<int64_t>(0));
EXPECT_EQ(registers_to_number(registers, 0, SensorValueType::BIT), std::optional<int64_t>(0));
}
TEST(ModbusHelpersTest, RegistersToNumberRejectsTruncatedMultiRegisterValue) {
const uint16_t registers[] = {0x1234};
EXPECT_FALSE(registers_to_number(registers, 1, SensorValueType::U_DWORD).has_value());
}
// --- registers_to_value ----------------------------------------------------
// registers_to_number() dispatches to registers_to_value(), so this checks the dispatch table picks
// the right specialisation for each type, not that two implementations agree. The independent check
// against the byte decoder is RegistersToNumberMatchesPayloadToNumber below.
template<SensorValueType VALUE_TYPE> void expect_matches_registers_to_number(const uint16_t *registers) {
const auto expected = registers_to_number(registers, register_width_for(VALUE_TYPE), VALUE_TYPE);
// Plain control flow rather than ASSERT_TRUE: the optional analysis does not see through the macro.
if (!expected.has_value()) {
ADD_FAILURE() << "registers_to_number() returned no value for value_type=" << static_cast<int>(VALUE_TYPE);
return;
}
const int64_t number = expected.value();
if constexpr (VALUE_TYPE == SensorValueType::FP32 || VALUE_TYPE == SensorValueType::FP32_R) {
EXPECT_FLOAT_EQ(registers_to_value<VALUE_TYPE>(registers), bit_cast<float>(static_cast<uint32_t>(number)))
<< "value_type=" << static_cast<int>(VALUE_TYPE);
} else {
EXPECT_EQ(static_cast<int64_t>(registers_to_value<VALUE_TYPE>(registers)), number)
<< "value_type=" << static_cast<int>(VALUE_TYPE);
}
}
TEST(ModbusHelpersTest, RegistersToValueMatchesRegistersToNumber) {
// A high bit in each word exercises sign handling and word order together.
const uint16_t registers[] = {0x8001, 0xFE02};
expect_matches_registers_to_number<SensorValueType::U_WORD>(registers);
expect_matches_registers_to_number<SensorValueType::S_WORD>(registers);
expect_matches_registers_to_number<SensorValueType::U_WORD_S>(registers);
expect_matches_registers_to_number<SensorValueType::S_WORD_S>(registers);
expect_matches_registers_to_number<SensorValueType::U_DWORD>(registers);
expect_matches_registers_to_number<SensorValueType::U_DWORD_R>(registers);
expect_matches_registers_to_number<SensorValueType::S_DWORD>(registers);
expect_matches_registers_to_number<SensorValueType::S_DWORD_R>(registers);
expect_matches_registers_to_number<SensorValueType::FP32>(registers);
expect_matches_registers_to_number<SensorValueType::FP32_R>(registers);
}
TEST(ModbusHelpersTest, RegistersToUint32CombinesWordsHighFirst) {
EXPECT_EQ(registers_to_uint32(0x1234, 0x5678), 0x12345678u);
}
// --- value_at ---------------------------------------------------------------
// Addresses are absolute; anything not wholly inside the response yields nullopt.
TEST(ModbusHelpersTest, ValueAtDecodesByAbsoluteAddress) {
const uint16_t registers[] = {0x1111, 0x2222, 0x3333};
const std::span<const uint16_t> span(registers, 3);
EXPECT_EQ(value_at<SensorValueType::U_WORD>(span, 100, 100), std::optional<uint16_t>(0x1111));
EXPECT_EQ(value_at<SensorValueType::U_WORD>(span, 100, 102), std::optional<uint16_t>(0x3333));
EXPECT_EQ(value_at<SensorValueType::U_DWORD>(span, 100, 101), std::optional<uint32_t>(0x22223333u));
// Types whose RegisterValueType<> is not an unsigned integer, and the widest bounds check.
const uint16_t floats[] = {0x4048, 0xF5C3, 0xF5C3, 0x4048};
const std::span<const uint16_t> float_span(floats, 4);
EXPECT_FLOAT_EQ(value_at<SensorValueType::FP32>(float_span, 10, 10).value_or(0.0f), 3.14f);
EXPECT_FLOAT_EQ(value_at<SensorValueType::FP32_R>(float_span, 10, 12).value_or(0.0f), 3.14f);
EXPECT_EQ(value_at<SensorValueType::U_QWORD>(float_span, 10, 10), std::optional<uint64_t>(0x4048F5C3F5C34048ULL));
EXPECT_FALSE(value_at<SensorValueType::U_QWORD>(float_span, 10, 11).has_value());
}
TEST(ModbusHelpersTest, ValueAtIsUsableInAConstantExpression) {
static constexpr uint16_t REGISTERS[] = {0x1234, 0x5678};
static_assert(value_at<SensorValueType::U_DWORD>(REGISTERS, 7, 7).value_or(0) == 0x12345678u);
static_assert(!value_at<SensorValueType::U_DWORD>(REGISTERS, 7, 6).has_value());
}
TEST(ModbusHelpersTest, ValueAtRejectsAddressesOutsideTheResponse) {
const uint16_t registers[] = {0x1111, 0x2222, 0x3333};
const std::span<const uint16_t> span(registers, 3);
// Below the response: must not wrap when the subtraction would go negative.
EXPECT_FALSE(value_at<SensorValueType::U_WORD>(span, 100, 99).has_value());
EXPECT_FALSE(value_at<SensorValueType::U_WORD>(span, 100, 0).has_value());
// Past the end, and a multi-register value truncated by the end of the response.
EXPECT_FALSE(value_at<SensorValueType::U_WORD>(span, 100, 103).has_value());
EXPECT_FALSE(value_at<SensorValueType::U_DWORD>(span, 100, 102).has_value());
EXPECT_TRUE(value_at<SensorValueType::U_DWORD>(span, 100, 101).has_value());
}
TEST(ModbusHelpersTest, ValueAtHandlesAnEmptyResponse) {
EXPECT_FALSE(value_at<SensorValueType::U_WORD>(std::span<const uint16_t>(), 0, 0).has_value());
}
// --- QWORD decoding ---------------------------------------------------------
TEST(ModbusHelpersTest, RegistersToValueDecodesQwordBothWordOrders) {
const uint16_t registers[] = {0x0123, 0x4567, 0x89AB, 0xCDEF};
EXPECT_EQ(registers_to_value<SensorValueType::U_QWORD>(registers), 0x0123456789ABCDEFULL);
const uint16_t reversed[] = {0xCDEF, 0x89AB, 0x4567, 0x0123};
EXPECT_EQ(registers_to_value<SensorValueType::U_QWORD_R>(reversed), 0x0123456789ABCDEFULL);
// Signed reading of the same bits, and the sign-extreme case.
EXPECT_EQ(registers_to_value<SensorValueType::S_QWORD>(registers), 0x0123456789ABCDEFLL);
const uint16_t negative[] = {0xFFFF, 0xFFFF, 0xFFFF, 0xFFFE};
EXPECT_EQ(registers_to_value<SensorValueType::S_QWORD>(negative), -2);
EXPECT_EQ(registers_to_value<SensorValueType::U_QWORD>(negative), 0xFFFFFFFFFFFFFFFEULL);
}
TEST(ModbusHelpersTest, RegistersToUint64CombinesWordsHighFirst) {
EXPECT_EQ(registers_to_uint64(0x0123, 0x4567, 0x89AB, 0xCDEF), 0x0123456789ABCDEFULL);
}
// --- packed bit helpers ------------------------------------------------------
TEST(ModbusHelpersTest, PackBitsAppendsToContainer) {
+13
View File
@@ -28,8 +28,21 @@ sensor:
accuracy_decimals: 1
nox_index:
name: NOx Index
algorithm_tuning:
index_offset: 8
learning_time_offset_hours: 6
learning_time_gain_hours: 24
gating_max_duration_minutes: 900
gain_factor: 180
voc_index:
name: VOC Index
algorithm_tuning:
index_offset: 120
learning_time_offset_hours: 6
learning_time_gain_hours: 24
gating_max_duration_minutes: 240
std_initial: 75
gain_factor: 180
co2:
name: Carbon Dioxide
formaldehyde:
@@ -0,0 +1,18 @@
# Config-only: partial algorithm_tuning blocks, so the schema defaults fill in the
# keys that are left out.
packages:
i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
sensor:
- platform: sen6x
id: sen6x_partial_tuning
type: SEN65
i2c_id: i2c_bus
voc_index:
name: VOC Index
algorithm_tuning:
index_offset: 60
nox_index:
name: NOx Index
algorithm_tuning:
gain_factor: 45
+2 -1
View File
@@ -78,7 +78,8 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
@pytest.fixture(scope="session")
def shared_platformio_cache() -> Generator[Path]:
"""Initialize a shared PlatformIO cache for all integration tests."""
# Use a dedicated directory for integration tests to avoid conflicts
# Use a dedicated directory for integration tests to avoid conflicts.
# CI caches parts of this path; keep in sync with ci.yml integration-tests.
test_cache_dir = Path.home() / ".esphome-integration-tests"
cache_dir = test_cache_dir / "platformio"
@@ -0,0 +1,26 @@
esphome:
name: api-action-metadata-test
host:
api:
batch_delay: 0ms
actions:
- action: play_buzzer
description: Play an RTTTL melody on the buzzer
variables:
song_str:
type: string
description: RTTTL melody string
example: "two_short:d=4,o=5,b=100:16e6,16e6"
volume:
type: int
then:
- logger.log:
format: "Buzzer: %s"
args: [song_str.c_str()]
- action: plain_action
variables:
value: int
then:
- logger.log: "Plain action called"
logger:
@@ -0,0 +1,59 @@
esphome:
name: ha-bs-initial
host:
api:
logger:
level: DEBUG
binary_sensor:
# trigger_on_initial_state: true must fire on_press for the first state from HA
- platform: homeassistant
name: Initial On
entity_id: binary_sensor.initial_on
trigger_on_initial_state: true
on_press:
- logger.log: "initial_on on_press"
on_release:
- logger.log: "initial_on on_release"
# Default (false) must not fire on the first state, only on later changes
- platform: homeassistant
name: Default
entity_id: binary_sensor.default
on_press:
- logger.log: "default on_press"
on_release:
- logger.log: "default on_release"
# Real HA startup shape: 'unavailable' arrives before the first real state
- platform: homeassistant
name: Unavailable First
entity_id: binary_sensor.unavailable_first
trigger_on_initial_state: true
on_press:
- logger.log: "unavailable_first on_press"
on_release:
- logger.log: "unavailable_first on_release"
# Initial 'off' must fire on_release when trigger_on_initial_state is set
- platform: homeassistant
name: Initial Off
entity_id: binary_sensor.initial_off
trigger_on_initial_state: true
on_press:
- logger.log: "initial_off on_press"
on_release:
- logger.log: "initial_off on_release"
# Same 'unavailable' first shape without the flag; must stay quiet on the
# first real state and only fire on the later change
- platform: homeassistant
name: Default Unavailable First
entity_id: binary_sensor.default_unavail
on_press:
- logger.log: "default_unavail on_press"
on_release:
- logger.log: "default_unavail on_release"
@@ -29,6 +29,8 @@ void WiFiComponent::set_sta(const WiFiAP &ap) { ESP_LOGI(TAG, "set_sta ssid=%s",
void WiFiComponent::start_connecting(const WiFiAP &ap) {
ESP_LOGI(TAG, "start_connecting ssid=%s", ap.get_ssid().c_str());
// Connecting succeeds immediately, so the requested network is the connected one
this->connected_ssid_ = ap.get_ssid().c_str();
}
void WiFiComponent::clear_sta() { ESP_LOGI(TAG, "clear_sta"); }
@@ -13,11 +13,15 @@
#include "esphome/core/component.h"
#include "esphome/core/string_ref.h"
#include <cstdio>
#include <span>
#include <string>
#include <vector>
namespace esphome::wifi {
static constexpr size_t SSID_BUFFER_SIZE = 33;
class WiFiAP {
public:
void set_ssid(const char *ssid) { this->ssid_ = ssid; }
@@ -58,6 +62,12 @@ class WiFiComponent : public Component {
bool is_disabled() const { return false; }
// Always connected so network::is_connected() keeps the API server accepting clients
bool is_connected() const { return true; }
// Reports the network start_connecting() was last asked for, so a consumer checking that it
// joined the network it requested (rather than an earlier one) sees the connect succeed
const char *wifi_ssid_to(std::span<char, SSID_BUFFER_SIZE> buffer) {
snprintf(buffer.data(), buffer.size(), "%s", this->connected_ssid_.c_str());
return buffer.data();
}
void start_scanning();
const std::vector<WiFiScanResult> &get_scan_result() const { return this->scan_result_; }
void set_sta(const WiFiAP &ap);
@@ -70,6 +80,7 @@ class WiFiComponent : public Component {
protected:
std::vector<WiFiScanResult> scan_result_;
std::string connected_ssid_;
};
extern WiFiComponent *global_wifi_component; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+5
View File
@@ -28,6 +28,11 @@ class LineWaiter:
self._future.set_result(line)
self._future = None
async def wait_for_each(self, *texts: str, timeout: float = 10.0) -> None:
"""Await each text in turn; a text may match a line already received."""
for text in texts:
await self.wait_for(text, timeout=timeout)
async def wait_for(self, *needles: str, timeout: float = 10.0) -> str:
"""Return the first line, past or future, containing every needle."""
for line in self.lines:
@@ -0,0 +1,65 @@
"""Integration test for user-defined action field metadata."""
from __future__ import annotations
import asyncio
import re
import pytest
from esphome.helpers import fnv1_hash
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_api_action_metadata(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Action and argument metadata reach the client and the actions still run."""
loop = asyncio.get_running_loop()
buzzer_called = loop.create_future()
plain_called = loop.create_future()
buzzer_pattern = re.compile(r"Buzzer: two_short")
plain_pattern = re.compile(r"Plain action called")
def check_output(line: str) -> None:
if not buzzer_called.done() and buzzer_pattern.search(line):
buzzer_called.set_result(True)
elif not plain_called.done() and plain_pattern.search(line):
plain_called.set_result(True)
async with (
run_compiled(yaml_config, line_callback=check_output),
api_client_connected() as client,
):
_, services = await client.list_entities_services()
by_name = {service.name: service for service in services}
assert set(by_name) == {"play_buzzer", "plain_action"}
# Keys are hashed at codegen time and must match what the client expects
for name, service in by_name.items():
assert service.key == fnv1_hash(name), name
buzzer = by_name["play_buzzer"]
assert buzzer.description == "Play an RTTTL melody on the buzzer"
args = {arg.name: arg for arg in buzzer.args}
assert args["song_str"].description == "RTTTL melody string"
assert args["song_str"].example == "two_short:d=4,o=5,b=100:16e6,16e6"
# An arg without metadata sends empty strings
assert args["volume"].description == ""
assert args["volume"].example == ""
# An action without metadata sends empty strings
plain = by_name["plain_action"]
assert plain.description == ""
assert plain.args[0].description == ""
await client.execute_service(
buzzer, {"song_str": "two_short:d=4,o=5,b=100:16e6,16e6", "volume": 3}
)
await client.execute_service(plain, {"value": 1})
await asyncio.wait_for(buzzer_called, timeout=5.0)
await asyncio.wait_for(plain_called, timeout=5.0)
@@ -0,0 +1,98 @@
"""Test on_press/on_release for homeassistant binary sensors on the first HA state."""
from __future__ import annotations
import asyncio
import pytest
from .log_utils import LineWaiter
from .types import APIClientConnectedFactory, RunCompiledFunction
ENTITIES = (
"binary_sensor.initial_on",
"binary_sensor.default",
"binary_sensor.unavailable_first",
"binary_sensor.initial_off",
"binary_sensor.default_unavail",
)
@pytest.mark.asyncio
async def test_api_homeassistant_binary_sensor_initial_state(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""The first state from HA fires on_press only with trigger_on_initial_state."""
loop = asyncio.get_running_loop()
waiter = LineWaiter()
subscribed: set[str] = set()
all_subscribed = loop.create_future()
def on_state_sub(entity_id: str, _attribute: str | None) -> None:
subscribed.add(entity_id)
if not all_subscribed.done() and subscribed.issuperset(ENTITIES):
all_subscribed.set_result(None)
async with (
run_compiled(yaml_config, line_callback=waiter.callback),
api_client_connected() as client,
):
client.subscribe_home_assistant_states(on_state_sub)
try:
await asyncio.wait_for(all_subscribed, timeout=5.0)
except TimeoutError:
pytest.fail(f"never subscribed: {set(ENTITIES) - subscribed}")
# First state from HA
client.send_home_assistant_state("binary_sensor.initial_on", "", "on")
client.send_home_assistant_state("binary_sensor.default", "", "on")
client.send_home_assistant_state(
"binary_sensor.unavailable_first", "", "unavailable"
)
client.send_home_assistant_state("binary_sensor.unavailable_first", "", "on")
client.send_home_assistant_state(
"binary_sensor.default_unavail", "", "unavailable"
)
client.send_home_assistant_state("binary_sensor.default_unavail", "", "on")
client.send_home_assistant_state("binary_sensor.initial_off", "", "off")
await waiter.wait_for("initial_on on_press", timeout=5.0)
await waiter.wait_for("unavailable_first on_press", timeout=5.0)
# Pin that the 'unavailable' message actually arrived and was rejected
await waiter.wait_for("Can't convert 'unavailable'", timeout=5.0)
# initial_off is the last state sent, so this wait also proves the
# earlier 'default' initial state was already processed
await waiter.wait_for("initial_off on_release", timeout=5.0)
# Both 'unavailable' senders must have been seen and rejected
assert sum("Can't convert 'unavailable'" in line for line in waiter.lines) == 2
# Guard every phase 2 needle against being satisfied by a stale
# phase 1 line, and pin that the initial states fired nothing else
for absent in (
"initial_on on_release",
"default on_press",
"default on_release",
"default_unavail on_press",
"default_unavail on_release",
"unavailable_first on_release",
"initial_off on_press",
):
assert not any(absent in line for line in waiter.lines), (
f"unexpected trigger before the second state change: {absent}"
)
# A later change fires for all of them
client.send_home_assistant_state("binary_sensor.initial_on", "", "off")
client.send_home_assistant_state("binary_sensor.default", "", "off")
client.send_home_assistant_state("binary_sensor.unavailable_first", "", "off")
client.send_home_assistant_state("binary_sensor.initial_off", "", "on")
client.send_home_assistant_state("binary_sensor.default_unavail", "", "off")
await waiter.wait_for_each(
"initial_on on_release",
"default on_release",
"default_unavail on_release",
"unavailable_first on_release",
"initial_off on_press",
timeout=5.0,
)
+37
View File
@@ -81,3 +81,40 @@ def test_read_file_bytes(tmp_path: Path) -> None:
result = clang_tidy_hash.read_file_bytes(test_file)
assert result == test_content
def test_calculate_idedata_cache_hash_changes_with_infra_code(tmp_path: Path) -> None:
_populate(tmp_path)
infra = tmp_path / "esphome" / "espidf" / "clang_tidy.py"
infra.parent.mkdir(parents=True)
infra.write_text("a")
before = clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
assert before == clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
infra.write_text("b")
assert clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path) != before
def test_calculate_idedata_cache_hash_includes_listed_files(tmp_path: Path) -> None:
_populate(tmp_path)
before = clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
listed = tmp_path / "esphome" / "platformio" / "library.py"
listed.parent.mkdir(parents=True)
listed.write_text("x")
assert clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path) != before
def test_idedata_cache_hash_only_widens_for_esp32(tmp_path: Path) -> None:
_populate(tmp_path)
infra = tmp_path / "esphome" / "espidf" / "clang_tidy.py"
infra.parent.mkdir(parents=True)
infra.write_text("a")
esp32_before = clang_tidy_hash.idedata_cache_hash("esp32-idf-tidy", tmp_path)
other_before = clang_tidy_hash.idedata_cache_hash("esp8266-arduino-tidy", tmp_path)
infra.write_text("b")
assert (
clang_tidy_hash.idedata_cache_hash("esp32-idf-tidy", tmp_path) != esp32_before
)
assert (
clang_tidy_hash.idedata_cache_hash("esp8266-arduino-tidy", tmp_path)
== other_before
)
+7
View File
@@ -552,6 +552,13 @@ def test_determine_integration_tests(
assert run_all is True
assert test_files == []
# Dependency pins and the session init fixture trigger run_all
for trigger in sorted(determine_jobs.INTEGRATION_TESTS_TRIGGER_FILES):
with patch.object(determine_jobs, "changed_files", return_value=[trigger]):
run_all, test_files = determine_jobs.determine_integration_tests()
assert run_all is True
assert test_files == []
# Python files directly in esphome/ do NOT trigger tests
with patch.object(
determine_jobs, "changed_files", return_value=["esphome/config.py"]
@@ -0,0 +1,14 @@
# Common UART configuration for ESP32 Arduino tests - 9600 baud, EVEN parity, 7 data bits
substitutions:
tx_pin: GPIO17
rx_pin: GPIO16
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 9600
parity: EVEN
data_bits: 7
stop_bits: 1
@@ -0,0 +1,14 @@
# Common UART configuration for ESP32 IDF tests - 9600 baud, EVEN parity, 7 data bits
substitutions:
tx_pin: GPIO17
rx_pin: GPIO16
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 9600
parity: EVEN
data_bits: 7
stop_bits: 1
@@ -0,0 +1,14 @@
# Common UART configuration for ESP8266 Arduino tests - 9600 baud even parity, 7 data bits
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 9600
parity: EVEN
data_bits: 7
stop_bits: 1
@@ -0,0 +1,14 @@
# Common UART configuration for RP2040 Arduino tests - 9600 baud even parity, 7 data bits
substitutions:
tx_pin: GPIO0
rx_pin: GPIO1
uart:
- id: uart_bus
tx_pin: ${tx_pin}
rx_pin: ${rx_pin}
baud_rate: 9600
parity: EVEN
data_bits: 7
stop_bits: 1
@@ -0,0 +1,72 @@
"""Tests for the esp32 sdkconfig write and its toolchain-gated clean."""
from __future__ import annotations
import os
from pathlib import Path
import time
from unittest.mock import patch
import pytest
from esphome.components.esp32 import _write_sdkconfig
from esphome.components.esp32.const import KEY_SDKCONFIG_OPTIONS
from esphome.const import KEY_CORE, KEY_ESP32, KEY_FRAMEWORK_VERSION, Toolchain
from esphome.core import CORE
from esphome.espidf.toolchain import has_outdated_files
def _setup_core(tmp_path: Path, toolchain: Toolchain | None) -> None:
CORE.config_path = tmp_path / "test.yaml"
CORE.build_path = tmp_path
CORE.toolchain = toolchain
CORE.data[KEY_ESP32] = {KEY_SDKCONFIG_OPTIONS: {"CONFIG_X": "y"}}
CORE.data[KEY_CORE] = {KEY_FRAMEWORK_VERSION: "5.5.5"}
def _seed_configured_build(tmp_path: Path) -> None:
"""A settled native build: configure outputs predate what comes next."""
build = tmp_path / "build"
(build / "config").mkdir(parents=True)
(build / "config" / "sdkconfig.h").write_text("")
(build / "CMakeCache.txt").write_text("")
(build / "build.ninja").write_text("")
# Explicitly older than what the test writes next: has_outdated_files()
# compares st_mtime with a strict >, so same-tick writes would pass
past = time.time() - 60
for f in build.rglob("*"):
os.utime(f, (past, past))
@pytest.mark.parametrize(
("toolchain", "clean_expected"),
[(Toolchain.ESP_IDF, False), (Toolchain.PLATFORMIO, True), (None, True)],
)
def test_write_sdkconfig_cleans_only_on_platformio(
tmp_path: Path, toolchain: Toolchain | None, clean_expected: bool
) -> None:
"""A changed sdkconfig forces a full clean only under PlatformIO; the
esp-idf toolchain reconfigures via has_outdated_files() instead; an
unresolved toolchain fails safe onto the clean."""
_setup_core(tmp_path, toolchain)
_seed_configured_build(tmp_path)
with (
patch.object(CORE, "name", "test"),
patch("esphome.components.esp32.clean_build") as clean,
):
_write_sdkconfig()
assert "CONFIG_X" in CORE.relative_build_path("sdkconfig.test").read_text()
assert clean.called is clean_expected
if clean_expected:
clean.assert_called_once_with(clear_pio_cache=False)
# The change must still trigger a reconfigure: the internal
# sdkconfig snapshot is now newer than build/CMakeCache.txt
assert has_outdated_files() is True
clean.reset_mock()
# A settled configure restamps the cache; an unchanged rewrite
# must then neither clean nor mark the build stale
future = time.time() + 60
os.utime(CORE.relative_build_path("build/CMakeCache.txt"), (future, future))
_write_sdkconfig()
clean.assert_not_called()
assert has_outdated_files() is False
@@ -53,9 +53,12 @@ def test_nonzero_indices_are_nonzero(gamma: float) -> None:
assert table[i] >= 1, f"gamma={gamma}, index {i}: got {table[i]}"
@pytest.mark.parametrize("gamma", [1.0, 2.0, 2.2, 2.8, 3.0])
@pytest.mark.parametrize("gamma", [1.0, 1.8, 2.0, 2.2, 2.8, 3.0, 4.0])
def test_table_monotonically_nondecreasing(gamma: float) -> None:
"""The gamma table must be monotonically non-decreasing."""
"""The gamma table must be monotonically non-decreasing.
gamma_table_reverse_search()'s binary search depends on this.
"""
table = generate_gamma_table(gamma)
for i in range(1, 256):
assert table[i] >= table[i - 1], (
@@ -115,3 +118,13 @@ def test_lut_output_monotonically_nondecreasing() -> None:
result = _simulate_gamma_correct_lut(table, value)
assert result >= prev, f"value={value}: result {result} < previous {prev}"
prev = result
def test_table_matches_raw_power_curve() -> None:
"""Check the gamma table against known good values for gamma=2.8."""
table = generate_gamma_table(2.8)
golden = {1: 1, 5: 1, 15: 24, 27: 122, 28: 135, 100: 4766, 200: 33193, 254: 64818}
for i, expected in golden.items():
assert table[i] == expected, (
f"index {i}: table[{i}]={table[i]} expected {expected}"
)
+28
View File
@@ -1127,6 +1127,34 @@ def test_config_hash_same_for_different_config_dirs(tmp_path: Path) -> None:
assert hash1 == hash2
def test_config_hash_same_for_different_data_dirs(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test that downloaded file paths hash the same wherever data_dir lives."""
config_dir = tmp_path / "config"
config_dir.mkdir()
CORE.reset()
CORE.config_path = config_dir / "device.yaml"
CORE.config = {
"esphome": {"name": "test"},
"file": config_dir / ".esphome" / "image" / "c44630d6",
}
hash1 = CORE.config_hash
other_data_dir = tmp_path / "data"
CORE.reset()
monkeypatch.setenv("ESPHOME_DATA_DIR", str(other_data_dir))
CORE.config_path = config_dir / "device.yaml"
CORE.config = {
"esphome": {"name": "test"},
"file": other_data_dir / "image" / "c44630d6",
}
hash2 = CORE.config_hash
assert hash1 == hash2
def test_make_app_name_cpp_no_mac_simple() -> None:
"""Test simple name without MAC suffix returns string literal."""
cpp_expr, global_decl, byte_len = make_app_name_cpp(
+52
View File
@@ -112,6 +112,7 @@ from esphome.const import (
PLATFORM_BK72XX,
PLATFORM_ESP32,
PLATFORM_ESP8266,
PLATFORM_HOST,
PLATFORM_NRF52,
PLATFORM_RP2,
Toolchain,
@@ -7254,3 +7255,54 @@ async def test_wrap_to_code_comment_is_insertion_order_independent() -> None:
assert first == second
assert second.index("alpha") < second.index("beta")
assert second.index("a: 2") < second.index("z: 1")
def test_host_program_path_platformio_toolchain() -> None:
"""Host + PlatformIO toolchain reads the memoized idedata path."""
setup_core(platform=PLATFORM_HOST)
idedata = SimpleNamespace(firmware_elf_path="/build/x/.pioenvs/x/program")
with patch(
"esphome.platformio.toolchain.get_idedata", return_value=idedata
) as mock_get:
assert main._host_program_path({}) == "/build/x/.pioenvs/x/program"
mock_get.assert_called_once_with({})
def test_host_program_path_esp_idf_toolchain() -> None:
"""Host + native ESP-IDF toolchain asks the espidf toolchain for the ELF."""
setup_core(platform=PLATFORM_HOST)
CORE.toolchain = Toolchain.ESP_IDF
with patch(
"esphome.espidf.toolchain.get_elf_path", return_value=Path("/b/app.elf")
):
assert main._host_program_path({}) == str(Path("/b/app.elf"))
def test_command_compile_host_logs_program_path(
caplog: pytest.LogCaptureFixture,
) -> None:
"""command_compile on host logs the compiled program path."""
setup_core(platform=PLATFORM_HOST)
with (
patch.object(main, "write_cpp", return_value=0),
patch.object(main, "compile_program", return_value=0),
patch.object(main, "_host_program_path", return_value="/b/program"),
caplog.at_level(logging.INFO),
):
assert main.command_compile(SimpleNamespace(only_generate=False), {}) == 0
assert "Successfully compiled program to path '/b/program'" in caplog.text
def test_command_run_host_executes_program(caplog: pytest.LogCaptureFixture) -> None:
"""command_run on host logs and executes the compiled program directly."""
setup_core(platform=PLATFORM_HOST)
with (
patch.object(main, "write_cpp", return_value=0),
patch.object(main, "compile_program", return_value=0),
patch.object(main, "_host_program_path", return_value="/b/program"),
patch.object(main, "run_external_process", return_value=0) as mock_run,
caplog.at_level(logging.INFO),
):
assert main.command_run(SimpleNamespace(), {}) == 0
mock_run.assert_called_with("/b/program")
assert "Running program from path '/b/program'" in caplog.text
@@ -1446,6 +1446,76 @@ def test_prefetch_installs_cached_archives_without_downloads(
assert not (tmp_path / pf._SENTINEL_NAME).exists()
@pytest.mark.parametrize(
("platform_group", "lib_group", "expected"),
[
(
[("toolchain-x@1", _FakeSpec(name="toolchain-x"))],
[],
["configure", "install", "configure"],
),
([], [("noise-c@1.0", _FakeSpec(name="noise-c"))], ["configure", "install"]),
],
)
def test_prefetch_reconfigures_only_after_platform_installs(
tmp_path: Path, platform_group: list, lib_group: list, expected: list[str]
) -> None:
"""Installed platform packages get a second configure pass; libraries do not."""
_write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
order: list[str] = []
fake_platform = MagicMock()
fake_platform.packages = {}
fake_platform.configure_project_packages.side_effect = lambda env, targets: (
order.append("configure")
)
config = _fake_config(
tmp_path, {"platform": "fake/p@1", "lib_deps": ["esphome/noise-c@1.0"]}
)
modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
with (
patch.dict("sys.modules", modules),
patch.object(
pf,
"_registry_jobs",
side_effect=[([], 0, platform_group), ([], 0, lib_group)],
),
patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
patch.object(pf, "_preinstall", side_effect=lambda *_: order.append("install")),
):
pf._prefetch(tmp_path, "testenv")
assert order == expected
@pytest.mark.parametrize(
"err", [RuntimeError("idf_tools.py failed"), SystemExit("postinstall exited")]
)
def test_prefetch_settle_failure_warns_and_continues(
tmp_path: Path, caplog: pytest.LogCaptureFixture, err: BaseException
) -> None:
"""A failing second configure pass only costs the speedup."""
_write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
fake_platform = MagicMock()
fake_platform.packages = {}
fake_platform.configure_project_packages.side_effect = [None, err]
config = _fake_config(tmp_path, {"platform": "fake/p@1"})
modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
with (
patch.dict("sys.modules", modules),
patch.object(
pf,
"_registry_jobs",
side_effect=[
([], 0, [("toolchain-x@1", _FakeSpec(name="toolchain-x"))]),
([], 0, []),
],
),
patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
patch.object(pf, "_preinstall"),
):
pf._prefetch(tmp_path, "testenv")
assert f"Could not settle platform packages: {err}" in caplog.text
def test_preinstall_extracts_in_parallel_under_one_lock(tmp_path: Path) -> None:
"""The manager lock wraps the whole batch; per-thread managers share
its package dir; one failing install leaves the rest alone."""
+47
View File
@@ -1706,6 +1706,53 @@ def test_dump_path_dotdot_reference_outside_anchor() -> None:
assert output.strip() == "file: ../shared/font.ttf"
@pytest.mark.parametrize(
"data_dir",
[
pytest.param(Path("/config/.esphome"), id="cli"),
pytest.param(Path("/data"), id="addon"),
],
)
def test_dump_path_under_data_dir_uses_default_location(data_dir: Path) -> None:
"""Test that Path values under data_dir dump as .esphome/<rest> for any layout."""
anchor = Path("/config").absolute()
path = data_dir.absolute() / "image" / "c44630d6"
output = yaml_util.dump(
{"file": path}, relative_to=anchor, data_dir=data_dir.absolute()
)
assert output.strip() == "file: .esphome/image/c44630d6"
def test_dump_path_equal_to_data_dir() -> None:
"""Test that the data dir itself dumps as .esphome, matching the default layout."""
anchor = Path("/config").absolute()
data_dir = Path("/data").absolute()
output = yaml_util.dump({"dir": data_dir}, relative_to=anchor, data_dir=data_dir)
assert output.strip() == "dir: .esphome"
default = yaml_util.dump(
{"dir": anchor / ".esphome"}, relative_to=anchor, data_dir=anchor / ".esphome"
)
assert default == output
def test_dump_path_outside_data_dir_still_relative_to_anchor() -> None:
"""Test that data_dir does not affect paths that are not under it."""
anchor = Path("/config").absolute()
path = anchor / "fonts" / "arial.ttf"
output = yaml_util.dump(
{"file": path}, relative_to=anchor, data_dir=Path("/data").absolute()
)
assert output.strip() == "file: fonts/arial.ttf"
def test_dump_path_data_dir_without_relative_to_is_unchanged() -> None:
"""Test that data_dir alone does not change the output."""
data_dir = Path("/data").absolute()
path = data_dir / "image" / "c44630d6"
output = yaml_util.dump({"file": path}, data_dir=data_dir)
assert output.strip() == f"file: {path}"
def test_dump_relative_to_does_not_leak_between_calls() -> None:
"""Test that the relative_to flag is scoped to a single dump call."""
anchor = Path("/config/esphome").absolute()