Merge branch 'dev' into jesserockz-2026-503

This commit is contained in:
Jesse Hills
2026-09-02 11:12:53 +12:00
committed by GitHub
1889 changed files with 60990 additions and 21606 deletions
@@ -263,6 +263,17 @@ def test_device_capabilities_response_has_id_150() -> None:
)
def test_z_wave_proxy_request_response_has_id_151() -> None:
body = _extract_proto_message(PROTO_TEXT, "ZWaveProxyRequestResponse")
match = re.search(r"option \(id\) = (\d+);", body)
assert match is not None, "ZWaveProxyRequestResponse is missing `option (id)`"
assert int(match.group(1)) == 151, (
f"ZWaveProxyRequestResponse has id {match.group(1)}, expected 151. "
"Message ids are part of the wire protocol and must not change once "
"assigned."
)
def test_superseded_fields_are_not_marked_deprecated_in_proto() -> None:
"""The six superseded fields must not carry `[deprecated = true]` in
api.proto, or the generator drops them and old clients stop receiving
@@ -15,7 +15,12 @@ import pytest
sys.path.insert(0, str(Path(__file__).parents[4] / "script" / "api_protobuf"))
from api_protobuf import _make_ifdef_line, get_varint64_ifdef # noqa: E402
from api_protobuf import ( # noqa: E402
MAX_MESSAGE_ID,
_make_ifdef_line,
get_varint64_ifdef,
validate_message_id,
)
from google.protobuf import descriptor_pb2 # noqa: E402
@@ -91,3 +96,14 @@ def test_make_ifdef_line_conjunction_and_negation() -> None:
assert (
_make_ifdef_line("USE_X && !USE_Y") == "#if defined(USE_X) && !defined(USE_Y)"
)
def test_message_id_at_maximum_is_accepted() -> None:
# 16383 is the largest ID whose plaintext type varint fits the 2 bytes
# budgeted in HEADER_PADDING.
validate_message_id(MAX_MESSAGE_ID, "MaxMessage")
def test_message_id_above_maximum_is_rejected() -> None:
with pytest.raises(ValueError, match="exceeds the plaintext"):
validate_message_id(MAX_MESSAGE_ID + 1, "TooBigMessage")
@@ -0,0 +1,64 @@
"""Tests for the bme68x_bsec2 prefetch extraction."""
from __future__ import annotations
from pathlib import Path
from esphome.components import bme68x_bsec2 as bsec
from esphome.loader import get_component
def test_prefetch_applies_defaults(setup_core: Path) -> None:
[files] = list(bsec.PREFETCH_FILES([{"model": "bme680"}]))
assert len(files) == 1
assert "bme680_iaq_33v_3s_28d" in files[0].url
assert files[0].path == bsec._compute_local_file_path(files[0].url)
def test_prefetch_normalizes_enum_case(setup_core: Path) -> None:
[files] = list(
bsec.PREFETCH_FILES(
[
{
"model": "BME688",
"sample_rate": "ulp",
"supply_voltage": "1.8v",
"algorithm_output": "REGRESSION",
"operating_age": "4D",
}
]
)
)
assert len(files) == 1
assert "bme688_reg_18v_300s_4d" in files[0].url
def test_prefetch_skips_unknown_values(setup_core: Path) -> None:
entries = [
{"model": "bme999"},
{"model": "bme680", "sample_rate": "TURBO"},
{"model": "bme680", "algorithm_output": "psychic"},
{},
]
assert list(bsec.PREFETCH_FILES(entries)) == [[]]
def test_prefetch_matches_validator_url(setup_core: Path) -> None:
"""The hook's URL equals _compute_url over the validated config shape."""
validated = {
"model": "bme688",
"operating_age": "28d",
"sample_rate": "LP",
"supply_voltage": "3.3V",
"algorithm_output": "classification",
}
[files] = list(bsec.PREFETCH_FILES([dict(validated)]))
assert files[0].url == bsec._compute_url(validated)
def test_hook_is_wired_to_the_user_facing_domain() -> None:
"""The i2c domain (the only user-facing one) exposes the hook."""
component = get_component("bme68x_bsec2_i2c")
assert component is not None
assert component.prefetch_files is bsec.PREFETCH_FILES
@@ -0,0 +1,72 @@
"""Tests for the esp32 sdkconfig write and its toolchain-gated clean."""
from __future__ import annotations
import os
from pathlib import Path
import time
from unittest.mock import patch
import pytest
from esphome.components.esp32 import _write_sdkconfig
from esphome.components.esp32.const import KEY_SDKCONFIG_OPTIONS
from esphome.const import KEY_CORE, KEY_ESP32, KEY_FRAMEWORK_VERSION, Toolchain
from esphome.core import CORE
from esphome.espidf.toolchain import has_outdated_files
def _setup_core(tmp_path: Path, toolchain: Toolchain | None) -> None:
CORE.config_path = tmp_path / "test.yaml"
CORE.build_path = tmp_path
CORE.toolchain = toolchain
CORE.data[KEY_ESP32] = {KEY_SDKCONFIG_OPTIONS: {"CONFIG_X": "y"}}
CORE.data[KEY_CORE] = {KEY_FRAMEWORK_VERSION: "5.5.5"}
def _seed_configured_build(tmp_path: Path) -> None:
"""A settled native build: configure outputs predate what comes next."""
build = tmp_path / "build"
(build / "config").mkdir(parents=True)
(build / "config" / "sdkconfig.h").write_text("")
(build / "CMakeCache.txt").write_text("")
(build / "build.ninja").write_text("")
# Explicitly older than what the test writes next: has_outdated_files()
# compares st_mtime with a strict >, so same-tick writes would pass
past = time.time() - 60
for f in build.rglob("*"):
os.utime(f, (past, past))
@pytest.mark.parametrize(
("toolchain", "clean_expected"),
[(Toolchain.ESP_IDF, False), (Toolchain.PLATFORMIO, True), (None, True)],
)
def test_write_sdkconfig_cleans_only_on_platformio(
tmp_path: Path, toolchain: Toolchain | None, clean_expected: bool
) -> None:
"""A changed sdkconfig forces a full clean only under PlatformIO; the
esp-idf toolchain reconfigures via has_outdated_files() instead; an
unresolved toolchain fails safe onto the clean."""
_setup_core(tmp_path, toolchain)
_seed_configured_build(tmp_path)
with (
patch.object(CORE, "name", "test"),
patch("esphome.components.esp32.clean_build") as clean,
):
_write_sdkconfig()
assert "CONFIG_X" in CORE.relative_build_path("sdkconfig.test").read_text()
assert clean.called is clean_expected
if clean_expected:
clean.assert_called_once_with(clear_pio_cache=False)
# The change must still trigger a reconfigure: the internal
# sdkconfig snapshot is now newer than build/CMakeCache.txt
assert has_outdated_files() is True
clean.reset_mock()
# A settled configure restamps the cache; an unchanged rewrite
# must then neither clean nor mark the build stale
future = time.time() + 60
os.utime(CORE.relative_build_path("build/CMakeCache.txt"), (future, future))
_write_sdkconfig()
clean.assert_not_called()
assert has_outdated_files() is False
@@ -0,0 +1,33 @@
"""Tests for the per-board linker-script rule."""
import pytest
from esphome.components.esp8266 import _choose_ld_script
from esphome.components.esp8266.boards import BOARDS, board_ld_script
import esphome.config_validation as cv
from esphome.core import EsphomeError
def test_d1_wroom_02_keeps_its_shipped_layout() -> None:
"""The override must survive a BOARDS regeneration or key typo: the
2m.ld default moves _FS_end and the preferences sector on deployed
devices."""
assert board_ld_script(BOARDS["d1_wroom_02"]) == "eagle.flash.2m64.ld"
def test_default_boards_use_the_flash_size_layout() -> None:
assert board_ld_script(BOARDS["d1_mini"]) == "eagle.flash.4m.ld"
assert board_ld_script(BOARDS["esp01_1m"]) == "eagle.flash.1m.ld"
def test_choose_ld_script_paths() -> None:
"""Old cores get the size default, overriding boards hard-error there
(a substituted layout would wipe flash-backed state), modern cores
honor the override."""
assert _choose_ld_script("nodemcuv2", cv.Version(2, 3, 0)) is None
assert _choose_ld_script("nodemcuv2", cv.Version(2, 4, 2)) == "eagle.flash.4m.ld"
assert _choose_ld_script("d1_wroom_02", cv.Version(2, 7, 4)) == (
"eagle.flash.2m64.ld"
)
with pytest.raises(EsphomeError, match="cannot honor"):
_choose_ld_script("d1_wroom_02", cv.Version(2, 4, 2))
@@ -0,0 +1,145 @@
"""Tests for the linker-script surgery shared with the native toolchain."""
from __future__ import annotations
import importlib.util
from pathlib import Path
import sys
import pytest
from esphome.components.esp8266 import build_surgery
from esphome.components.esp8266.boards import BOARDS, ESP8266_BOARD_BUILD
from esphome.components.esp8266.build_surgery import (
RATETABLE_RULE,
apply_testing_memory_patches,
relocate_ratetable,
segment_length,
)
_COMMON_LD_SNIPPET = """\
.dport0.data : ALIGN(4)
{
_dport0_data_start = ABSOLUTE(.);
} >dport0_0_seg :dport0_0_phdr
.data : ALIGN(4)
{
_data_start = ABSOLUTE(.);
*(.data)
} >dram0_0_seg :dram0_0_phdr
"""
# Shaped like the real SDK flash ld scripts: no iram1_0_seg (that lives in
# the generated common ld only)
_FLASH_LD_SNIPPET = """\
MEMORY
{
dport0_0_seg : org = 0x3FF00000, len = 0x10
dram0_0_seg : org = 0x3FFE8000, len = 0x14000
irom0_0_seg : org = 0x40201010, len = 0xfeff0
}
"""
# Shaped like the preprocessed common ld: MMU_IRAM_SIZE expands with a ul
# suffix the patcher must leave in place
_COMMON_LD_MEMORY_SNIPPET = """\
MEMORY
{
iram1_0_seg : org = 0x40100000, len = 0x8000ul
}
"""
def test_relocate_ratetable_inserts_after_data_start() -> None:
patched = relocate_ratetable(_COMMON_LD_SNIPPET)
assert RATETABLE_RULE in patched
# Inserted after the .data section's anchor, not the .dport0.data one
# (whose closing brace bounds the decoy block)
assert RATETABLE_RULE not in patched[: patched.index("} >dport0_0_seg")]
assert patched.index(RATETABLE_RULE) < patched.index("*(.data)")
# Idempotent on an already-patched script
assert relocate_ratetable(patched) == patched
def test_relocate_ratetable_requires_anchor() -> None:
with pytest.raises(RuntimeError, match="_data_start"):
relocate_ratetable("SECTIONS { }")
def test_testing_memory_patches_enlarge_segments() -> None:
patched = apply_testing_memory_patches(
_FLASH_LD_SNIPPET, ("dram0_0_seg", "irom0_0_seg")
)
assert segment_length(patched, "dram0_0_seg") == 0x200000
assert segment_length(patched, "irom0_0_seg") == 0x2000000
# Untouched segments keep their sizes
assert segment_length(patched, "dport0_0_seg") == 0x10
def test_testing_memory_patches_keep_ul_suffix() -> None:
"""The common ld's preprocessed sizes carry a ul suffix; the patch must
replace only the hex digits, as testing_mode.py.script does."""
patched = apply_testing_memory_patches(_COMMON_LD_MEMORY_SNIPPET, ("iram1_0_seg",))
assert "len = 0x200000ul" in patched
assert segment_length(patched, "iram1_0_seg") == 0x200000
def test_segment_length_requires_whole_name() -> None:
"""A name must match its own line, never inside a longer segment name."""
assert segment_length(_FLASH_LD_SNIPPET, "ram0_0_seg") is None
def test_testing_memory_patches_unknown_segment_raises() -> None:
with pytest.raises(RuntimeError, match="Unknown testing-mode segment"):
apply_testing_memory_patches("MEMORY { }", ("bogus_seg",))
def test_segment_length() -> None:
assert segment_length(_FLASH_LD_SNIPPET, "irom0_0_seg") == 0xFEFF0
assert segment_length(_FLASH_LD_SNIPPET, "missing_seg") is None
def test_testing_memory_patches_missing_segment_raises() -> None:
"""A named segment the patch could not find raises instead of silently
keeping the real memory limits."""
with pytest.raises(RuntimeError, match="dram0_0_seg"):
apply_testing_memory_patches("MEMORY { }", ("dram0_0_seg",))
def test_board_build_covers_every_board() -> None:
"""Every supported board has native build metadata (the table may carry
extras that BOARDS does not expose)."""
assert set(BOARDS) <= set(ESP8266_BOARD_BUILD)
def test_surgery_fingerprint_is_stable_and_sensitive(tmp_path) -> None:
"""The properties the linker-script cache depends on: the fingerprint is
stable across calls and changes when the module's source changes."""
first = build_surgery.surgery_fingerprint()
assert first == build_surgery.surgery_fingerprint()
assert len(first) == 64
int(first, 16) # sha256 hex digest
# A modified copy of the module must fingerprint differently
copy = tmp_path / "build_surgery_variant.py"
copy.write_text(
Path(build_surgery.__file__).read_text(encoding="utf-8")
+ "\nEXTRA_BEHAVIORAL_INPUT = 1\n",
encoding="utf-8",
)
spec = importlib.util.spec_from_file_location("build_surgery_variant", copy)
variant = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = variant
try:
spec.loader.exec_module(variant)
assert variant.surgery_fingerprint() != first
finally:
del sys.modules[spec.name]
def test_testing_memory_patches_present_but_unselected_raises() -> None:
"""A known segment left off the caller's list must fail, not silently
keep its real memory limit."""
with pytest.raises(RuntimeError, match="not selected"):
apply_testing_memory_patches(_FLASH_LD_SNIPPET, ("dram0_0_seg",))
@@ -0,0 +1,75 @@
"""Tests for the file image platform's prefetch extraction."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from esphome.components.file import image as file_image
from esphome.external_files import RemoteFile
from esphome.loader import get_component, get_platform
def test_extract_mdi_shorthand(setup_core: Path) -> None:
ref = file_image._extract_file_ref("mdi:home")
assert ref is not None
assert ref.url == file_image.MDI_SOURCES["mdi"] + "home.svg"
assert ref.path.name == "home.svg"
assert ref.path.parent.name == "mdi"
def test_extract_web_url(setup_core: Path) -> None:
url = "https://example.com/img.png"
ref = file_image._extract_file_ref(url)
assert ref == RemoteFile(url, file_image.compute_local_image_path(url))
def test_extract_typed_dicts(setup_core: Path) -> None:
url = "https://example.com/img.png"
assert file_image._extract_file_ref({"source": "web", "url": url}) == RemoteFile(
url, file_image.compute_local_image_path(url)
)
ref = file_image._extract_file_ref({"source": "mdil", "icon": "home"})
assert ref is not None
assert ref.url == file_image.MDI_SOURCES["mdil"] + "home.svg"
def test_extract_skips_local_and_garbage(setup_core: Path) -> None:
assert file_image._extract_file_ref("images/local.png") is None
assert file_image._extract_file_ref("mdi:not a valid icon!") is None
assert file_image._extract_file_ref({"source": "local", "path": "x.png"}) is None
assert file_image._extract_file_ref(42) is None
assert file_image._extract_file_ref(None) is None
def test_prefetch_files_yields_remote_refs(setup_core: Path) -> None:
entries = [
{"file": "mdi:home"},
{"file": "images/local.png"},
{"file": "https://example.com/img.png"},
{"no_file_key": True},
]
[files] = list(file_image.PREFETCH_FILES(entries))
assert len(files) == 2
assert files[0].url.endswith("home.svg")
assert files[1].url == "https://example.com/img.png"
def test_extractor_matches_validator_path(setup_core: Path) -> None:
"""The path the validator downloads to equals the extractor's path."""
with patch(
"esphome.components.file.image.external_files.download_content"
) as mock_download:
file_image.validate_file_shorthand("mdi:home")
validated_path = mock_download.call_args[0][1]
assert validated_path == file_image._extract_file_ref("mdi:home").path
def test_hook_is_wired_to_both_animation_domains() -> None:
"""Both animation entry points expose the shared image hook."""
assert get_component("animation").prefetch_files is file_image.PREFETCH_FILES
assert (
get_platform("image", "animation").prefetch_files is file_image.PREFETCH_FILES
)
@@ -0,0 +1,229 @@
"""Tests for the font component's prefetch extraction."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from esphome import external_files
from esphome.components import font
import esphome.config_validation as cv
from esphome.external_files import RemoteFile
def _gspec(family: str, weight: int = 400, italic: bool = False) -> dict:
return {"family": family, "weight": weight, "italic": italic}
def test_extract_gfonts_shorthand_defaults(setup_core: Path) -> None:
spec = font._extract_remote_font("gfonts://Roboto")
assert spec is not None
assert spec[font.CONF_FAMILY] == "Roboto"
assert spec[font.CONF_WEIGHT] == 400
assert spec[font.CONF_ITALIC] is False
def test_extract_gfonts_shorthand_weight_variants(setup_core: Path) -> None:
assert font._extract_remote_font("gfonts://Roboto@bold")[font.CONF_WEIGHT] == 700
assert font._extract_remote_font("gfonts://Roboto@500")[font.CONF_WEIGHT] == 500
def test_extract_gfonts_normalizes_quoted_italic(setup_core: Path) -> None:
"""Boolean spellings the schema accepts are accepted by the extractor."""
spec = font._extract_remote_font(
{"type": "gfonts", "family": "Roboto", "italic": "true"}
)
assert spec is not None
assert spec[font.CONF_ITALIC] is True
assert (
font._extract_remote_font(
{"type": "gfonts", "family": "Roboto", "italic": "maybe"}
)
is None
)
def test_extract_typed_gfonts_dict(setup_core: Path) -> None:
spec = font._extract_remote_font(
{"type": "gfonts", "family": "Roboto", "weight": "medium", "italic": True}
)
assert spec is not None
assert spec[font.CONF_WEIGHT] == 500
assert spec[font.CONF_ITALIC] is True
def test_extract_web_font(setup_core: Path) -> None:
url = "https://example.com/font.ttf"
for value in (url, {"type": "web", "url": url}):
spec = font._extract_remote_font(value)
assert spec is not None
assert spec[font.CONF_URL] == url
def test_extract_skips_local_and_garbage(setup_core: Path) -> None:
assert font._extract_remote_font("fonts/local.ttf") is None
assert font._extract_remote_font({"type": "local", "path": "x.ttf"}) is None
assert (
font._extract_remote_font({"type": "gfonts", "family": "R", "weight": "no"})
is None
)
assert font._extract_remote_font(42) is None
def test_prefetch_yields_css_for_stale_gfont(setup_core: Path) -> None:
entries = [
{"file": "gfonts://Roboto"},
{"file": "fonts/local.ttf"},
{
"file": "https://example.com/font.ttf",
"extras": [{"file": "gfonts://Monocraft"}],
},
]
batches = list(font.PREFETCH_FILES(entries))
urls = [file.url for file in batches[0]]
assert font._gfonts_css_url(_gspec("Roboto")) in urls
assert font._gfonts_css_url(_gspec("Monocraft")) in urls
assert "https://example.com/font.ttf" in urls
assert len(batches[0]) == 3
def test_prefetch_skips_recent_ttf(setup_core: Path) -> None:
path = font._gfonts_ttf_path(_gspec("Roboto"))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"cached ttf")
batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
assert batches == [[], []]
def test_stage2_parses_cached_css(setup_core: Path) -> None:
css_path = font._gfonts_css_path(_gspec("Roboto"))
css_path.parent.mkdir(parents=True, exist_ok=True)
css_path.write_text(
"src: url(https://fonts.gstatic.com/roboto.ttf) format('truetype');"
)
# Stage two only trusts CSS confirmed fetched this run.
external_files._run_data().fresh_paths.add(css_path)
batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
assert batches[1] == [
RemoteFile(
"https://fonts.gstatic.com/roboto.ttf",
font._gfonts_ttf_path(_gspec("Roboto")),
)
]
def test_stage2_skips_missing_css(setup_core: Path) -> None:
batches = list(font.PREFETCH_FILES([{"file": "gfonts://NoCss"}]))
assert batches[1] == []
def test_prefetch_handles_bare_mapping_extras(setup_core: Path) -> None:
"""A bare-mapping extras value (valid raw config) is scanned."""
entries = [
{
"file": "fonts/local.ttf",
"extras": {"file": "gfonts://Roboto", "glyphs": "ABC"},
}
]
batches = list(font.PREFETCH_FILES(entries))
assert [file.url for file in batches[0]] == [font._gfonts_css_url(_gspec("Roboto"))]
def test_unparseable_gfonts_css_is_evicted(setup_core: Path) -> None:
"""A CSS body that fails to parse is removed from the cache."""
spec = {
"family": "Roboto",
"weight": 400,
"italic": False,
"refresh": font._REFRESH_VALIDATOR("0s"),
}
css_path = font._gfonts_css_path(spec)
with (
patch(
"esphome.components.font.external_files.download_content",
return_value=b"no truetype url here",
),
patch(
"esphome.components.font.external_files.is_fresh_this_run",
return_value=True,
),
pytest.raises(cv.Invalid, match="please report this"),
):
font.download_gfont(spec)
assert not css_path.exists()
with (
patch(
"esphome.components.font.external_files.download_content",
return_value=b"\xff\xfe\x00\x01binary",
),
patch(
"esphome.components.font.external_files.is_fresh_this_run",
return_value=True,
),
pytest.raises(cv.Invalid, match="not a text document"),
):
font.download_gfont(spec)
assert not css_path.exists()
def test_unrevalidated_gfonts_css_uses_cached_font(setup_core: Path) -> None:
"""A CSS body that could not be revalidated is not parsed for a ttf
URL; the cached font is used instead."""
spec = {
"family": "Roboto",
"weight": 400,
"italic": False,
"refresh": font._REFRESH_VALIDATOR("0s"),
}
ttf_path = font._gfonts_ttf_path(spec)
ttf_path.parent.mkdir(parents=True, exist_ok=True)
ttf_path.write_bytes(b"cached ttf")
cache = MagicMock()
with (
patch.object(font, "FONT_CACHE", cache),
patch(
"esphome.components.font.external_files.download_content",
return_value=b"stale css",
),
):
assert font.download_gfont(spec) is spec
cache.__setitem__.assert_called_once_with(spec, ttf_path)
def test_unrevalidated_gfonts_css_without_cached_font_errors(
setup_core: Path,
) -> None:
"""No verified CSS and no cached font is a clear error."""
spec = {
"family": "Roboto",
"weight": 500,
"italic": False,
"refresh": font._REFRESH_VALIDATOR("0s"),
}
with (
patch(
"esphome.components.font.external_files.download_content",
return_value=b"stale css",
),
pytest.raises(cv.Invalid, match="no cached font"),
):
font.download_gfont(spec)
def test_stage2_skips_css_not_fetched_this_run(setup_core: Path) -> None:
"""A leftover CSS from an earlier run is not trusted for stage two."""
css_path = font._gfonts_css_path(_gspec("Roboto"))
css_path.parent.mkdir(parents=True, exist_ok=True)
css_path.write_text(
"src: url(https://fonts.gstatic.com/rotated.ttf) format('truetype');"
)
batches = list(font.PREFETCH_FILES([{"file": "gfonts://Roboto"}]))
assert batches[1] == []
@@ -0,0 +1,35 @@
"""Tests for the gsl3670 touchscreen prefetch extraction."""
from __future__ import annotations
from pathlib import Path
from esphome.components.gsl3670 import touchscreen as gsl
from esphome.external_files import RemoteFile
def test_prefetch_explicit_url(setup_core: Path) -> None:
url = "https://example.com/fw.bin"
entries = [{"platform": "gsl3670", "firmware": {"url": url}}]
assert list(gsl.PREFETCH_FILES(entries)) == [
[RemoteFile(url, gsl._cache_path(url))]
]
def test_prefetch_model_default_firmware(setup_core: Path) -> None:
entries = [{"platform": "gsl3670", "model": "seeed-reterminal-d1001"}]
[files] = list(gsl.PREFETCH_FILES(entries))
assert len(files) == 1
assert (
files[0].url == gsl.MODELS["SEEED-RETERMINAL-D1001"][gsl.CONF_FIRMWARE]["url"]
)
assert files[0].path == gsl._cache_path(files[0].url)
def test_prefetch_skips_local_file_and_custom(setup_core: Path) -> None:
entries = [
{"platform": "gsl3670", "firmware": {"file": "fw.bin"}},
{"platform": "gsl3670", "model": "CUSTOM"},
{"platform": "gsl3670"},
]
assert list(gsl.PREFETCH_FILES(entries)) == [[]]
@@ -0,0 +1,144 @@
"""Tests for the shared addressable-strip channel order helpers."""
import logging
import pytest
from esphome.components.const import CONF_CHANNEL_COLORS, CONF_IS_WRGB
from esphome.components.light import (
channel_colors_struct,
migrate_channel_colors,
validate_channel_colors,
)
import esphome.config_validation as cv
from esphome.const import CONF_IS_RGBW, CONF_RGB_ORDER
from esphome.types import ConfigType
NO_WHITE = "light::ChannelColors::NO_WHITE"
@pytest.mark.parametrize(
("value", "expected"),
[
("RGB", "RGB"),
("grb", "GRB"),
("BRG", "BRG"),
("rgbw", "RGBW"),
("WRGB", "WRGB"),
("GWRB", "GWRB"),
],
)
def test_validate_channel_colors(value: str, expected: str) -> None:
assert validate_channel_colors(value) == expected
@pytest.mark.parametrize(
"value",
[
"RG", # missing a channel
"RGBB", # duplicate channel
"RRGB", # duplicate channel, correct length
"RGBWW", # two white channels
"RGBX", # unknown channel
"RGBWX", # unknown channel, correct length
"",
],
)
def test_validate_channel_colors_rejects_invalid(value: str) -> None:
with pytest.raises(cv.Invalid, match="is not a valid channel order"):
validate_channel_colors(value)
@pytest.mark.parametrize(
("value", "expected"),
[
("RGB", (0, 1, 2, NO_WHITE)),
("GRB", (1, 0, 2, NO_WHITE)),
("BRG", (1, 2, 0, NO_WHITE)),
("RGBW", (0, 1, 2, 3)),
("GRBW", (1, 0, 2, 3)),
("WRGB", (1, 2, 3, 0)),
("GWRB", (2, 0, 3, 1)),
],
)
def test_channel_colors_struct(value: str, expected: tuple[int, int, int, int]) -> None:
struct = channel_colors_struct(value)
assert str(struct.base) == "light::ChannelColors"
assert tuple(str(arg) for arg in struct.args.values()) == tuple(
str(field) for field in expected
)
def _migrate(config: ConfigType) -> ConfigType:
return migrate_channel_colors(removed_in="2027.3.0", component="test_strip")(config)
def test_migrate_passes_through_channel_colors() -> None:
config = {CONF_CHANNEL_COLORS: "GRBW"}
assert _migrate(config) == {CONF_CHANNEL_COLORS: "GRBW"}
@pytest.mark.parametrize(
("deprecated", "expected", "named"),
[
({}, "GRB", "'rgb_order' is"),
(
{CONF_IS_RGBW: False, CONF_IS_WRGB: False},
"GRB",
"'rgb_order', 'is_rgbw' and 'is_wrgb' are",
),
({CONF_IS_RGBW: True}, "GRBW", "'rgb_order' and 'is_rgbw' are"),
({CONF_IS_WRGB: True}, "WGRB", "'rgb_order' and 'is_wrgb' are"),
],
)
def test_migrate_folds_deprecated_keys(
deprecated: ConfigType,
expected: str,
named: str,
caplog: pytest.LogCaptureFixture,
) -> None:
config = {CONF_RGB_ORDER: "GRB", "num_leds": 1, **deprecated}
with caplog.at_level(logging.WARNING):
result = _migrate(config)
assert result == {CONF_CHANNEL_COLORS: expected, "num_leds": 1}
assert f"[test_strip] {named} deprecated" in caplog.text
assert f"'{CONF_CHANNEL_COLORS}: {expected}'" in caplog.text
assert "2027.3.0" in caplog.text
def test_migrate_does_not_mutate_input() -> None:
config = {CONF_RGB_ORDER: "GRB", CONF_IS_RGBW: True}
_migrate(config)
assert config == {CONF_RGB_ORDER: "GRB", CONF_IS_RGBW: True}
@pytest.mark.parametrize("deprecated", [CONF_RGB_ORDER, CONF_IS_RGBW, CONF_IS_WRGB])
def test_migrate_rejects_mixing_old_and_new(deprecated: str) -> None:
config = {CONF_CHANNEL_COLORS: "GRBW", deprecated: "GRB"}
with pytest.raises(cv.Invalid, match=f"cannot be combined with '{deprecated}'"):
_migrate(config)
def test_migrate_reports_every_conflicting_key() -> None:
config = {
CONF_CHANNEL_COLORS: "GRBW",
CONF_RGB_ORDER: "GRB",
CONF_IS_RGBW: True,
CONF_IS_WRGB: False,
}
with pytest.raises(
cv.Invalid, match="cannot be combined with 'rgb_order', 'is_rgbw' and 'is_wrgb'"
):
_migrate(config)
def test_migrate_requires_channel_colors() -> None:
with pytest.raises(cv.Invalid, match=f"'{CONF_CHANNEL_COLORS}' is required"):
_migrate({"num_leds": 1})
def test_migrate_rejects_is_rgbw_with_is_wrgb() -> None:
config = {CONF_RGB_ORDER: "GRB", CONF_IS_RGBW: True, CONF_IS_WRGB: True}
with pytest.raises(cv.Invalid, match="cannot both be enabled"):
_migrate(config)
@@ -53,9 +53,12 @@ def test_nonzero_indices_are_nonzero(gamma: float) -> None:
assert table[i] >= 1, f"gamma={gamma}, index {i}: got {table[i]}"
@pytest.mark.parametrize("gamma", [1.0, 2.0, 2.2, 2.8, 3.0])
@pytest.mark.parametrize("gamma", [1.0, 1.8, 2.0, 2.2, 2.8, 3.0, 4.0])
def test_table_monotonically_nondecreasing(gamma: float) -> None:
"""The gamma table must be monotonically non-decreasing."""
"""The gamma table must be monotonically non-decreasing.
gamma_table_reverse_search()'s binary search depends on this.
"""
table = generate_gamma_table(gamma)
for i in range(1, 256):
assert table[i] >= table[i - 1], (
@@ -115,3 +118,13 @@ def test_lut_output_monotonically_nondecreasing() -> None:
result = _simulate_gamma_correct_lut(table, value)
assert result >= prev, f"value={value}: result {result} < previous {prev}"
prev = result
def test_table_matches_raw_power_curve() -> None:
"""Check the gamma table against known good values for gamma=2.8."""
table = generate_gamma_table(2.8)
golden = {1: 1, 5: 1, 15: 24, 27: 122, 28: 135, 100: 4766, 200: 33193, 254: 64818}
for i, expected in golden.items():
assert table[i] == expected, (
f"index {i}: table[{i}]={table[i]} expected {expected}"
)
@@ -0,0 +1,206 @@
"""Tests for the LVGL table widget's C++ code generation."""
from __future__ import annotations
import pytest
from esphome.automation import ACTION_REGISTRY
from esphome.components.lvgl.defines import set_widgets_completed
from esphome.components.lvgl.lvcode import LvContext
from esphome.components.lvgl.schemas import container_schema
from esphome.components.lvgl.trigger import generate_triggers
from esphome.components.lvgl.widgets import Widget, widget_to_code
from esphome.components.lvgl.widgets.table import table_spec
from esphome.const import (
CONF_AUTOMATION_ID,
CONF_ON_VALUE,
CONF_THEN,
CONF_TRIGGER_ID,
CONF_TYPE_ID,
)
from esphome.core import CORE, ID
from esphome.cpp_generator import MockObj, TemplateArguments
from esphome.yaml_util import make_data_base
async def _create_table(raw_config: dict) -> Widget:
"""Validate `raw_config` as a table widget and generate its creation code."""
config = container_schema(table_spec)(raw_config)
parent = MockObj("parent_obj")
async with LvContext():
return await widget_to_code(config, table_spec, parent)
def _statements() -> list[str]:
return [str(s) for s in CORE.main_statements]
@pytest.mark.asyncio
async def test_create_table_sets_row_and_column_count(setup_core) -> None:
await _create_table(
{"id": "table_counts", "rows": [["Name", "Value"], ["Temp", "22.5"]]}
)
statements = _statements()
assert any("lv_table_set_row_count(table_counts->obj, 2)" in s for s in statements)
assert any(
"lv_table_set_column_count(table_counts->obj, 2)" in s for s in statements
)
@pytest.mark.asyncio
async def test_create_table_writes_cell_values(setup_core) -> None:
await _create_table({"id": "table_cells", "rows": [["Name", "Value"]]})
statements = _statements()
assert any(
'lv_table_set_cell_value(table_cells->obj, 0, 0, "Name")' in s
for s in statements
)
assert any(
'lv_table_set_cell_value(table_cells->obj, 0, 1, "Value")' in s
for s in statements
)
@pytest.mark.asyncio
async def test_create_table_sets_cell_control_flags(setup_core) -> None:
await _create_table(
{
"id": "table_ctrl",
"rows": [
{
"cells": [
{"text": "wide", "merge_right": True},
{"text": "cropped", "text_crop": True},
]
}
],
}
)
statements = _statements()
assert any(
"lv_table_set_cell_ctrl(table_ctrl->obj, 0, 0, LV_TABLE_CELL_CTRL_MERGE_RIGHT)"
in s
for s in statements
)
assert any(
"lv_table_set_cell_ctrl(table_ctrl->obj, 0, 1, LV_TABLE_CELL_CTRL_TEXT_CROP)"
in s
for s in statements
)
# text_crop omitted for cell 0: no clear_cell_ctrl() should be emitted.
assert not any(
"table_ctrl->obj, 0, 0, LV_TABLE_CELL_CTRL_TEXT_CROP" in s for s in statements
)
@pytest.mark.asyncio
async def test_pixel_column_width_calls_lvgl_directly(setup_core) -> None:
await _create_table({"id": "table_px", "columns": [{"width": 96}]})
statements = _statements()
assert any(
"lv_table_set_column_width(table_px->obj, 0, 96)" in s for s in statements
)
@pytest.mark.asyncio
async def test_percent_column_width_uses_the_dynamic_helper(setup_core) -> None:
"""Regression test: lv_table_set_column_width() only accepts a literal
pixel count, so a percentage width must not be passed to it directly -
it has to go through the LvTableType helper that recomputes it at
runtime from the table's actual content width.
"""
await _create_table({"id": "table_pct", "columns": [{"width": "40%"}]})
statements = _statements()
assert any("table_pct->init_column_pct(1)" in s for s in statements)
assert any("table_pct->add_column_width_pct(0, 40)" in s for s in statements)
assert not any(
"lv_table_set_column_width(table_pct->obj, 0" in s for s in statements
)
@pytest.mark.asyncio
async def test_selected_cell_with_both_indices(setup_core) -> None:
await _create_table(
{"id": "table_sel_both", "selected_row": 1, "selected_column": 2}
)
statements = _statements()
assert any(
"lv_table_set_selected_cell(table_sel_both->obj, 1, 2)" in s for s in statements
)
@pytest.mark.asyncio
async def test_selected_cell_with_only_row_selects_whole_row(setup_core) -> None:
await _create_table({"id": "table_sel_row", "selected_row": 1})
statements = _statements()
assert any(
"lv_table_set_selected_cell(table_sel_row->obj, 1, LV_TABLE_CELL_NONE)" in s
for s in statements
)
@pytest.mark.asyncio
async def test_selected_cell_omitted_entirely_when_not_configured(
setup_core,
) -> None:
await _create_table({"id": "table_no_selection", "rows": [["a"]]})
statements = _statements()
assert not any("lv_table_set_selected_cell" in s for s in statements)
@pytest.mark.asyncio
async def test_cell_update_action_writes_only_the_given_fields(setup_core) -> None:
await _create_table({"id": "table_update", "rows": [["a", "b"], ["c", "d"]]})
set_widgets_completed(True)
# Only inspect statements emitted by the action below, not by creation.
before = len(_statements())
entry = ACTION_REGISTRY["lvgl.table.cell.update"]
config = entry.schema(
{"id": "table_update", "row": 1, "column": 1, "text": "new value"}
)
action_id = ID("test_cell_update_action", is_declaration=True, type=entry.type_id)
await entry.coroutine_fun(config, action_id, TemplateArguments(), [])
statements = _statements()[before:]
assert any(
'lv_table_set_cell_value(table_update->obj, 1, 1, "new value")' in s
for s in statements
)
# Neither control flag was specified, so neither call should be emitted.
assert not any("LV_TABLE_CELL_CTRL" in s for s in statements)
@pytest.mark.asyncio
async def test_on_value_registers_a_value_changed_event_callback(setup_core) -> None:
config = container_schema(table_spec)(
{
"id": "table_on_value",
"rows": [["a"]],
"on_value": [
{"lambda": make_data_base("id(table_on_value).get_selected_row();")}
],
}
)
# Auto-generated IDs (trigger/automation/action) are normally resolved to
# unique names by esphome's full config pass before code generation; do
# that by hand here since this test only exercises the widget/trigger
# codegen slice in isolation.
automation_conf = config[CONF_ON_VALUE][0]
automation_conf[CONF_TRIGGER_ID].resolve([])
automation_conf[CONF_AUTOMATION_ID].resolve([])
automation_conf[CONF_THEN][0][CONF_TYPE_ID].resolve([])
parent = MockObj("parent_obj")
async with LvContext():
await widget_to_code(config, table_spec, parent)
set_widgets_completed(True)
await generate_triggers()
statements = _statements()
assert any(
"table_on_value->obj" in s
and "add_event_cb" in s
and "LV_EVENT_VALUE_CHANGED" in s
for s in statements
)
@@ -0,0 +1,142 @@
"""Tests for the LVGL table widget's configuration validation."""
from __future__ import annotations
import pytest
from esphome import config_validation as cv
from esphome.automation import ACTION_REGISTRY
from esphome.components.lvgl.widgets.table import (
CONF_MERGE_RIGHT,
CONF_TEXT_CROP,
TABLE_SCHEMA,
)
def test_minimal_config_is_valid() -> None:
assert TABLE_SCHEMA({}) == {}
def test_row_shorthand_expands_to_plain_cells() -> None:
config = TABLE_SCHEMA({"rows": [["Name", "Value"]]})
[row] = config["rows"]
assert row["cells"] == [{"text": "Name"}, {"text": "Value"}]
def test_row_dict_form_with_cell_overrides() -> None:
config = TABLE_SCHEMA(
{
"rows": [
{
"cells": [
"Temp",
{"text": "22.5", "text_crop": True, "merge_right": True},
]
}
]
}
)
[row] = config["rows"]
assert row["cells"][0] == {"text": "Temp"}
assert row["cells"][1] == {
"text": "22.5",
"merge_right": True,
"text_crop": True,
}
def test_row_count_defaults_are_not_injected_by_the_schema() -> None:
# Inference of row/column counts from `rows` happens at code generation
# time, not during validation - the schema should leave them unset.
config = TABLE_SCHEMA({"rows": [["a", "b"], ["c"]]})
assert "row_count" not in config
assert "column_count" not in config
def test_explicit_row_and_column_count_are_kept() -> None:
config = TABLE_SCHEMA({"row_count": 5, "column_count": 3})
assert config["row_count"] == 5
assert config["column_count"] == 3
def test_row_count_too_small_for_given_rows_raises() -> None:
with pytest.raises(cv.Invalid, match="row_count"):
TABLE_SCHEMA({"rows": [["a"], ["b"], ["c"]], "row_count": 2})
def test_column_count_too_small_for_given_cells_raises() -> None:
with pytest.raises(cv.Invalid, match="column_count"):
TABLE_SCHEMA({"rows": [["a", "b", "c"]], "column_count": 2})
def test_columns_list_longer_than_column_count_raises() -> None:
with pytest.raises(cv.Invalid, match="columns"):
TABLE_SCHEMA(
{
"column_count": 1,
"columns": [{"width": 10}, {"width": 20}],
}
)
def test_columns_list_matching_inferred_column_count_is_valid() -> None:
config = TABLE_SCHEMA(
{
"rows": [["a", "b"]],
"columns": [{"width": 10}, {"width": 20}],
}
)
assert [c["width"] for c in config["columns"]] == [10, 20]
@pytest.mark.parametrize(
("width", "expected"),
[
(100, 100),
("50%", 0.5),
("32px", 32),
],
)
def test_column_width_accepts_pixels_and_percent(width, expected) -> None:
config = TABLE_SCHEMA({"columns": [{"width": width}]})
assert config["columns"][0]["width"] == expected
def test_columns_percent_widths_summing_over_100_percent_raises() -> None:
with pytest.raises(cv.Invalid, match="columns"):
TABLE_SCHEMA({"columns": [{"width": "60%"}, {"width": "50%"}]})
def test_columns_percent_widths_summing_to_100_percent_is_valid() -> None:
config = TABLE_SCHEMA({"columns": [{"width": "60%"}, {"width": "40%"}]})
assert [c["width"] for c in config["columns"]] == [0.6, 0.4]
def test_columns_mixed_pixel_and_percent_widths_ignore_pixels_in_the_total() -> None:
# Pixel widths aren't part of the percentage budget, so they shouldn't
# count towards the 100% limit.
config = TABLE_SCHEMA(
{"columns": [{"width": 200}, {"width": "80%"}, {"width": "20%"}]}
)
assert [c["width"] for c in config["columns"]] == [200, 0.8, 0.2]
def test_selected_row_and_selected_column_are_independently_optional() -> None:
config = TABLE_SCHEMA({"selected_row": 1})
assert config["selected_row"] == 1
assert "selected_column" not in config
def test_cell_update_action_requires_at_least_one_field() -> None:
entry = ACTION_REGISTRY["lvgl.table.cell.update"]
with pytest.raises(cv.Invalid):
entry.schema({"id": "some_table", "row": 0, "column": 0})
def test_cell_update_action_accepts_a_single_field() -> None:
entry = ACTION_REGISTRY["lvgl.table.cell.update"]
config = entry.schema(
{"id": "some_table", "row": 0, "column": 0, "merge_right": True}
)
assert config[CONF_MERGE_RIGHT] is True
assert CONF_TEXT_CROP not in config
@@ -16,6 +16,7 @@ from esphome.const import (
CONF_TYPE,
CONF_URL,
)
from esphome.external_files import RemoteFile
@pytest.fixture
@@ -114,12 +115,16 @@ def test_download_http_models_batches_manifests_then_models(
assert mock_download_content_many.call_count == 2
manifest_items = list(mock_download_content_many.call_args_list[0].args[0])
assert manifest_items == [
(f"https://example.com/models/{name}.json", paths[name] / "manifest.json")
RemoteFile(
f"https://example.com/models/{name}.json", paths[name] / "manifest.json"
)
for name in names
]
model_items = list(mock_download_content_many.call_args_list[1].args[0])
assert model_items == [
(f"https://example.com/models/{name}.tflite", paths[name] / f"{name}.tflite")
RemoteFile(
f"https://example.com/models/{name}.tflite", paths[name] / f"{name}.tflite"
)
for name in names
]
@@ -1,239 +0,0 @@
"""Tests for the MQTT object_id conflict filter.
MQTT still builds default topics and discovery topics from the sanitized
object_id, so entity names that only differ in characters lost during
sanitizing conflict there; _topics_conflict() exempts entities that never
use an object_id-derived topic. See https://github.com/esphome/backlog/issues/85
"""
from pathlib import Path
import pytest
from esphome.components.mqtt import (
_COMMAND_TOPIC_PLATFORMS,
_SUB_TOPIC_PLATFORMS,
_topics_conflict,
)
from esphome.config_validation import Invalid
from esphome.const import (
CONF_COMMAND_TOPIC,
CONF_DISCOVERY,
CONF_NAME,
CONF_STATE_TOPIC,
CONF_TOPIC_PREFIX,
)
from esphome.core import CORE
from esphome.core.entity_helpers import (
entity_duplicate_validator,
validate_no_object_id_conflicts,
)
COMPONENTS_DIR = Path(__file__).parents[4] / "esphome" / "components"
REASON = "mqtt builds default topics from the entity object_id"
# MQTT infrastructure sources, not entity components
_NON_ENTITY_MQTT_SOURCES = {"mqtt_client", "mqtt_component"}
# The date, time and datetime MQTT components all belong to the datetime platform
_DATETIME_STEMS = {"date", "time", "datetime"}
def test_command_topic_platforms_in_sync() -> None:
"""Verify _COMMAND_TOPIC_PLATFORMS matches the MQTT components that subscribe.
Drift silently reintroduces shared subscribe topics, so this derives the set
from the C++ components that actually call subscribe(); that also catches
platforms like text that subscribe a command topic without exposing a
command_topic key in their schema.
"""
expected: set[str] = set()
for path in (COMPONENTS_DIR / "mqtt").glob("mqtt_*.cpp"):
if path.stem in _NON_ENTITY_MQTT_SOURCES:
continue
if "this->subscribe" not in path.read_text(encoding="utf-8"):
continue
stem = path.stem.removeprefix("mqtt_")
expected.add("datetime" if stem in _DATETIME_STEMS else stem)
assert expected == _COMMAND_TOPIC_PLATFORMS
def test_sub_topic_platforms_in_sync() -> None:
"""Verify _SUB_TOPIC_PLATFORMS matches the MQTT components with sub-topics.
Platforms whose MQTT headers use MQTT_COMPONENT_CUSTOM_TOPIC derive extra
topics such as position/command from the object_id.
"""
expected = {
path.stem.removeprefix("mqtt_")
for path in (COMPONENTS_DIR / "mqtt").glob("mqtt_*.h")
if path.stem != "mqtt_component"
and "MQTT_COMPONENT_CUSTOM_TOPIC" in path.read_text(encoding="utf-8")
}
assert expected == _SUB_TOPIC_PLATFORMS
def test_conflict_filter_exempts_custom_topics() -> None:
"""Test that custom state topics with discovery off avoid the conflict."""
validator = entity_duplicate_validator("sensor")
# Both entities have custom state topics and discovery disabled per entity,
# so no object_id-derived MQTT topic is used
validator(
{
CONF_NAME: "Датчик открытия",
CONF_STATE_TOPIC: "custom/topic/a",
CONF_DISCOVERY: False,
}
)
validator(
{
CONF_NAME: "Датчик закрытия",
CONF_STATE_TOPIC: "custom/topic/b",
CONF_DISCOVERY: False,
}
)
component_validator = validate_no_object_id_conflicts(
REASON, conflict_filter=_topics_conflict
)
config: dict = {CONF_DISCOVERY: True, CONF_TOPIC_PREFIX: "test-device"}
assert component_validator(config) is config
# Without the filter the same conflicts are fatal
with pytest.raises(Invalid, match=r"mqtt builds default topics"):
validate_no_object_id_conflicts(REASON)({})
def test_conflict_on_default_command_topic() -> None:
"""Test that commandable platforms conflict through their default command topic.
Custom state topics with discovery off are not enough for platforms that also
subscribe to an object_id-derived command topic.
"""
validator = entity_duplicate_validator("switch")
validator(
{
CONF_NAME: "Датчик открытия",
CONF_STATE_TOPIC: "custom/topic/a",
CONF_DISCOVERY: False,
}
)
validator(
{
CONF_NAME: "Датчик закрытия",
CONF_STATE_TOPIC: "custom/topic/b",
CONF_DISCOVERY: False,
}
)
component_validator = validate_no_object_id_conflicts(
REASON, conflict_filter=_topics_conflict
)
mqtt_config: dict = {CONF_DISCOVERY: True, CONF_TOPIC_PREFIX: "test-device"}
# Both switches share the default command topic: rejected
with pytest.raises(Invalid, match=r"mqtt builds default topics"):
component_validator(mqtt_config)
# With custom command topics as well, nothing derives from the object_id
CORE.reset()
validator = entity_duplicate_validator("switch")
validator(
{
CONF_NAME: "Датчик открытия",
CONF_STATE_TOPIC: "custom/topic/a",
CONF_COMMAND_TOPIC: "custom/cmd/a",
CONF_DISCOVERY: False,
}
)
validator(
{
CONF_NAME: "Датчик закрытия",
CONF_STATE_TOPIC: "custom/topic/b",
CONF_COMMAND_TOPIC: "custom/cmd/b",
CONF_DISCOVERY: False,
}
)
assert component_validator(mqtt_config) is mqtt_config
def test_conflict_on_sub_topic_platforms() -> None:
"""Test that platforms with extra object_id sub-topics always conflict.
Covers derive topics like position/command from the object_id through their
own config keys, so custom state and command topics cannot exempt them.
"""
validator = entity_duplicate_validator("cover")
validator(
{
CONF_NAME: "Датчик открытия",
CONF_STATE_TOPIC: "custom/topic/a",
CONF_COMMAND_TOPIC: "custom/cmd/a",
CONF_DISCOVERY: False,
}
)
validator(
{
CONF_NAME: "Датчик закрытия",
CONF_STATE_TOPIC: "custom/topic/b",
CONF_COMMAND_TOPIC: "custom/cmd/b",
CONF_DISCOVERY: False,
}
)
component_validator = validate_no_object_id_conflicts(
REASON, conflict_filter=_topics_conflict
)
with pytest.raises(Invalid, match=r"mqtt builds default topics"):
component_validator({CONF_DISCOVERY: True, CONF_TOPIC_PREFIX: "test-device"})
def test_no_conflict_on_disjoint_default_topics() -> None:
"""Test that entities whose default topics are disjoint do not conflict.
One entity uses only the default command topic and the other only the default
state topic, so they never share a topic.
"""
validator = entity_duplicate_validator("switch")
validator(
{
CONF_NAME: "Датчик открытия",
CONF_STATE_TOPIC: "custom/topic/a",
CONF_DISCOVERY: False,
}
)
validator(
{
CONF_NAME: "Датчик закрытия",
CONF_COMMAND_TOPIC: "custom/cmd/b",
CONF_DISCOVERY: False,
}
)
component_validator = validate_no_object_id_conflicts(
REASON, conflict_filter=_topics_conflict
)
config: dict = {CONF_DISCOVERY: True, CONF_TOPIC_PREFIX: "test-device"}
assert component_validator(config) is config
def test_no_conflict_on_empty_topic_prefix() -> None:
"""Test that an empty topic_prefix disables the default topic conflict.
With topic_prefix set to null no default topics exist at runtime, so entities
without custom state topics cannot conflict; only discovery still matters.
"""
validator = entity_duplicate_validator("sensor")
validator({CONF_NAME: "Датчик открытия"})
validator({CONF_NAME: "Датчик закрытия"})
component_validator = validate_no_object_id_conflicts(
REASON, conflict_filter=_topics_conflict
)
# No default topics and no discovery: valid
config: dict = {CONF_DISCOVERY: False, CONF_TOPIC_PREFIX: ""}
assert component_validator(config) is config
# Discovery still uses object_id-derived config topics: rejected
with pytest.raises(Invalid, match=r"mqtt builds default topics"):
component_validator({CONF_DISCOVERY: True, CONF_TOPIC_PREFIX: ""})
@@ -0,0 +1,154 @@
"""Tests for the shelly_dimmer firmware download and prefetch extraction."""
from __future__ import annotations
import hashlib
from pathlib import Path
from unittest.mock import patch
import pytest
from esphome import external_files
from esphome.components.shelly_dimmer import light as shd
from esphome.config_validation import Invalid
from esphome.external_files import RemoteFile
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def test_prefetch_known_version(setup_core: Path) -> None:
entries = [{"firmware": {"version": "51.6", "update": True}}]
stages = list(shd.PREFETCH_FILES(entries))
url, sha = shd.KNOWN_FIRMWARE["51.6"]
assert stages == [[RemoteFile(url, shd._firmware_cache_path(sha))]]
def test_prefetch_normalizes_update_like_the_schema(setup_core: Path) -> None:
"""Quoted booleans behave as the schema will normalize them."""
url, sha = shd.KNOWN_FIRMWARE["51.6"]
off = [{"firmware": {"version": "51.6", "update": "false"}}]
assert list(shd.PREFETCH_FILES(off)) == [[]]
on = [{"firmware": {"version": "51.6", "update": "true"}}]
assert list(shd.PREFETCH_FILES(on)) == [
[RemoteFile(url, shd._firmware_cache_path(sha))]
]
def test_prefetch_rejects_malformed_sha256(setup_core: Path) -> None:
"""A raw sha256 that is not a hash never becomes a path component."""
entries = [
{
"firmware": {
"url": "https://example.com/fw.bin",
"sha256": "/tmp/payload",
"update": True,
}
}
]
assert list(shd.PREFETCH_FILES(entries)) == [[]]
def test_prefetch_skips_content_addressed_blob_on_disk(setup_core: Path) -> None:
"""A sha-keyed cache file needs no revalidation; get_firmware hashes it."""
url, sha = shd.KNOWN_FIRMWARE["51.6"]
shd._firmware_cache_path(sha).write_bytes(b"pinned firmware")
entries = [{"firmware": {"version": "51.6", "update": True}}]
assert list(shd.PREFETCH_FILES(entries)) == [[]]
def test_prefetch_explicit_url_without_sha(setup_core: Path) -> None:
url = "https://example.com/fw.bin"
entries = [{"firmware": {"url": url, "update": True}}]
stages = list(shd.PREFETCH_FILES(entries))
key = external_files.url_cache_key(url)
# No sha means the bytes cannot be verified, so the prefetch itself
# must carry the validator's strict no-stale policy.
assert stages == [
[RemoteFile(url, shd._firmware_cache_path(key), allow_stale=False)]
]
def test_prefetch_skips_no_update(setup_core: Path) -> None:
entries = [
{"firmware": {"version": "51.6"}},
{"firmware": "51.6"},
{"firmware": {"version": "0.0", "update": True}},
{},
]
assert list(shd.PREFETCH_FILES(entries)) == [[]]
def test_get_firmware_rejects_corrupted_cache(setup_core: Path) -> None:
"""A cached blob failing its hash check is discarded and re-downloaded."""
good = b"good firmware"
expected = _sha(good)
path = shd._firmware_cache_path(expected)
path.write_bytes(b"corrupted blob")
with patch(
"esphome.components.shelly_dimmer.light.external_files.download_content",
return_value=good,
) as mock_download:
result = shd.get_firmware(
{
"update": True,
"url": "https://example.com/fw.bin",
"sha256": expected,
}
)
mock_download.assert_called_once()
assert result == [int(b) for b in good]
def test_get_firmware_trusts_valid_cache(setup_core: Path) -> None:
"""A cached blob passing its hash check is used with zero network."""
good = b"good firmware"
expected = _sha(good)
shd._firmware_cache_path(expected).write_bytes(good)
with patch(
"esphome.components.shelly_dimmer.light.external_files.download_content"
) as mock_download:
result = shd.get_firmware(
{
"update": True,
"url": "https://example.com/fw.bin",
"sha256": expected,
}
)
mock_download.assert_not_called()
assert result == [int(b) for b in good]
def test_get_firmware_hash_mismatch_raises_and_uncaches(setup_core: Path) -> None:
"""A fresh download failing its hash check raises and is not cached."""
expected = _sha(b"expected firmware")
path = shd._firmware_cache_path(expected)
with (
patch(
"esphome.components.shelly_dimmer.light.external_files.download_content",
return_value=b"wrong firmware",
),
pytest.raises(Invalid, match="Hash mismatch"),
):
shd.get_firmware(
{"update": True, "url": "https://example.com/fw.bin", "sha256": expected}
)
assert not path.exists()
def test_get_firmware_without_sha_rejects_stale(setup_core: Path) -> None:
"""The unverifiable no-hash branch must not accept a stale copy."""
with patch(
"esphome.components.shelly_dimmer.light.external_files.download_content",
return_value=b"fw",
) as mock_download:
shd.get_firmware({"update": True, "url": "https://example.com/fw.bin"})
assert mock_download.call_args.kwargs["allow_stale"] is False
@@ -1,57 +0,0 @@
import pytest
from esphome.components.esp32_rmt_led_strip.light import (
CONF_IS_WRGB,
CONF_RGBW_ORDER,
_split_rgbw_order,
_validate_rgbw_order,
_validate_rgbw_order_exclusivity,
)
import esphome.config_validation as cv
from esphome.const import CONF_IS_RGBW
def test_validate_rgbw_order() -> None:
assert _validate_rgbw_order("rwgb") == "RWGB"
@pytest.mark.parametrize("rgbw_order", ["RGB", "RRGB", "RGBWW"])
def test_validate_rgbw_order_rejects_invalid_order(rgbw_order: str) -> None:
with pytest.raises(cv.Invalid, match="permutation of RGBW"):
_validate_rgbw_order(rgbw_order)
@pytest.mark.parametrize(
("rgbw_order", "expected"),
[
("WRGB", ("RGB", 0)),
("RWGB", ("RGB", 1)),
("GWRB", ("GRB", 1)),
("RGBW", ("RGB", 3)),
],
)
def test_split_rgbw_order(rgbw_order: str, expected: tuple[str, int]) -> None:
assert _split_rgbw_order(rgbw_order) == expected
@pytest.mark.parametrize("conflict", [CONF_IS_RGBW, CONF_IS_WRGB])
def test_rgbw_order_is_mutually_exclusive(conflict: str) -> None:
with pytest.raises(cv.Invalid, match="cannot be used with"):
_validate_rgbw_order_exclusivity(
{
CONF_RGBW_ORDER: "RGBW",
CONF_IS_RGBW: conflict == CONF_IS_RGBW,
CONF_IS_WRGB: conflict == CONF_IS_WRGB,
}
)
@pytest.mark.parametrize("legacy_option", [CONF_IS_RGBW, CONF_IS_WRGB])
def test_rgbw_order_allows_disabled_legacy_options(legacy_option: str) -> None:
config = {
CONF_RGBW_ORDER: "RGBW",
CONF_IS_RGBW: False,
CONF_IS_WRGB: False,
}
config[legacy_option] = False
assert _validate_rgbw_order_exclusivity(config) is config
+97 -10
View File
@@ -13,25 +13,24 @@ itself (Python imports, YAML key rename, deprecation warning) is covered
by the framework tests under ``tests/unit_tests/``.
"""
from pathlib import Path
import re
from esphome.components import rp2
def test_board_id_has_wifi_for_known_wifi_board() -> None:
"""``rpipicow`` is the canonical Pico W → True."""
from esphome.components import rp2
assert rp2.board_id_has_wifi("rpipicow") is True
def test_board_id_has_wifi_for_known_non_wifi_board() -> None:
"""Plain ``rpipico`` has no CYW43 → False."""
from esphome.components import rp2
assert rp2.board_id_has_wifi("rpipico") is False
def test_board_id_has_wifi_for_rp2350_w_variant() -> None:
"""``rpipico2w`` is the RP2350 Pico 2 W → True."""
from esphome.components import rp2
assert rp2.board_id_has_wifi("rpipico2w") is True
@@ -43,8 +42,6 @@ def test_board_id_has_wifi_for_unknown_board_returns_true() -> None:
block and any genuinely-unsupported config trips the existing
"no CYW43" guard at compile time.
"""
from esphome.components import rp2
assert rp2.board_id_has_wifi("not-a-real-board-id") is True
@@ -55,8 +52,6 @@ def test_rp2_declares_rp2040_as_alias() -> None:
opts in via ``ALIASES``; without this declaration the rename
framework wouldn't route legacy configs.
"""
from esphome.components import rp2
assert "rp2040" in rp2.ALIASES
assert rp2.ALIAS_REMOVAL_VERSION == "2027.7.0"
@@ -93,3 +88,95 @@ def test_rp2040_submodule_imports_resolve_to_rp2_submodules() -> None:
assert rp2040_boards is rp2_boards
assert rp2040_generate is rp2_generate
def test_lwip_segment_pool_exceeds_per_pcb_queue() -> None:
"""The segment pool is global while the send queue is per-PCB.
lwIP's sanity check only requires ``MEMP_NUM_TCP_SEG >= TCP_SND_QUEUELEN``,
which is the floor for a *single* connection: at equality one busy PCB can
drain the pool for every other PCB. Dropping back to that floor would
rebuild the starvation this sizing exists to prevent, and nothing in the
build would complain.
"""
assert rp2.LWIP_MEMP_NUM_TCP_SEG >= 2 * rp2.LWIP_TCP_SND_QUEUELEN
def test_lwip_mem_size_keeps_mem_size_t_narrow() -> None:
"""``lwip/mem.h`` widens ``mem_size_t`` to ``u32_t`` on
``MEM_SIZE > 64000L``, growing the header on every heap block. Raising the
heap past that bound is a real option, but it should be a deliberate one
rather than a side effect of tuning.
"""
assert rp2.LWIP_MEM_SIZE <= 64000
def test_lwip_mem_size_holds_the_concurrent_senders_it_claims() -> None:
"""Pin the floor as well as the ceiling.
The ceiling above is satisfied by arduino-pico's own 16 KB, which is the
value this change exists to move off, so on its own it would let a revert
through. Derive the floor from the sizing comment on the constant: with
TCP_OVERSIZE at TCP_MSS every queued segment takes a full MSS-sized block
(pbuf header + PBUF_TRANSPORT offset + 1460 + heap block header, ~1.5 KB),
a PCB at a full 4xMSS TCP_SND_BUF holds four of them, and api's
max_connections on rp2 is 4. Room for three concurrent senders is the
minimum that makes the change worth making; 16 KB does not reach it.
"""
segments_per_full_send_buf = 4
bytes_per_mss_block = 1536
concurrent_senders = 3
assert (
concurrent_senders * segments_per_full_send_buf * bytes_per_mss_block
<= rp2.LWIP_MEM_SIZE
)
def test_lwip_defines_carry_the_sizing_into_the_header() -> None:
"""The constants above only matter if they reach the generated header.
``build_lwip_defines()`` is what feeds lwipopts.h.jinja, so assert on it
rather than on the constants alone: dropping a key here would silently
fall back to arduino-pico's own value while every other assertion in this
file stayed green.
"""
defines = rp2.build_lwip_defines(tcp_sockets=8, udp_sockets=6, listening_tcp=2)
assert defines["MEM_SIZE"] == str(rp2.LWIP_MEM_SIZE)
assert defines["MEMP_NUM_TCP_SEG"] == str(rp2.LWIP_MEMP_NUM_TCP_SEG)
assert defines["TCP_SND_QUEUELEN"] == str(rp2.LWIP_TCP_SND_QUEUELEN)
# Socket-derived counts pass through untouched.
assert defines["MEMP_NUM_TCP_PCB"] == "8"
assert defines["MEMP_NUM_UDP_PCB"] == "6"
assert defines["MEMP_NUM_TCP_PCB_LISTEN"] == "2"
def test_lwipopts_template_renders_every_sizing_value() -> None:
"""Render the template the way _generate_lwipopts_h() does and check the
header that actually ships.
Covers both directions. A ``#define`` block deleted from the template
leaves the value at arduino-pico's own, which for MEM_SIZE is the 16 KB
heap this change exists to move off, and the loop below catches that. A
placeholder with no dict key would otherwise render empty and emit a bare
``#define FOO``; StrictUndefined turns that into an error instead.
Matching on text also survives a filter or conditional appearing in the
template later, which a placeholder regex would not.
"""
from jinja2 import Environment, StrictUndefined
defines = rp2.build_lwip_defines(tcp_sockets=8, udp_sockets=6, listening_tcp=2)
template_text = (Path(rp2.__file__).parent / "lwipopts.h.jinja").read_text(
encoding="utf-8"
)
rendered = (
Environment(keep_trailing_newline=True, undefined=StrictUndefined)
.from_string(template_text)
.render(**defines)
)
for name, value in defines.items():
assert re.search(
rf"^#define {re.escape(name)} +{re.escape(value)}$", rendered, re.MULTILINE
), f"{name} did not reach the generated header as {value!r}"