Merge branch 'dev' into platformio-skip-private-probe

This commit is contained in:
J. Nick Koston
2026-08-30 19:31:25 -05:00
committed by GitHub
159 changed files with 11182 additions and 10677 deletions
+43
View File
@@ -363,4 +363,47 @@ static void Snprintf_Uint32_Large(benchmark::State &state) {
}
BENCHMARK(Snprintf_Uint32_Large);
// --- step_to_accuracy_decimals() ---
// Called from climate traits and web_server for every number/climate step.
static void StepToAccuracyDecimals_Tenth(benchmark::State &state) {
for (auto _ : state) {
int result = 0;
for (int i = 0; i < kInnerIterations; i++) {
result += step_to_accuracy_decimals(0.1f);
}
benchmark::DoNotOptimize(result);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(StepToAccuracyDecimals_Tenth);
static void StepToAccuracyDecimals_Whole(benchmark::State &state) {
for (auto _ : state) {
int result = 0;
for (int i = 0; i < kInnerIterations; i++) {
result += step_to_accuracy_decimals(1.0f);
}
benchmark::DoNotOptimize(result);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(StepToAccuracyDecimals_Whole);
static void StepToAccuracyDecimals_Mixed(benchmark::State &state) {
static constexpr float steps[] = {
0.001f, 0.01f, 0.05f, 0.1f, 0.25f, 0.5f, 1.0f, 2.5f, 5.0f, 10.0f,
};
static constexpr int num_steps = sizeof(steps) / sizeof(steps[0]);
for (auto _ : state) {
int result = 0;
for (int i = 0; i < kInnerIterations; i++) {
result += step_to_accuracy_decimals(steps[i % num_steps]);
}
benchmark::DoNotOptimize(result);
}
state.SetItemsProcessed(state.iterations() * kInnerIterations);
}
BENCHMARK(StepToAccuracyDecimals_Mixed);
} // namespace esphome::benchmarks
@@ -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
espnow:
channel: 1
auto_add_peer: true
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
sensor:
- platform: internal_temperature
name: Internal Temperature
@@ -0,0 +1,13 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ssid: MySSID
password: password1
esp32_ble_tracker:
@@ -0,0 +1,11 @@
esphome:
name: test
esp32:
board: esp32-s3-devkitc-1
framework:
type: esp-idf
sensor:
- platform: internal_temperature
name: Internal Temperature
@@ -6,6 +6,8 @@ esp32:
framework:
type: esp-idf
mdns:
ethernet:
type: W5500
clk_pin: 19
@@ -6,6 +6,8 @@ esp32:
framework:
type: esp-idf
mdns:
wifi:
ssid: "test_ssid"
password: "test_password"
+44 -4
View File
@@ -24,6 +24,7 @@ from esphome.components.esp32 import (
)
from esphome.components.esp32.const import (
KEY_ESP32,
KEY_EXCLUDE_COMPONENTS,
KEY_NETWORK_SDKCONFIG,
KEY_SDKCONFIG_OPTIONS,
KEY_VARIANT,
@@ -298,6 +299,26 @@ def test_esp32_configuration_errors(
("esp-tls", "esp_http_client"),
id="nextion",
),
pytest.param(
# esp_wifi/wpa_supplicant from request_wifi(), bt from
# request_bluetooth(), esp_coex from esp32_ble_tracker's software
# coexistence (defaults on with wifi). esp_phy stays excluded;
# IDF requirement expansion pulls it back via esp_wifi.
"exclusion_reincludes_wifi_ble.yaml",
("esp_wifi", "wpa_supplicant", "bt", "esp_coex"),
id="wifi_ble",
),
pytest.param(
"exclusion_reincludes_espnow.yaml",
("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(
@@ -309,8 +330,6 @@ def test_default_exclusions_reincluded_by_owning_components(
"""Components whose IDF driver is excluded by default must re-include it
during codegen; a dropped include_builtin_idf_component() call would only
surface as a missing-header failure in a full compile job."""
from esphome.components.esp32.const import KEY_EXCLUDE_COMPONENTS
generate_main(component_config_path(config_file))
excluded = CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS]
@@ -324,13 +343,20 @@ 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],
) -> None:
"""An explicit CONFIG_NVS_ENCRYPTION=n keeps nvs_sec_provider excluded."""
from esphome.components.esp32.const import KEY_EXCLUDE_COMPONENTS
generate_main(component_config_path("exclusion_stays_nvs_sdkconfig_off.yaml"))
assert "nvs_sec_provider" in CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS]
@@ -939,6 +965,14 @@ def test_network_wifi_only_reconciles_end_to_end(
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert sdkconfig.get("CONFIG_ESP_WIFI_SOFTAP_SUPPORT") is False
assert sdkconfig.get("CONFIG_LWIP_DHCPS") is False
# request_wifi() also puts the WiFi components back in the build set;
# esp_phy stays excluded, IDF requirement expansion pulls it back.
excluded = CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS]
assert "esp_wifi" not in excluded
assert "wpa_supplicant" not in excluded
assert "esp_phy" in excluded
# With wifi present mdns keeps its predefined interfaces.
assert "CONFIG_MDNS_PREDEF_NETIF_STA" not in sdkconfig
# WiFi stack stays enabled (no ethernet) and no Bluetooth requested.
assert "CONFIG_ESP_WIFI_ENABLED" not in sdkconfig
assert "CONFIG_BT_ENABLED" not in sdkconfig
@@ -954,6 +988,12 @@ def test_network_ethernet_only_reconciles_end_to_end(
sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
assert sdkconfig.get("CONFIG_ESP_WIFI_ENABLED") is False
assert sdkconfig.get("CONFIG_SW_COEXIST_ENABLE") is False
# The whole radio stack stays out of the build set as well.
excluded = CORE.data[KEY_ESP32][KEY_EXCLUDE_COMPONENTS]
assert {"esp_wifi", "wpa_supplicant", "esp_phy", "esp_coex", "bt"} <= excluded
# Without wifi, mdns drops its predefined STA/AP interfaces.
assert sdkconfig.get("CONFIG_MDNS_PREDEF_NETIF_STA") is False
assert sdkconfig.get("CONFIG_MDNS_PREDEF_NETIF_AP") is False
def test_network_wifi_ble_coexistence_reconciles_end_to_end(
+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:
+68
View File
@@ -1,4 +1,5 @@
#include <gtest/gtest.h>
#include <cmath>
#include <cstring>
#include "esphome/core/alloc_helpers.h"
@@ -280,4 +281,71 @@ TEST(Base64, Rfc4648Vectors) {
}
}
// --- step_to_accuracy_decimals() ---
TEST(StepToAccuracyDecimals, TypicalSteps) {
EXPECT_EQ(step_to_accuracy_decimals(0.001f), 3);
EXPECT_EQ(step_to_accuracy_decimals(0.005f), 3);
EXPECT_EQ(step_to_accuracy_decimals(0.01f), 2);
EXPECT_EQ(step_to_accuracy_decimals(0.025f), 3);
EXPECT_EQ(step_to_accuracy_decimals(0.05f), 2);
EXPECT_EQ(step_to_accuracy_decimals(0.1f), 1);
EXPECT_EQ(step_to_accuracy_decimals(0.25f), 2);
EXPECT_EQ(step_to_accuracy_decimals(0.5f), 1);
EXPECT_EQ(step_to_accuracy_decimals(1.5f), 1);
EXPECT_EQ(step_to_accuracy_decimals(2.5f), 1);
}
TEST(StepToAccuracyDecimals, WholeSteps) {
EXPECT_EQ(step_to_accuracy_decimals(1.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(2.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(5.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(10.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(100.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(1000.0f), 0);
}
TEST(StepToAccuracyDecimals, FiveSignificantDigits) {
EXPECT_EQ(step_to_accuracy_decimals(1.23456f), 4);
EXPECT_EQ(step_to_accuracy_decimals(12.345f), 3);
EXPECT_EQ(step_to_accuracy_decimals(123.45f), 2);
EXPECT_EQ(step_to_accuracy_decimals(1234.5f), 1);
EXPECT_EQ(step_to_accuracy_decimals(12345.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(0.33333f), 5);
EXPECT_EQ(step_to_accuracy_decimals(0.0001f), 4);
}
TEST(StepToAccuracyDecimals, TrailingZerosDropped) {
EXPECT_EQ(step_to_accuracy_decimals(0.3f), 1);
EXPECT_EQ(step_to_accuracy_decimals(0.7f), 1);
EXPECT_EQ(step_to_accuracy_decimals(0.125f), 3);
EXPECT_EQ(step_to_accuracy_decimals(0.0625f), 4);
}
TEST(StepToAccuracyDecimals, RoundsUpToWholeNumber) {
// Rounds to five significant digits first, so this becomes 10 with no decimals.
EXPECT_EQ(step_to_accuracy_decimals(9.999999f), 0);
}
TEST(StepToAccuracyDecimals, OutsideFixedNotationRange) {
// %.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), 0);
EXPECT_EQ(step_to_accuracy_decimals(1000000.0f), 0);
}
TEST(StepToAccuracyDecimals, SignIgnored) {
EXPECT_EQ(step_to_accuracy_decimals(-0.1f), 1);
EXPECT_EQ(step_to_accuracy_decimals(-0.25f), 2);
EXPECT_EQ(step_to_accuracy_decimals(-1.0f), 0);
}
TEST(StepToAccuracyDecimals, NonFiniteAndZero) {
EXPECT_EQ(step_to_accuracy_decimals(0.0f), 0);
EXPECT_EQ(step_to_accuracy_decimals(NAN), 0);
EXPECT_EQ(step_to_accuracy_decimals(INFINITY), 0);
EXPECT_EQ(step_to_accuracy_decimals(-INFINITY), 0);
}
} // namespace esphome::core::testing
@@ -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; }
@@ -775,7 +775,7 @@ TEST(ModbusClientHubBroadcast, DeliversNoTerminalToTypedDevice) {
// A broadcast is only meaningful for a command that changes state; a broadcast READ could never be
// answered, so the hub refuses it at the door (false return, no entry queued) rather than silently
// retiring it. Writes, 0x17, and custom codes still go through (covered above).
// retiring it. Writes and custom/unknown codes still go through (covered in the neighboring tests).
TEST(ModbusClientHubBroadcast, RefusesReadBroadcast) {
NullUART uart;
NoResponseProbeHub hub;
@@ -814,9 +814,8 @@ TEST(ModbusClientHubBroadcast, AcceptsCustomBroadcast) {
EXPECT_EQ(hub.entries(), 0u); // the entry is gone
}
// An exception-flagged custom code (0x80 bit set) is not a real request: is_function_code_custom() masks
// the bit away and would accept it, but the broadcast guard excludes it, matching classify()'s handling
// of an exception-flagged write.
// An exception-flagged code (0x80 bit set) is never a valid request - that bit is response-only - so
// queue_pdu refuses it up front, before the broadcast guard, whatever its base code.
TEST(ModbusClientHubBroadcast, RefusesExceptionFlaggedCustomBroadcast) {
NullUART uart;
NoResponseProbeHub hub;
@@ -833,6 +832,50 @@ TEST(ModbusClientHubBroadcast, RefusesExceptionFlaggedCustomBroadcast) {
EXPECT_EQ(device.sent_count_, 0); // never transmitted
}
// FC23 (read/write multiple) has a read half that expects a reply, so the Modbus spec does not allow it
// as a broadcast. is_function_code_read() covers it, so the broadcast guard refuses it despite its write
// half.
TEST(ModbusClientHubBroadcast, RefusesReadWriteMultipleBroadcast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
// fc, read start+qty, write start+qty, byte count, one data word.
const uint8_t read_write_multiple[] = {0x17, 0x00, 0x00, 0x00, 0x01, 0x00, 0x10, 0x00, 0x01, 0x02, 0xBE, 0xEF};
EXPECT_FALSE(device.queue_pdu(read_write_multiple)); // its read half could never be answered
EXPECT_EQ(hub.entries(), 0u);
}
// FC 0x18 (read FIFO queue) is not a "read" by is_function_code_read(), but the hub has an explicit
// response-length rule for it - it demonstrably expects a reply, so it cannot broadcast.
TEST(ModbusClientHubBroadcast, RefusesKnownLengthNonWriteBroadcast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t read_fifo[] = {0x18, 0x00, 0x10}; // fc, FIFO pointer address
EXPECT_FALSE(device.queue_pdu(read_fifo));
EXPECT_EQ(hub.entries(), 0u);
}
// A code that is neither a read nor exception-flagged (here 0x63, unassigned) is fire-and-forget on a
// broadcast: the hub can't know it isn't a vendor write, so it is accepted and delivered to all devices.
TEST(ModbusClientHubBroadcast, AcceptsNonReadUnknownBroadcast) {
NullUART uart;
NoResponseProbeHub hub;
hub.set_uart_parent(&uart);
hub.setup();
BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
const uint8_t unknown[] = {0x63, 0x00, 0x01};
EXPECT_TRUE(device.queue_pdu(unknown)); // not a read, so not refused
EXPECT_EQ(hub.entries(), 1u);
}
namespace {
// tx_blocked() clear for send_next_frame_'s gate, then blocked for send_frame_'s post-delay re-check.
class RejectPostDelayHub : public NoResponseProbeHub {
@@ -1882,30 +1925,20 @@ TEST(ModbusClientHubPriority, ResendFromOnResponseAbsorbsIntoCompletingCommand)
EXPECT_FALSE(hub.queued(0).options.continuous); // the one-shot re-send downgraded the poll
}
// An exception-flagged function code is never silently re-sendable, even though the read check
// masks the exception bit: its duplicate takes the drop path like any other non-read.
TEST(ModbusClientHubPriority, ExceptionFlaggedDuplicateDroppedNotPromoted) {
// The exception bit marks a response, so a request carrying it is refused outright.
TEST(ModbusClientHubPriority, ExceptionFlaggedPduRefused) {
NoResponseProbeHub hub;
SentCountingDevice device(&hub, 0x02);
const uint8_t weird[] = {0x83, 0x01, 0x00, 0x00, 0x02}; // read-shaped but exception-flagged
EXPECT_TRUE(device.queue_pdu(weird));
EXPECT_FALSE(device.queue_pdu(weird)); // non-requeueable: cap of one, so the duplicate is refused
// The 0x80 exception flag is a response-only bit; a request must never set it. queue_pdu refuses an
// exception-flagged PDU up front - nothing is queued - whether its base code reads (0x83 = 0x03 | 0x80)
// or writes (0x86 = 0x06 | 0x80).
const uint8_t read_shaped[] = {0x83, 0x01, 0x00, 0x00, 0x02};
const uint8_t write_shaped[] = {0x86, 0x00, 0x10, 0xBE, 0xEF};
EXPECT_FALSE(device.queue_pdu(read_shaped));
EXPECT_FALSE(device.queue_pdu(write_shaped));
hub.sweep_for_test();
ASSERT_EQ(hub.queued_frames(), 1u);
EXPECT_EQ(hub.queued(0).pending, 1u);
EXPECT_EQ(device.not_sent_count_, 0);
// The write-shaped twin (0x86 masks to WRITE_SINGLE_REGISTER) must not take WRITE-class
// ordering either: exception-flagged codes are excluded from the mutates classification.
const uint8_t weird_write[] = {0x86, 0x00, 0x10, 0xBE, 0xEF};
device.queue_pdu(weird_write);
ASSERT_EQ(hub.queued_frames(), 2u);
EXPECT_EQ(hub.queued(1).priority(), CommandPriority::READ); // not WRITE
const ModbusDeviceCommand *next = hub.next_ready();
ASSERT_NE(next, nullptr);
EXPECT_EQ(next->frame.pdu()[0], 0x83); // FIFO by age: it did not jump the older entry
EXPECT_EQ(hub.queued_frames(), 0u);
}
namespace {
@@ -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
@@ -63,6 +63,12 @@ TEST(ModbusClientFrameLength, TooShortReturnsMinimum) {
EXPECT_EQ(client_frame_length(frame, 1), MIN_FRAME_SIZE);
}
TEST(ModbusClientFrameLength, ExceptionFlaggedIsTheExceptionShape) {
// Sized at 2 so an exception-flagged request fails its CRC at once instead of being scanned for.
const uint8_t exception_request[] = {0x83, 0x02};
EXPECT_EQ(client_pdu_length(exception_request, sizeof(exception_request)), 2);
}
TEST(ModbusClientFrameLength, ReadAndWriteSingleAreFixed) {
// basic_register request fixture is a read-holding request -> 8 bytes
const uint8_t read[] = {0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A};
@@ -421,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) {
@@ -483,6 +610,28 @@ TEST(ModbusTypedBuilders, WriteRegistersPduRejectsOverLimit) {
EXPECT_FALSE(create_write_registers_pdu(0x0000, values).empty());
}
TEST(ModbusTypedBuilders, WriteFewRegistersPduMatchesFullSizeBuilder) {
static_assert(sizeof(WriteFewRegistersPdu) < sizeof(PduBuffer) / 4,
"WriteFewRegistersPdu must be meaningfully smaller");
const uint16_t values[] = {0x000B, 0x0016, 0xABCD, 0xFF00};
for (size_t count = 1; count <= MAX_FEW_REGISTERS; count++) {
auto small = create_write_few_registers_pdu(0x0102, std::span<const uint16_t>(values, count));
auto full = create_write_registers_pdu(0x0102, std::span<const uint16_t>(values, count));
EXPECT_EQ(std::vector<uint8_t>(small.begin(), small.end()), std::vector<uint8_t>(full.begin(), full.end()))
<< count << " registers";
EXPECT_EQ(small.size(), 6u + 2 * count);
EXPECT_TRUE(is_client_pdu_standard(small.data(), small.size()));
}
}
TEST(ModbusTypedBuilders, WriteFewRegistersPduRejectsInvalidInput) {
const uint16_t values[MAX_FEW_REGISTERS + 1] = {0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA, 0xAAAA};
EXPECT_TRUE(create_write_few_registers_pdu(0x0000, values).empty());
EXPECT_FALSE(create_write_few_registers_pdu(0x0000, std::span<const uint16_t>(values, MAX_FEW_REGISTERS)).empty());
EXPECT_TRUE(create_write_few_registers_pdu(0x0000, std::span<const uint16_t>()).empty());
EXPECT_TRUE(create_write_few_registers_pdu(0xFFFF, std::span<const uint16_t>(values, 2)).empty());
}
TEST(ModbusTypedBuilders, ReadWriteMultipleRegistersPduWireBytes) {
const uint16_t write_values[] = {0x000B, 0x0016};
// Read 2 registers at 0x0010, write 2 registers at 0x0020.
@@ -54,7 +54,8 @@ class TestServerHub : public ModbusServerHub {
// The frame-length parsers have explicit cases for exactly these 13 codes; every other value - the
// assigned-but-unimplemented management codes, both user-defined ranges, and all unassigned codes -
// must classify as unknown length. The exception flag masks off first.
// must classify as unknown length. Exception replies are always the 2-byte spec shape, so every
// 0x80-set code is known length.
TEST(ModbusUnknownFunction, HelperMatchesParserCoverage) {
for (uint8_t fc : {0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x0F, 0x10, 0x14, 0x15, 0x16, 0x17, 0x18}) {
EXPECT_FALSE(helpers::is_function_code_unknown_length(fc)) << "fc 0x" << std::hex << int(fc);
@@ -62,11 +63,13 @@ TEST(ModbusUnknownFunction, HelperMatchesParserCoverage) {
for (uint8_t fc : {0x07, 0x08, 0x0B, 0x0C, 0x11, 0x2A, 0x41, 0x48, 0x49, 0x64, 0x6E, 0x00, 0x7F}) {
EXPECT_TRUE(helpers::is_function_code_unknown_length(fc)) << "fc 0x" << std::hex << int(fc);
}
// Exception replies classify by their base code.
// Every exception-flagged code is known length (the 2-byte spec exception shape), whatever its base.
EXPECT_FALSE(helpers::is_function_code_unknown_length(0x83));
EXPECT_TRUE(helpers::is_function_code_unknown_length(0x87));
// Strictly wider than the user-defined ranges: every custom code is unknown-length, but not vice versa.
for (int fc = 0; fc <= 0xFF; fc++) {
EXPECT_FALSE(helpers::is_function_code_unknown_length(0x87));
EXPECT_FALSE(helpers::is_function_code_unknown_length(0xC9));
// Strictly wider than the user-defined ranges below 0x80: every non-exception custom code is
// unknown-length, but not vice versa.
for (int fc = 0; fc <= 0x7F; fc++) {
if (helpers::is_function_code_custom(fc))
EXPECT_TRUE(helpers::is_function_code_unknown_length(fc)) << "fc 0x" << std::hex << fc;
}
@@ -75,10 +78,10 @@ TEST(ModbusUnknownFunction, HelperMatchesParserCoverage) {
// Derived contract check: the helper must say "unknown" exactly when both length parsers fall
// through to default. With a zero-filled max-size PDU every explicit case returns at least 2
// (file records bottom out at 2, FIFO at 3) and only default returns MIN_PDU_SIZE, so comparing
// against MIN_PDU_SIZE detects a case added to either switch without updating the helper. The
// loop stops at 0x7F: above it the helper masks the exception flag off while client_pdu_length()
// switches on the unmasked byte and server_pdu_length() early-returns the exception length.
for (int fc = 0; fc <= 0x7F; fc++) {
// against MIN_PDU_SIZE detects a case added to either switch without updating the helper. Both
// parsers early-return the 2-byte exception shape above 0x7F, which the helper's own exception
// early-return mirrors, so the whole byte range is covered.
for (int fc = 0; fc <= 0xFF; fc++) {
const uint8_t pdu[MAX_PDU_SIZE] = {static_cast<uint8_t>(fc)}; // zero header fields
EXPECT_EQ(helpers::is_function_code_unknown_length(fc),
helpers::client_pdu_length(pdu, sizeof(pdu)) == MIN_PDU_SIZE)
@@ -89,6 +92,17 @@ TEST(ModbusUnknownFunction, HelperMatchesParserCoverage) {
}
}
// Broadcastable = writes plus unknown codes (possible vendor writes); everything known to expect a
// reply is not. Classifies the underlying code: the exception bit masks off first (0x85 as 0x05).
TEST(ModbusUnknownFunction, BroadcastableClassification) {
for (uint8_t fc : {0x05, 0x06, 0x0F, 0x10, 0x16, 0x49, 0x63, 0x6E, 0x85, 0xC9}) {
EXPECT_TRUE(helpers::is_function_code_broadcastable(fc)) << "fc 0x" << std::hex << int(fc);
}
for (uint8_t fc : {0x01, 0x02, 0x03, 0x04, 0x14, 0x15, 0x17, 0x18, 0x83, 0x97}) {
EXPECT_FALSE(helpers::is_function_code_broadcastable(fc)) << "fc 0x" << std::hex << int(fc);
}
}
// A response with a function code outside the user-defined ranges (0x49) has no length case in
// server_pdu_length(), so the parser must find the frame end by CRC scan - the same way it already
// handles user-defined codes. Frame: address + FC 0x49 + 3 data bytes + CRC = 7 bytes. Without the
@@ -5,6 +5,11 @@
#include "esphome/components/modbus_controller/modbus_controller.h"
// These tests pin the behaviour of the deprecated ModbusCommandItem until its removal.
// Remove with ModbusCommandItem before 2027.3.0.
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
namespace esphome::modbus_controller::testing {
// The coil write factory packs into an exact-size payload. Pinned at one past the protocol maximum
@@ -29,3 +34,5 @@ TEST(ModbusCommandPayload, CoilWritePacksLsbFirstWithZeroPad) {
}
} // namespace esphome::modbus_controller::testing
#pragma GCC diagnostic pop
+11 -7
View File
@@ -21,6 +21,7 @@ binary_sensor:
name: Test Binary Sensor with Lambda
register_type: input
address: 0x3201
reuse_previous_range: false
lambda: |-
return x;
@@ -85,6 +86,7 @@ select:
name: Test Select with Lambda
address: 1001
value_type: U_WORD
reuse_previous_range: auto
optionsmap:
"Off": 0
"On": 1
@@ -140,9 +142,10 @@ sensor:
register_type: holding
address: 0x9002
value_type: U_WORD
reuse_previous_range: true
lambda: |-
return x / 10.0;
# Non-mergeable sensor sharing the start address of modbus_sensor1 (different register_count):
# Non-mergeable sensor sharing the start address of modbus_sensor1 (different value type width):
# must join the same range, never open a second range keyed on the same (address, type).
- platform: modbus_controller
modbus_controller_id: modbus_controller1
@@ -187,8 +190,8 @@ sensor:
value_type: U_WORD
lambda: |-
return modbus_controller::get_data<uint16_t>(data, item->offset) * 0.1f;
# force_new_range sensors sort before plain ones, so this high-address forced sensor is grouped
# first and the lower-address plain sensors above must still get their own ranges.
# The deprecated force_new_range migrates to reuse_previous_range: false, so this sensor never
# joins a range built before it and the lower-address sensors above keep their own ranges.
- platform: modbus_controller
modbus_controller_id: modbus_controller1
id: modbus_sensor_forced_high
@@ -224,7 +227,6 @@ text_sensor:
name: Test Text Sensor
register_type: holding
address: 0x9013
register_count: 3
raw_encode: HEXBYTES
response_size: 6
- platform: modbus_controller
@@ -233,12 +235,13 @@ text_sensor:
name: Test Text Sensor with Lambda
register_type: holding
address: 0x9014
register_count: 2
response_size: 4
lambda: |-
return "Modified: " + x;
# A register reporting FEWER bytes than 2*register_count (response_size: 3 for 2 registers), followed
# by a contiguous sensor: the follower's byte position must track the actual 3 bytes, not underflow.
# A register reporting FEWER bytes than two per register (response_size: 3 over 2 registers), followed
# by a contiguous reuse:true sensor (auto never joins past a response_size register): the follower's
# byte position must track the actual 3 bytes, not underflow.
# register_count matches the derived width, so it migrates with a deprecation warning.
- platform: modbus_controller
modbus_controller_id: modbus_controller1
id: modbus_text_sensor_narrow
@@ -255,4 +258,5 @@ text_sensor:
register_type: holding
address: 0x9032
register_count: 1
reuse_previous_range: true
raw_encode: HEXBYTES
+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"
@@ -27,8 +27,8 @@ uart_mock:
# so these also pin the grouping: an extra or differently shaped read fails the test.
- expect_tx: [0x01, 0x01, 0x00, 0x10, 0x00, 0x02, 0xBC, 0x0E] # coils 0x10 count 2
inject_rx: [0x01, 0x01, 0x01, 0x01, 0x90, 0x48] # bit0 set, bit1 clear
- expect_tx: [0x01, 0x03, 0x01, 0x60, 0x00, 0x01, 0x85, 0xE8] # holding 0x160 count 1
inject_rx: [0x01, 0x03, 0x02, 0x01, 0x60, 0xB9, 0xFC] # 352
- expect_tx: [0x01, 0x03, 0x01, 0x60, 0x00, 0x02, 0xC5, 0xE9] # holding 0x160 count 2
inject_rx: [0x01, 0x03, 0x04, 0x01, 0x60, 0x01, 0x61, 0x3B, 0xA9] # 352, 353
- expect_tx: [0x01, 0x03, 0x01, 0x00, 0x00, 0x01, 0x85, 0xF6] # holding 0x100 count 1
inject_rx: [0x01, 0x03, 0x04, 0x01, 0x11, 0x02, 0x22, 0x2A, 0xB3] # 4 bytes: 273 then 546
- expect_tx: [0x01, 0x03, 0x01, 0x20, 0x00, 0x04, 0x44, 0x3F] # holding 0x120 count 4
@@ -49,8 +49,6 @@ uart_mock:
inject_rx: [0x01, 0x03, 0x02, 0x33, 0x33, 0xEC, 0xA1] # 13107
- expect_tx: [0x01, 0x03, 0x01, 0x70, 0x00, 0x03, 0x05, 0xEC] # holding 0x170 count 3
inject_rx: [0x01, 0x03, 0x06, 0x00, 0x2A, 0x1B, 0x2C, 0x03, 0x0D, 0x3E, 0xAB] # 6 bytes
- expect_tx: [0x01, 0x03, 0x01, 0x61, 0x00, 0x01, 0xD4, 0x28] # holding 0x161 count 1
inject_rx: [0x01, 0x03, 0x02, 0x01, 0x61, 0x78, 0x3C] # 353
modbus:
uart_id: virtual_uart_dev
@@ -104,8 +102,9 @@ sensor:
value_type: U_DWORD
modbus_controller_id: modbus_controller_ok
# D - a wide (response_size) register followed by a contiguous one: the follower must start after the
# bytes the wide register actually returned, not after 2 * register_count.
# D - a wide (response_size) register followed by a contiguous reuse:true one (auto never joins past
# a response_size register): the follower must start after the bytes the wide register actually
# returned, not after two per register.
- platform: modbus_controller
name: "wide_first"
address: 0x130
@@ -118,6 +117,7 @@ sensor:
address: 0x131
register_type: holding
value_type: U_WORD
reuse_previous_range: true
modbus_controller_id: modbus_controller_ok
# E - a gap: these must never share a range.
@@ -195,13 +195,14 @@ sensor:
value_type: U_WORD
modbus_controller_id: modbus_controller_ok
# H - a sensor pinned to its own range, followed by a contiguous one.
# H - a sensor that never joins the range built before it (reuse_previous_range: false), followed
# by a contiguous plain item that extends the new range it started.
- platform: modbus_controller
name: "forced_first"
address: 0x160
register_type: holding
value_type: U_WORD
force_new_range: true
reuse_previous_range: false
modbus_controller_id: modbus_controller_ok
- platform: modbus_controller
name: "forced_next"
@@ -0,0 +1,227 @@
esphome:
name: uart-mock-modbus-ranges-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
# Each expect_tx below pins the exact read request the controller's range builder emits, so this
# fixture is a wire-level test of reuse_previous_range (auto/yes/no), gap joins, same-register reuse,
# response_size surplus accounting, and RAW/text block reads.
uart_mock:
- id: virtual_uart_dev
baud_rate: 9600
rx_full_threshold: 120
rx_timeout: 2
# auto_start must be false to avoid races: the test presses the
# "Start Scenario" button only after subscribing to states.
auto_start: false
debug:
responses:
- expect_tx: [0x01, 0x03, 0x00, 0x00, 0x00, 0x03, 0x05, 0xCB] # auto adjacency: one read covers 0x00-0x02
inject_rx: [0x01, 0x03, 0x06, 0x00, 0x01, 0x00, 0x02, 0x00, 0x03, 0xFD, 0x74]
- expect_tx: [0x01, 0x03, 0x00, 0x10, 0x00, 0x01, 0x85, 0xCF] # auto gap: 0x10 alone
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x04, 0xB9, 0x87]
- expect_tx: [0x01, 0x03, 0x00, 0x13, 0x00, 0x01, 0x75, 0xCF] # auto gap: 0x13 alone
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x05, 0x78, 0x47]
- expect_tx: [0x01, 0x03, 0x00, 0x20, 0x00, 0x04, 0x45, 0xC3] # yes across gap: one read 0x20-0x23, gap registers ignored
inject_rx: [0x01, 0x03, 0x08, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x09, 0x33, 0xD1]
- expect_tx: [0x01, 0x03, 0x00, 0x30, 0x00, 0x01, 0x84, 0x05] # no isolation: 0x30 alone despite adjacency
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x0A, 0x38, 0x43]
- expect_tx: [0x01, 0x03, 0x00, 0x31, 0x00, 0x01, 0xD5, 0xC5] # no isolation: 0x31 alone
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x0B, 0xF9, 0x83]
- expect_tx: [0x01, 0x03, 0x00, 0x3F, 0x00, 0x01, 0xB4, 0x06] # open NEVER: 0x3F alone (the reuse:false item split off)
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x0C, 0xB8, 0x41]
- expect_tx: [0x01, 0x03, 0x00, 0x40, 0x00, 0x02, 0xC5, 0xDF] # open NEVER: 0x40 (reuse: false) still extended by the auto item at 0x41
inject_rx: [0x01, 0x03, 0x04, 0x00, 0x0D, 0x00, 0x0E, 0xEA, 0x34]
- expect_tx: [0x01, 0x03, 0x00, 0x50, 0x00, 0x01, 0x84, 0x1B] # same-address reuse: one read, two sensors on 0x50
inject_rx: [0x01, 0x03, 0x02, 0x12, 0x34, 0xB5, 0x33]
- expect_tx: [0x01, 0x03, 0x00, 0x60, 0x00, 0x04, 0x44, 0x17] # text block + adjacent word: one read 0x60-0x63
inject_rx: [0x01, 0x03, 0x08, 0x41, 0x42, 0x43, 0x44, 0x45, 0x46, 0x00, 0x0F, 0x78, 0x0E]
- expect_tx: [0x01, 0x03, 0x00, 0x70, 0x00, 0x02, 0xC5, 0xD0] # response_size surplus: 0x70 answers 4 bytes, reuse:true word at 0x71 shifted along
inject_rx: [0x01, 0x03, 0x06, 0x00, 0x10, 0xAA, 0xBB, 0x00, 0x11, 0x71, 0x47]
- expect_tx: [0x01, 0x03, 0x00, 0x90, 0x00, 0x01, 0x84, 0x27] # auto after surplus: 0x90 alone (auto never joins past response_size)
inject_rx: [0x01, 0x03, 0x04, 0x00, 0x18, 0xCC, 0xDD, 0xEF, 0x6D]
- expect_tx: [0x01, 0x03, 0x00, 0x91, 0x00, 0x01, 0xD5, 0xE7] # auto after surplus: 0x91 alone
inject_rx: [0x01, 0x03, 0x02, 0x00, 0x19, 0x79, 0x8E]
- expect_tx: [0x01, 0x03, 0x00, 0x80, 0x00, 0x04, 0x45, 0xE1] # RAW block via response_size: 8 bytes = 4 registers in one read
inject_rx: [0x01, 0x03, 0x08, 0x00, 0x14, 0x00, 0x15, 0x00, 0x16, 0x00, 0x17, 0x6D, 0xDF]
modbus:
uart_id: virtual_uart_dev
send_wait_time: 200ms
turnaround_time: 10ms
modbus_controller:
- address: 1
id: ranges_controller
max_cmd_retries: 0
# The test triggers a single poll by pressing the "Start Scenario" button
update_interval: never
sensor:
# Case 1: three adjacent registers merge into one read (auto default)
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "adjacent_a"
register_type: holding
address: 0x00
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "adjacent_b"
register_type: holding
address: 0x01
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "adjacent_c"
register_type: holding
address: 0x02
# Case 2: a gap keeps auto items apart
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "gap_a"
register_type: holding
address: 0x10
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "gap_b"
register_type: holding
address: 0x13
# Case 3: reuse_previous_range: true bridges the gap into one read
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "bridge_a"
register_type: holding
address: 0x20
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "bridge_b"
register_type: holding
address: 0x23
reuse_previous_range: true
# Case 4: reuse_previous_range: false splits adjacent registers
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "split_a"
register_type: holding
address: 0x30
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "split_b"
register_type: holding
address: 0x31
reuse_previous_range: false
# Case 5: a reuse:false item starts its own range but stays open for later auto items
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "open_prev"
register_type: holding
address: 0x3F
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "open_never"
register_type: holding
address: 0x40
reuse_previous_range: false
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "open_tagalong"
register_type: holding
address: 0x41
# Case 6: two sensors on the same register share one read
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "shared_lo"
register_type: holding
address: 0x50
bitmask: 0x00FF
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "shared_hi"
register_type: holding
address: 0x50
bitmask: 0xFF00
# Case 10 (text block, see text_sensor below) shares the range with this word at 0x63
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "after_text"
register_type: holding
address: 0x63
# Case 11: response_size surplus — the device answers 4 bytes for this single register, so the
# following sensor's data sits 2 bytes later than its address alone implies. Joining past a
# non-standard response_size takes an explicit reuse_previous_range: true.
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "surplus"
register_type: holding
address: 0x70
response_size: 4
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "after_surplus"
register_type: holding
address: 0x71
reuse_previous_range: true
# Case 13: auto never joins past a response_size register — despite adjacency these poll separately
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "surplus_split"
register_type: holding
address: 0x90
response_size: 4
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "after_surplus_split"
register_type: holding
address: 0x91
# Case 12: RAW + response_size reads a block of ceil(8/2) = 4 registers
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "raw_block"
register_type: holding
address: 0x80
value_type: RAW
response_size: 8
lambda: |-
return (float) data.size();
text_sensor:
# Case 10: text sensor reads 3 registers (response_size 6)
- platform: modbus_controller
modbus_controller_id: ranges_controller
name: "text_block"
register_type: holding
address: 0x60
response_size: 6
raw_encode: NONE
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: |-
id(virtual_uart_dev).start_scenario();
id(ranges_controller).set_update_interval(1000);
id(ranges_controller).start_poller();
@@ -32,9 +32,9 @@ uart_mock:
# duplicate or overlapping range would put an extra frame on the bus and fail to match.
- expect_tx: [0x01, 0x03, 0x90, 0x01, 0x00, 0x02, 0xB8, 0xCB] # Read holding 0x9001 count 2 on device 1
inject_rx: [0x01, 0x03, 0x04, 0x03, 0x97, 0x02, 0x91, 0x8B, 0x57] # 0x9001=0x0397, 0x9002=0x0291
# A force_new_range sensor at a HIGH address (0x30) sorts before the plain sensor at a LOW address
# (0x10). The two must poll as separate ranges: the covered branch's lower-bound check prevents the
# 0x10 sensor from being absorbed into the forced 0x30 range with a wrapped byte offset.
# A sensor at 0x30 with the deprecated force_new_range (migrates to reuse_previous_range: false)
# and a plain sensor at 0x10. The two must poll as separate ranges: the 0x10 sensor must not be
# absorbed into the isolated 0x30 range.
- expect_tx: [0x01, 0x03, 0x00, 0x30, 0x00, 0x01, 0x84, 0x05] # Read holding 0x30 count 1 (forced range)
inject_rx: [0x01, 0x03, 0x02, 0x01, 0x11, 0x79, 0xD8] # 0x30 = 0x0111 = 273
- expect_tx: [0x01, 0x03, 0x00, 0x10, 0x00, 0x01, 0x85, 0xCF] # Read holding 0x10 count 1 (own range)
@@ -87,7 +87,7 @@ sensor:
register_type: holding
value_type: U_WORD
modbus_controller_id: modbus_controller_ok
# Forced sensor at a high address: sorts first, opens its own isolated range
# Isolated sensor (deprecated spelling, migrates to reuse_previous_range: false): own range
- platform: modbus_controller
name: "forced_high"
address: 0x30
@@ -95,7 +95,7 @@ sensor:
value_type: U_WORD
force_new_range: true
modbus_controller_id: modbus_controller_ok
# Plain sensor at a lower address: must get its own range, never absorbed into the forced one
# Plain sensor at a lower address: must get its own range, never absorbed into the isolated one
- platform: modbus_controller
name: "plain_low"
address: 0x10
+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,
)
+76 -1
View File
@@ -21,7 +21,7 @@ import asyncio
from collections.abc import Callable
from dataclasses import dataclass
from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo
from aioesphomeapi import ButtonInfo, NumberInfo, SwitchInfo, TextSensorState
import pytest
from .state_utils import SensorTracker, find_entity, wait_for_state
@@ -1158,3 +1158,78 @@ async def test_uart_mock_modbus_deprecated_write_buffer(
assert warn_count == 1, (
f"deprecation warning should fire exactly once per entity, got {warn_count}"
)
@pytest.mark.asyncio
async def test_uart_mock_modbus_ranges(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Wire-level test of the range builder's reuse_previous_range semantics.
Every expect_tx in the fixture pins the exact read request the controller emits, so a
wrongly merged or split range fails on the mock before any value arrives. Covers: auto
adjacency merging, auto gap splitting, reuse:true bridging a gap (with correct data
offsets past the gap), reuse:false splitting adjacent registers while staying open for
later auto items, two sensors sharing one register, a text block read sized by
response_size with a following word, response_size surplus shifting a later reuse:true
sensor's bytes while an auto sensor refuses to join past the surplus, and a RAW block
read of ceil(response_size / 2) registers.
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {
"adjacent_a": 1,
"adjacent_b": 2,
"adjacent_c": 3,
"gap_a": 4,
"gap_b": 5,
"bridge_a": 6,
"bridge_b": 9,
"split_a": 10,
"split_b": 11,
"open_prev": 12,
"open_never": 13,
"open_tagalong": 14,
"shared_lo": 0x34,
"shared_hi": 0x12,
"after_text": 15,
"surplus": 16,
"after_surplus": 17,
"surplus_split": 24,
"after_surplus_split": 25,
"raw_block": 8, # the RAW lambda publishes data.size(): 4 registers = 8 bytes
}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
# The tracker only handles numeric sensors; capture the text block separately.
text_future: asyncio.Future = asyncio.get_running_loop().create_future()
tracker_on_state = tracker.on_state
def on_state(state) -> None:
if (
isinstance(state, TextSensorState)
and not state.missing_state
and state.state == "ABCDEF"
and not text_future.done()
):
text_future.set_result(True)
tracker_on_state(state)
tracker.on_state = on_state
async with (
run_compiled(yaml_config, line_callback=line_callback),
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
# text_block is not tracker-registered (non-numeric), so time out explicitly.
try:
await asyncio.wait_for(text_future, timeout=5.0)
except TimeoutError:
pytest.fail("text_block never published 'ABCDEF'")
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
+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,246 @@
"""Tests for the ninja build-tool helper script."""
from __future__ import annotations
from pathlib import Path
import subprocess
import sys
from unittest.mock import MagicMock, patch
import pytest
from esphome.build_gen import build_tool
def test_ar_removes_stale_archive(tmp_path: Path) -> None:
archive = tmp_path / "lib.a"
archive.write_text("stale")
rsp = tmp_path / "lib.a.rsp"
rsp.write_text("a.o\n")
with (
patch.object(
build_tool.sys,
"argv",
["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
),
patch.object(
build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
) as mock_run,
):
assert build_tool.main() == 0
assert not archive.exists()
# The rspfile is expanded by the shim (GNU ar would escape backslashes)
assert mock_run.call_args[0][0] == ["ar-bin", "rcs", str(archive), "a.o"]
def test_copy(tmp_path: Path) -> None:
src = tmp_path / "firmware.bin"
src.write_text("data")
dst = tmp_path / "firmware.factory.bin"
with patch.object(
build_tool.sys, "argv", ["build_tool", "copy", str(src), str(dst)]
):
assert build_tool.main() == 0
assert dst.read_text() == "data"
def test_unknown_mode(capsys: pytest.CaptureFixture[str]) -> None:
with patch.object(build_tool.sys, "argv", ["build_tool", "bogus"]):
assert build_tool.main() == 1
assert "unknown build_tool mode" in capsys.readouterr().err
def test_runs_as_script(tmp_path: Path) -> None:
"""The ninja rules invoke the file as a plain script."""
src = tmp_path / "a.bin"
src.write_text("x")
dst = tmp_path / "b.bin"
result = subprocess.run(
[sys.executable, build_tool.__file__, "copy", str(src), str(dst)],
check=False,
)
assert result.returncode == 0
assert dst.read_text() == "x"
def test_ar_expands_rspfile_without_escaping(tmp_path) -> None:
"""Backslash paths survive: the shim expands the rspfile itself instead
of letting GNU ar treat backslashes as escapes."""
rsp = tmp_path / "objs.rsp"
rsp.write_text("obj/a.o\nsub\\b.o\n")
with (
patch.object(
build_tool.sys,
"argv",
["build_tool", "ar", "ar-bin", str(tmp_path / "lib.a"), str(rsp)],
),
patch.object(
build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
) as mock_run,
):
assert build_tool.main() == 0
assert mock_run.call_args[0][0] == [
"ar-bin",
"rcs",
str(tmp_path / "lib.a"),
"obj/a.o",
"sub\\b.o",
]
def test_ar_unquotes_ninja_escaped_paths(tmp_path: Path) -> None:
"""The shim strips a simple surrounding quote, since ninja shell-
quotes special rsp paths, so ar sees the real filename."""
rsp = tmp_path / "t.rsp"
rsp.write_text("'obj/a b.o'\nobj/c.o\n")
with (
patch.object(
build_tool.sys, "argv", ["bt", "ar", "/usr/bin/ar", "lib.a", str(rsp)]
),
patch.object(build_tool.subprocess, "run") as mock_run,
):
mock_run.return_value.returncode = 0
rc = build_tool.main()
assert rc == 0
assert mock_run.call_args.args[0] == [
"/usr/bin/ar",
"rcs",
"lib.a",
"obj/a b.o",
"obj/c.o",
]
def test_ar_empty_object_list_fails(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A lost object list is an error here, not undefined symbols at link."""
rsp = tmp_path / "t.rsp"
rsp.write_text("\n\n")
with patch.object(
build_tool.sys, "argv", ["bt", "ar", "/usr/bin/ar", "lib.a", str(rsp)]
):
rc = build_tool.main()
assert rc == 1
assert "no objects listed" in capsys.readouterr().err
def test_ar_batches_long_object_lists(tmp_path: Path) -> None:
"""The expanded argv must stay under the Windows 32767-char limit: a
long object list creates with rcs, then appends with qs."""
archive = tmp_path / "lib.a"
rsp = tmp_path / "lib.a.rsp"
objects = [f"dir/{'x' * 120}_{i}.o" for i in range(400)]
rsp.write_text("\n".join(objects) + "\n")
with (
patch.object(
build_tool.sys,
"argv",
["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
),
patch.object(
build_tool.subprocess, "run", return_value=MagicMock(returncode=0)
) as mock_run,
):
assert build_tool.main() == 0
calls = [c[0][0] for c in mock_run.call_args_list]
assert len(calls) > 1
assert calls[0][1] == "rcs"
assert all(c[1] == "qs" for c in calls[1:])
assert [o for c in calls for o in c[3:]] == objects
assert all(sum(len(a) + 1 for a in c) < 32000 for c in calls)
def test_ar_batch_failure_stops(tmp_path: Path) -> None:
"""A failing batch propagates its exit code without running the rest."""
archive = tmp_path / "lib.a"
rsp = tmp_path / "lib.a.rsp"
rsp.write_text("\n".join(f"{'y' * 200}_{i}.o" for i in range(300)) + "\n")
with (
patch.object(
build_tool.sys,
"argv",
["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
),
patch.object(
build_tool.subprocess,
"run",
side_effect=lambda cmd, **kw: (
archive.write_text("partial"),
MagicMock(returncode=3),
)[1],
) as mock_run,
):
assert build_tool.main() == 3
assert mock_run.call_count == 1
# The failed batch must not leave a truncated archive behind
assert not archive.exists()
def test_ar_exception_leaves_no_partial_archive(tmp_path: Path) -> None:
"""A missing ar binary mid-loop must not leave a truncated archive from
earlier successful batches."""
archive = tmp_path / "lib.a"
rsp = tmp_path / "lib.a.rsp"
rsp.write_text("a.o\n")
with (
patch.object(
build_tool.sys,
"argv",
["build_tool", "ar", "ar-bin", str(archive), str(rsp)],
),
patch.object(
build_tool.subprocess,
"run",
side_effect=lambda cmd, **kw: (
archive.write_text("partial"),
(_ for _ in ()).throw(FileNotFoundError("no ar")),
),
),
pytest.raises(FileNotFoundError),
):
build_tool.main()
assert not archive.exists()
def test_surplus_arguments_error(capsys: pytest.CaptureFixture[str]) -> None:
"""A mis-specified ninja rule passing extra operands errors instead of
silently dropping them."""
with patch.object(
build_tool.sys, "argv", ["build_tool", "copy", "a", "b", "extra"]
):
assert build_tool.main() == 1
assert "expected 2 arguments, got 3" in capsys.readouterr().err
def test_copy_same_file_keeps_the_input(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A same-file copy (dst IS src) must not unlink the input, and fails
with a message and exit code like the other shim paths."""
src = tmp_path / "firmware.bin"
src.write_bytes(b"image")
with patch.object(
build_tool.sys, "argv", ["build_tool", "copy", str(src), str(src)]
):
assert build_tool.main() == 1
assert src.read_bytes() == b"image"
assert "failed" in capsys.readouterr().err
def test_copy_failure_leaves_no_partial_output(tmp_path: Path) -> None:
"""A failed copy unlinks the destination; a partial firmware image must
never be left on disk."""
dst = tmp_path / "firmware.factory.bin"
dst.write_text("stale")
with (
patch.object(build_tool.shutil, "copyfile", side_effect=OSError("disk full")),
patch.object(
build_tool.sys,
"argv",
["build_tool", "copy", str(tmp_path / "src.bin"), str(dst)],
),
):
assert build_tool.main() == 1
assert not dst.exists()
@@ -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(
File diff suppressed because it is too large Load Diff
+27
View File
@@ -520,6 +520,33 @@ def test_check_esp_idf_install_feature_failure(espidf_mocks: SimpleNamespace) ->
check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
def test_python_deps_use_uv_when_available(
espidf_mocks: SimpleNamespace, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The python env installs go through uv when on the PATH, pip otherwise."""
monkeypatch.delenv("UV_HTTP_RETRIES", raising=False)
with patch(
"esphome.espidf.framework.shutil.which",
# Keyed on the name: the same which() also probes the default tools
side_effect=lambda name: "/usr/bin/uv" if name == "uv" else None,
):
check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
upgrade_call, feature_call = espidf_mocks.run_ok.call_args_list[1:3]
upgrade_cmd, feature_cmd = upgrade_call.args[0], feature_call.args[0]
assert upgrade_cmd[:3] == ["/usr/bin/uv", "pip", "install"]
assert "--python" in upgrade_cmd
assert feature_cmd[:3] == ["/usr/bin/uv", "pip", "install"]
assert upgrade_call.kwargs["env"]["UV_HTTP_RETRIES"] == "10"
espidf_mocks.run_ok.reset_mock()
monkeypatch.setenv("UV_HTTP_RETRIES", "3") # an explicit user value wins
with patch("esphome.espidf.framework.shutil.which", return_value=None):
check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])
upgrade_call = espidf_mocks.run_ok.call_args_list[1]
assert upgrade_call.args[0][1:4] == ["-m", "pip", "install"]
assert upgrade_call.kwargs["env"]["UV_HTTP_RETRIES"] == "3"
def _mark_installed() -> None:
"""Create the extracted marker and python-env interpreter so the install
check takes the already-installed path rather than force-installing."""
+73 -1
View File
@@ -1,3 +1,4 @@
import errno
import io
import logging
import os
@@ -5,7 +6,7 @@ from pathlib import Path
import socket
import stat
import types
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock, call, patch
from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
from hypothesis import given, settings
@@ -966,6 +967,77 @@ def test_copy_file_if_changed_nonexistent_source(tmp_path: Path) -> None:
helpers.copy_file_if_changed(src, dst)
def test_rmtree_removes_tree(tmp_path: Path) -> None:
"""Test rmtree removes a populated directory tree."""
target = tmp_path / "target"
(target / "sub").mkdir(parents=True)
(target / "sub" / "file.txt").write_text("content")
helpers.rmtree(target)
assert not target.exists()
def test_rmtree_nonexistent_path(tmp_path: Path) -> None:
"""Test rmtree on an already-removed path is a no-op."""
helpers.rmtree(tmp_path / "gone")
def test_rmtree_retries_when_directory_repopulated(tmp_path: Path) -> None:
"""Test rmtree retries when a file appears mid-delete (Finder .DS_Store race)."""
target = tmp_path / "target"
(target / "sub").mkdir(parents=True)
real_rmdir = os.rmdir
repopulated = False
def racy_rmdir(path, **kwargs):
nonlocal repopulated
if not repopulated and Path(path).name == "target":
repopulated = True
(target / ".DS_Store").write_text("x") # Finder wins the race
real_rmdir(path, **kwargs)
with patch("os.rmdir", side_effect=racy_rmdir), patch("time.sleep"):
helpers.rmtree(target)
assert repopulated
assert not target.exists()
def test_rmtree_raises_after_retries_exhausted(tmp_path: Path) -> None:
"""Test rmtree gives up on a persistent ENOTEMPTY once attempts run out."""
target = tmp_path / "target"
target.mkdir()
errs = [
OSError(errno.ENOTEMPTY, "Directory not empty", str(target))
for _ in range(helpers.RMTREE_MAX_ATTEMPTS)
]
with (
patch("shutil.rmtree", side_effect=errs) as mock_rmtree,
patch("time.sleep") as mock_sleep,
pytest.raises(OSError, match="Directory not empty") as excinfo,
):
helpers.rmtree(target)
assert mock_rmtree.call_count == helpers.RMTREE_MAX_ATTEMPTS
assert mock_sleep.call_args_list == [call(0.05), call(0.1)]
# Final failure chains to the last retried race
assert excinfo.value is errs[-1]
assert excinfo.value.__cause__ is errs[-2]
def test_rmtree_does_not_retry_other_oserror(tmp_path: Path) -> None:
"""Test rmtree raises non-ENOTEMPTY errors immediately."""
target = tmp_path / "target"
target.mkdir()
err = OSError(errno.EACCES, "Permission denied", str(target))
with (
patch("shutil.rmtree", side_effect=err) as mock_rmtree,
pytest.raises(OSError, match="Permission denied"),
):
helpers.rmtree(target)
assert mock_rmtree.call_count == 1
def test_resolve_ip_address_sorting() -> None:
"""Test that results are sorted by preference."""
# Create multiple address infos with different preferences
+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
+206 -3
View File
@@ -11,7 +11,7 @@ from unittest.mock import Mock
import pytest
from esphome.core import EsphomeError, Library
from esphome.core import CORE, EsphomeError, Library
import esphome.platformio.library as lib
from esphome.platformio.library import (
SOURCE_KIND_FOR_SUFFIX,
@@ -30,9 +30,13 @@ from esphome.platformio.library import (
)
def _backend(emit=lambda component: None) -> LibraryBackend:
def _backend(emit=lambda component: None, provides=None) -> LibraryBackend:
return LibraryBackend(
platform="espressif32", framework="espidf", emit=emit, cache_key="idf"
platform="espressif32",
framework="espidf",
emit=emit,
cache_key="idf",
provides=provides,
)
@@ -971,3 +975,202 @@ def test_source_kind_map_shape() -> None:
assert SOURCE_KIND_FOR_SUFFIX[".S"] == "aspp"
assert SOURCE_KIND_FOR_SUFFIX[".c"] == "c"
assert SOURCE_KIND_FOR_SUFFIX[".cpp"] == "cxx"
# SCons's case-sensitive C++ suffixes: PIO compiles .C as C++
assert SOURCE_KIND_FOR_SUFFIX[".C"] == "cxx"
assert SOURCE_KIND_FOR_SUFFIX[".C++"] == "cxx"
def test_versionless_platform_filtered_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A version-less dependency the platform filter excludes is
deliberately absent, not a drop to warn about."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [{"name": "Hash", "platforms": "espressif8266"}],
}
},
)
convert_libraries([Library("esphome/A", None, None)], _backend())
assert "has no version to resolve" not in caplog.text
def test_versionless_ignored_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A lib_ignore'd version-less dependency is deliberately excluded, not
a drop; no reconciliation warning."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]}},
)
CORE.platformio_options = {"lib_ignore": ["Hash"]}
convert_libraries([Library("esphome/A", None, None)], _backend())
assert "has no version to resolve" not in caplog.text
def test_versionless_dependency_without_provider_warns(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A backend whose tree could supply the name warns on the drop; one
without provides() can never act on it, so it stays at debug."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
# The duplicate entry warns once (reconciliation dedup)
"dependencies": [{"name": "Hash"}, {"name": "Hash"}],
}
},
)
convert_libraries(
[Library("esphome/A", None, None)], _backend(provides=lambda name: False)
)
assert (
caplog.text.count(
"Hash of esphome/A has no version to resolve and nothing provides it"
)
== 1
)
caplog.clear()
with caplog.at_level(logging.DEBUG):
convert_libraries([Library("esphome/A", None, None)], _backend())
records = [
r
for r in caplog.records
if "has no version to resolve and nothing provides it" in r.message
]
assert records and all(r.levelno == logging.DEBUG for r in records)
def test_url_version_dependency_is_not_substituted_by_provides(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A URL-valued version names one specific source; the backend-provided
skip must not replace it with the bundled copy."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [
{"name": "Hash", "version": "https://github.com/o/Hash.git"}
],
},
"o/Hash": {"name": "Hash"},
},
)
emitted: list[str] = []
convert_libraries(
[Library("esphome/A", "1.0.0", None)],
_backend(emit=lambda c: emitted.append(c.name), provides=lambda name: True),
)
assert "Skip backend-provided" not in caplog.text
assert "using the library bundled" not in caplog.text
assert any("o/hash" in n.lower() for n in emitted)
def test_versionless_owner_qualified_dependency_warns_despite_provides(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An owner-qualified version-less dependency is not satisfied by
provides(); it must still warn."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {
"name": "A",
"dependencies": [{"name": "Wire", "owner": "Foo"}],
}
},
)
convert_libraries(
[Library("esphome/A", None, None)],
_backend(provides=lambda name: name == "Wire"),
)
assert "Wire of esphome/A has no version to resolve" in caplog.text
def test_versionless_provided_dependency_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An owner-less version-less dependency the backend provides is added
by the backend after emit; no reconciliation warning."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "Wire"}]}},
)
convert_libraries(
[Library("esphome/A", None, None)],
_backend(provides=lambda name: name == "Wire"),
)
assert "has no version to resolve" not in caplog.text
def test_versionless_dependency_requested_top_level_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A version-less dependency the config also requests top-level is in
the build; no drop warning even without a provides backend."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "Hash"}]},
"Hash": {"name": "Hash"},
},
)
convert_libraries(
[Library("esphome/A", None, None), Library("Hash", None, None)],
_backend(),
)
assert "has no version to resolve" not in caplog.text
def test_versionless_url_ish_dependency_name_warns_cleanly(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A malformed URL-ish dependency name falls to the drop warning, never
a RuntimeError out of the key parser."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{"esphome/A": {"name": "A", "dependencies": [{"name": "file://"}]}},
)
convert_libraries(
[Library("esphome/A", None, None)], _backend(provides=lambda name: False)
)
assert (
"file:// of esphome/A has no version to resolve and nothing provides it"
in caplog.text
)
def test_versionless_dependency_matching_resolved_manifest_name_stays_quiet(
tmp_path, monkeypatch, caplog: pytest.LogCaptureFixture
) -> None:
"""A bare name satisfied by an owner-qualified component's manifest
name is not a drop."""
_patch_download_with_manifests(
monkeypatch,
tmp_path,
{
"esphome/A": {"name": "A", "dependencies": [{"name": "B"}]},
"esphome/B": {"name": "B"},
},
)
convert_libraries(
[Library("esphome/A", None, None), Library("esphome/B", None, None)],
_backend(),
)
assert "has no version to resolve" not in caplog.text
+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()