github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-20 11:38:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into partition-table-ota

Browse Source
This commit is contained in:
J. Nick Koston
2026-04-29 14:33:22 -05:00
committed by GitHub
parent 8fd9af7f2a 5a146ab6b7
commit c293869e6e
60 changed files with 1370 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/api/test_proto_mac_varint.cpp
+1 -1
View File
@@ -112,7 +112,7 @@ TEST(ProtoMacVarint, AllOnes) { verify_mac(0xFFFFFFFFFFFFULL, 7); } // F
// 100 deterministic-random 48-bit MACs to catch regressions across the space.
TEST(ProtoMacVarint, RandomSample) {
// NOLINTNEXTLINE(cert-msc32-c,cert-msc51-cpp) -- intentional fixed seed for reproducibility.
// NOLINTNEXTLINE(cert-msc32-c,cert-msc51-cpp,bugprone-random-generator-seed) -- fixed seed for reproducibility
std::mt19937_64 rng(0xC0FFEE);
for (int i = 0; i < 100; i++) {
uint64_t mac = rng() & 0xFFFFFFFFFFFFULL;
tests/components/fan/common.yaml
+31
View File
@@ -57,3 +57,34 @@ binary_sensor:
return true;
}
return false;
# Exercise fan.turn_on with various field combinations so the
# TurnOnAction codegen paths get build coverage.
button:
- platform: template
name: "Fan Speed Only"
on_press:
- fan.turn_on:
id: test_fan
speed: 2
- platform: template
name: "Fan Oscillating + Direction"
on_press:
- fan.turn_on:
id: test_fan
oscillating: true
direction: REVERSE
- platform: template
name: "Fan All Fields"
on_press:
- fan.turn_on:
id: test_fan
oscillating: false
speed: 3
direction: FORWARD
- platform: template
name: "Fan Lambda Speed"
on_press:
- fan.turn_on:
id: test_fan
speed: !lambda 'return 1;'
tests/components/template/common-base.yaml
+14
View File
@@ -442,6 +442,20 @@ valve:
state: CLOSED
stop_action:
- logger.log: stop_action
# Exercise valve.control with various field combinations so the
# ControlAction codegen paths get build coverage.
- valve.control:
id: template_valve
stop: true
- valve.control:
id: template_valve
position: 50%
- valve.control:
id: template_valve
state: OPEN
- valve.control:
id: template_valve
position: !lambda 'return 0.25f;'
optimistic: true
text:
tests/integration/fixtures/fan_turn_on_action.yaml
+59
View File
@@ -0,0 +1,59 @@
esphome:
name: fan-turn-on-action-test
host:
api:
logger:
level: DEBUG
globals:
- id: test_speed
type: int
initial_value: "2"
fan:
- platform: template
id: test_fan
name: "Test Fan"
has_oscillating: true
has_direction: true
speed_count: 5
button:
# fan.turn_on: speed only
- platform: template
id: btn_speed
name: "Set Speed"
on_press:
- fan.turn_on:
id: test_fan
speed: 3
# fan.turn_on: oscillating + direction (no speed)
- platform: template
id: btn_oscillate_direction
name: "Set Oscillate Direction"
on_press:
- fan.turn_on:
id: test_fan
oscillating: true
direction: REVERSE
# fan.turn_on: all three fields
- platform: template
id: btn_all_fields
name: "Set All Fields"
on_press:
- fan.turn_on:
id: test_fan
oscillating: false
speed: 4
direction: FORWARD
# fan.turn_on: lambda for speed (exercises lambda path)
- platform: template
id: btn_lambda_speed
name: "Lambda Speed"
on_press:
- fan.turn_on:
id: test_fan
speed: !lambda "return id(test_speed);"
tests/integration/fixtures/scheduler_self_keyed.yaml
+112
View File
@@ -0,0 +1,112 @@
esphome:
debug_scheduler: true # Enable scheduler leak detection
name: scheduler-self-keyed-test
on_boot:
priority: -100
then:
- logger.log: "Starting scheduler self-keyed tests"
host:
api:
logger:
level: VERBOSE
globals:
- id: tests_done
type: bool
initial_value: 'false'
script:
- id: test_self_keyed
then:
- logger.log: "Testing self-keyed scheduler API"
- lambda: |-
// Two distinct keys backed by addresses of static markers — they
// must not collide even though both are self-keyed and share no
// Component pointer. Static storage gives them stable, unique
// addresses for the lifetime of the program.
static int key_a_marker = 0;
static int key_b_marker = 0;
void *key_a = &key_a_marker;
void *key_b = &key_b_marker;
// ---- Test 1: Self-keyed timeout fires ----
App.scheduler.set_timeout(key_a, 50, []() {
ESP_LOGI("test", "Self timeout A fired");
});
// ---- Test 2: Self-keyed cancel cancels only that key ----
App.scheduler.set_timeout(key_b, 100, []() {
ESP_LOGE("test", "ERROR: Self timeout B should have been cancelled");
});
App.scheduler.cancel_timeout(key_b);
// ---- Test 3: Two independent self keys don't collide ----
// Using fresh static markers so neither matches key_a / key_b.
static int key_c_marker = 0;
static int key_d_marker = 0;
void *key_c = &key_c_marker;
void *key_d = &key_d_marker;
App.scheduler.set_timeout(key_c, 150, []() {
ESP_LOGI("test", "Self timeout C fired");
});
App.scheduler.set_timeout(key_d, 150, []() {
ESP_LOGI("test", "Self timeout D fired");
});
// ---- Test 4: Self-keyed and component-keyed don't collide ----
// Use a self pointer that happens to look like a Component-attached id.
// The scheduler must treat them as separate namespaces.
static int shared_marker = 0;
void *self_shared = &shared_marker;
App.scheduler.set_timeout(self_shared, 200, []() {
ESP_LOGI("test", "Self timeout shared fired");
});
App.scheduler.set_timeout(id(test_sensor), 7777U, 200, []() {
ESP_LOGI("test", "Component timeout 7777 fired");
});
// ---- Test 5: Self-keyed interval fires multiple times then cancels ----
static int interval_count = 0;
static int key_e_marker = 0;
void *key_e = &key_e_marker;
App.scheduler.set_interval(key_e, 80, [key_e]() {
interval_count++;
if (interval_count == 2) {
ESP_LOGI("test", "Self interval E fired twice");
App.scheduler.cancel_interval(key_e);
}
});
// ---- Test 6: Re-registering same self-key replaces the timer ----
// The old timer must NOT fire; only the new one does.
static int key_f_marker = 0;
void *key_f = &key_f_marker;
App.scheduler.set_timeout(key_f, 250, []() {
ESP_LOGE("test", "ERROR: Self timeout F first registration should have been replaced");
});
App.scheduler.set_timeout(key_f, 300, []() {
ESP_LOGI("test", "Self timeout F replacement fired");
});
// Log completion after all timers should have fired
App.scheduler.set_timeout(id(test_sensor), 9999U, 1500, []() {
ESP_LOGI("test", "All self-keyed tests complete");
});
sensor:
- platform: template
name: Test Sensor
id: test_sensor
lambda: return 1.0;
update_interval: never
interval:
- interval: 0.1s
then:
- if:
condition:
lambda: 'return id(tests_done) == false;'
then:
- lambda: 'id(tests_done) = true;'
- script.execute: test_self_keyed
tests/integration/fixtures/valve_control_action.yaml
+69
View File
@@ -0,0 +1,69 @@
esphome:
name: valve-control-action-test
host:
api:
logger:
level: DEBUG
globals:
- id: test_position
type: float
initial_value: "0.42"
valve:
- platform: template
name: "Test Valve"
id: test_valve
has_position: true
optimistic: true
assumed_state: true
open_action:
- valve.template.publish:
id: test_valve
position: 1.0
close_action:
- valve.template.publish:
id: test_valve
position: 0.0
stop_action:
- valve.template.publish:
id: test_valve
current_operation: IDLE
button:
# valve.control: position only
- platform: template
id: btn_position
name: "Set Position"
on_press:
- valve.control:
id: test_valve
position: 50%
# valve.control: state alias for position 1.0
- platform: template
id: btn_open_state
name: "Open State"
on_press:
- valve.control:
id: test_valve
state: OPEN
# valve.control: lambda position (exercises lambda path)
- platform: template
id: btn_lambda_position
name: "Lambda Position"
on_press:
- valve.control:
id: test_valve
position: !lambda "return id(test_position);"
# valve.control: stop only — template valve's stop_action publishes
# current_operation: IDLE.
- platform: template
id: btn_stop
name: "Stop Valve"
on_press:
- valve.control:
id: test_valve
stop: true
tests/integration/test_fan_turn_on_action.py
+75
View File
@@ -0,0 +1,75 @@
"""Integration test for fan TurnOnAction.
Tests that fan.turn_on automation actions work correctly across multiple
field combinations and the lambda path.
"""
from __future__ import annotations
import asyncio
from aioesphomeapi import ButtonInfo, EntityState, FanDirection, FanInfo, FanState
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_fan_turn_on_action(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test fan TurnOnAction with constants and a lambda."""
loop = asyncio.get_running_loop()
async with run_compiled(yaml_config), api_client_connected() as client:
fan_state_future: asyncio.Future[FanState] | None = None
def on_state(state: EntityState) -> None:
if (
isinstance(state, FanState)
and fan_state_future is not None
and not fan_state_future.done()
):
fan_state_future.set_result(state)
async def wait_for_fan_state(timeout: float = 5.0) -> FanState:
nonlocal fan_state_future
fan_state_future = loop.create_future()
try:
return await asyncio.wait_for(fan_state_future, timeout)
finally:
fan_state_future = None
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
require_entity(entities, "test_fan", FanInfo)
async def press_and_wait(name: str) -> FanState:
btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
client.button_command(btn.key)
return await wait_for_fan_state()
# speed only
state = await press_and_wait("Set Speed")
assert state.state is True
assert state.speed_level == 3
# oscillating + direction
state = await press_and_wait("Set Oscillate Direction")
assert state.oscillating is True
assert state.direction == FanDirection.REVERSE
# all three fields
state = await press_and_wait("Set All Fields")
assert state.oscillating is False
assert state.speed_level == 4
assert state.direction == FanDirection.FORWARD
# lambda path: speed computed at runtime (test_speed global = 2)
state = await press_and_wait("Lambda Speed")
assert state.speed_level == 2
tests/integration/test_scheduler_self_keyed.py
+96
View File
@@ -0,0 +1,96 @@
"""Test the self-keyed scheduler API.
Verifies that `Scheduler::set_timeout(const void *, ...)` /
`set_interval(const void *, ...)` and the matching `cancel_*(const void *)`
overloads behave correctly: callbacks fire, distinct keys don't collide,
self-keyed and component-keyed namespaces are independent, and re-registering
the same key replaces the existing timer.
"""
import asyncio
import re
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_scheduler_self_keyed(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test self-keyed scheduler API."""
self_a_fired = asyncio.Event()
self_b_error = asyncio.Event()
self_c_fired = asyncio.Event()
self_d_fired = asyncio.Event()
self_shared_fired = asyncio.Event()
component_7777_fired = asyncio.Event()
self_interval_done = asyncio.Event()
self_f_first_error = asyncio.Event()
self_f_replacement_fired = asyncio.Event()
all_tests_complete = asyncio.Event()
def on_log_line(line: str) -> None:
clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
if "Self timeout A fired" in clean_line:
self_a_fired.set()
elif "ERROR: Self timeout B" in clean_line:
self_b_error.set()
elif "Self timeout C fired" in clean_line:
self_c_fired.set()
elif "Self timeout D fired" in clean_line:
self_d_fired.set()
elif "Self timeout shared fired" in clean_line:
self_shared_fired.set()
elif "Component timeout 7777 fired" in clean_line:
component_7777_fired.set()
elif "Self interval E fired twice" in clean_line:
self_interval_done.set()
elif "ERROR: Self timeout F first registration" in clean_line:
self_f_first_error.set()
elif "Self timeout F replacement fired" in clean_line:
self_f_replacement_fired.set()
elif "All self-keyed tests complete" in clean_line:
all_tests_complete.set()
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "scheduler-self-keyed-test"
try:
await asyncio.wait_for(all_tests_complete.wait(), timeout=5.0)
except TimeoutError:
pytest.fail("Not all self-keyed tests completed within 5 seconds")
# Test 1: self-keyed timeout fires
assert self_a_fired.is_set(), "Self timeout A should have fired"
# Test 2: cancel_timeout(self) actually cancels
assert not self_b_error.is_set(), "Self timeout B should have been cancelled"
# Test 3: distinct self keys don't collide
assert self_c_fired.is_set(), "Self timeout C should have fired"
assert self_d_fired.is_set(), "Self timeout D should have fired"
# Test 4: self-keyed and component-keyed namespaces are independent
assert self_shared_fired.is_set(), "Self timeout shared should have fired"
assert component_7777_fired.is_set(), "Component timeout 7777 should have fired"
# Test 5: self-keyed interval fires repeatedly and cancels cleanly
assert self_interval_done.is_set(), "Self interval E should have fired twice"
# Test 6: re-registering same self-key replaces the previous timer
assert not self_f_first_error.is_set(), (
"Self timeout F first registration should have been replaced"
)
assert self_f_replacement_fired.is_set(), (
"Self timeout F replacement should have fired"
)
tests/integration/test_valve_control_action.py
+72
View File
@@ -0,0 +1,72 @@
"""Integration test for valve ControlAction.
Tests that valve.control automation actions work correctly across multiple
field combinations and the lambda path.
"""
from __future__ import annotations
import asyncio
from aioesphomeapi import ButtonInfo, EntityState, ValveInfo, ValveOperation, ValveState
import pytest
from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_valve_control_action(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test valve ControlAction with constants and a lambda."""
loop = asyncio.get_running_loop()
async with run_compiled(yaml_config), api_client_connected() as client:
valve_state_future: asyncio.Future[ValveState] | None = None
def on_state(state: EntityState) -> None:
if (
isinstance(state, ValveState)
and valve_state_future is not None
and not valve_state_future.done()
):
valve_state_future.set_result(state)
async def wait_for_valve_state(timeout: float = 5.0) -> ValveState:
nonlocal valve_state_future
valve_state_future = loop.create_future()
try:
return await asyncio.wait_for(valve_state_future, timeout)
finally:
valve_state_future = None
entities, _ = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
await initial_state_helper.wait_for_initial_states()
require_entity(entities, "test_valve", ValveInfo)
async def press_and_wait(name: str) -> ValveState:
btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
client.button_command(btn.key)
return await wait_for_valve_state()
# valve.control: position only
state = await press_and_wait("Set Position")
assert state.position == pytest.approx(0.5, abs=0.01)
# valve.control: state alias for position 1.0
state = await press_and_wait("Open State")
assert state.position == pytest.approx(1.0, abs=0.01)
# valve.control: lambda position (test_position global = 0.42)
state = await press_and_wait("Lambda Position")
assert state.position == pytest.approx(0.42, abs=0.01)
# valve.control: stop only — template valve's stop_action publishes
# current_operation: IDLE.
state = await press_and_wait("Stop Valve")
assert state.current_operation == ValveOperation.IDLE
tests/unit_tests/test_external_files.py
+63
View File
@@ -469,6 +469,69 @@ def test_download_content_with_network_error_no_cache_fails(
external_files.download_content(url, test_file)
class _BodyReadErrorResponse:
"""Stand-in for `requests.Response` whose `.content` raises on access.
A small dedicated stub avoids mutating `MagicMock`'s class with a
`property` (which would leak across every other MagicMock-based test
in this file).
"""
def __init__(self, exc: Exception) -> None:
self._exc = exc
self.headers: dict[str, str] = {}
def raise_for_status(self) -> None:
return None
@property
def content(self) -> bytes:
raise self._exc
def test_download_content_with_body_read_error_uses_cache(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
setup_core: Path,
) -> None:
"""Body-read errors (chunked-decode/gzip-decode/mid-stream connection
drop) raise RequestException subclasses on `.content` access, not from
`requests.get` itself. They must follow the same fall-back-to-cache
path as a connect-time failure.
"""
test_file = setup_core / "cached.txt"
cached_content = b"cached content"
test_file.write_bytes(cached_content)
mock_has_remote_file_changed.return_value = True
mock_requests_get.return_value = _BodyReadErrorResponse(
requests.exceptions.ChunkedEncodingError("body truncated")
)
result = external_files.download_content("https://example.com/file.txt", test_file)
assert result == cached_content
def test_download_content_with_body_read_error_no_cache_fails(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
setup_core: Path,
) -> None:
"""A body-read failure with no cache available must surface as a
cv.Invalid, same as a connect-time failure with no cache.
"""
test_file = setup_core / "nonexistent.txt"
mock_has_remote_file_changed.return_value = True
mock_requests_get.return_value = _BodyReadErrorResponse(
requests.exceptions.ChunkedEncodingError("body truncated")
)
with pytest.raises(Invalid, match="Could not download from.*body truncated"):
external_files.download_content("https://example.com/file.txt", test_file)
def test_download_content_skip_external_update_uses_cache(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
tests/unit_tests/test_main.py
+4 -4
View File
@@ -2692,7 +2692,7 @@ def test_choose_upload_log_host_discovers_mac_suffix_devices(tmp_path: Path) ->
}
with (
patch(
"esphome.__main__.discover_mdns_devices", return_value=discovered
"esphome.zeroconf.discover_mdns_devices", return_value=discovered
) as mock_discover,
patch(
"esphome.__main__.choose_prompt", return_value="mydevice-abc123.local"
@@ -2740,7 +2740,7 @@ def test_choose_upload_log_host_mac_suffix_no_devices_found(
)
with (
patch("esphome.__main__.discover_mdns_devices", return_value={}),
patch("esphome.zeroconf.discover_mdns_devices", return_value={}),
caplog.at_level(logging.WARNING, logger="esphome.__main__"),
pytest.raises(EsphomeError),
):
@@ -2773,7 +2773,7 @@ def test_choose_upload_log_host_default_ota_discovers_mac_suffix(
"mydevice-def456.local": ["10.0.0.2"],
}
with patch(
"esphome.__main__.discover_mdns_devices", return_value=discovered
"esphome.zeroconf.discover_mdns_devices", return_value=discovered
) as mock_discover:
result = choose_upload_log_host(
default="OTA",
@@ -2802,7 +2802,7 @@ def test_choose_upload_log_host_default_ota_no_suffix_discovery(
name="mydevice",
)
with patch("esphome.__main__.discover_mdns_devices") as mock_discover:
with patch("esphome.zeroconf.discover_mdns_devices") as mock_discover:
result = choose_upload_log_host(
default="OTA",
check_default=None,