Merge remote-tracking branch 'upstream/dev' into integration

# Conflicts:
#	esphome/__main__.py
#	esphome/components/api/client.py
#	esphome/components/bl0942/bl0942.cpp
#	esphome/components/cover/__init__.py
#	esphome/components/cover/automation.h
#	esphome/components/esphome/ota/ota_esphome.cpp
#	esphome/components/ota/ota_backend_esp_idf.cpp
#	esphome/components/ota/ota_partitions_esp_idf.cpp
#	esphome/espota2.py
#	tests/components/template/common-base.yaml
#	tests/unit_tests/components/api/test_client.py
#	tests/unit_tests/test_main.py
This commit is contained in:
J. Nick Koston
2026-05-06 16:47:31 -05:00
125 changed files with 4761 additions and 1254 deletions
@@ -0,0 +1,84 @@
esphome:
name: test-line
esp32:
board: lolin_c3_mini
spi:
mosi_pin:
number: GPIO2
ignore_strapping_warning: true
clk_pin: GPIO1
display:
- platform: mipi_spi
data_rate: 20MHz
model: st7735
cs_pin:
number: GPIO8
ignore_strapping_warning: true
dc_pin:
number: GPIO3
lvgl:
widgets:
# Dict format
- line:
id: line_dict
points:
- x: 10
y: 20
- x: 100
y: 200
- x: 0
y: 0
# List format
- line:
id: line_list
points:
- [10, 20]
- [100, 200]
- [0, 0]
# String format
- line:
id: line_string
points:
- "10, 20"
- "100, 200"
- "0, 0"
# Percentage - dict format
- line:
id: line_pct_dict
points:
- x: "50%"
y: "75%"
# Percentage - list format
- line:
id: line_pct_list
points:
- ["50%", "75%"]
# Percentage - string format
- line:
id: line_pct_string
points:
- "50%, 75%"
# Mixed integer and percentage
- line:
id: line_mixed_dict
points:
- x: 10
y: "50%"
- x: "25%"
y: 200
- line:
id: line_mixed_list
points:
- [10, "50%"]
- ["25%", 200]
+147
View File
@@ -0,0 +1,147 @@
"""Tests for the LVGL line widget point schema and code generation."""
from __future__ import annotations
import re
import pytest
from esphome.components.lvgl.schemas import point_schema
from esphome.config_validation import Invalid
from esphome.const import CONF_X, CONF_Y
# ---------------------------------------------------------------------------
# Validation: point_schema normalises dict / list / string to same result
# ---------------------------------------------------------------------------
class TestPointSchemaValidation:
"""Test that all point input formats normalise to the same dict."""
@pytest.mark.parametrize(
"dict_input,list_input,string_input",
[
({CONF_X: 10, CONF_Y: 20}, [10, 20], "10, 20"),
({CONF_X: 0, CONF_Y: 0}, [0, 0], "0, 0"),
({CONF_X: 100, CONF_Y: 200}, [100, 200], "100, 200"),
({CONF_X: -5, CONF_Y: -10}, [-5, -10], "-5, -10"),
],
)
def test_integer_formats_produce_same_result(
self, dict_input, list_input, string_input
):
result_dict = point_schema(dict_input)
result_list = point_schema(list_input)
result_string = point_schema(string_input)
assert result_dict == result_list
assert result_dict == result_string
def test_percentage_formats_produce_same_result(self):
result_dict = point_schema({CONF_X: "50%", CONF_Y: "75%"})
result_list = point_schema(["50%", "75%"])
result_string = point_schema("50%, 75%")
assert result_dict == result_list
assert result_dict == result_string
def test_pixel_suffix_matches_plain_integer(self):
result_px = point_schema({CONF_X: "10px", CONF_Y: "20px"})
result_int = point_schema({CONF_X: 10, CONF_Y: 20})
assert result_px == result_int
@pytest.mark.parametrize(
"value",
[
{CONF_X: 50, CONF_Y: 75},
[50, 75],
"50, 75",
],
)
def test_output_contains_x_and_y(self, value):
result = point_schema(value)
assert CONF_X in result
assert CONF_Y in result
def test_list_wrong_length_raises(self):
with pytest.raises(Invalid, match="Invalid point"):
point_schema([1])
with pytest.raises(Invalid, match="Invalid point"):
point_schema([1, 2, 3])
def test_string_without_comma_raises(self):
with pytest.raises(Invalid, match="Invalid point"):
point_schema("garbage")
def test_string_extra_commas_raises(self):
with pytest.raises(Invalid, match="Invalid point"):
point_schema("1,2,3")
# ---------------------------------------------------------------------------
# Code generation: different point formats produce identical C++ output
# ---------------------------------------------------------------------------
_SET_POINTS_RE = re.compile(r"(\w+)->set_points\((.+?)\);")
def _extract_set_points(main_cpp: str) -> dict[str, str]:
"""Return {var_name: args_text} for every set_points() call found."""
return {m.group(1): m.group(2) for m in _SET_POINTS_RE.finditer(main_cpp)}
class TestLineCodeGeneration:
"""Verify that alternative point formats generate identical C++ code."""
@pytest.fixture()
def main_cpp(self, generate_main, component_config_path) -> str:
return generate_main(component_config_path("line_points.yaml"))
@pytest.fixture()
def set_points_calls(self, main_cpp) -> dict[str, str]:
return _extract_set_points(main_cpp)
def test_integer_points_all_formats_match(self, set_points_calls):
"""Dict, list, and string formats with integer points produce same set_points call."""
assert set_points_calls["line_dict"] == set_points_calls["line_list"]
assert set_points_calls["line_dict"] == set_points_calls["line_string"]
def test_percentage_points_all_formats_match(self, set_points_calls):
"""Dict, list, and string formats with percentage points produce same set_points call."""
assert set_points_calls["line_pct_dict"] == set_points_calls["line_pct_list"]
assert set_points_calls["line_pct_dict"] == set_points_calls["line_pct_string"]
def test_mixed_points_formats_match(self, set_points_calls):
"""Dict and list formats with mixed int/percent points produce same set_points call."""
assert (
set_points_calls["line_mixed_dict"] == set_points_calls["line_mixed_list"]
)
def test_integer_points_contain_expected_values(self, set_points_calls):
"""Integer points appear literally in the generated code."""
args = set_points_calls["line_dict"]
for val in ("10", "20", "100", "200"):
assert val in args
def test_percentage_points_use_lv_pct(self, set_points_calls):
"""Percentage points are generated using the lv_pct() macro."""
args = set_points_calls["line_pct_dict"]
assert "lv_pct(50)" in args
assert "lv_pct(75)" in args
def test_all_lines_present(self, set_points_calls):
"""All expected line IDs have a set_points call."""
expected = {
"line_dict",
"line_list",
"line_string",
"line_pct_dict",
"line_pct_list",
"line_pct_string",
"line_mixed_dict",
"line_mixed_list",
}
assert expected.issubset(set_points_calls.keys())
@@ -14,6 +14,7 @@ from esphome.components.packages import (
do_packages_pass,
is_package_definition,
merge_packages,
resolve_packages,
)
from esphome.components.substitutions import ContextVars, do_substitution_pass
import esphome.config as config_module
@@ -1621,3 +1622,122 @@ def test_remote_package_vars_resolved_against_sibling_package_substitutions(
actual = packages_pass(config)
assert actual[CONF_SENSOR][0]["pin"] == "GPIO5"
# ---------------------------------------------------------------------------
# resolve_packages — single-call wrapper around do_packages_pass + merge_packages
# ---------------------------------------------------------------------------
def test_resolve_packages_returns_config_unchanged_without_packages() -> None:
"""No ``packages:`` key → no-op, same dict back."""
config = {CONF_ESPHOME: {CONF_NAME: "test"}, CONF_WIFI: {CONF_SSID: "x"}}
result = resolve_packages(config)
assert result is config
assert CONF_PACKAGES not in result
def test_resolve_packages_loads_and_merges_in_one_call() -> None:
"""End-to-end: a config with one local-dict package gets its blocks flattened."""
config = {
CONF_ESPHOME: {CONF_NAME: "main"},
CONF_PACKAGES: {
"shared": {
CONF_WIFI: {CONF_SSID: "from_package"},
CONF_SENSOR: [
{CONF_PLATFORM: "template", CONF_NAME: "from_package_sensor"},
],
}
},
}
result = resolve_packages(config)
# ``packages:`` is gone — it was consumed by the merge.
assert CONF_PACKAGES not in result
# Blocks contributed by the package are now top-level.
assert result[CONF_WIFI][CONF_SSID] == "from_package"
assert result[CONF_SENSOR][0][CONF_NAME] == "from_package_sensor"
# The main config's own keys survive untouched.
assert result[CONF_ESPHOME][CONF_NAME] == "main"
def test_resolve_packages_preserves_main_config_overrides() -> None:
"""Main-config values win over package values for the same key.
Pinning the precedence ESPHome's compiler uses so any future
refactor of the wrapper doesn't accidentally flip the order.
"""
config = {
CONF_ESPHOME: {CONF_NAME: "main"},
CONF_WIFI: {CONF_SSID: "main_wins"},
CONF_PACKAGES: {
"shared": {CONF_WIFI: {CONF_SSID: "package_loses"}},
},
}
result = resolve_packages(config)
assert result[CONF_WIFI][CONF_SSID] == "main_wins"
def test_resolve_packages_forwards_command_line_substitutions() -> None:
"""``command_line_substitutions`` reaches the underlying ``do_packages_pass``.
The wrapper exists so external tools have one stable seam; if
that seam silently dropped a kwarg the underlying call accepts,
callers would see surprising behaviour. This pins the
pass-through.
"""
config = {
CONF_ESPHOME: {CONF_NAME: "main"},
CONF_PACKAGES: {"shared": {CONF_WIFI: {CONF_SSID: "from_package"}}},
}
with patch(
"esphome.components.packages.do_packages_pass",
wraps=do_packages_pass,
) as spy:
resolve_packages(config, command_line_substitutions={"foo": "bar"})
spy.assert_called_once()
_, kwargs = spy.call_args
assert kwargs.get("command_line_substitutions") == {"foo": "bar"}
def test_resolve_packages_does_not_run_substitutions() -> None:
"""``${var}`` placeholders inside package content stay literal.
The full ``validate_config`` pipeline runs ``do_substitution_pass``
BETWEEN ``do_packages_pass`` and ``merge_packages``; this wrapper
skips it on purpose. Pin that contract so a future refactor can't
silently start resolving substitutions and break callers that
deliberately compose the passes themselves.
"""
config = {
CONF_ESPHOME: {CONF_NAME: "main"},
CONF_SUBSTITUTIONS: {"ssid_value": "resolved_ssid"},
CONF_PACKAGES: {
"shared": {CONF_WIFI: {CONF_SSID: "${ssid_value}"}},
},
}
result = resolve_packages(config)
# Without ``do_substitution_pass`` the placeholder is preserved.
assert result[CONF_WIFI][CONF_SSID] == "${ssid_value}"
def test_resolve_packages_does_not_apply_extend_remove() -> None:
"""Top-level ``!remove`` / ``!extend`` markers stay in the merged dict.
The full ``validate_config`` pipeline runs ``resolve_extend_remove``
AFTER ``merge_packages``; this wrapper skips it on purpose. Pin
that contract: a package-contributed block paired with a top-level
``!remove`` is left as-is for callers to handle (or for them to
call ``resolve_extend_remove`` themselves).
"""
config = {
CONF_ESPHOME: {CONF_NAME: "main"},
CONF_WIFI: Remove(),
CONF_PACKAGES: {
"shared": {CONF_WIFI: {CONF_SSID: "from_package"}},
},
}
result = resolve_packages(config)
# ``merge_packages`` keeps the top-level ``!remove`` (it wins
# over the package value during merge), and the marker is not
# resolved by this wrapper.
assert isinstance(result[CONF_WIFI], Remove)
@@ -50,12 +50,33 @@ esphome:
format: "After delay, body still: %s"
args:
- body.c_str()
# Regression test for esphome/esphome#16224: a LightControlAction
# nested inside on_response with capture_response: true puts
# `std::string &` into the trigger's Ts..., which exposed a codegen
# bug where the apply lambda's parameter list did not match the
# ApplyFn signature.
- light.turn_on:
id: test_regression_light
brightness: 100%
effect: "None"
http_request:
useragent: esphome/tagreader
timeout: 10s
verify_ssl: ${verify_ssl}
output:
- platform: template
id: test_regression_output
type: float
write_action:
- logger.log: "set"
light:
- platform: monochromatic
id: test_regression_light
output: test_regression_output
script:
- id: does_not_compile
parameters:
+14 -3
View File
@@ -649,11 +649,15 @@ lvgl:
on_scroll_begin:
logger.log: Button clicked
on_release:
logger.log: Button clicked
logger.log:
format: Button released at %d/%d
args: [point.x, point.y]
on_long_press_repeat:
logger.log: Button clicked
on_pressing:
logger.log: Button pressing
logger.log:
format: Button pressing at %d/%d
args: [point.x, point.y]
on_press_lost:
logger.log: Button press lost
on_single_click:
@@ -925,6 +929,10 @@ lvgl:
value: !lambda |-
static float yyy = 83.0;
return yyy + .8;
on_release:
logger.log:
format: Slider released at %d/%d with value %.0f
args: [point.x, point.y, x]
- button:
styles: spin_button
id: spin_up
@@ -1038,7 +1046,10 @@ lvgl:
- 5, 5
- x: !lambda return random_uint32() % 100;
y: !lambda return random_uint32() % 100;
- 70, 70
- x: 10%
y: 50%
- 70%, 70%
- [75%, 75%]
- 120, 10
- 180, 60
- 240, 10
@@ -17,3 +17,16 @@ media_player:
volume_max: 0.95
volume_min: 0.0
task_stack_in_psram: true
files:
- id: speaker_test_audio
file:
type: local
path: $component_dir/test.wav
script:
- id: play_built_in_file
then:
- media_player.speaker.play_on_device_media_file:
id: speaker_media_player_id
media_file: speaker_test_audio
announcement: true
Binary file not shown.
+18 -2
View File
@@ -202,6 +202,11 @@ sensor:
value: last
- timeout:
timeout: 1d
- to_ntc_temperature:
calibration:
b_constant: 3950
reference_temperature: 25.0°C
reference_resistance: 10kOhm
- to_ntc_resistance:
calibration:
- 10.0kOhm -> 25°C
@@ -270,8 +275,6 @@ cover:
stop_action:
- logger.log: stop_action
optimistic: true
on_open:
- logger.log: "Cover on_open (deprecated)"
on_opened:
- logger.log: "Cover fully opened"
on_closed:
@@ -369,6 +372,19 @@ number:
- valve.control:
id: template_valve
position: !lambda "return x / 100.0f;"
# Same regression test for cover.control: forces the apply-lambda
# codegen to handle a non-empty trigger Ts (float).
- platform: template
id: template_cover_position_number
optimistic: true
min_value: 0
max_value: 100
step: 1
on_value:
then:
- cover.control:
id: template_cover_with_triggers
position: !lambda "return x / 100.0f;"
select:
- platform: template
+8
View File
@@ -0,0 +1,8 @@
esphome:
on_boot:
- lambda: |-
ESP_LOGD("test", "millis=%u micros=%u cycles=%u",
(unsigned) millis(), (unsigned) micros(),
(unsigned) arch_get_cpu_cycle_count());
delay(1);
delayMicroseconds(1);
@@ -0,0 +1 @@
<<: !include common.yaml
+54 -57
View File
@@ -501,14 +501,15 @@ async def _read_stream_lines(
@asynccontextmanager
async def run_binary_and_wait_for_port(
async def run_binary(
binary_path: Path,
host: str,
port: int,
timeout: float = PORT_WAIT_TIMEOUT,
line_callback: Callable[[str], None] | None = None,
) -> AsyncGenerator[None]:
"""Run a binary, wait for it to open a port, and clean up on exit."""
) -> AsyncGenerator[tuple[asyncio.subprocess.Process, list[str]]]:
"""Run a binary under a PTY, capture log output, and clean up on exit.
Yields the running ``Process`` and a live list of captured log lines.
No port wait -- callers that need that should use
``run_binary_and_wait_for_port``."""
# Create a pseudo-terminal to make the binary think it's running interactively
# This is needed because the ESPHome host logger checks isatty()
controller_fd, device_fd = pty.openpty()
@@ -535,7 +536,6 @@ async def run_binary_and_wait_for_port(
controller_transport, _ = await loop.connect_read_pipe(
lambda: controller_protocol, os.fdopen(controller_fd, "rb", 0)
)
output_reader = controller_reader
if process.returncode is not None:
raise RuntimeError(
@@ -543,27 +543,59 @@ async def run_binary_and_wait_for_port(
"Ensure the binary is valid and can run successfully."
)
# Wait for the API server to start listening
loop = asyncio.get_running_loop()
start_time = loop.time()
# Start collecting output
stdout_lines: list[str] = []
output_tasks: list[asyncio.Task] = []
output_task = asyncio.create_task(
_read_stream_lines(controller_reader, stdout_lines, sys.stdout, line_callback)
)
try:
# Read from output stream
output_tasks = [
asyncio.create_task(
_read_stream_lines(
output_reader, stdout_lines, sys.stdout, line_callback
)
)
]
# Small yield to ensure the process has a chance to start
await asyncio.sleep(0)
yield process, stdout_lines
finally:
output_task.cancel()
result = await asyncio.gather(output_task, return_exceptions=True)
if isinstance(result[0], Exception) and not isinstance(
result[0], asyncio.CancelledError
):
print(f"Error reading from PTY: {result[0]}", file=sys.stderr)
# Close the PTY transport (Unix only)
if controller_transport is not None:
controller_transport.close()
# Cleanup: terminate the process gracefully
if process.returncode is None:
# Send SIGINT (Ctrl+C) for graceful shutdown
process.send_signal(signal.SIGINT)
try:
await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
except TimeoutError:
# If SIGINT didn't work, try SIGTERM
process.terminate()
try:
await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
except TimeoutError:
# Last resort: SIGKILL
process.kill()
await process.wait()
@asynccontextmanager
async def run_binary_and_wait_for_port(
binary_path: Path,
host: str,
port: int,
timeout: float = PORT_WAIT_TIMEOUT,
line_callback: Callable[[str], None] | None = None,
) -> AsyncGenerator[None]:
"""Run a binary, wait for it to open a port, and clean up on exit."""
async with run_binary(binary_path, line_callback=line_callback) as (
process,
stdout_lines,
):
loop = asyncio.get_running_loop()
start_time = loop.time()
while loop.time() - start_time < timeout:
try:
# Try to connect to the port
@@ -593,41 +625,6 @@ async def run_binary_and_wait_for_port(
raise TimeoutError(error_msg)
finally:
# Cancel output collection tasks
for task in output_tasks:
task.cancel()
# Wait for tasks to complete and check for exceptions
results = await asyncio.gather(*output_tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception) and not isinstance(
result, asyncio.CancelledError
):
print(
f"Error reading from PTY: {result}",
file=sys.stderr,
)
# Close the PTY transport (Unix only)
if controller_transport is not None:
controller_transport.close()
# Cleanup: terminate the process gracefully
if process.returncode is None:
# Send SIGINT (Ctrl+C) for graceful shutdown
process.send_signal(signal.SIGINT)
try:
await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
except TimeoutError:
# If SIGINT didn't work, try SIGTERM
process.terminate()
try:
await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
except TimeoutError:
# Last resort: SIGKILL
process.kill()
await process.wait()
@asynccontextmanager
async def run_compiled_context(
@@ -0,0 +1,40 @@
esphome:
name: test-autorepeat-filter
host:
api:
batch_delay: 0ms # Disable batching to receive every state transition
logger:
level: DEBUG
binary_sensor:
# The autorepeat filter is applied directly to the template sensor, so each
# write through `binary_sensor.template.publish` runs through the filter
# chain. With the source true the filter must oscillate after `delay`; once
# the source returns to false the filter must cancel both timers and emit a
# final false.
- platform: template
name: "Autorepeat Sensor"
id: autorepeat_sensor
filters:
- autorepeat:
- delay: 200ms
time_off: 100ms
time_on: 100ms
button:
- platform: template
name: "Press"
id: press_button
on_press:
- binary_sensor.template.publish:
id: autorepeat_sensor
state: true
- platform: template
name: "Release"
id: release_button
on_press:
- binary_sensor.template.publish:
id: autorepeat_sensor
state: false
@@ -1,5 +1,5 @@
esphome:
name: host-climate-test
name: host-climate-basic-state
host:
api:
logger:
@@ -10,6 +10,7 @@ climate:
name: Dual-mode Thermostat
sensor: host_thermostat_temperature_sensor
humidity_sensor: host_thermostat_humidity_sensor
on_boot_restore_from: default_preset
humidity_hysteresis: 1.0
min_cooling_off_time: 20s
min_cooling_run_time: 20s
@@ -28,10 +29,6 @@ climate:
min_temperature: 15.0
max_temperature: 32.0
temperature_step: 0.1
# Don't restore previous state from flash — this fixture shares the
# `host-climate-test` build dir with host_mode_climate_control.yaml, so a
# prior run of that test could leave the thermostat in HEAT/COOL.
on_boot_restore_from: default_preset
default_preset: home
preset:
- name: "away"
@@ -0,0 +1,25 @@
esphome:
name: safe-mode-loop-runs
host:
logger:
safe_mode:
num_attempts: 10
on_safe_mode:
- lambda: |-
// Spawn a detached thread that logs a unique marker. The
// non-main-thread log goes through the task log buffer, which
// is only drained by Logger::loop(). If looping components
// weren't initialized (the bug fixed in #16269), the buffer is
// never read and the marker never reaches the console.
struct MarkerThread {
static void *thread_func(void *) {
ESP_LOGI("safe_mode_test", "looping component ran in safe mode");
return nullptr;
}
};
pthread_t t;
pthread_create(&t, nullptr, MarkerThread::thread_func, nullptr);
pthread_detach(t);
+39
View File
@@ -0,0 +1,39 @@
"""Helpers for manipulating the host platform's preferences file.
ESPHome's host platform stores preferences in
``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
mirrors ``HostPreferences::sync()``:
``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
Tests use these helpers to pre-populate state the binary will see at
boot (e.g. forcing safe mode) or to clear stale state between runs.
"""
from __future__ import annotations
from pathlib import Path
import struct
def host_prefs_path(device_name: str) -> Path:
"""Return the on-disk prefs file path for a host-platform device."""
return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
def clear_host_prefs(device_name: str) -> None:
"""Delete the prefs file for a host-platform device, if it exists."""
host_prefs_path(device_name).unlink(missing_ok=True)
def write_host_pref(device_name: str, key: int, data: bytes) -> Path:
"""Write a single preference entry, replacing the file's contents.
Returns the path that was written.
"""
if len(data) > 255:
raise ValueError(f"Preference data too long: {len(data)} bytes (max 255)")
path = host_prefs_path(device_name)
path.parent.mkdir(parents=True, exist_ok=True)
payload = struct.pack("<IB", key, len(data)) + data
path.write_bytes(payload)
return path
+37
View File
@@ -8,6 +8,7 @@ import logging
from typing import TypeVar
from aioesphomeapi import (
APIClient,
BinarySensorState,
ButtonInfo,
EntityInfo,
@@ -19,6 +20,42 @@ from aioesphomeapi import (
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T", bound=EntityInfo)
S = TypeVar("S", bound=EntityState)
async def wait_for_state(
client: APIClient,
predicate: Callable[[EntityState], bool],
timeout: float = 5.0,
) -> EntityState:
"""Subscribe to states and wait for one matching ``predicate``.
Resolves with the first :class:`EntityState` for which ``predicate``
returns ``True``. Useful when a component publishes multiple states
during setup (e.g. before sensor readings arrive) and the test needs
to wait for the state to converge to expected values rather than
capturing whichever state happens to arrive first.
Args:
client: Connected API client.
predicate: Callable invoked for every received state; the first
state for which it returns ``True`` is returned.
timeout: Maximum time to wait in seconds.
Returns:
The first state matching ``predicate``.
Raises:
asyncio.TimeoutError: If no matching state arrives within ``timeout``.
"""
future: asyncio.Future[EntityState] = asyncio.get_running_loop().create_future()
def on_state(state: EntityState) -> None:
if not future.done() and predicate(state):
future.set_result(state)
client.subscribe_states(on_state)
return await asyncio.wait_for(future, timeout=timeout)
def find_entity(
@@ -0,0 +1,123 @@
"""Integration test for the binary_sensor autorepeat filter.
Verifies that the autorepeat filter:
1. Passes the initial true through unchanged.
2. Begins oscillating after the configured ``delay`` while the source stays true.
3. Stops oscillating and emits a final false when the source goes false.
This exercises both scheduled timers in ``AutorepeatFilter`` (the per-step
``delay`` timer keyed off the filter ``this`` pointer and the on/off toggle
timer keyed off ``&active_timing_``).
"""
from __future__ import annotations
import asyncio
import pytest
from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_binary_sensor_autorepeat_filter(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Drive the source true and verify the downstream sensor oscillates."""
collector = SensorStateCollector(
sensor_names=[],
binary_sensor_names=["autorepeat_sensor"],
)
async with (
run_compiled(yaml_config),
api_client_connected() as client,
):
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "test-autorepeat-filter"
entities, _ = await client.list_entities_services()
collector.build_key_mapping(entities)
press_button = require_entity(entities, "press", description="Press button")
release_button = require_entity(
entities, "release", description="Release button"
)
initial_state_helper = InitialStateHelper(entities)
client.subscribe_states(
initial_state_helper.on_state_wrapper(collector.on_state)
)
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
autorepeat_states = collector.binary_states["autorepeat_sensor"]
# Press: source becomes true, autorepeat passes the initial true through
# and then oscillates after the configured delay.
# Configured timings: delay=200ms, time_on=100ms, time_off=100ms.
# Expected within ~700ms:
# true (0ms), false (200ms), true (300ms), false (400ms),
# true (500ms), false (600ms)
client.button_command(press_button.key)
# Wait for at least 5 transitions to verify the oscillation pattern.
oscillation_seen = collector.add_waiter(lambda: len(autorepeat_states) >= 5)
try:
await asyncio.wait_for(oscillation_seen, timeout=2.0)
except TimeoutError:
pytest.fail(
f"Expected at least 5 autorepeat transitions, got {autorepeat_states}"
)
assert autorepeat_states[0] is True, (
f"First transition should be the pass-through true, got {autorepeat_states}"
)
# After the initial true and the configured delay, the filter must
# toggle false/true/false/... — verify the alternation pattern.
for index, value in enumerate(autorepeat_states):
expected = index % 2 == 0
assert value is expected, (
f"Expected alternating values starting with True, "
f"got {autorepeat_states} (mismatch at index {index})"
)
# Release: source becomes false, autorepeat must cancel both timers
# and settle on false. If the most recent oscillation was already
# false, the binary sensor will dedup and not emit a new state event;
# if it was true, exactly one final false transition arrives. Either
# way, the steady state must be false and no further toggles should
# arrive after a settle window longer than time_on + time_off.
was_true_before_release = autorepeat_states[-1] is True
before_count = len(autorepeat_states)
client.button_command(release_button.key)
if was_true_before_release:
settle_seen = collector.add_waiter(
lambda: len(autorepeat_states) > before_count
)
try:
await asyncio.wait_for(settle_seen, timeout=2.0)
except TimeoutError:
pytest.fail("Timeout waiting for autorepeat to settle to false")
assert autorepeat_states[-1] is False, (
f"After release, final state should be False, got {autorepeat_states}"
)
steady_count = len(autorepeat_states)
await asyncio.sleep(0.5)
assert len(autorepeat_states) == steady_count, (
f"Expected no further toggles after release, "
f"got {autorepeat_states[steady_count:]}"
)
assert autorepeat_states[-1] is False, (
f"Final autorepeat state should be False, got {autorepeat_states}"
)
@@ -2,11 +2,17 @@
from __future__ import annotations
import aioesphomeapi
from aioesphomeapi import ClimateAction, ClimateInfo, ClimateMode, ClimatePreset
from aioesphomeapi import (
ClimateAction,
ClimateInfo,
ClimateMode,
ClimatePreset,
ClimateState,
EntityState,
)
import pytest
from .state_utils import InitialStateHelper
from .state_utils import wait_for_state
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -18,32 +24,30 @@ async def test_host_mode_climate_basic_state(
) -> None:
"""Test basic climate state reporting."""
async with run_compiled(yaml_config), api_client_connected() as client:
# Get entities and set up state synchronization
entities, services = await client.list_entities_services()
initial_state_helper = InitialStateHelper(entities)
entities, _ = await client.list_entities_services()
climate_infos = [e for e in entities if isinstance(e, ClimateInfo)]
assert len(climate_infos) >= 1, "Expected at least 1 climate entity"
# Subscribe with the wrapper (no-op callback since we just want initial states)
client.subscribe_states(initial_state_helper.on_state_wrapper(lambda _: None))
# Wait for all initial states to be broadcast
try:
await initial_state_helper.wait_for_initial_states()
except TimeoutError:
pytest.fail("Timeout waiting for initial states")
# Get the climate entity and its initial state
test_climate = climate_infos[0]
climate_state = initial_state_helper.initial_states.get(test_climate.key)
assert climate_state is not None, "Climate initial state not found"
assert isinstance(climate_state, aioesphomeapi.ClimateState)
assert climate_state.mode == ClimateMode.OFF
assert climate_state.action == ClimateAction.OFF
assert climate_state.current_temperature == 22.0
assert climate_state.target_temperature_low == 18.0
assert climate_state.target_temperature_high == 24.0
assert climate_state.preset == ClimatePreset.HOME
assert climate_state.current_humidity == 42.0
assert climate_state.target_humidity == 20.0
# The thermostat publishes multiple states during setup as the
# temperature/humidity sensors come online. Wait for the state to
# converge to the expected default values rather than relying on
# whichever state happens to arrive first.
def is_default_state(state: EntityState) -> bool:
return (
isinstance(state, ClimateState)
and state.key == test_climate.key
and state.mode == ClimateMode.OFF
and state.action == ClimateAction.OFF
and state.current_temperature == 22.0
and state.target_temperature_low == 18.0
and state.target_temperature_high == 24.0
and state.preset == ClimatePreset.HOME
and state.current_humidity == 42.0
and state.target_humidity == 20.0
)
try:
await wait_for_state(client, is_default_state)
except TimeoutError:
pytest.fail("Climate did not converge to expected default state")
@@ -0,0 +1,94 @@
"""Regression test for safe_mode + looping_components init ordering.
Reproduces the bug fixed in https://github.com/esphome/esphome/pull/16269:
``App.looping_components_.init(...)`` was emitted at ``CoroPriority.FINAL``,
which placed it *after* the ``safe_mode`` early-return in ``setup_app()``.
When safe mode was entered, the ``FixedVector`` backing the looping-component
list was never sized, ``looping_components_active_end_`` stayed at 0, and
``loop()`` iterated zero components -- so any looping component above
``CoroPriority.APPLICATION`` (e.g. wifi, logger) never ran.
The test forces safe mode by writing ``ENTER_SAFE_MODE_MAGIC`` to the host
preferences file before booting, then asserts that ``Logger::loop()`` runs
by logging from a non-main thread. Non-main-thread logs are buffered in
``TaskLogBuffer`` and only emitted to the console when ``Logger::loop()``
drains the buffer. Without the fix, the marker stays in the buffer
forever; with the fix, it reaches the console.
The API server (``CoroPriority.WEB``, 40) is registered below safe_mode
(``CoroPriority.APPLICATION``, 50), so it's never set up when safe mode
is active and ``run_compiled`` would hang waiting for the API port.
This test uses ``run_binary`` directly to skip the port wait.
"""
from __future__ import annotations
import asyncio
import re
import struct
import pytest
from .conftest import run_binary
from .host_prefs import clear_host_prefs, write_host_pref
from .types import CompileFunction, ConfigWriter
# Must match esphome::safe_mode::RTC_KEY in safe_mode.h
SAFE_MODE_RTC_KEY = 233825507
# Must match esphome::safe_mode::SafeModeComponent::ENTER_SAFE_MODE_MAGIC
ENTER_SAFE_MODE_MAGIC = 0x5AFE5AFE
DEVICE_NAME = "safe-mode-loop-runs"
THREAD_LOG_MARKER = "looping component ran in safe mode"
@pytest.mark.asyncio
async def test_safe_mode_loop_runs(
yaml_config: str,
write_yaml_config: ConfigWriter,
compile_esphome: CompileFunction,
) -> None:
"""When safe mode is active, ``App.loop()`` must still iterate looping
components -- proven here by a thread-logged marker reaching the
console (which requires ``Logger::loop()`` to run)."""
config_path = await write_yaml_config(yaml_config)
binary_path = await compile_esphome(config_path)
# Compile finished successfully; pre-populate prefs so the *next* run
# enters safe mode immediately.
write_host_pref(
DEVICE_NAME, SAFE_MODE_RTC_KEY, struct.pack("<I", ENTER_SAFE_MODE_MAGIC)
)
try:
loop = asyncio.get_running_loop()
safe_mode_active = loop.create_future()
thread_log_seen = loop.create_future()
safe_mode_pattern = re.compile(r"SAFE MODE IS ACTIVE")
thread_log_pattern = re.compile(re.escape(THREAD_LOG_MARKER))
def on_log(line: str) -> None:
if not safe_mode_active.done() and safe_mode_pattern.search(line):
safe_mode_active.set_result(True)
if not thread_log_seen.done() and thread_log_pattern.search(line):
thread_log_seen.set_result(True)
async with run_binary(binary_path, line_callback=on_log):
try:
await asyncio.wait_for(safe_mode_active, timeout=15.0)
except TimeoutError:
pytest.fail(
"Did not observe 'SAFE MODE IS ACTIVE' -- safe mode "
"didn't trigger, so this test isn't exercising the bug."
)
try:
await asyncio.wait_for(thread_log_seen, timeout=10.0)
except TimeoutError:
pytest.fail(
f"Did not observe thread-logged marker {THREAD_LOG_MARKER!r} "
"within timeout. Logger::loop() never drained the task "
"log buffer, meaning App.looping_components_ was never "
"sized -- this is the regression #16269 fixed."
)
finally:
clear_host_prefs(DEVICE_NAME)
+6 -9
View File
@@ -9,7 +9,6 @@ Tests that:
from __future__ import annotations
import asyncio
from pathlib import Path
import socket
from typing import Any
@@ -17,9 +16,12 @@ from aioesphomeapi import TextInfo, TextState
import pytest
from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
from .host_prefs import clear_host_prefs
from .state_utils import InitialStateHelper, require_entity
from .types import CompileFunction, ConfigWriter
DEVICE_NAME = "host-template-text-save-test"
@pytest.mark.asyncio
async def test_template_text_save(
@@ -32,11 +34,7 @@ async def test_template_text_save(
port, port_socket = reserved_tcp_port
# Clean up any stale preference file from previous runs
prefs_file = (
Path.home() / ".esphome" / "prefs" / "host-template-text-save-test.prefs"
)
if prefs_file.exists():
prefs_file.unlink()
clear_host_prefs(DEVICE_NAME)
# Write and compile once
config_path = await write_yaml_config(yaml_config)
@@ -59,7 +57,7 @@ async def test_template_text_save(
wait_and_connect_api_client(port=port) as client,
):
device_info = await client.device_info()
assert device_info.name == "host-template-text-save-test"
assert device_info.name == DEVICE_NAME
entities, _ = await client.list_entities_services()
text_entity = require_entity(
@@ -127,5 +125,4 @@ async def test_template_text_save(
)
# Clean up preference file
if prefs_file.exists():
prefs_file.unlink()
clear_host_prefs(DEVICE_NAME)
+93
View File
@@ -63,6 +63,13 @@ def mock_should_run_import_time() -> Generator[Mock, None, None]:
yield mock
@pytest.fixture
def mock_should_run_device_builder() -> Generator[Mock, None, None]:
"""Mock should_run_device_builder from determine_jobs."""
with patch.object(determine_jobs, "should_run_device_builder") as mock:
yield mock
@pytest.fixture
def mock_determine_cpp_unit_tests() -> Generator[Mock, None, None]:
"""Mock determine_cpp_unit_tests from helpers."""
@@ -99,6 +106,7 @@ def test_main_all_tests_should_run(
mock_should_run_clang_format: Mock,
mock_should_run_python_linters: Mock,
mock_should_run_import_time: Mock,
mock_should_run_device_builder: Mock,
mock_changed_files: Mock,
mock_determine_cpp_unit_tests: Mock,
capsys: pytest.CaptureFixture[str],
@@ -113,6 +121,7 @@ def test_main_all_tests_should_run(
mock_should_run_clang_format.return_value = True
mock_should_run_python_linters.return_value = True
mock_should_run_import_time.return_value = True
mock_should_run_device_builder.return_value = True
mock_determine_cpp_unit_tests.return_value = (False, ["wifi", "api", "sensor"])
# Mock changed_files to return non-component files (to avoid memory impact)
@@ -193,6 +202,7 @@ def test_main_all_tests_should_run(
assert output["clang_format"] is True
assert output["python_linters"] is True
assert output["import_time"] is True
assert output["device_builder"] is True
assert output["changed_components"] == ["wifi", "api", "sensor"]
# changed_components_with_tests will only include components that actually have test files
assert "changed_components_with_tests" in output
@@ -225,6 +235,7 @@ def test_main_no_tests_should_run(
mock_should_run_clang_format: Mock,
mock_should_run_python_linters: Mock,
mock_should_run_import_time: Mock,
mock_should_run_device_builder: Mock,
mock_changed_files: Mock,
mock_determine_cpp_unit_tests: Mock,
capsys: pytest.CaptureFixture[str],
@@ -239,6 +250,7 @@ def test_main_no_tests_should_run(
mock_should_run_clang_format.return_value = False
mock_should_run_python_linters.return_value = False
mock_should_run_import_time.return_value = False
mock_should_run_device_builder.return_value = False
mock_determine_cpp_unit_tests.return_value = (False, [])
# Mock changed_files to return no component files
@@ -278,6 +290,7 @@ def test_main_no_tests_should_run(
assert output["clang_format"] is False
assert output["python_linters"] is False
assert output["import_time"] is False
assert output["device_builder"] is False
assert output["changed_components"] == []
assert output["changed_components_with_tests"] == []
assert output["component_test_count"] == 0
@@ -299,6 +312,7 @@ def test_main_with_branch_argument(
mock_should_run_clang_format: Mock,
mock_should_run_python_linters: Mock,
mock_should_run_import_time: Mock,
mock_should_run_device_builder: Mock,
mock_changed_files: Mock,
mock_determine_cpp_unit_tests: Mock,
capsys: pytest.CaptureFixture[str],
@@ -313,6 +327,7 @@ def test_main_with_branch_argument(
mock_should_run_clang_format.return_value = False
mock_should_run_python_linters.return_value = True
mock_should_run_import_time.return_value = True
mock_should_run_device_builder.return_value = True
mock_determine_cpp_unit_tests.return_value = (False, ["mqtt"])
# Mock changed_files to return non-component files (to avoid memory impact)
@@ -350,6 +365,7 @@ def test_main_with_branch_argument(
mock_should_run_clang_format.assert_called_once_with("main")
mock_should_run_python_linters.assert_called_once_with("main")
mock_should_run_import_time.assert_called_once_with("main")
mock_should_run_device_builder.assert_called_once_with("main")
# Check output
captured = capsys.readouterr()
@@ -362,6 +378,7 @@ def test_main_with_branch_argument(
assert output["clang_format"] is False
assert output["python_linters"] is True
assert output["import_time"] is True
assert output["device_builder"] is True
assert output["changed_components"] == ["mqtt"]
# changed_components_with_tests will only include components that actually have test files
assert "changed_components_with_tests" in output
@@ -734,6 +751,82 @@ def test_should_run_import_time_with_branch() -> None:
mock_changed.assert_called_once_with("release")
@pytest.mark.parametrize(
("changed_files", "expected_result"),
[
# esphome Python files trigger downstream device-builder tests
(["esphome/__main__.py"], True),
(["esphome/components/wifi/__init__.py"], True),
(["esphome/core/config.py"], True),
(["esphome/types.pyi"], True),
# Runtime dependency changes trigger
(["requirements.txt"], True),
(["pyproject.toml"], True),
# Non-C++ files packaged with esphome trigger -- device-builder
# picks them up because esphome's pyproject sets
# include-package-data = true.
(["esphome/idf_component.yml"], True),
(["esphome/dashboard/templates/index.html"], True),
(["esphome/components/api/api_pb2_service.json"], True),
# Mixed: any triggering file is enough
(["docs/README.md", "esphome/config.py"], True),
# Dev/test-only dependency changes don't trigger device-builder
# (they don't affect the importable surface device-builder uses)
(["requirements_dev.txt"], False),
(["requirements_test.txt"], False),
# Files outside esphome/ don't trigger
(["script/some_other_script.py"], False),
(["tests/script/test_determine_jobs.py"], False),
# C++ files under esphome/ don't trigger -- they only affect
# compiled firmware, not the Python install device-builder pulls in.
(["esphome/core/component.cpp"], False),
(["esphome/core/component.h"], False),
(["esphome/components/wifi/wifi_component.cpp"], False),
# Files outside esphome/ entirely
(["tests/components/wifi/test.esp32-idf.yaml"], False),
(["README.md"], False),
([], False),
],
)
def test_should_run_device_builder(
changed_files: list[str], expected_result: bool
) -> None:
"""Test should_run_device_builder function (non-beta/release target)."""
with (
patch.object(determine_jobs, "changed_files", return_value=changed_files),
# Mock target branch to "dev" so the beta/release skip is bypassed
# for these per-file behavior checks.
patch.object(determine_jobs, "get_target_branch", return_value="dev"),
):
result = determine_jobs.should_run_device_builder()
assert result == expected_result
def test_should_run_device_builder_with_branch() -> None:
"""Test should_run_device_builder with branch argument."""
with (
patch.object(determine_jobs, "changed_files") as mock_changed,
patch.object(determine_jobs, "get_target_branch", return_value="dev"),
):
mock_changed.return_value = []
determine_jobs.should_run_device_builder("release")
mock_changed.assert_called_once_with("release")
@pytest.mark.parametrize("target_branch", ["beta", "release", "release-2026.5"])
def test_should_run_device_builder_skips_beta_release(target_branch: str) -> None:
"""Beta/release target branches skip device-builder (lag behind device-builder@main)."""
with (
patch.object(determine_jobs, "get_target_branch", return_value=target_branch),
patch.object(determine_jobs, "changed_files") as mock_changed,
):
# Even with a triggering file present, the target-branch guard wins.
mock_changed.return_value = ["esphome/__main__.py"]
assert determine_jobs.should_run_device_builder() is False
# changed_files shouldn't even be consulted -- the guard short-circuits.
mock_changed.assert_not_called()
@pytest.mark.parametrize(
("changed_files", "expected_result"),
[
+12 -11
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from unittest.mock import patch
from esphome.components import esp32
from esphome.components.api import client as api_client
from esphome.core import EsphomeError
@@ -18,11 +19,11 @@ def test_decoder_swallows_esphome_error() -> None:
reconnect.
"""
config = {"esphome": {"name": "test"}}
processor = api_client._LogLineProcessor(config, None)
with patch.object(
api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
) as mock_process:
processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
processor.process_line("PC: 0x4010496e")
assert mock_process.called
@@ -47,9 +48,9 @@ def test_decoder_warning_uses_fallback_for_empty_error(caplog) -> None:
must show a useful explanation rather than empty parens.
"""
config = {"esphome": {"name": "test"}}
processor = api_client._LogLineProcessor(config, None)
with patch.object(api_client, "process_stacktrace", side_effect=EsphomeError()):
with patch.object(esp32, "process_stacktrace", side_effect=EsphomeError()):
processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
processor.process_line("PC: 0x4010496e")
warnings = [r.message for r in caplog.records if r.levelname == "WARNING"]
@@ -65,11 +66,11 @@ def test_decoder_short_circuits_after_failure() -> None:
stall log streaming.
"""
config = {"esphome": {"name": "test"}}
processor = api_client._LogLineProcessor(config, None)
with patch.object(
api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
esp32, "process_stacktrace", side_effect=EsphomeError("no idedata")
) as mock_process:
processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
processor.process_line("PC: 0x4010496e")
processor.process_line("BT0: 0x4010496e")
processor.process_line("BT1: 0x401049aa")
@@ -80,18 +81,18 @@ def test_decoder_short_circuits_after_failure() -> None:
def test_decoder_threads_backtrace_state() -> None:
"""When decoding succeeds, backtrace_state is threaded across calls."""
config = {"esphome": {"name": "test"}}
processor = api_client._LogLineProcessor(config, None)
with patch.object(
api_client, "process_stacktrace", side_effect=[True, False]
esp32, "process_stacktrace", side_effect=[True, False]
) as mock_process:
processor = api_client._LogLineProcessor(config, esp32.process_stacktrace)
processor.process_line(">>>stack>>>")
assert processor.backtrace_state is True
processor.process_line("<<<stack<<<")
assert processor.backtrace_state is False
assert mock_process.call_args_list[0].kwargs == {"backtrace_state": False}
assert mock_process.call_args_list[1].kwargs == {"backtrace_state": True}
assert not mock_process.call_args_list[0].args[-1]
assert mock_process.call_args_list[1].args[-1]
def test_decoder_uses_platform_handler_when_provided() -> None:
@@ -105,7 +106,7 @@ def test_decoder_uses_platform_handler_when_provided() -> None:
processor = api_client._LogLineProcessor(config, platform_handler)
with patch.object(api_client, "process_stacktrace") as mock_generic:
with patch.object(esp32, "process_stacktrace") as mock_generic:
processor.process_line("BT0: 0x4010496e")
assert calls == [(config, "BT0: 0x4010496e", False)]
@@ -0,0 +1,109 @@
"""Tests for ESP32 component."""
from pathlib import Path
from unittest.mock import Mock
def test_process_stacktrace_esp8266_exception(setup_core: Path, caplog) -> None:
"""Test process_stacktrace handles ESP8266 exceptions."""
from esphome.components.esp8266 import process_stacktrace
config = {"name": "test"}
# Test exception type parsing
line = "Exception (28):"
backtrace_state = False
result = process_stacktrace(config, line, backtrace_state)
assert "Access to invalid address: LOAD (wild pointer?)" in caplog.text
assert result is False
def test_process_stacktrace_esp8266_backtrace(
setup_core: Path, mock_esp8266_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP8266 multi-line backtrace."""
from esphome.components.esp8266 import process_stacktrace
config = {"name": "test"}
# Start of backtrace
line1 = ">>>stack>>>"
state = process_stacktrace(config, line1, False)
assert state is True
# Backtrace content with addresses
line2 = "40201234 40205678"
state = process_stacktrace(config, line2, state)
assert state is True
assert mock_esp8266_decode_pc.call_count == 2
# End of backtrace
line3 = "<<<stack<<<"
state = process_stacktrace(config, line3, state)
assert state is False
def test_process_stacktrace_esp32_backtrace(
setup_core: Path, mock_esp32_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP32 single-line backtrace."""
from esphome.components.esp32 import process_stacktrace
config = {"name": "test"}
line = "Backtrace: 0x40081234:0x3ffb1234 0x40085678:0x3ffb5678"
state = process_stacktrace(config, line, False)
# Should decode both addresses
assert mock_esp32_decode_pc.call_count == 2
mock_esp32_decode_pc.assert_any_call(config, "40081234")
mock_esp32_decode_pc.assert_any_call(config, "40085678")
assert state is False
def test_process_stacktrace_bad_alloc(
setup_core: Path, mock_esp32_decode_pc: Mock, caplog
) -> None:
"""Test process_stacktrace handles bad alloc messages."""
from esphome.components.esp32 import process_stacktrace
config = {"name": "test"}
line = "last failed alloc call: 40201234(512)"
state = process_stacktrace(config, line, False)
assert "Memory allocation of 512 bytes failed at 40201234" in caplog.text
mock_esp32_decode_pc.assert_called_once_with(config, "40201234")
assert state is False
def test_process_stacktrace_esp32_crash_handler(
setup_core: Path, mock_esp32_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP32 crash handler backtrace lines."""
from esphome.components.esp32 import process_stacktrace
config = {"name": "test"}
# Simulate crash handler log lines as they appear from the API/serial
line_pc = "[E][esp32.crash:078]: PC: 0x400D1234 (fault location)"
state = process_stacktrace(config, line_pc, False)
# PC line is matched by existing STACKTRACE_ESP32_PC_RE
mock_esp32_decode_pc.assert_called_with(config, "400D1234")
assert state is False
mock_esp32_decode_pc.reset_mock()
line_bt0 = "[E][esp32.crash:080]: BT0: 0x400D5678 (backtrace)"
state = process_stacktrace(config, line_bt0, False)
mock_esp32_decode_pc.assert_called_once_with(config, "400D5678")
assert state is False
mock_esp32_decode_pc.reset_mock()
line_bt1 = "[E][esp32.crash:080]: BT1: 0x42005ABC (backtrace)"
state = process_stacktrace(config, line_bt1, False)
mock_esp32_decode_pc.assert_called_once_with(config, "42005ABC")
assert state is False
+10 -3
View File
@@ -77,9 +77,16 @@ def mock_run_platformio_cli_run() -> Generator[Mock, None, None]:
@pytest.fixture
def mock_decode_pc() -> Generator[Mock, None, None]:
"""Mock _decode_pc for platformio_api."""
with patch("esphome.platformio_api._decode_pc") as mock:
def mock_esp32_decode_pc() -> Generator[Mock, None, None]:
"""Mock _decode_pc for esp32."""
with patch("esphome.components.esp32._decode_pc") as mock:
yield mock
@pytest.fixture
def mock_esp8266_decode_pc() -> Generator[Mock, None, None]:
"""Mock _decode_pc for esp8266."""
with patch("esphome.components.esp8266._decode_pc") as mock:
yield mock
+70
View File
@@ -10,6 +10,7 @@ from unittest.mock import MagicMock, Mock, patch
import pytest
from esphome import config_validation as cv, core
from esphome.components.safe_mode import to_code as safe_mode_to_code
from esphome.const import (
CONF_AREA,
CONF_AREAS,
@@ -312,6 +313,75 @@ def test_add_platform_defines_priority() -> None:
)
def test_to_code_priority_above_safe_mode() -> None:
"""Test that core to_code emits the looping_components_ init before safe_mode.
Regression test for https://github.com/esphome/esphome/issues/16262.
safe_mode emits an `if (should_enter_safe_mode(...)) return;` line in main()
at APPLICATION priority. The `App.looping_components_.init(...)` call must be
emitted at a higher priority than APPLICATION so it lands in main() before
the early return; otherwise the FixedVector is never sized when safe mode is
active and loop() never runs (Wi-Fi never connects).
"""
assert config.to_code.priority > safe_mode_to_code.priority, (
f"core to_code priority ({config.to_code.priority}) must be greater than "
f"safe_mode to_code priority ({safe_mode_to_code.priority}) so that "
"App.looping_components_.init() is emitted before safe_mode's early return"
)
@pytest.mark.asyncio
async def test_add_looping_components_handles_empty_entries() -> None:
"""Test that _add_looping_components emits a valid constexpr when there are
no looping component entries.
With zero entries the generated constexpr must still be syntactically valid
C++ (`= 0;`), not an empty expression (`= ;`). This guards the empty-list
case that would otherwise produce uncompilable main.cpp output.
"""
CORE.data["looping_component_entries"] = []
await config._add_looping_components()
constexpr_lines = [
str(s)
for s in CORE.global_statements
if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
]
assert len(constexpr_lines) == 1
text = constexpr_lines[0]
assert "static constexpr size_t ESPHOME_LOOPING_COMPONENT_COUNT" in text
# The right-hand side must contain a literal `0`, not be empty.
rhs = text.split("=", 1)[1]
assert "0" in rhs
assert rhs.strip().rstrip(";").strip(), (
f"constexpr right-hand side must not be empty, got: {text!r}"
)
@pytest.mark.asyncio
async def test_add_looping_components_with_entries() -> None:
"""Test that _add_looping_components builds a HasLoopOverride sum from entries."""
CORE.data["looping_component_entries"] = [
"esphome::wifi::WiFiComponent",
"esphome::logger::Logger",
"esphome::wifi::WiFiComponent",
]
await config._add_looping_components()
constexpr_lines = [
str(s)
for s in CORE.global_statements
if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
]
assert len(constexpr_lines) == 1
text = constexpr_lines[0]
# Deduplicated by type, with per-type counts as multiplier.
assert "(2 * HasLoopOverride<esphome::wifi::WiFiComponent>::value)" in text
assert "(1 * HasLoopOverride<esphome::logger::Logger>::value)" in text
def test_valid_include_with_angle_brackets() -> None:
"""Test valid_include accepts angle bracket includes."""
assert valid_include("<ArduinoJson.h>") == "<ArduinoJson.h>"
@@ -0,0 +1,16 @@
substitutions:
device:
manufacturer: espressif
model: esp32
mac_suffix: ffffff
name: espressif-esp32-ffffff
network:
host: example.com
port: 8080
url: http://example.com:8080/api
esphome:
name: espressif-esp32-ffffff
test_list:
- espressif-esp32-ffffff
- http://example.com:8080/api
- espressif/esp32
@@ -0,0 +1,18 @@
substitutions:
device:
manufacturer: "espressif"
model: "esp32"
mac_suffix: "ffffff"
name: ${device.manufacturer}-${device.model}-${device.mac_suffix}
network:
host: "example.com"
port: 8080
url: "http://${network.host}:${network.port}/api"
esphome:
name: ${device.name}
test_list:
- ${device.name}
- ${network.url}
- "${device.manufacturer}/${device.model}"
@@ -0,0 +1,9 @@
binary_sensor:
- platform: template
id: front_door_enrolling
name: Front Door Enrolling
substitutions:
enrolling_id: front_door_enrolling
enrolling_name: Front Door Enrolling
esphome:
name: test
@@ -0,0 +1,9 @@
esphome:
name: test
packages:
fingerprint: !include
file: 18-package_vars_in_subs_inc.yaml
vars:
sensor_name: "Front Door"
sensor_id_prefix: "front_door"
@@ -0,0 +1,8 @@
substitutions:
enrolling_id: ${sensor_id_prefix}_enrolling
enrolling_name: ${sensor_name} Enrolling
binary_sensor:
- platform: template
id: ${enrolling_id}
name: ${enrolling_name}
+46
View File
@@ -170,6 +170,23 @@ def test_find_used_secret_keys_deduplicates(tmp_path: Path) -> None:
assert keys == {"key1"}
def test_find_used_secret_keys_quoted(tmp_path: Path) -> None:
"""Quoted !secret keys should resolve to the same key as unquoted form.
YAML strips surrounding quotes during parsing, so the secrets.yaml
lookup uses the unquoted key. The bundle scan must do the same.
"""
yaml1 = tmp_path / "a.yaml"
yaml1.write_text(
"single: !secret 'wifi_ssid'\n"
'double: !secret "wifi_pw"\n'
"bare: !secret api_key\n"
)
keys = _find_used_secret_keys([yaml1])
assert keys == {"wifi_ssid", "wifi_pw", "api_key"}
# ---------------------------------------------------------------------------
# _add_bytes_to_tar
# ---------------------------------------------------------------------------
@@ -1217,6 +1234,35 @@ def test_create_bundle_filters_secrets(tmp_path: Path) -> None:
assert "should_not_appear" not in secrets_data
def test_create_bundle_filters_secrets_quoted(tmp_path: Path) -> None:
"""Bundling must include secrets.yaml when !secret keys are quoted.
Regression test for issue 16259: quoted !secret references previously
captured the quotes as part of the key, so no key matched secrets.yaml
entries and the secrets file was dropped from the bundle entirely.
"""
config_dir = _setup_config_dir(tmp_path)
secrets = config_dir / "secrets.yaml"
secrets.write_text("ota_password: hunter2\nunused: should_not_appear\n")
config_yaml = "ota:\n password: !secret 'ota_password'\n"
(config_dir / "test.yaml").write_text(config_yaml)
creator = ConfigBundleCreator({})
result = creator.create_bundle()
assert result.manifest[ManifestKey.HAS_SECRETS] is True
buf = io.BytesIO(result.data)
with tarfile.open(fileobj=buf, mode="r:gz") as tar:
secrets_data = tar.extractfile("secrets.yaml").read().decode()
assert "ota_password" in secrets_data
assert "hunter2" in secrets_data
assert "unused" not in secrets_data
def test_create_bundle_no_secrets(tmp_path: Path) -> None:
_setup_config_dir(tmp_path)
+203
View File
@@ -0,0 +1,203 @@
"""Unit tests for ``esphome.components.dashboard_import.import_config``.
Locks the YAML shape that ``import_config`` materialises on disk for
adopted factory firmware. Both the legacy dashboard and the new
device-builder backend (esphome/device-builder) call this function
during the adoption flow and depend on the output's ``esphome.name``
/ ``packages:`` keys to route subsequent compile + flash operations.
"""
from __future__ import annotations
from pathlib import Path
import pytest
import yaml as pyyaml
from esphome.components.dashboard_import import import_config
def _load_plain_yaml(path: Path) -> dict:
"""Load YAML without invoking ESPHome's ``CORE``-aware loader.
``esphome.yaml_util.load_yaml`` resolves ``!include`` /
``!secret`` against ``CORE.config_path`` which isn't set in
these tests. We're only asserting on plain key/value structure,
so ``pyyaml.load`` with a custom loader subclassing
``pyyaml.SafeLoader`` (and empty fallbacks for the secret/include
tags) is enough.
"""
class _Loader(pyyaml.SafeLoader):
pass
_Loader.add_constructor("!secret", lambda loader, node: f"!secret {node.value}")
_Loader.add_constructor("!include", lambda loader, node: f"!include {node.value}")
return pyyaml.load(path.read_text(encoding="utf-8"), Loader=_Loader)
def test_basic_import_writes_expected_yaml_shape(tmp_path: Path) -> None:
"""A minimal Wi-Fi import emits the substitutions / packages / esphome triad.
These three top-level blocks are the contract: substitutions
holds the device-specific name, packages pulls in the upstream
firmware via the import URL, and esphome.name interpolates from
substitutions. Anything that depends on this output (frontend
config viewer, follow-up edits, version checks) reads those
keys directly.
"""
yaml_path = tmp_path / "kitchen.yaml"
import_config(
path=str(yaml_path),
name="kitchen",
friendly_name="Kitchen",
project_name="acme.kitchen-light",
import_url="github://acme/firmware/kitchen.yaml@main",
)
assert yaml_path.exists()
config = _load_plain_yaml(yaml_path)
assert config["substitutions"] == {
"name": "kitchen",
"friendly_name": "Kitchen",
}
assert config["packages"] == {
"acme.kitchen-light": "github://acme/firmware/kitchen.yaml@main"
}
assert config["esphome"] == {
"name": "${name}",
"name_add_mac_suffix": False,
"friendly_name": "${friendly_name}",
}
def test_import_appends_wifi_config_when_network_is_wifi(tmp_path: Path) -> None:
"""Wi-Fi devices get a ``wifi:`` block templated with secrets references.
Adopted Wi-Fi devices need a ``wifi:`` section so they can
actually connect on the user's LAN — the boilerplate references
``!secret wifi_ssid`` / ``!secret wifi_password`` so the
user's existing secrets file plugs in. Devices on other
networks (Ethernet) shouldn't get the Wi-Fi block.
"""
yaml_path = tmp_path / "kitchen.yaml"
import_config(
path=str(yaml_path),
name="kitchen",
friendly_name=None,
project_name="acme.kitchen-light",
import_url="github://acme/firmware/kitchen.yaml@main",
)
contents = yaml_path.read_text()
assert "wifi:" in contents
assert "!secret wifi_ssid" in contents
assert "!secret wifi_password" in contents
def test_import_omits_wifi_block_for_ethernet_network(tmp_path: Path) -> None:
"""Ethernet devices get no ``wifi:`` block — caller wires Ethernet separately.
The ``network`` parameter exists specifically so non-Wi-Fi
devices (PoE / Ethernet, etc.) skip the Wi-Fi templating —
otherwise their generated YAML would carry an unused ``wifi:``
section the user has to clean up by hand.
"""
yaml_path = tmp_path / "olimex-poe.yaml"
import_config(
path=str(yaml_path),
name="olimex-poe",
friendly_name=None,
project_name="acme.poe-monitor",
import_url="github://acme/firmware/poe.yaml@main",
network="ethernet",
)
contents = yaml_path.read_text()
assert "wifi:" not in contents
def test_import_with_encryption_writes_api_key(tmp_path: Path) -> None:
"""``encryption=True`` generates a fresh Noise PSK in the api block.
Used during the adoption flow when the device-builder UI
explicitly opts the new device into encrypted API. Each
invocation must produce a fresh 32-byte PSK base64-encoded into
the YAML; subsequent compiles and the dashboard's encryption
indicator both read it from there.
"""
yaml_path_1 = tmp_path / "a.yaml"
yaml_path_2 = tmp_path / "b.yaml"
import_config(
path=str(yaml_path_1),
name="a",
friendly_name=None,
project_name="acme.dev",
import_url="github://acme/firmware/dev.yaml@main",
encryption=True,
)
import_config(
path=str(yaml_path_2),
name="b",
friendly_name=None,
project_name="acme.dev",
import_url="github://acme/firmware/dev.yaml@main",
encryption=True,
)
config_1 = _load_plain_yaml(yaml_path_1)
config_2 = _load_plain_yaml(yaml_path_2)
assert "api" in config_1 and "encryption" in config_1["api"]
key_1 = config_1["api"]["encryption"]["key"]
key_2 = config_2["api"]["encryption"]["key"]
# Fresh per-call PSK, not a hardcoded value.
assert key_1 != key_2
# Base64-encoded 32 bytes → length 44 with one trailing `=`.
assert len(key_1) == 44
def test_import_without_friendly_name_omits_friendly_substitution(
tmp_path: Path,
) -> None:
"""``friendly_name=None`` skips the friendly_name substitution.
Some imported configs don't carry a friendly name. The output
shouldn't pretend they do — the substitutions block must omit
``friendly_name`` so the dashboard renders blank rather than
the literal substitution token.
"""
yaml_path = tmp_path / "noname.yaml"
import_config(
path=str(yaml_path),
name="noname",
friendly_name=None,
project_name="acme.dev",
import_url="github://acme/firmware/dev.yaml@main",
)
config = _load_plain_yaml(yaml_path)
assert config["substitutions"] == {"name": "noname"}
assert "friendly_name" not in config["esphome"]
def test_import_refuses_to_overwrite_existing_yaml(tmp_path: Path) -> None:
"""An already-present file raises rather than clobbering the user's edits.
Both the legacy dashboard and device-builder rely on the
``FileExistsError`` to surface a "config already exists" message
instead of silently destroying user data.
"""
yaml_path = tmp_path / "existing.yaml"
yaml_path.write_text("# user's hand-edited config\n", encoding="utf-8")
with pytest.raises(FileExistsError):
import_config(
path=str(yaml_path),
name="existing",
friendly_name=None,
project_name="acme.dev",
import_url="github://acme/firmware/dev.yaml@main",
)
# Original content survives unchanged.
assert yaml_path.read_text() == "# user's hand-edited config\n"
+45
View File
@@ -90,6 +90,51 @@ def test_cpp_string_escape(string, expected):
assert actual == expected
@pytest.mark.parametrize(
"value, expected",
(
# Basic underscore→dash conversion.
("Living Room Sensor", "living-room-sensor"),
# Already-slugified input passes through with dash output.
("kitchen_light", "kitchen-light"),
# Accents are stripped (matches the underlying ``slugify``).
("Café Caché", "cafe-cache"),
# Mixed casing + multiple separators collapse correctly.
("Foo Bar__Baz", "foo-bar-baz"),
# Empty input yields empty output.
("", ""),
# Numbers survive intact.
("Sensor 42", "sensor-42"),
),
)
def test_friendly_name_slugify(value, expected):
"""Friendly-name → URL-safe dash-slug.
Stable mapping is part of the cross-tool contract
(legacy dashboard + device-builder both depend on it for
filename → device-name routing). Lock the cases here so a
refactor can't accidentally change a slug shape and break
on-disk filenames in already-deployed installs.
"""
assert helpers.friendly_name_slugify(value) == expected
def test_friendly_name_slugify_back_compat_shim():
"""``esphome.dashboard.util.text`` keeps re-exporting for back-compat.
The function moved to ``esphome.helpers`` so the new
device-builder dashboard backend can import it without depending
on the legacy dashboard package, but downstream code that still
imports from the old path keeps working until the dashboard
module is removed.
"""
from esphome.dashboard.util.text import (
friendly_name_slugify as legacy_friendly_name_slugify,
)
assert legacy_friendly_name_slugify is helpers.friendly_name_slugify
@pytest.mark.parametrize(
"host",
(
+321 -11
View File
@@ -43,6 +43,7 @@ from esphome.__main__ import (
has_non_ip_address,
has_ota,
has_resolvable_address,
has_web_server_ota,
mqtt_get_ip,
run_esphome,
run_miniterm,
@@ -54,9 +55,11 @@ from esphome.__main__ import (
)
from esphome.address_cache import AddressCache
from esphome.bundle import BUNDLE_EXTENSION, BundleFile, BundleResult
from esphome.components import esp32
from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
from esphome.const import (
CONF_API,
CONF_AUTH,
CONF_BAUD_RATE,
CONF_BROKER,
CONF_DISABLED,
@@ -75,6 +78,8 @@ from esphome.const import (
CONF_SUBSTITUTIONS,
CONF_TOPIC,
CONF_USE_ADDRESS,
CONF_USERNAME,
CONF_WEB_SERVER,
CONF_WIFI,
KEY_CORE,
KEY_TARGET_PLATFORM,
@@ -85,7 +90,7 @@ from esphome.const import (
)
from esphome.core import CORE, EsphomeError
from esphome.espota2 import OTA_TYPE_UPDATE_APP, OTA_TYPE_UPDATE_PARTITION_TABLE
from esphome.util import BootselResult
from esphome.util import BootselResult, FlashImage
from esphome.zeroconf import _await_discovery, discover_mdns_devices
@@ -212,6 +217,13 @@ def mock_run_ota() -> Generator[Mock]:
yield mock
@pytest.fixture
def mock_run_web_server_ota() -> Generator[Mock]:
"""Mock web_server_ota.run_ota for testing."""
with patch("esphome.web_server_ota.run_ota") as mock:
yield mock
@pytest.fixture
def mock_is_ip_address() -> Generator[Mock]:
"""Mock is_ip_address for testing."""
@@ -1113,6 +1125,7 @@ class MockArgs:
reset: bool = False
list_only: bool = False
output: str | None = None
ota_platform: str | None = None
partition_table: bool = False
@@ -1181,8 +1194,8 @@ def test_upload_using_esptool_path_conversion(
mock_idedata = MagicMock(spec=platformio_api.IDEData)
mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
mock_idedata.extra_flash_images = [
platformio_api.FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
platformio_api.FlashImage(path=tmp_path / "partitions.bin", offset="0x8000"),
FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
FlashImage(path=tmp_path / "partitions.bin", offset="0x8000"),
]
mock_get_idedata.return_value = mock_idedata
@@ -1259,8 +1272,8 @@ def test_upload_using_esptool_skips_missing_extra_flash_images(
mock_idedata = MagicMock(spec=platformio_api.IDEData)
mock_idedata.firmware_bin_path = tmp_path / "firmware.bin"
mock_idedata.extra_flash_images = [
platformio_api.FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
platformio_api.FlashImage(path=missing_path, offset="0x2d0000"),
FlashImage(path=tmp_path / "bootloader.bin", offset="0x1000"),
FlashImage(path=missing_path, offset="0x2d0000"),
]
mock_get_idedata.return_value = mock_idedata
@@ -1877,6 +1890,277 @@ def test_upload_program_ota_no_config(
upload_program(config, args, devices)
def test_has_web_server_ota_detects_platform() -> None:
"""has_web_server_ota returns True when web_server OTA platform is configured."""
setup_core(
config={
CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
}
)
assert has_web_server_ota() is True
assert has_ota() is True
def test_has_web_server_ota_returns_false_without_config() -> None:
"""has_web_server_ota returns False when only native OTA is configured."""
setup_core(
config={
CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}],
}
)
assert has_web_server_ota() is False
assert has_ota() is True
def test_upload_program_web_server_only_auto_dispatches(
mock_run_web_server_ota: Mock,
mock_run_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""When only web_server OTA is configured, upload_program picks it automatically."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_web_server_ota.return_value = (0, "192.168.1.100")
config = {
CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
CONF_WEB_SERVER: {
CONF_PORT: 80,
CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
},
}
args = MockArgs()
devices = ["192.168.1.100"]
exit_code, host = upload_program(config, args, devices)
assert exit_code == 0
assert host == "192.168.1.100"
expected_firmware = (
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_web_server_ota.assert_called_once_with(
["192.168.1.100"], 80, "admin", "pw", expected_firmware
)
mock_run_ota.assert_not_called()
def test_upload_program_web_server_no_auth(
mock_run_web_server_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""web_server OTA works without an auth block (passes None for credentials)."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_web_server_ota.return_value = (0, "192.168.1.100")
config = {
CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
CONF_WEB_SERVER: {CONF_PORT: 8080},
}
args = MockArgs()
devices = ["192.168.1.100"]
exit_code, host = upload_program(config, args, devices)
assert exit_code == 0
assert host == "192.168.1.100"
expected_firmware = (
tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
)
mock_run_web_server_ota.assert_called_once_with(
["192.168.1.100"], 8080, None, None, expected_firmware
)
def test_upload_program_both_platforms_default_prefers_native(
mock_run_ota: Mock,
mock_run_web_server_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""When both OTA platforms are configured, default selection is native API."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_ota.return_value = (0, "192.168.1.100")
config = {
CONF_OTA: [
{
CONF_PLATFORM: CONF_ESPHOME,
CONF_PORT: 3232,
CONF_PASSWORD: "secret",
},
{CONF_PLATFORM: CONF_WEB_SERVER},
],
CONF_WEB_SERVER: {CONF_PORT: 80},
}
args = MockArgs()
devices = ["192.168.1.100"]
exit_code, host = upload_program(config, args, devices)
assert exit_code == 0
assert host == "192.168.1.100"
mock_run_ota.assert_called_once()
mock_run_web_server_ota.assert_not_called()
def test_upload_program_ota_platform_override_to_web_server(
mock_run_ota: Mock,
mock_run_web_server_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""--ota-platform web_server forces web_server OTA even when native is configured."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_web_server_ota.return_value = (0, "192.168.1.100")
config = {
CONF_OTA: [
{
CONF_PLATFORM: CONF_ESPHOME,
CONF_PORT: 3232,
CONF_PASSWORD: "secret",
},
{CONF_PLATFORM: CONF_WEB_SERVER},
],
CONF_WEB_SERVER: {CONF_PORT: 80},
}
args = MockArgs(ota_platform=CONF_WEB_SERVER)
devices = ["192.168.1.100"]
exit_code, host = upload_program(config, args, devices)
assert exit_code == 0
assert host == "192.168.1.100"
mock_run_ota.assert_not_called()
mock_run_web_server_ota.assert_called_once()
def test_upload_program_ota_platform_unavailable(
mock_get_port_type: Mock,
) -> None:
"""--ota-platform must reference a platform that is actually configured."""
setup_core(platform=PLATFORM_ESP32)
mock_get_port_type.return_value = "NETWORK"
config = {
CONF_OTA: [
{
CONF_PLATFORM: CONF_ESPHOME,
CONF_PORT: 3232,
CONF_PASSWORD: "secret",
}
],
}
args = MockArgs(ota_platform=CONF_WEB_SERVER)
devices = ["192.168.1.100"]
with pytest.raises(EsphomeError, match="--ota-platform web_server"):
upload_program(config, args, devices)
def test_upload_program_web_server_missing_component(
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""web_server OTA without a web_server component fails with a clear error."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
config = {
CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
# No CONF_WEB_SERVER
}
args = MockArgs()
devices = ["192.168.1.100"]
with pytest.raises(EsphomeError, match="web_server.*not configured"):
upload_program(config, args, devices)
def test_upload_program_unrelated_ota_platform_ignored(
mock_run_ota: Mock,
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""OTA list entries that are neither esphome nor web_server are ignored.
Covers the false branch in _choose_ota_platform's filter loop and the
no-match branch in _upload_via_native_api's lookup loop.
"""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
mock_run_ota.return_value = (0, "192.168.1.100")
config = {
CONF_OTA: [
{CONF_PLATFORM: "http_request"}, # unrelated platform; ignored
{
CONF_PLATFORM: CONF_ESPHOME,
CONF_PORT: 3232,
CONF_PASSWORD: "secret",
},
],
}
args = MockArgs()
devices = ["192.168.1.100"]
exit_code, host = upload_program(config, args, devices)
assert exit_code == 0
assert host == "192.168.1.100"
mock_run_ota.assert_called_once()
def test_upload_program_duplicate_platform_dedup_in_error(
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""Duplicate same-platform OTA entries don't repeat in --ota-platform errors."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
config = {
CONF_OTA: [
{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232},
{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3233},
],
}
args = MockArgs(ota_platform=CONF_WEB_SERVER)
devices = ["192.168.1.100"]
with pytest.raises(EsphomeError) as excinfo:
upload_program(config, args, devices)
# Error mentions esphome once in the platform list, not "esphome, esphome".
msg = str(excinfo.value)
assert "esphome, esphome" not in msg
assert msg.endswith(": esphome")
def test_upload_program_only_unrelated_ota_platforms(
mock_get_port_type: Mock,
tmp_path: Path,
) -> None:
"""Only unrelated OTA platforms configured -> raises like missing OTA."""
setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
mock_get_port_type.return_value = "NETWORK"
config = {
CONF_OTA: [{CONF_PLATFORM: "http_request"}],
}
args = MockArgs()
devices = ["192.168.1.100"]
with pytest.raises(EsphomeError, match="Cannot upload Over the Air"):
upload_program(config, args, devices)
def test_upload_program_ota_with_mqtt_resolution(
mock_mqtt_get_ip: Mock,
mock_is_ip_address: Mock,
@@ -4225,7 +4509,7 @@ def test_run_miniterm_batches_lines_with_same_timestamp(
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(platformio_api, "process_stacktrace") as mock_bt,
patch.object(esp32, "process_stacktrace") as mock_bt,
):
mock_bt.return_value = False
result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4264,7 +4548,7 @@ def test_run_miniterm_different_chunks_different_timestamps(
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(platformio_api, "process_stacktrace") as mock_bt,
patch.object(esp32, "process_stacktrace") as mock_bt,
):
mock_bt.return_value = False
result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4295,7 +4579,7 @@ def test_run_miniterm_handles_split_lines() -> None:
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(platformio_api, "process_stacktrace") as mock_bt,
patch.object(esp32, "process_stacktrace") as mock_bt,
patch("esphome.__main__.safe_print") as mock_print,
):
mock_bt.return_value = False
@@ -4349,7 +4633,7 @@ def test_run_miniterm_backtrace_state_maintained() -> None:
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(
platformio_api,
esp32,
"process_stacktrace",
side_effect=track_backtrace_state,
),
@@ -4400,7 +4684,7 @@ def test_run_miniterm_handles_empty_reads(
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(platformio_api, "process_stacktrace") as mock_bt,
patch.object(esp32, "process_stacktrace") as mock_bt,
):
mock_bt.return_value = False
result = run_miniterm(config, "/dev/ttyUSB0", args)
@@ -4473,7 +4757,7 @@ def test_run_miniterm_buffer_limit_prevents_unbounded_growth() -> None:
with (
patch("serial.Serial", return_value=mock_serial),
patch.object(platformio_api, "process_stacktrace") as mock_bt,
patch.object(esp32, "process_stacktrace") as mock_bt,
patch("esphome.__main__.safe_print") as mock_print,
patch("esphome.__main__.SERIAL_BUFFER_MAX_SIZE", test_buffer_limit),
):
@@ -4754,6 +5038,32 @@ def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
assert result == 2
@pytest.mark.parametrize(
("command", "expected_skip"),
[
("logs", True),
("clean", True),
("compile", False),
("config", False),
("run", False),
("clean-mqtt", False),
],
)
def test_run_esphome_skip_external_update_per_command(
tmp_path: Path, command: str, expected_skip: bool
) -> None:
"""read_config is invoked with skip_external_update=True only for commands
that don't need fresh external components (logs, clean)."""
yaml_file = tmp_path / "device.yaml"
yaml_file.write_text("esphome:\n name: test\n")
with patch("esphome.__main__.read_config", return_value=None) as mock_read:
run_esphome(["esphome", command, str(yaml_file)])
mock_read.assert_called_once()
assert mock_read.call_args.kwargs["skip_external_update"] is expected_skip
def test_get_configured_xtal_freq_reads_sdkconfig(tmp_path: Path) -> None:
"""Test reading XTAL_FREQ from sdkconfig."""
CORE.name = "test-device"
+3 -97
View File
@@ -13,6 +13,7 @@ import pytest
from esphome import platformio_api, platformio_runner
from esphome.core import CORE, EsphomeError
from esphome.util import FlashImage
def test_idedata_firmware_elf_path(setup_core: Path) -> None:
@@ -70,7 +71,7 @@ def test_idedata_extra_flash_images(setup_core: Path) -> None:
images = idedata.extra_flash_images
assert len(images) == 2
assert all(isinstance(img, platformio_api.FlashImage) for img in images)
assert all(isinstance(img, FlashImage) for img in images)
assert images[0].path == Path("/path/to/bootloader.bin")
assert images[0].offset == "0x1000"
assert images[1].path == Path("/path/to/partition.bin")
@@ -106,7 +107,7 @@ def test_idedata_cc_path(setup_core: Path) -> None:
def test_flash_image_dataclass() -> None:
"""Test FlashImage dataclass stores path and offset correctly."""
image = platformio_api.FlashImage(path=Path("/path/to/image.bin"), offset="0x10000")
image = FlashImage(path=Path("/path/to/image.bin"), offset="0x10000")
assert image.path == Path("/path/to/image.bin")
assert image.offset == "0x10000"
@@ -708,101 +709,6 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
assert build_dir.exists()
def test_process_stacktrace_esp8266_exception(setup_core: Path, caplog) -> None:
"""Test process_stacktrace handles ESP8266 exceptions."""
config = {"name": "test"}
# Test exception type parsing
line = "Exception (28):"
backtrace_state = False
result = platformio_api.process_stacktrace(config, line, backtrace_state)
assert "Access to invalid address: LOAD (wild pointer?)" in caplog.text
assert result is False
def test_process_stacktrace_esp8266_backtrace(
setup_core: Path, mock_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP8266 multi-line backtrace."""
config = {"name": "test"}
# Start of backtrace
line1 = ">>>stack>>>"
state = platformio_api.process_stacktrace(config, line1, False)
assert state is True
# Backtrace content with addresses
line2 = "40201234 40205678"
state = platformio_api.process_stacktrace(config, line2, state)
assert state is True
assert mock_decode_pc.call_count == 2
# End of backtrace
line3 = "<<<stack<<<"
state = platformio_api.process_stacktrace(config, line3, state)
assert state is False
def test_process_stacktrace_esp32_backtrace(
setup_core: Path, mock_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP32 single-line backtrace."""
config = {"name": "test"}
line = "Backtrace: 0x40081234:0x3ffb1234 0x40085678:0x3ffb5678"
state = platformio_api.process_stacktrace(config, line, False)
# Should decode both addresses
assert mock_decode_pc.call_count == 2
mock_decode_pc.assert_any_call(config, "40081234")
mock_decode_pc.assert_any_call(config, "40085678")
assert state is False
def test_process_stacktrace_bad_alloc(
setup_core: Path, mock_decode_pc: Mock, caplog
) -> None:
"""Test process_stacktrace handles bad alloc messages."""
config = {"name": "test"}
line = "last failed alloc call: 40201234(512)"
state = platformio_api.process_stacktrace(config, line, False)
assert "Memory allocation of 512 bytes failed at 40201234" in caplog.text
mock_decode_pc.assert_called_once_with(config, "40201234")
assert state is False
def test_process_stacktrace_esp32_crash_handler(
setup_core: Path, mock_decode_pc: Mock
) -> None:
"""Test process_stacktrace handles ESP32 crash handler backtrace lines."""
config = {"name": "test"}
# Simulate crash handler log lines as they appear from the API/serial
line_pc = "[E][esp32.crash:078]: PC: 0x400D1234 (fault location)"
state = platformio_api.process_stacktrace(config, line_pc, False)
# PC line is matched by existing STACKTRACE_ESP32_PC_RE
mock_decode_pc.assert_called_with(config, "400D1234")
assert state is False
mock_decode_pc.reset_mock()
line_bt0 = "[E][esp32.crash:080]: BT0: 0x400D5678 (backtrace)"
state = platformio_api.process_stacktrace(config, line_bt0, False)
mock_decode_pc.assert_called_once_with(config, "400D5678")
assert state is False
mock_decode_pc.reset_mock()
line_bt1 = "[E][esp32.crash:080]: BT1: 0x42005ABC (backtrace)"
state = platformio_api.process_stacktrace(config, line_bt1, False)
mock_decode_pc.assert_called_once_with(config, "42005ABC")
assert state is False
def test_patch_file_downloader_succeeds_first_try() -> None:
"""Test patch_file_downloader succeeds on first attempt."""
mock_exception_cls = type("PackageException", (Exception,), {})
+116
View File
@@ -709,3 +709,119 @@ def test_detect_rp2040_bootsel_timeout() -> None:
result = util.detect_rp2040_bootsel("/usr/bin/picotool")
assert result.device_count == 0
assert result.permission_error is False
class TestSafePrint:
"""Tests for ``safe_print`` and its UnicodeEncodeError fallback chain."""
@pytest.fixture(autouse=True)
def _no_dashboard(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Default ``CORE.dashboard`` to False so each test starts hermetic."""
from esphome.core import CORE
monkeypatch.setattr(CORE, "dashboard", False)
def test_prints_plain_message(self, capsys: pytest.CaptureFixture[str]) -> None:
"""ASCII-only messages take the fast path through native ``print``."""
util.safe_print("hello world")
assert capsys.readouterr().out == "hello world\n"
def test_prints_unicode_on_utf8_stdout(
self, capsys: pytest.CaptureFixture[str]
) -> None:
"""Non-ASCII goes straight through when stdout can encode it."""
util.safe_print("bars: \u2582\u2584\u2586\u2588")
assert capsys.readouterr().out == "bars: \u2582\u2584\u2586\u2588\n"
def test_dashboard_escapes_esc_byte(
self,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
r"""Dashboard mode escapes raw ``\033`` ESC bytes to literal ``\\033``."""
from esphome.core import CORE
monkeypatch.setattr(CORE, "dashboard", True)
util.safe_print("\033[0;32mhi\033[0m")
assert capsys.readouterr().out == "\\033[0;32mhi\\033[0m\n"
def test_fallback_writes_string_not_bytes_repr(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Regression: cp1252 fallback must produce a printable str, not ``b'...'``.
On Windows, when stdout is a redirected pipe (e.g. the dashboard),
Python uses cp1252, which cannot encode the wifi signal-bar block
characters (U+2582..U+2588). The previous fallback path called
``print(message.encode(...))`` with a ``bytes`` object, which
Python's ``print`` rendered as a literal ``b'...'`` repr — visible
in the user's dashboard output. The fix re-encodes through the
stream's encoding with ``backslashreplace`` and decodes back to
``str``.
"""
buf = io.BytesIO()
cp1252_stream = io.TextIOWrapper(buf, encoding="cp1252", errors="strict")
monkeypatch.setattr(sys, "stdout", cp1252_stream)
util.safe_print("bars: \u2582\u2584\u2586\u2588 done")
cp1252_stream.flush()
output = buf.getvalue().decode("cp1252")
# Output is a clean line, not the bytes repr.
assert not output.startswith("b'")
assert "b'bars" not in output
# Unencodable codepoints become readable backslash escapes.
assert "\\u2582\\u2584\\u2586\\u2588" in output
# Encodable parts survive unchanged.
assert "bars: " in output
assert " done" in output
assert output.endswith("\n")
def test_fallback_with_dashboard_escaped_message(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Dashboard ESC escaping + cp1252 fallback compose correctly."""
from esphome.core import CORE
monkeypatch.setattr(CORE, "dashboard", True)
buf = io.BytesIO()
cp1252_stream = io.TextIOWrapper(buf, encoding="cp1252", errors="strict")
monkeypatch.setattr(sys, "stdout", cp1252_stream)
util.safe_print("\033[0;32m\u2582\u2584\u2586\u2588\033[0m")
cp1252_stream.flush()
output = buf.getvalue().decode("cp1252")
# Dashboard escaping turned ESC into literal "\033" (5 chars), which
# cp1252 can encode, so it survives the round-trip verbatim.
assert "\\033[0;32m" in output
assert "\\033[0m" in output
# Block characters became backslash escapes via backslashreplace.
assert "\\u2582\\u2584\\u2586\\u2588" in output
def test_final_message_when_locale_is_invalid(
self,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
"""If every encoding path fails, surface the locale-error sentinel."""
original_print = print
call_count = 0
def fake_print(*args: Any, **kwargs: Any) -> None:
nonlocal call_count
call_count += 1
# The first three calls are: native print, stream-encoding
# fallback, ASCII fallback. Make all three raise so we reach
# the final sentinel "Cannot print line..." which is expected
# to succeed (no encoding required).
if call_count <= 3:
raise UnicodeEncodeError("ascii", "x", 0, 1, "boom")
original_print(*args, **kwargs)
monkeypatch.setattr("builtins.print", fake_print)
util.safe_print("x")
assert call_count == 4
assert (
capsys.readouterr().out == "Cannot print line because of invalid locale!\n"
)
+670
View File
@@ -0,0 +1,670 @@
"""Unit tests for esphome.web_server_ota module."""
from __future__ import annotations
import io
import logging
from pathlib import Path
import socket
from unittest.mock import MagicMock, patch
import pytest
import requests
from requests.auth import HTTPBasicAuth
from esphome.core import CORE, EsphomeError
from esphome.helpers import ProgressBar
from esphome.web_server_ota import (
OTA_PATH,
WebServerOTAError,
_MultipartStreamer,
run_ota,
)
@pytest.fixture
def firmware(tmp_path: Path) -> Path:
binary = tmp_path / "firmware.bin"
binary.write_bytes(b"\x00\x01\x02FIRMWARE\xff" * 64)
return binary
def _make_response(status: int, body: str) -> MagicMock:
response = MagicMock(spec=requests.Response)
response.status_code = status
response.text = body
response.reason = ""
return response
def _patch_resolve(
monkeypatch: pytest.MonkeyPatch, hosts: list[tuple[str, int]]
) -> None:
"""Replace resolve_ip_address so tests don't actually do DNS."""
addr_infos = [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", (host, port))
for host, port in hosts
]
monkeypatch.setattr(
"esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
)
# ---------------------------------------------------------------------------
# _MultipartStreamer
# ---------------------------------------------------------------------------
def test_multipart_streamer_emits_full_body() -> None:
"""Streaming the whole body in one call yields prefix + file + suffix."""
data = b"abcdef" * 100
streamer = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
body = streamer.read()
while True:
chunk = streamer.read()
if not chunk:
break
body += chunk
assert body.startswith(f"--{streamer.boundary}\r\n".encode())
assert b'name="update"' in body
assert b'filename="fw.bin"' in body
assert data in body
assert body.endswith(f"\r\n--{streamer.boundary}--\r\n".encode())
def test_multipart_streamer_chunked_read_matches_full_read() -> None:
"""Chunked reads (urllib3 calls read(8192) repeatedly) yield the same body."""
data = b"abcdef" * 1000 # 6000 bytes
full = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin").read()
streamed = bytearray()
s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
# Same boundary lengths -> identical total length.
while True:
chunk = s.read(64)
if not chunk:
break
streamed += chunk
# Boundaries are random per instance, so compare lengths and structure.
assert len(streamed) == len(full)
assert streamed.startswith(f"--{s.boundary}\r\n".encode())
assert streamed.endswith(f"\r\n--{s.boundary}--\r\n".encode())
def test_multipart_streamer_len_matches_emitted_bytes() -> None:
"""``__len__`` is what urllib3 uses to set Content-Length, so it must
equal the total bytes emitted by ``read``."""
data = b"x" * 12345
s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
declared = len(s)
emitted = 0
while True:
chunk = s.read(1024)
if not chunk:
break
emitted += len(chunk)
assert emitted == declared
def test_multipart_streamer_progress_ticks_during_read() -> None:
"""Each read advances the progress bar (this is the whole point of
streaming via ``data=``: progress reflects bytes leaving the host)."""
data = b"x" * 1000
s = _MultipartStreamer(io.BytesIO(data), len(data), "fw.bin")
updates: list[float] = []
s.progress.update = updates.append # type: ignore[method-assign]
while True:
chunk = s.read(128)
if not chunk:
break
assert updates, "progress.update was never called"
# Strictly non-decreasing.
assert updates == sorted(updates)
# Final update reaches (within FP) 1.0 because all bytes were read.
assert updates[-1] == pytest.approx(1.0, abs=1e-9)
def test_multipart_streamer_content_type_includes_boundary() -> None:
s = _MultipartStreamer(io.BytesIO(b""), 0, "fw.bin")
assert s.content_type == f"multipart/form-data; boundary={s.boundary}"
def test_multipart_streamer_zero_size_file() -> None:
"""A zero-byte file still produces a well-formed body and progress is
skipped (avoiding a divide-by-zero on the empty file segment)."""
s = _MultipartStreamer(io.BytesIO(b""), 0, "empty.bin")
body = b""
while True:
chunk = s.read(64)
if not chunk:
break
body += chunk
assert body.startswith(f"--{s.boundary}".encode())
assert body.endswith(f"--{s.boundary}--\r\n".encode())
def test_multipart_streamer_unique_boundary_per_instance() -> None:
a = _MultipartStreamer(io.BytesIO(b""), 0, "a")
b = _MultipartStreamer(io.BytesIO(b""), 0, "a")
assert a.boundary != b.boundary
def test_multipart_streamer_zero_size_read_returns_empty() -> None:
"""``read(0)`` short-circuits without touching state."""
s = _MultipartStreamer(io.BytesIO(b"x" * 10), 10, "fw.bin")
assert s.read(0) == b""
# No bytes consumed.
assert s._sent == 0
# ---------------------------------------------------------------------------
# run_ota
# ---------------------------------------------------------------------------
def test_run_ota_success(monkeypatch: pytest.MonkeyPatch, firmware: Path) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 0
assert host == "192.168.1.50"
post.assert_called_once()
args, kwargs = post.call_args
assert args == (f"http://192.168.1.50:80{OTA_PATH}",)
assert kwargs["auth"] is None
# Streaming body, not files=, so progress fires during transmission.
assert "files" not in kwargs
assert isinstance(kwargs["data"], _MultipartStreamer)
assert kwargs["headers"]["Content-Type"] == kwargs["data"].content_type
assert kwargs["headers"]["Connection"] == "close"
def test_run_ota_logs_device_response_body(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""The device's HTTP response body is surfaced on success."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
caplog.set_level(logging.INFO, logger="esphome.web_server_ota")
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
):
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert "Device response: Update Successful!" in caplog.text
assert "OTA successful" in caplog.text
def test_run_ota_log_says_via_web_server(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""The upload-start log line names the transport explicitly."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
caplog.set_level(logging.INFO, logger="esphome.web_server_ota")
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
):
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert "via web_server OTA" in caplog.text
def test_run_ota_sends_basic_auth(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
exit_code, _ = run_ota(["192.168.1.50"], 80, "admin", "secret", firmware)
assert exit_code == 0
auth = post.call_args.kwargs["auth"]
assert isinstance(auth, HTTPBasicAuth)
assert auth.username == "admin"
assert auth.password == "secret"
def test_run_ota_skips_auth_when_no_credentials(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert post.call_args.kwargs["auth"] is None
def test_run_ota_skips_auth_when_only_username(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""Both username and password are required to send Basic auth."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
run_ota(["192.168.1.50"], 80, "admin", None, firmware)
assert post.call_args.kwargs["auth"] is None
def test_run_ota_uses_update_url(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 8080)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
run_ota(["192.168.1.50"], 8080, None, None, firmware)
url = post.call_args.args[0]
assert url == f"http://192.168.1.50:8080{OTA_PATH}"
assert OTA_PATH == "/update"
def test_run_ota_failure_response(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Failed!"),
):
exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
assert "OTA failure" in caplog.text
def test_run_ota_failure_response_empty_body(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, ""),
):
exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
assert "no response body" in caplog.text
def test_run_ota_auth_failed(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(401, "Unauthorized"),
):
exit_code, host = run_ota(["192.168.1.50"], 80, "user", "wrong", firmware)
assert exit_code == 1
assert host is None
assert "Authentication failed" in caplog.text
def test_run_ota_unexpected_status_code(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(500, "Internal Error"),
):
exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
assert "Unexpected HTTP 500" in caplog.text
def test_run_ota_unexpected_status_empty_body_falls_back(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Empty response body uses response.reason / a fallback in the error."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
response = _make_response(503, "")
response.reason = "Service Unavailable"
with patch(
"esphome.web_server_ota.requests.post",
return_value=response,
):
exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
assert "Service Unavailable" in caplog.text
def test_run_ota_unexpected_status_no_body_no_reason(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Empty body and empty reason still produce a usable error message."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
response = _make_response(599, "")
response.reason = ""
with patch(
"esphome.web_server_ota.requests.post",
return_value=response,
):
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert "no response body" in caplog.text
def test_run_ota_connection_error_then_success(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""First resolved address fails to connect, second succeeds."""
_patch_resolve(
monkeypatch,
[("192.168.1.10", 80), ("192.168.1.50", 80)],
)
with patch(
"esphome.web_server_ota.requests.post",
side_effect=[
requests.ConnectionError("refused"),
_make_response(200, "Update Successful!"),
],
) as post:
exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 0
assert host == "192.168.1.50"
assert post.call_count == 2
def test_run_ota_request_exception_falls_through(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""A non-ConnectionError RequestException (e.g. timeout) falls through too."""
_patch_resolve(
monkeypatch,
[("192.168.1.10", 80), ("192.168.1.50", 80)],
)
with patch(
"esphome.web_server_ota.requests.post",
side_effect=[
requests.Timeout("read timeout"),
_make_response(200, "Update Successful!"),
],
):
exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 0
assert host == "192.168.1.50"
def test_run_ota_all_addresses_unreachable(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""When every resolved address fails to connect, run_ota returns failure."""
_patch_resolve(
monkeypatch,
[("192.168.1.10", 80), ("192.168.1.20", 80)],
)
with patch(
"esphome.web_server_ota.requests.post",
side_effect=requests.ConnectionError("refused"),
):
exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
# Per-address failure is logged for each attempt; final summary follows.
assert caplog.text.count("OTA upload to ") >= 2
assert "OTA upload failed." in caplog.text
def test_run_ota_no_resolved_addresses(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""If resolve_ip_address returns no candidates, log and return failure."""
_patch_resolve(monkeypatch, [])
exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
assert "Could not resolve 192.168.1.50" in caplog.text
def test_run_ota_resolution_failure(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
def _raise(*_args, **_kwargs):
raise EsphomeError("dns failed")
monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _raise)
exit_code, host = run_ota(["does.not.exist"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
def test_run_ota_resolution_failure_dashboard_mode(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Dashboard mode skips the '--device <IP>' tip on resolution failure."""
def _raise(*_args, **_kwargs):
raise EsphomeError("dns failed")
monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _raise)
monkeypatch.setattr(CORE, "dashboard", True)
try:
exit_code, host = run_ota(["does.not.exist"], 80, None, None, firmware)
finally:
monkeypatch.setattr(CORE, "dashboard", False)
assert exit_code == 1
assert host is None
assert "--device <IP>" not in caplog.text
def test_run_ota_empty_hosts(firmware: Path) -> None:
exit_code, host = run_ota([], 80, None, None, firmware)
assert exit_code == 1
assert host is None
def test_run_ota_string_host_accepted(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""A bare string is accepted in addition to a list of hosts."""
_patch_resolve(monkeypatch, [("10.0.0.5", 80)])
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
):
exit_code, host = run_ota("10.0.0.5", 80, None, None, firmware)
assert exit_code == 0
assert host == "10.0.0.5"
def test_run_ota_multiple_hosts_first_fails(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""Multi-host fallthrough: first host's addresses all fail, second host wins."""
addr_lookup = {
"primary.local": [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.10", 80)),
],
"secondary.local": [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.50", 80)),
],
}
def _resolve(host, port, address_cache=None): # noqa: ARG001
return addr_lookup[host]
monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _resolve)
with patch(
"esphome.web_server_ota.requests.post",
side_effect=[
requests.ConnectionError("refused"),
_make_response(200, "Update Successful!"),
],
):
exit_code, host = run_ota(
["primary.local", "secondary.local"], 80, None, None, firmware
)
assert exit_code == 0
assert host == "192.168.1.50"
def test_run_ota_all_hosts_return_failure_no_exception(
monkeypatch: pytest.MonkeyPatch, firmware: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""All hosts resolve to no addresses; run_ota cleanly returns failure."""
addr_lookup = {
"a.local": [],
"b.local": [],
}
def _resolve(host, port, address_cache=None): # noqa: ARG001
return addr_lookup[host]
monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _resolve)
exit_code, host = run_ota(["a.local", "b.local"], 80, None, None, firmware)
assert exit_code == 1
assert host is None
# Each host gets its own "Could not resolve" log line + final summary.
assert caplog.text.count("Could not resolve") == 2
assert "OTA upload failed." in caplog.text
def test_web_server_ota_error_is_esphome_error() -> None:
assert issubclass(WebServerOTAError, EsphomeError)
def test_run_ota_finalizes_progress_bar_on_success(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""progress.done() fires on the success path (finally block)."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
done_called: list[bool] = []
with (
patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
),
patch.object(ProgressBar, "done", lambda self: done_called.append(True)),
):
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert done_called
def test_run_ota_finalizes_progress_bar_on_failure(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""progress.done() fires when the request itself raises (finally block)."""
_patch_resolve(monkeypatch, [("192.168.1.50", 80)])
done_called: list[bool] = []
with (
patch(
"esphome.web_server_ota.requests.post",
side_effect=requests.ConnectionError("boom"),
),
patch.object(ProgressBar, "done", lambda self: done_called.append(True)),
):
run_ota(["192.168.1.50"], 80, None, None, firmware)
assert done_called
def test_run_ota_ipv6_url_brackets_host(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""IPv6 candidates are bracketed in the URL so the port parses correctly."""
addr_infos = [
(socket.AF_INET6, socket.SOCK_STREAM, 0, "", ("2001:db8::1", 80, 0, 0)),
]
monkeypatch.setattr(
"esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
)
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
exit_code, host = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 0
assert host == "2001:db8::1"
url = post.call_args.args[0]
assert url == f"http://[2001:db8::1]:80{OTA_PATH}"
def test_run_ota_ipv6_link_local_includes_scope_id(
monkeypatch: pytest.MonkeyPatch, firmware: Path
) -> None:
"""Link-local IPv6 candidates include the percent-encoded zone index."""
addr_infos = [
(socket.AF_INET6, socket.SOCK_STREAM, 0, "", ("fe80::1", 80, 0, 3)),
]
monkeypatch.setattr(
"esphome.web_server_ota.resolve_ip_address", lambda *a, **kw: addr_infos
)
with patch(
"esphome.web_server_ota.requests.post",
return_value=_make_response(200, "Update Successful!"),
) as post:
exit_code, _ = run_ota(["device.local"], 80, None, None, firmware)
assert exit_code == 0
url = post.call_args.args[0]
assert url == f"http://[fe80::1%253]:80{OTA_PATH}"
+237
View File
@@ -0,0 +1,237 @@
"""Unit tests for ``esphome.zeroconf`` device-discovery primitives.
Covers ``DashboardImportDiscovery`` (state transitions for adoption /
import flows) and ``DiscoveredImport`` (TXT-record parse shape). Both
are part of the cross-tool contract between the legacy dashboard and
the new device-builder backend (esphome/device-builder); changes to
the callback signature, the ``import_state`` dict shape, or the
``DiscoveredImport`` field set will break downstream consumers.
"""
from __future__ import annotations
from unittest.mock import MagicMock
from zeroconf import ServiceStateChange
from esphome.zeroconf import (
ESPHOME_SERVICE_TYPE,
DashboardImportDiscovery,
DiscoveredImport,
)
def _make_service_info(
package_import_url: str = "github://esphome/example/example.yaml",
project_name: str = "esphome.example",
project_version: str = "1.0.0",
network: str | None = "wifi",
friendly_name: str | None = "Living Room",
version: str | None = "2025.1.0",
) -> MagicMock:
"""Build a fake ``AsyncServiceInfo`` with the TXT records we care about.
The real callback path resolves a service via zeroconf and then
reads ``info.properties`` (a ``dict[bytes, bytes | None]``). Mock
that shape so we can drive ``_process_service_info`` directly
without spinning up a real zeroconf instance.
"""
info = MagicMock()
properties: dict[bytes, bytes | None] = {
b"package_import_url": package_import_url.encode(),
b"project_name": project_name.encode(),
b"project_version": project_version.encode(),
}
if network is not None:
properties[b"network"] = network.encode()
if friendly_name is not None:
properties[b"friendly_name"] = friendly_name.encode()
if version is not None:
properties[b"version"] = version.encode()
info.properties = properties
info.load_from_cache.return_value = True
return info
def test_added_service_populates_import_state_and_fires_callback() -> None:
"""An ADD with the required TXT records lands a ``DiscoveredImport`` and notifies.
Mirrors what both the legacy dashboard and device-builder rely
on — the callback is the only signal that an importable device
has appeared on the LAN, and ``import_state`` is the snapshot
they read on demand.
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
info = _make_service_info()
name = f"living-room.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
assert name in discovery.import_state
entry = discovery.import_state[name]
assert isinstance(entry, DiscoveredImport)
assert entry.device_name == "living-room"
assert entry.package_import_url == "github://esphome/example/example.yaml"
assert entry.project_name == "esphome.example"
assert entry.project_version == "1.0.0"
assert entry.network == "wifi"
assert entry.friendly_name == "Living Room"
on_update.assert_called_once_with(name, entry)
def test_added_service_without_required_txt_is_ignored() -> None:
"""A device that doesn't carry ``package_import_url`` etc. isn't importable.
The dashboard browser also fires for plain ``_esphomelib._tcp``
services that happen to match the type but aren't dashboard
imports. Those must not land in ``import_state`` or fire the
update callback — otherwise the dashboard would surface every
API-enabled device on the LAN as "ready to adopt".
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
info = MagicMock()
# Empty TXT records — no import URL, no version. ``version``-only
# services hit a separate ``update_device_mdns`` path that talks
# to ``StorageJSON``; that's covered elsewhere.
info.properties = {}
info.load_from_cache.return_value = True
discovery._process_service_info(f"plain.{ESPHOME_SERVICE_TYPE}", info)
assert discovery.import_state == {}
on_update.assert_not_called()
def test_repeated_add_does_not_re_fire_callback() -> None:
"""Re-resolving the same service doesn't spam the on_update callback.
The dashboard re-resolves periodically; without the ``is_new``
guard, every refresh would fire ``IMPORTABLE_DEVICE_ADDED`` and
the dashboard's UI would re-render endlessly.
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
info = _make_service_info()
name = f"living-room.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
discovery._process_service_info(name, info)
on_update.assert_called_once()
def test_removed_service_clears_state_and_fires_none_callback() -> None:
"""A ServiceStateChange.Removed pops the entry and notifies with ``None``.
Both consumers rely on the ``(name, None)`` callback shape to
distinguish "device gone" from "device updated". Coordinate
before changing the second-arg semantics.
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
info = _make_service_info()
name = f"living-room.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
on_update.reset_mock()
discovery.browser_callback(
zeroconf=MagicMock(),
service_type=ESPHOME_SERVICE_TYPE,
name=name,
state_change=ServiceStateChange.Removed,
)
assert name not in discovery.import_state
on_update.assert_called_once_with(name, None)
def test_remove_for_unknown_service_does_not_fire_callback() -> None:
"""A spurious Removed for a service we never tracked is a silent no-op.
The browser can fire Removed for any matching service type,
not just the importable ones we're tracking. Don't let those
confuse the callback consumer.
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
discovery.browser_callback(
zeroconf=MagicMock(),
service_type=ESPHOME_SERVICE_TYPE,
name=f"never-seen.{ESPHOME_SERVICE_TYPE}",
state_change=ServiceStateChange.Removed,
)
on_update.assert_not_called()
def test_updated_service_for_unknown_name_is_ignored() -> None:
"""Updates without a prior Add don't seed ``import_state``.
The dashboard counts on Add to introduce the device and Update
to refresh it. Letting Update silently introduce new state would
let an unrelated TXT change bypass the Add-time validation.
"""
on_update = MagicMock()
discovery = DashboardImportDiscovery(on_update=on_update)
discovery.browser_callback(
zeroconf=MagicMock(),
service_type=ESPHOME_SERVICE_TYPE,
name=f"living-room.{ESPHOME_SERVICE_TYPE}",
state_change=ServiceStateChange.Updated,
)
assert discovery.import_state == {}
on_update.assert_not_called()
def test_network_defaults_to_wifi_when_txt_absent() -> None:
"""Older firmware that doesn't broadcast ``network`` defaults to ``wifi``.
The TXT record was added in a later release; pre-existing
factory firmwares advertise without it. ``DiscoveredImport``
has to default cleanly so adoption flows can still produce a
valid YAML for those devices.
"""
discovery = DashboardImportDiscovery()
info = _make_service_info(network=None)
name = f"older.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
assert discovery.import_state[name].network == "wifi"
def test_friendly_name_optional() -> None:
"""``friendly_name`` may be ``None`` if the device doesn't broadcast it.
Both consumers handle the ``None`` case (rendering the device
name as fallback in the UI). Locking this in keeps the
optionality explicit so a future refactor doesn't accidentally
coerce it into an empty string.
"""
discovery = DashboardImportDiscovery()
info = _make_service_info(friendly_name=None)
name = f"no-friendly.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
assert discovery.import_state[name].friendly_name is None
def test_callback_is_optional() -> None:
"""``on_update=None`` lets ``import_state`` track silently.
Used by callers that read the dict directly rather than
subscribing to events.
"""
discovery = DashboardImportDiscovery(on_update=None)
info = _make_service_info()
name = f"silent.{ESPHOME_SERVICE_TYPE}"
discovery._process_service_info(name, info)
# No callback to assert against; just verify state landed.
assert name in discovery.import_state