github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-30 16:30:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
# Conflicts:
#	esphome/external_files.py
#	tests/components/wifi/test.esp8266-ard.yaml
This commit is contained in:
J. Nick Koston
2026-04-29 06:21:06 -05:00
parent 291eb22231 8ceada8d04
commit 34b66dca55
5 changed files with 325 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/script/test_test_helpers.py
+158
View File
@@ -258,3 +258,161 @@ def test_load_wraps_platform_component(tmp_path: Path) -> None:
assert key == "bthome.sensor"
assert isinstance(installed, ComponentManifestOverride)
assert installed.to_code is None
# ---------------------------------------------------------------------------
# populate_dependency_config
# ---------------------------------------------------------------------------
def _make_component_stub(
*,
multi_conf: bool = False,
is_platform_component: bool = False,
config_schema=None,
) -> MagicMock:
stub = MagicMock()
stub.multi_conf = multi_conf
stub.is_platform_component = is_platform_component
stub.config_schema = config_schema
return stub
def test_populate_platform_component_listed_alone_uses_list() -> None:
"""Regression: a platform component (sensor) with no `sensor.x` siblings
must land as `[]` in config. Previously it was populated as a dict via
`schema({})`, which then crashed the sibling `domain.platform` branch
when later dependencies tried `config.setdefault('sensor', []).append(...)`.
"""
sensor = _make_component_stub(is_platform_component=True)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["sensor"],
get_component_fn=lambda name: sensor if name == "sensor" else None,
register_platform_fn=lambda _: None,
)
assert config["sensor"] == []
def test_populate_platform_component_then_platform_entry() -> None:
"""When `sensor` is processed before `sensor.gpio` (sorted order),
the bare-component branch must leave `config['sensor']` as a list so
the platform-entry branch can append into it.
"""
sensor = _make_component_stub(is_platform_component=True)
gpio = _make_component_stub() # the bare `gpio` component
components: dict[str, object] = {"sensor": sensor, "gpio": gpio}
config: dict = {}
build_helpers.populate_dependency_config(
config,
["gpio", "sensor", "sensor.gpio"],
get_component_fn=components.get,
register_platform_fn=lambda _: None,
)
assert config["sensor"] == [{"platform": "gpio"}]
def test_populate_multi_conf_component_uses_list() -> None:
multi = _make_component_stub(multi_conf=True)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["multi"],
get_component_fn=lambda name: multi if name == "multi" else None,
register_platform_fn=lambda _: None,
)
assert config["multi"] == []
def test_populate_plain_component_uses_schema_defaults() -> None:
schema = MagicMock(return_value={"default_key": 42})
plain = _make_component_stub(config_schema=schema)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
schema.assert_called_once_with({})
assert config["plain"] == {"default_key": 42}
def test_populate_plain_component_falls_back_when_schema_raises() -> None:
def picky_schema(_):
raise ValueError("required field missing")
plain = _make_component_stub(config_schema=picky_schema)
config: dict = {}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
assert config["plain"] == {}
def test_populate_skips_unresolvable_pseudo_components() -> None:
"""`core` and other names that get_component returns None for are skipped
silently without inserting anything into the config.
"""
config: dict = {}
build_helpers.populate_dependency_config(
config,
["core"],
get_component_fn=lambda _: None,
register_platform_fn=lambda _: None,
)
assert config == {}
def test_populate_preserves_existing_plain_component_config() -> None:
"""If a plain component already has a config entry (e.g. from the user's
YAML), the schema-defaults branch must not overwrite it.
"""
schema = MagicMock()
plain = _make_component_stub(config_schema=schema)
config: dict = {"plain": {"user_key": "set_by_user"}}
build_helpers.populate_dependency_config(
config,
["plain"],
get_component_fn=lambda name: plain if name == "plain" else None,
register_platform_fn=lambda _: None,
)
schema.assert_not_called()
assert config["plain"] == {"user_key": "set_by_user"}
def test_populate_registers_platform_for_platform_entry() -> None:
"""Each `domain.platform` entry triggers register_platform_fn(domain) so
USE_<DOMAIN> defines get emitted later in the build pipeline.
"""
registered: list[str] = []
config: dict = {}
build_helpers.populate_dependency_config(
config,
["sensor.gpio", "binary_sensor.gpio"],
get_component_fn=lambda _: None,
register_platform_fn=registered.append,
)
assert registered == ["sensor", "binary_sensor"]
assert config["sensor"] == [{"platform": "gpio"}]
assert config["binary_sensor"] == [{"platform": "gpio"}]
tests/unit_tests/test_external_files.py
+94 -63
View File
@@ -9,7 +9,7 @@ import pytest
import requests
from esphome import external_files
from esphome.config_validation import Invalid
from esphome.config_validation import Invalid, MultipleInvalid
from esphome.core import CORE, EsphomeError, TimePeriod
@@ -574,65 +574,6 @@ def test_download_content_skip_external_update_downloads_when_missing(
assert test_file.read_bytes() == new_content
def test_download_content_saves_etag(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
setup_core: Path,
) -> None:
"""Test download_content writes the ETag sidecar after a successful download."""
test_file = setup_core / "fresh.txt"
new_content = b"fresh content"
mock_has_remote_file_changed.return_value = True
mock_response = MagicMock()
mock_response.content = new_content
mock_response.headers = {external_files.ETAG: '"deadbeef"'}
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
url = "https://example.com/file.txt"
external_files.download_content(url, test_file)
assert external_files._etag_sidecar_path(test_file).read_text() == '"deadbeef"'
def test_download_content_atomic_write_no_partial_on_failure(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
mock_write_file: MagicMock,
setup_core: Path,
) -> None:
"""If `write_file` (the atomic-write helper) fails, the existing cache
file must remain untouched and no temp files may be left behind. Patching
`write_file` directly exercises the atomic-rename path -- a failure inside
`write_file` is the only reason the rename wouldn't have happened.
"""
from esphome.core import EsphomeError
test_file = setup_core / "cached.txt"
original_content = b"original content"
test_file.write_bytes(original_content)
mock_has_remote_file_changed.return_value = True
mock_response = MagicMock()
mock_response.content = b"new content"
mock_response.headers = {}
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
mock_write_file.side_effect = EsphomeError("disk full")
with pytest.raises(EsphomeError, match="disk full"):
external_files.download_content("https://example.com/file.txt", test_file)
# Original file is untouched -- write_file aborted before its rename step.
assert test_file.read_bytes() == original_content
# write_file is responsible for cleaning its own temp files; nothing leaks
# into the cache directory either way.
leftover_tmps = list(setup_core.glob("tmp*"))
assert leftover_tmps == []
def test_download_content_many_empty_is_noop(
mock_download_content: MagicMock, setup_core: Path
) -> None:
@@ -676,10 +617,12 @@ def test_download_content_many_runs_in_parallel(
assert mock_download_content.call_count == 3
def test_download_content_many_propagates_errors(
def test_download_content_many_propagates_single_error(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""An exception from any worker must propagate out of download_content_many."""
"""A single failing worker should raise its `Invalid` directly, not wrap
it in a `MultipleInvalid` that the caller would have to unpack.
"""
def fake_download(url: str, path: Path, timeout: int) -> bytes:
if url.endswith("bad"):
@@ -691,8 +634,37 @@ def test_download_content_many_propagates_errors(
("https://example.com/ok", setup_core / "ok"),
("https://example.com/bad", setup_core / "bad"),
]
with pytest.raises(Invalid, match="could not download"):
with pytest.raises(Invalid, match="could not download") as exc_info:
external_files.download_content_many(items)
assert not isinstance(exc_info.value, MultipleInvalid)
def test_download_content_many_aggregates_multiple_errors(
mock_download_content: MagicMock, setup_core: Path
) -> None:
"""Every failing worker should be reported in a single MultipleInvalid so
the user sees all broken URLs in one validation pass instead of fixing
them one network round-trip at a time.
"""
def fake_download(url: str, path: Path, timeout: int) -> bytes:
if url.endswith("ok"):
return b""
raise Invalid(f"could not download {url}")
mock_download_content.side_effect = fake_download
items = [
("https://example.com/ok", setup_core / "ok"),
("https://example.com/bad1", setup_core / "bad1"),
("https://example.com/bad2", setup_core / "bad2"),
]
with pytest.raises(MultipleInvalid) as exc_info:
external_files.download_content_many(items)
messages = {str(e) for e in exc_info.value.errors}
assert messages == {
"could not download https://example.com/bad1",
"could not download https://example.com/bad2",
}
def test_download_content_many_dedupes_by_path(
@@ -767,3 +739,62 @@ def test_download_web_files_in_config_no_web_entries(
external_files.download_web_files_in_config(config, lambda _: setup_core / "x")
mock_download_content_many.assert_called_once()
assert list(mock_download_content_many.call_args[0][0]) == []
def test_download_content_saves_etag(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
setup_core: Path,
) -> None:
"""Test download_content writes the ETag sidecar after a successful download."""
test_file = setup_core / "fresh.txt"
new_content = b"fresh content"
mock_has_remote_file_changed.return_value = True
mock_response = MagicMock()
mock_response.content = new_content
mock_response.headers = {external_files.ETAG: '"deadbeef"'}
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
url = "https://example.com/file.txt"
external_files.download_content(url, test_file)
assert external_files._etag_sidecar_path(test_file).read_text() == '"deadbeef"'
def test_download_content_atomic_write_no_partial_on_failure(
mock_has_remote_file_changed: MagicMock,
mock_requests_get: MagicMock,
mock_write_file: MagicMock,
setup_core: Path,
) -> None:
"""If `write_file` (the atomic-write helper) fails, the existing cache
file must remain untouched and no temp files may be left behind. Patching
`write_file` directly exercises the atomic-rename path -- a failure inside
`write_file` is the only reason the rename wouldn't have happened.
"""
from esphome.core import EsphomeError
test_file = setup_core / "cached.txt"
original_content = b"original content"
test_file.write_bytes(original_content)
mock_has_remote_file_changed.return_value = True
mock_response = MagicMock()
mock_response.content = b"new content"
mock_response.headers = {}
mock_response.raise_for_status = MagicMock()
mock_requests_get.return_value = mock_response
mock_write_file.side_effect = EsphomeError("disk full")
with pytest.raises(EsphomeError, match="disk full"):
external_files.download_content("https://example.com/file.txt", test_file)
# Original file is untouched -- write_file aborted before its rename step.
assert test_file.read_bytes() == original_content
# write_file is responsible for cleaning its own temp files; nothing leaks
# into the cache directory either way.
leftover_tmps = list(setup_core.glob("tmp*"))
assert leftover_tmps == []