github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-06 19:06:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into libretiny-lwip-tune

Browse Source
This commit is contained in:
J. Nick Koston
2026-02-21 19:44:17 -06:00
committed by GitHub
parent 30a2af0d54 d5c9c56fdf
commit 623bbc0dbe
22 changed files with 466 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/dsmr/common.yaml
+9
View File
@@ -1,4 +1,13 @@
esphome:
on_boot:
then:
- lambda: |-
// Test deprecated std::string overload still compiles
std::string key = "00112233445566778899aabbccddeeff";
id(dsmr_instance).set_decryption_key(key);
dsmr:
id: dsmr_instance
decryption_key: 00112233445566778899aabbccddeeff
max_telegram_length: 1000
request_pin: ${request_pin}
tests/components/homeassistant/common.yaml
+13
View File
@@ -90,6 +90,19 @@ text_sensor:
id: ha_hello_world_text2
attribute: some_attribute
event:
- platform: template
name: Test Event
id: test_event
event_types:
- test_event_type
on_event:
- homeassistant.event:
event: esphome.test_event
data:
event_name: !lambda |-
return event_type;
time:
- platform: homeassistant
on_time:
tests/unit_tests/test_main.py
+6
View File
@@ -2951,6 +2951,7 @@ def test_run_miniterm_batches_lines_with_same_timestamp(
mock_serial = MockSerial([chunk, MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
@@ -2989,6 +2990,7 @@ def test_run_miniterm_different_chunks_different_timestamps(
mock_serial = MockSerial([chunk1, chunk2, MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
@@ -3019,6 +3021,7 @@ def test_run_miniterm_handles_split_lines() -> None:
mock_serial = MockSerial([chunk1, chunk2, MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
@@ -3057,6 +3060,7 @@ def test_run_miniterm_backtrace_state_maintained() -> None:
mock_serial = MockSerial([backtrace_chunk, MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
@@ -3122,6 +3126,7 @@ def test_run_miniterm_handles_empty_reads(
mock_serial = MockSerial([b"", chunk, b"", MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
@@ -3194,6 +3199,7 @@ def test_run_miniterm_buffer_limit_prevents_unbounded_growth() -> None:
mock_serial = MockSerial([large_data_no_newline, final_line, MOCK_SERIAL_END])
CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
config = {
CONF_LOGGER: {
CONF_BAUD_RATE: 115200,
tests/unit_tests/test_platformio_api.py
+195 -1
View File
@@ -6,7 +6,7 @@ import os
from pathlib import Path
import shutil
from types import SimpleNamespace
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import MagicMock, Mock, call, patch
import pytest
@@ -673,6 +673,200 @@ def test_process_stacktrace_bad_alloc(
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,), {})
original_init = MagicMock()
with patch.dict(
"sys.modules",
{
"platformio": MagicMock(),
"platformio.package": MagicMock(),
"platformio.package.download": SimpleNamespace(
FileDownloader=type("FileDownloader", (), {"__init__": original_init})
),
"platformio.package.exception": SimpleNamespace(
PackageException=mock_exception_cls
),
},
):
platformio_api.patch_file_downloader()
from platformio.package.download import FileDownloader
instance = object.__new__(FileDownloader)
FileDownloader.__init__(instance, "http://example.com/file.zip")
original_init.assert_called_once()
def test_patch_file_downloader_retries_on_failure() -> None:
"""Test patch_file_downloader retries with backoff on PackageException."""
mock_exception_cls = type("PackageException", (Exception,), {})
call_count = 0
def failing_init(self, *args, **kwargs):
nonlocal call_count
call_count += 1
if call_count < 3:
raise mock_exception_cls(f"502 error attempt {call_count}")
with (
patch.dict(
"sys.modules",
{
"platformio": MagicMock(),
"platformio.package": MagicMock(),
"platformio.package.download": SimpleNamespace(
FileDownloader=type(
"FileDownloader", (), {"__init__": failing_init}
)
),
"platformio.package.exception": SimpleNamespace(
PackageException=mock_exception_cls
),
},
),
patch("time.sleep") as mock_sleep,
):
platformio_api.patch_file_downloader()
from platformio.package.download import FileDownloader
instance = object.__new__(FileDownloader)
FileDownloader.__init__(instance, "http://example.com/file.zip")
# Should have been called 3 times (2 failures + 1 success)
assert call_count == 3
# Should have slept with exponential backoff: 2s, 4s
assert mock_sleep.call_count == 2
mock_sleep.assert_any_call(2)
mock_sleep.assert_any_call(4)
def test_patch_file_downloader_raises_after_max_retries() -> None:
"""Test patch_file_downloader raises after exhausting all retries."""
mock_exception_cls = type("PackageException", (Exception,), {})
def always_failing_init(self, *args, **kwargs):
raise mock_exception_cls("502 error")
with (
patch.dict(
"sys.modules",
{
"platformio": MagicMock(),
"platformio.package": MagicMock(),
"platformio.package.download": SimpleNamespace(
FileDownloader=type(
"FileDownloader", (), {"__init__": always_failing_init}
)
),
"platformio.package.exception": SimpleNamespace(
PackageException=mock_exception_cls
),
},
),
patch("time.sleep") as mock_sleep,
):
platformio_api.patch_file_downloader()
from platformio.package.download import FileDownloader
instance = object.__new__(FileDownloader)
with pytest.raises(mock_exception_cls, match="502 error"):
FileDownloader.__init__(instance, "http://example.com/file.zip")
# Should have slept 4 times (before attempts 2-5), not on final attempt
assert mock_sleep.call_count == 4
mock_sleep.assert_has_calls([call(2), call(4), call(8), call(16)])
def test_patch_file_downloader_closes_session_and_response_between_retries() -> None:
"""Test patch_file_downloader closes HTTP session and response between retries."""
mock_exception_cls = type("PackageException", (Exception,), {})
mock_session = MagicMock()
mock_response = MagicMock()
call_count = 0
def failing_init_with_session(self, *args, **kwargs):
nonlocal call_count
call_count += 1
self._http_session = mock_session
self._http_response = mock_response
if call_count < 2:
raise mock_exception_cls("502 error")
with (
patch.dict(
"sys.modules",
{
"platformio": MagicMock(),
"platformio.package": MagicMock(),
"platformio.package.download": SimpleNamespace(
FileDownloader=type(
"FileDownloader",
(),
{"__init__": failing_init_with_session},
)
),
"platformio.package.exception": SimpleNamespace(
PackageException=mock_exception_cls
),
},
),
patch("time.sleep"),
):
platformio_api.patch_file_downloader()
from platformio.package.download import FileDownloader
instance = object.__new__(FileDownloader)
FileDownloader.__init__(instance, "http://example.com/file.zip")
# Both response and session should have been closed between retries
mock_response.close.assert_called_once()
mock_session.close.assert_called_once()
def test_patch_file_downloader_idempotent() -> None:
"""Test patch_file_downloader does not stack wrappers when called multiple times."""
mock_exception_cls = type("PackageException", (Exception,), {})
call_count = 0
def counting_init(self, *args, **kwargs):
nonlocal call_count
call_count += 1
with patch.dict(
"sys.modules",
{
"platformio": MagicMock(),
"platformio.package": MagicMock(),
"platformio.package.download": SimpleNamespace(
FileDownloader=type("FileDownloader", (), {"__init__": counting_init})
),
"platformio.package.exception": SimpleNamespace(
PackageException=mock_exception_cls
),
},
):
# Patch multiple times
platformio_api.patch_file_downloader()
platformio_api.patch_file_downloader()
platformio_api.patch_file_downloader()
from platformio.package.download import FileDownloader
instance = object.__new__(FileDownloader)
FileDownloader.__init__(instance, "http://example.com/file.zip")
# Should only be called once, not 3 times from stacked wrappers
assert call_count == 1
def test_platformio_log_filter_allows_non_platformio_messages() -> None:
"""Test that non-platformio logger messages are allowed through."""
log_filter = platformio_api.PlatformioLogFilter()