Merge branch 'dev' into store-yaml-firmware

This commit is contained in:
J. Nick Koston
2026-05-26 09:17:39 -05:00
committed by GitHub
243 changed files with 6371 additions and 1686 deletions
@@ -2,6 +2,8 @@
from unittest.mock import patch
import pytest
from esphome.components.display import (
DisplayMetaData,
add_metadata,
@@ -74,8 +76,5 @@ def test_add_metadata_overwrites_existing():
def test_metadata_is_frozen():
"""Test that DisplayMetaData instances are immutable (frozen dataclass)."""
meta = DisplayMetaData(320, 240, True, False)
try:
with pytest.raises(AttributeError):
meta.width = 640
assert False, "Expected FrozenInstanceError"
except AttributeError:
pass
@@ -9,13 +9,17 @@ import pytest
from esphome import config_validation as cv
from esphome.components.light import (
EffectCycleRef,
EffectRef,
_final_validate,
_get_data,
available_effects_str,
find_effect_index,
)
from esphome.components.light.automation import _record_effect_ref
from esphome.components.light.automation import (
_record_effect_cycle_ref,
_record_effect_ref,
)
from esphome.config import Config, path_context
from esphome.const import CONF_EFFECT, CONF_EFFECTS, CONF_ID, CONF_NAME
from esphome.core import ID, Lambda
@@ -215,6 +219,111 @@ def test_final_validate_drains_refs() -> None:
fv.full_config.reset(token)
# --- _final_validate: EffectCycleRef ---
def _setup_cycle_final_validate(
cycle_refs: list[EffectCycleRef],
light_configs: list[ConfigType],
declare_ids: list[tuple[ID, list[str | int]]],
) -> Token:
"""Set up CORE.data and fv.full_config for EffectCycleRef final_validate tests."""
data = _get_data()
data.effect_cycle_refs = cycle_refs
full_conf = Config()
full_conf["light"] = light_configs
for id_, path in declare_ids:
full_conf.declare_ids.append((id_, path))
return fv.full_config.set(full_conf)
def test_final_validate_cycle_accepts_light_with_effects() -> None:
"""Cycle ref against a light with effects should not raise."""
light_id = ID("led1", is_declaration=True)
token = _setup_cycle_final_validate(
cycle_refs=[
EffectCycleRef(light_id=light_id, component_path=["esphome"]),
],
light_configs=[{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse")}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_cycle_rejects_light_without_effects_key() -> None:
"""Cycle ref against a light with no CONF_EFFECTS key should raise."""
light_id = ID("led1", is_declaration=True)
token = _setup_cycle_final_validate(
cycle_refs=[
EffectCycleRef(light_id=light_id, component_path=["esphome"]),
],
light_configs=[{CONF_ID: light_id}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
with pytest.raises(cv.FinalExternalInvalid, match="no effects configured"):
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_cycle_rejects_light_with_empty_effects() -> None:
"""Cycle ref against a light with empty effects list should raise."""
light_id = ID("led1", is_declaration=True)
token = _setup_cycle_final_validate(
cycle_refs=[
EffectCycleRef(light_id=light_id, component_path=["esphome"]),
],
light_configs=[{CONF_ID: light_id, CONF_EFFECTS: []}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
with pytest.raises(cv.FinalExternalInvalid, match="no effects configured"):
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_cycle_unknown_light_id_skipped() -> None:
"""Cycle refs to unknown light IDs should be silently skipped."""
data = _get_data()
data.effect_cycle_refs = [
EffectCycleRef(
light_id=ID("nonexistent", is_declaration=True),
component_path=["esphome"],
)
]
full_conf = Config()
token = fv.full_config.set(full_conf)
try:
_final_validate({})
finally:
fv.full_config.reset(token)
def test_final_validate_drains_cycle_refs() -> None:
"""Cycle refs should be drained after validation to avoid redundant runs."""
light_id = ID("led1", is_declaration=True)
token = _setup_cycle_final_validate(
cycle_refs=[
EffectCycleRef(light_id=light_id, component_path=["esphome"]),
],
light_configs=[{CONF_ID: light_id, CONF_EFFECTS: _make_effects("Fast Pulse")}],
declare_ids=[(light_id, ["light", 0, CONF_ID])],
)
try:
_final_validate({})
assert _get_data().effect_cycle_refs == []
finally:
fv.full_config.reset(token)
# --- _record_effect_ref ---
@@ -278,3 +387,19 @@ def test_record_effect_ref_skips_no_effect_key() -> None:
config: ConfigType = {CONF_ID: ID("led1", is_declaration=True)}
_record_effect_ref(config)
assert _get_data().effect_refs == []
# --- _record_effect_cycle_ref ---
@pytest.mark.usefixtures("_path_ctx")
def test_record_effect_cycle_ref() -> None:
"""Cycle-action config should be recorded with light_id and path."""
light_id = ID("led1", is_declaration=True)
config: ConfigType = {CONF_ID: light_id}
result = _record_effect_cycle_ref(config)
assert result is config
data = _get_data()
assert len(data.effect_cycle_refs) == 1
assert data.effect_cycle_refs[0].light_id is light_id
assert data.effect_cycle_refs[0].component_path == ["esphome"]
@@ -0,0 +1,71 @@
"""Tests for lvgl automation_schema lazy validate_automation build."""
from __future__ import annotations
from unittest.mock import patch
import esphome.components.lvgl # noqa: F401
from esphome.components.lvgl import schemas as lvgl_schemas
from esphome.components.lvgl.schemas import (
WIDGET_TYPES,
_lazy_validate_automation,
automation_schema,
)
from esphome.components.lvgl.widgets import WidgetType
from esphome.config_validation import GenerateID, declare_id
from esphome.const import CONF_TRIGGER_ID
from esphome.core.config import StartupTrigger
def _widget_type(name: str = "obj") -> WidgetType:
wt = WIDGET_TYPES.get(name)
assert wt is not None, f"widget type {name!r} not registered"
return wt
def _trigger_extra_schema() -> dict:
return {GenerateID(CONF_TRIGGER_ID): declare_id(StartupTrigger)}
def test_lazy_validator_defers_build_until_first_call() -> None:
with patch(
"esphome.components.lvgl.schemas.validate_automation",
wraps=lvgl_schemas.validate_automation,
) as va_mock:
validator = _lazy_validate_automation(_trigger_extra_schema())
assert va_mock.call_count == 0
validator({"then": []})
assert va_mock.call_count == 1
validator({"then": []})
assert va_mock.call_count == 1
def test_eager_build_when_schema_extraction_enabled() -> None:
with (
patch("esphome.components.lvgl.schemas.EnableSchemaExtraction", True),
patch(
"esphome.components.lvgl.schemas.validate_automation",
wraps=lvgl_schemas.validate_automation,
) as va_mock,
):
_lazy_validate_automation(_trigger_extra_schema())
assert va_mock.call_count == 1
def test_lazy_and_eager_produce_equivalent_validation() -> None:
extra = _trigger_extra_schema()
with patch("esphome.components.lvgl.schemas.EnableSchemaExtraction", True):
eager = _lazy_validate_automation(extra)
lazy = _lazy_validate_automation(_trigger_extra_schema())
sample = {"then": []}
assert lazy(sample) == eager(sample)
def test_automation_schema_uses_lazy_validators() -> None:
wt = _widget_type("obj")
with patch(
"esphome.components.lvgl.schemas.validate_automation",
wraps=lvgl_schemas.validate_automation,
) as va_mock:
automation_schema(wt.w_type)
assert va_mock.call_count == 0
@@ -0,0 +1,87 @@
"""Tests for container_schema() memoization and lazy build."""
from __future__ import annotations
from collections.abc import Generator
from unittest.mock import patch
import pytest
from esphome import config_validation as cv
import esphome.components.lvgl # noqa: F401
from esphome.components.lvgl import schemas as lvgl_schemas
from esphome.components.lvgl.schemas import WIDGET_TYPES, container_schema
@pytest.fixture(autouse=True)
def _clear_container_schema_cache() -> Generator[None]:
cache = getattr(lvgl_schemas, "_CONTAINER_SCHEMA_CACHE", None)
if cache is not None:
cache.clear()
yield
if cache is not None:
cache.clear()
def _widget_type(name: str = "obj"):
wt = WIDGET_TYPES.get(name)
assert wt is not None, f"widget type {name!r} not registered"
return wt
def test_same_args_return_same_validator() -> None:
wt = _widget_type("obj")
assert container_schema(wt) is container_schema(wt)
def test_extras_none_vs_truthy_get_different_validators() -> None:
wt = _widget_type("obj")
no_extras = container_schema(wt)
extras = {cv.Optional("custom_extra"): cv.string}
assert no_extras is not container_schema(wt, extras)
def test_different_widget_types_get_different_validators() -> None:
assert container_schema(_widget_type("obj")) is not container_schema(
_widget_type("label")
)
def test_schema_build_is_deferred_until_first_validation() -> None:
wt = _widget_type("obj")
with patch.object(
lvgl_schemas, "obj_schema", wraps=lvgl_schemas.obj_schema
) as obj_schema_mock:
validator = container_schema(wt)
assert obj_schema_mock.call_count == 0
validator({})
assert obj_schema_mock.call_count == 1
validator({})
assert obj_schema_mock.call_count == 1
def test_cached_validator_produces_equivalent_output() -> None:
wt = _widget_type("obj")
cached = container_schema(wt)
cached_result = cached({})
lvgl_schemas._CONTAINER_SCHEMA_CACHE.clear()
reference = container_schema(wt)
assert cached is not reference
assert cached_result == reference({})
def test_id_recycling_is_caught_by_identity_guard() -> None:
wt = _widget_type("obj")
real_extras = {cv.Optional("a"): cv.int_}
validator_a = container_schema(wt, real_extras)
cache_key = (id(wt), id(real_extras))
cached_entry = lvgl_schemas._CONTAINER_SCHEMA_CACHE[cache_key]
sentinel = {cv.Optional("a"): cv.int_}
lvgl_schemas._CONTAINER_SCHEMA_CACHE[cache_key] = (
cached_entry[0],
sentinel,
cached_entry[2],
)
assert container_schema(wt, real_extras) is not validator_a
@@ -0,0 +1,67 @@
"""Tests for obj_schema() memoization."""
from __future__ import annotations
from collections.abc import Generator
import pytest
import esphome.components.lvgl # noqa: F401
from esphome.components.lvgl import schemas as lvgl_schemas
from esphome.components.lvgl.schemas import WIDGET_TYPES, obj_schema
@pytest.fixture(autouse=True)
def _clear_obj_schema_cache() -> Generator[None]:
cache = getattr(lvgl_schemas, "_OBJ_SCHEMA_CACHE", None)
if cache is not None:
cache.clear()
yield
if cache is not None:
cache.clear()
def _widget_type(name: str = "obj"):
wt = WIDGET_TYPES.get(name)
assert wt is not None, f"widget type {name!r} not registered"
return wt
def test_same_widget_type_returns_same_schema() -> None:
wt = _widget_type("obj")
assert obj_schema(wt) is obj_schema(wt)
def test_different_widget_types_return_different_schemas() -> None:
assert obj_schema(_widget_type("obj")) is not obj_schema(_widget_type("label"))
def test_cache_is_populated_after_first_call() -> None:
wt = _widget_type("obj")
assert id(wt) not in lvgl_schemas._OBJ_SCHEMA_CACHE
obj_schema(wt)
assert id(wt) in lvgl_schemas._OBJ_SCHEMA_CACHE
def test_cached_schema_produces_equivalent_output() -> None:
wt = _widget_type("obj")
cached_result = obj_schema(wt)({})
lvgl_schemas._OBJ_SCHEMA_CACHE.clear()
fresh_result = obj_schema(wt)({})
assert cached_result == fresh_result
def test_id_recycling_is_caught_by_identity_guard() -> None:
wt = _widget_type("obj")
real_schema = obj_schema(wt)
cached_widget_type, _ = lvgl_schemas._OBJ_SCHEMA_CACHE[id(wt)]
sentinel_schema = object()
lvgl_schemas._OBJ_SCHEMA_CACHE[id(wt)] = (cached_widget_type, sentinel_schema)
assert obj_schema(wt) is sentinel_schema
other = _widget_type("label")
lvgl_schemas._OBJ_SCHEMA_CACHE[id(wt)] = (other, sentinel_schema)
rebuilt = obj_schema(wt)
assert rebuilt is not sentinel_schema
assert rebuilt is not real_schema
@@ -0,0 +1,236 @@
"""Tests for part_dict / obj_dict / part_schema / obj_schema mapping contracts.
These guard the dict-merge refactor: the dict helpers must keep returning the
same logical mapping as the chained-extend version produced, and the
corresponding Schema(...) wrappers must accept and reject the same configs.
"""
from __future__ import annotations
from collections.abc import Generator
import pytest
import voluptuous as vol
from esphome import config_validation as cv
import esphome.components.lvgl
from esphome.components.lvgl import (
_theme_schema,
defines as df,
schemas as lvgl_schemas,
)
from esphome.components.lvgl.schemas import (
ALIGN_TO_SCHEMA,
FLAG_SCHEMA,
FULL_STYLE_SCHEMA,
STATE_SCHEMA,
STYLE_SCHEMA,
WIDGET_TYPES,
automation_schema,
obj_dict,
obj_schema,
part_dict,
part_schema,
)
from esphome.components.lvgl.types import LvType
from esphome.components.lvgl.widgets import WidgetType
@pytest.fixture(autouse=True)
def _clear_obj_dict_cache() -> Generator[None]:
cache = getattr(lvgl_schemas, "_OBJ_DICT_CACHE", None)
if cache is not None:
cache.clear()
# The lazily-built theme schema is cached on _build_theme_schema; clear it
# too so each test starts from a clean slate.
build_theme = getattr(esphome.components.lvgl, "_build_theme_schema", None)
if build_theme is not None and hasattr(build_theme, "cache_clear"):
build_theme.cache_clear()
yield
if cache is not None:
cache.clear()
if build_theme is not None and hasattr(build_theme, "cache_clear"):
build_theme.cache_clear()
def _marker_names(mapping) -> set[str]:
"""Return the underlying string names of every voluptuous Marker key."""
names: set[str] = set()
for key in mapping:
if isinstance(key, vol.Marker):
schema = key.schema
if isinstance(schema, str):
names.add(schema)
return names
def _widget_type(name: str = "obj"):
wt = WIDGET_TYPES.get(name)
assert wt is not None, f"widget type {name!r} not registered"
return wt
def test_part_dict_includes_state_flag_and_part_keys() -> None:
parts = ("indicator", "knob")
keys = _marker_names(part_dict(parts))
assert {"indicator", "knob"} <= keys
assert _marker_names(STATE_SCHEMA.schema) <= keys
assert _marker_names(FLAG_SCHEMA.schema) <= keys
def test_obj_dict_extends_part_dict_with_align_automation_state_group() -> None:
wt = _widget_type("obj")
part_keys = _marker_names(part_dict(wt.parts))
obj_keys = _marker_names(obj_dict(wt))
assert part_keys <= obj_keys
assert _marker_names(ALIGN_TO_SCHEMA) <= obj_keys
assert _marker_names(automation_schema(wt.w_type)) <= obj_keys
assert {"state", "group"} <= obj_keys
def test_obj_dict_is_memoized_by_widget_type() -> None:
wt = _widget_type("obj")
first = obj_dict(wt)
second = obj_dict(wt)
assert first is second
# Different widget type, different dict.
assert obj_dict(_widget_type("label")) is not first
def test_part_schema_round_trips_known_state_and_part_settings() -> None:
schema = part_schema(("indicator",))
out = schema(
{
"bg_color": 0x112233,
"checked": {"bg_color": 0x445566},
"indicator": {"bg_color": 0x778899},
}
)
assert out["bg_color"] == 0x112233
assert out["checked"]["bg_color"] == 0x445566
assert out["indicator"]["bg_color"] == 0x778899
def test_part_schema_rejects_unknown_part() -> None:
schema = part_schema(("indicator",))
with pytest.raises(vol.Invalid):
schema({"definitely_not_a_part": {}})
@pytest.mark.parametrize("name", sorted(WIDGET_TYPES))
def test_obj_schema_accepts_empty_config_for_every_widget_type(name: str) -> None:
obj_schema(_widget_type(name))({})
def test_obj_schema_accepts_align_to_and_state_group() -> None:
schema = obj_schema(_widget_type("obj"))
out = schema(
{
df.CONF_ALIGN_TO: {
"id": "some_other_widget",
df.CONF_ALIGN: "TOP_LEFT",
},
"state": {"checked": True},
}
)
assert out[df.CONF_ALIGN_TO][df.CONF_ALIGN] == "LV_ALIGN_TOP_LEFT"
assert out["state"]["checked"] is True
def test_obj_schema_rejects_unknown_top_level_key() -> None:
with pytest.raises(vol.Invalid):
obj_schema(_widget_type("obj"))({"definitely_not_a_real_key": 1})
def test_part_schema_returns_cv_schema_for_extend_callers() -> None:
schema = part_schema(("indicator",))
extended = schema.extend({cv.Optional("extra_key"): cv.string})
out = extended({"extra_key": "value", "bg_color": 0xAABBCC})
assert out["extra_key"] == "value"
assert out["bg_color"] == 0xAABBCC
def test_obj_schema_returns_cv_schema_for_extend_callers() -> None:
schema = obj_schema(_widget_type("obj"))
extended = schema.extend({cv.Optional("extra_key"): cv.string})
extended({"extra_key": "value"})
@pytest.mark.parametrize(
"schema",
[STATE_SCHEMA, FLAG_SCHEMA, STYLE_SCHEMA, FULL_STYLE_SCHEMA],
)
def test_spread_sources_carry_no_extra_schemas(schema: cv.Schema) -> None:
# part_dict / obj_dict reach into .schema and rebuild via cv.Schema(...),
# which silently drops _extra_schemas and any non-default extra/required.
# Lock the invariant so a future add_extra() on these sources fails CI
# instead of quietly removing validation from part/obj/theme schemas.
assert not schema._extra_schemas
assert schema.extra is vol.PREVENT_EXTRA
assert schema.required is False
def test_theme_schema_merges_obj_dict_and_full_style_props() -> None:
# _theme_schema is the riskiest merge: obj_dict(w) and FULL_STYLE_SCHEMA.schema
# share many STYLE_SCHEMA marker instances. Exercise the merged schema
# end-to-end with one key from each side (a STATE_SCHEMA part from obj_dict
# and a FULL_STYLE-only property) to lock the behaviour against future
# regressions in either source.
out = _theme_schema(
{
df.CONF_DARK_MODE: True,
"obj": {
"bg_color": 0x112233,
"checked": {"bg_color": 0x445566},
df.CONF_PAD_ROW: 4,
df.CONF_GRID_CELL_X_ALIGN: "CENTER",
},
}
)
assert out[df.CONF_DARK_MODE] is True
obj_out = out["obj"]
assert obj_out["bg_color"] == 0x112233
assert obj_out["checked"]["bg_color"] == 0x445566
assert obj_out[df.CONF_PAD_ROW] == 4
assert obj_out[df.CONF_GRID_CELL_X_ALIGN] == "LV_GRID_ALIGN_CENTER"
def test_theme_schema_self_heals_when_a_widget_type_is_registered_later() -> None:
# _build_theme_schema is functools.cached on a snapshot of WIDGET_TYPES.
# any_widget_schema explicitly supports external components registering
# widgets lazily, and the device builder revalidates in-process, so a
# widget registered after first use must invalidate the cached snapshot.
_theme_schema({df.CONF_DARK_MODE: True}) # populate the cache
name = "test_self_heal_widget"
assert name not in WIDGET_TYPES
# is_mock=True skips registration side-effects; insert into WIDGET_TYPES
# manually so the next theme call sees the new entry.
WIDGET_TYPES[name] = WidgetType(name, LvType("test_fake_t"), (), is_mock=True)
try:
out = _theme_schema({df.CONF_DARK_MODE: False, name: {"bg_color": 0x010203}})
assert out[name]["bg_color"] == 0x010203
finally:
WIDGET_TYPES.pop(name, None)
@pytest.mark.parametrize(
"schema",
[STATE_SCHEMA, FLAG_SCHEMA, STYLE_SCHEMA, FULL_STYLE_SCHEMA],
)
def test_spread_sources_have_no_top_level_marker_defaults(schema: cv.Schema) -> None:
# _theme_schema merges obj_dict(w) with FULL_STYLE_SCHEMA.schema; on a key
# collision, dict-spread keeps the first source's marker (and its default)
# but the last source's value, whereas .extend() would take both from the
# later source. The two are equivalent today because the overlapping
# markers are the same instances (both derive from STYLE_SCHEMA) and none
# carry a top-level default. Lock that so a future divergent default would
# fail CI rather than silently drift the merged validation.
offenders = [
marker.schema
for marker in schema.schema
if isinstance(marker, vol.Optional) and marker.default is not vol.UNDEFINED
]
assert not offenders, f"top-level Optional with default: {offenders}"
@@ -0,0 +1,53 @@
"""Tests for lvgl.<widget>.update lazy schema build."""
from __future__ import annotations
from unittest.mock import patch
from esphome.automation import ACTION_REGISTRY
import esphome.components.lvgl # noqa: F401
from esphome.components.lvgl.schemas import WIDGET_TYPES
from esphome.components.lvgl.widgets import _update_action_schema
from esphome.config_validation import Schema
def _widget_type(name: str = "obj"):
wt = WIDGET_TYPES.get(name)
assert wt is not None, f"widget type {name!r} not registered"
return wt
def test_registry_entry_uses_lazy_validator() -> None:
entry = ACTION_REGISTRY["lvgl.label.update"]
assert callable(entry.raw_schema)
assert not isinstance(entry.raw_schema, Schema)
def test_lazy_validator_defers_build_until_first_call() -> None:
wt = _widget_type("label")
with patch(
"esphome.components.lvgl.widgets._build_update_schema",
wraps=lambda w: Schema({}),
) as build_mock:
validator = _update_action_schema(wt)
assert build_mock.call_count == 0
validator({})
assert build_mock.call_count == 1
validator({})
assert build_mock.call_count == 1
def test_eager_build_when_schema_extraction_enabled() -> None:
wt = _widget_type("label")
with patch("esphome.components.lvgl.widgets.EnableSchemaExtraction", True):
result = _update_action_schema(wt)
assert isinstance(result, Schema)
def test_lazy_and_eager_produce_equivalent_validation() -> None:
wt = _widget_type("label")
with patch("esphome.components.lvgl.widgets.EnableSchemaExtraction", True):
eager = _update_action_schema(wt)
lazy = _update_action_schema(wt)
sample = {"id": "label_id"}
assert lazy(sample) == eager(sample)
@@ -510,15 +510,9 @@ def test_package_merge_by_missing_id() -> None:
],
}
error_raised = False
try:
with pytest.raises(cv.Invalid) as exc_info:
packages_pass(config)
assert False, "Expected validation error for missing ID"
except cv.Invalid as err:
error_raised = True
assert err.path == [CONF_SENSOR, 2]
assert error_raised
assert exc_info.value.path == [CONF_SENSOR, 2]
def test_package_list_remove_by_id() -> None:
+1
View File
@@ -0,0 +1 @@
"""Tests for the time component."""
+369
View File
@@ -0,0 +1,369 @@
"""Tests for time component – ha-timezone branch changes.
Covers:
- detect_tz() platform guard (returns None for unsupported platforms)
- detect_tz() result caching (avoids duplicate log messages)
- detect_tz() error paths (tzlocal None, tzdata missing)
- validate_tz() accepts/rejects POSIX timezone strings and IANA keys
- TIME_SCHEMA: timezone is now truly optional (was SplitDefault)
- homeassistant/time: USE_HOMEASSISTANT_TIMEZONE define emitted iff
CONF_TIMEZONE is absent from the config
"""
from __future__ import annotations
from unittest import mock
import pytest
from esphome.components.time import DOMAIN, TIME_SCHEMA, detect_tz, validate_tz
import esphome.config_validation as cv
from esphome.const import (
CONF_ID,
CONF_TIMEZONE,
KEY_CORE,
KEY_TARGET_FRAMEWORK,
KEY_TARGET_PLATFORM,
Platform,
PlatformFramework,
)
from esphome.core import CORE, EsphomeError
from tests.component_tests.types import SetCoreConfigCallable
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
# A minimal TZif v2/v3 file that encodes "EST5EDT" as the footer line.
# The binary content is not validated at this level – what matters is that
# _extract_tz_string() picks up the last-but-one newline-terminated line.
_FAKE_TZFILE = b"\x00" * 44 + b"TZif2\x00" * 1 + b"\n" + b"EST5EDT,M3.2.0,M11.1.0\n"
def _set_platform(platform: Platform) -> None:
"""Set CORE.data so that CORE.target_platform returns *platform*."""
CORE.data[KEY_CORE] = {
KEY_TARGET_PLATFORM: platform,
KEY_TARGET_FRAMEWORK: "arduino",
}
# ---------------------------------------------------------------------------
# detect_tz – platform guard
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"platform_framework",
[
PlatformFramework.NRF52_ZEPHYR,
],
)
def test_detect_tz_returns_none_for_unsupported_platform(
platform_framework: PlatformFramework,
set_core_config: SetCoreConfigCallable,
) -> None:
"""detect_tz() must return None for platforms that do not support TZ auto-detection."""
set_core_config(platform_framework)
result = detect_tz()
assert result is None
@pytest.mark.parametrize(
"platform_framework",
[
PlatformFramework.ESP32_IDF,
PlatformFramework.ESP32_ARDUINO,
PlatformFramework.ESP8266_ARDUINO,
PlatformFramework.RP2040_ARDUINO,
PlatformFramework.BK72XX_ARDUINO,
PlatformFramework.RTL87XX_ARDUINO,
PlatformFramework.LN882X_ARDUINO,
PlatformFramework.HOST_NATIVE,
],
)
def test_detect_tz_calls_tzlocal_for_supported_platform(
platform_framework: PlatformFramework,
set_core_config: SetCoreConfigCallable,
) -> None:
"""detect_tz() must call tzlocal for every supported platform."""
set_core_config(platform_framework)
with (
mock.patch(
"esphome.components.time.tzlocal.get_localzone_name",
return_value="America/New_York",
),
mock.patch(
"esphome.components.time._load_tzdata",
return_value=_FAKE_TZFILE,
),
):
result = detect_tz()
assert result is not None
assert isinstance(result, str)
assert len(result) > 0
# ---------------------------------------------------------------------------
# detect_tz – caching
# ---------------------------------------------------------------------------
def test_detect_tz_caches_result(
set_core_config: SetCoreConfigCallable,
caplog: pytest.LogCaptureFixture,
) -> None:
"""detect_tz() must cache the TZ string after the first call so that
subsequent invocations (e.g. when multiple time platforms are configured)
skip tzlocal and avoid duplicate INFO messages."""
set_core_config(PlatformFramework.ESP32_IDF)
with (
mock.patch(
"esphome.components.time.tzlocal.get_localzone_name",
return_value="America/New_York",
) as mock_tz,
mock.patch(
"esphome.components.time._load_tzdata",
return_value=_FAKE_TZFILE,
) as mock_load,
):
first = detect_tz()
second = detect_tz()
assert first == second
# tzlocal and _load_tzdata must be called exactly once despite two detect_tz() calls
mock_tz.assert_called_once()
mock_load.assert_called_once()
def test_detect_tz_cache_stored_in_core_data(
set_core_config: SetCoreConfigCallable,
) -> None:
"""The cached TZ string should be stored under CORE.data[DOMAIN][CONF_TIMEZONE]."""
set_core_config(PlatformFramework.ESP32_IDF)
with (
mock.patch(
"esphome.components.time.tzlocal.get_localzone_name",
return_value="Europe/London",
),
mock.patch(
"esphome.components.time._load_tzdata",
return_value=_FAKE_TZFILE,
),
):
result = detect_tz()
assert CORE.data.get(DOMAIN, {}).get(CONF_TIMEZONE) == result
def test_detect_tz_returns_pre_seeded_cache(
set_core_config: SetCoreConfigCallable,
) -> None:
"""If CORE.data already has a cached TZ string, detect_tz() must return it
without calling tzlocal at all."""
set_core_config(PlatformFramework.ESP32_IDF)
CORE.data[DOMAIN] = {CONF_TIMEZONE: "CET-1CEST,M3.5.0,M10.5.0/3"}
with mock.patch("esphome.components.time.tzlocal.get_localzone_name") as mock_tz:
result = detect_tz()
assert result == "CET-1CEST,M3.5.0,M10.5.0/3"
mock_tz.assert_not_called()
# ---------------------------------------------------------------------------
# detect_tz – error paths
# ---------------------------------------------------------------------------
def test_detect_tz_raises_when_tzlocal_returns_none(
set_core_config: SetCoreConfigCallable,
) -> None:
"""detect_tz() must raise EsphomeError when the local timezone cannot be determined."""
set_core_config(PlatformFramework.ESP32_IDF)
with (
mock.patch(
"esphome.components.time.tzlocal.get_localzone_name",
return_value=None,
),
pytest.raises(EsphomeError, match="Could not automatically determine timezone"),
):
detect_tz()
def test_detect_tz_raises_when_tzdata_not_found(
set_core_config: SetCoreConfigCallable,
) -> None:
"""detect_tz() must raise EsphomeError when tzdata has no entry for the IANA key."""
set_core_config(PlatformFramework.ESP32_IDF)
with (
mock.patch(
"esphome.components.time.tzlocal.get_localzone_name",
return_value="Antarctica/Troll",
),
mock.patch(
"esphome.components.time._load_tzdata",
return_value=None,
),
pytest.raises(EsphomeError, match="Could not automatically determine timezone"),
):
detect_tz()
# ---------------------------------------------------------------------------
# validate_tz
# ---------------------------------------------------------------------------
def test_validate_tz_accepts_valid_posix_string() -> None:
"""validate_tz() must accept a syntactically valid POSIX TZ string."""
result = validate_tz("UTC0")
assert result == "UTC0"
def test_validate_tz_accepts_posix_string_with_dst() -> None:
"""validate_tz() must accept a full POSIX TZ string with DST rules."""
tz = "EST5EDT,M3.2.0,M11.1.0"
result = validate_tz(tz)
assert result == tz
def test_validate_tz_accepts_iana_key_and_converts() -> None:
"""validate_tz() must accept an IANA timezone key and return the POSIX string."""
with mock.patch(
"esphome.components.time._load_tzdata",
return_value=_FAKE_TZFILE,
):
result = validate_tz("America/New_York")
# Should have been converted from IANA to POSIX via _extract_tz_string
assert result == "EST5EDT,M3.2.0,M11.1.0"
def test_validate_tz_rejects_invalid_posix_string() -> None:
"""validate_tz() must raise cv.Invalid for a malformed POSIX TZ string."""
with pytest.raises(cv.Invalid, match="Invalid POSIX timezone string"):
validate_tz("NOTAVALIDTZ!!!")
def test_validate_tz_accepts_empty_string() -> None:
"""An empty string is accepted by validate_tz() and signals 'disable timezone'."""
result = validate_tz("")
assert result == ""
# ---------------------------------------------------------------------------
# TIME_SCHEMA – timezone is now cv.Optional (no SplitDefault)
# ---------------------------------------------------------------------------
def test_time_schema_timezone_is_optional(
set_core_config: SetCoreConfigCallable,
) -> None:
"""TIME_SCHEMA must accept a config with no timezone key on a supported platform."""
set_core_config(PlatformFramework.ESP32_IDF)
# Should not raise
config = TIME_SCHEMA({})
assert CONF_TIMEZONE not in config
def test_time_schema_explicit_timezone_accepted(
set_core_config: SetCoreConfigCallable,
) -> None:
"""TIME_SCHEMA must accept an explicit valid POSIX timezone on Arduino/IDF."""
set_core_config(PlatformFramework.ESP32_IDF)
config = TIME_SCHEMA({CONF_TIMEZONE: "UTC0"})
assert config[CONF_TIMEZONE] == "UTC0"
def test_time_schema_explicit_empty_timezone_accepted(
set_core_config: SetCoreConfigCallable,
) -> None:
"""An empty timezone string (timezone-disable sentinel) must pass TIME_SCHEMA."""
set_core_config(PlatformFramework.ESP32_IDF)
config = TIME_SCHEMA({CONF_TIMEZONE: ""})
assert config[CONF_TIMEZONE] == ""
def test_time_schema_timezone_rejected_on_zephyr(
set_core_config: SetCoreConfigCallable,
) -> None:
"""TIME_SCHEMA must reject a timezone value on Zephyr with the framework error.
The platform check (cv.only_with_framework) must run BEFORE validate_tz so
that users receive an actionable "unsupported framework" message rather than a
confusing TZ-parsing error.
"""
set_core_config(PlatformFramework.NRF52_ZEPHYR)
with pytest.raises(cv.Invalid, match="only available with framework"):
TIME_SCHEMA({CONF_TIMEZONE: "UTC0"})
def test_time_schema_invalid_tz_on_zephyr_gives_framework_error(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Even a syntactically invalid TZ string must produce the framework error on Zephyr.
This specifically tests that cv.only_with_framework is evaluated before
validate_tz: if the order were reversed, an invalid POSIX string would
generate a misleading TZ-parsing error instead.
"""
set_core_config(PlatformFramework.NRF52_ZEPHYR)
with pytest.raises(cv.Invalid, match="only available with framework"):
TIME_SCHEMA({CONF_TIMEZONE: "NOTAVALIDTZ!!!"})
# ---------------------------------------------------------------------------
# homeassistant/time: USE_HOMEASSISTANT_TIMEZONE define
# ---------------------------------------------------------------------------
@pytest.fixture
def mock_ha_cg():
"""Mock codegen functions used by homeassistant/time to_code."""
with (
mock.patch(
"esphome.components.homeassistant.time.cg.new_Pvariable",
return_value=mock.MagicMock(),
),
mock.patch(
"esphome.components.homeassistant.time.cg.add_define",
) as mock_add_define,
mock.patch(
"esphome.components.homeassistant.time.cg.register_component",
new_callable=mock.AsyncMock,
),
mock.patch(
"esphome.components.homeassistant.time.time_.register_time",
new_callable=mock.AsyncMock,
),
):
yield mock_add_define
@pytest.mark.asyncio
async def test_ha_time_defines_ha_timezone_when_no_explicit_tz(mock_ha_cg) -> None:
"""When CONF_TIMEZONE is absent from the config, to_code() must call
cg.add_define('USE_HOMEASSISTANT_TIMEZONE')."""
from esphome.components.homeassistant.time import to_code
await to_code({CONF_ID: mock.MagicMock()})
mock_ha_cg.assert_any_call("USE_HOMEASSISTANT_TIMEZONE")
@pytest.mark.asyncio
async def test_ha_time_no_ha_timezone_define_when_explicit_tz(mock_ha_cg) -> None:
"""When CONF_TIMEZONE is present in the config, to_code() must NOT call
cg.add_define('USE_HOMEASSISTANT_TIMEZONE')."""
from esphome.components.homeassistant.time import to_code
await to_code({CONF_ID: mock.MagicMock(), CONF_TIMEZONE: "UTC0"})
define_calls = [call.args[0] for call in mock_ha_cg.call_args_list]
assert "USE_HOMEASSISTANT_TIME" in define_calls
assert "USE_HOMEASSISTANT_TIMEZONE" not in define_calls
+14 -14
View File
@@ -120,12 +120,12 @@ api:
lambda: 'return condition;'
then:
- logger.log:
format: "Condition true, value: %d"
args: ['value']
format: "Condition true, value: %ld"
args: ['(long) value']
else:
- logger.log:
format: "Condition false, value: %d"
args: ['value']
format: "Condition false, value: %ld"
args: ['(long) value']
- logger.log: "After if/else"
# Test nested IfAction (multiple ContinuationAction instances)
- action: test_nested_if
@@ -171,8 +171,8 @@ api:
count: !lambda 'return count;'
then:
- logger.log:
format: "Repeat iteration: %d"
args: ['iteration']
format: "Repeat iteration: %lu"
args: ['(unsigned long) iteration']
- logger.log: "After repeat"
# Test combined continuations (if + while + repeat)
- action: test_combined_continuations
@@ -193,8 +193,8 @@ api:
lambda: 'return id(api_continuation_test_counter) > 0;'
then:
- logger.log:
format: "Combined: repeat=%d, while=%d"
args: ['iteration', 'id(api_continuation_test_counter)']
format: "Combined: repeat=%lu, while=%d"
args: ['(unsigned long) iteration', 'id(api_continuation_test_counter)']
- lambda: 'id(api_continuation_test_counter)--;'
else:
- logger.log: "Skipped loops"
@@ -208,8 +208,8 @@ api:
- api.respond:
success: true
- logger.log:
format: "Status response sent (call_id=%d)"
args: [call_id]
format: "Status response sent (call_id=%lu)"
args: ['(unsigned long) call_id']
- action: test_respond_status_error
variables:
@@ -229,8 +229,8 @@ api:
value: float
then:
- logger.log:
format: "Optional response (call_id=%d, return_response=%d)"
args: [call_id, return_response]
format: "Optional response (call_id=%lu, return_response=%lu)"
args: ['(unsigned long) call_id', '(unsigned long) return_response']
- api.respond:
data: !lambda |-
root["sensor"] = sensor_name;
@@ -264,8 +264,8 @@ api:
input: string
then:
- logger.log:
format: "Only response (call_id=%d)"
args: [call_id]
format: "Only response (call_id=%lu)"
args: ['(unsigned long) call_id']
- api.respond:
data: !lambda |-
root["input"] = input;
@@ -0,0 +1,11 @@
audio_file:
- id: test_audio
file:
type: local
path: $component_dir/test.wav
media_source:
- platform: audio_file
id: audio_file_source
# task_stack_in_psram: false must validate without a psram: component
task_stack_in_psram: false
@@ -1,6 +1,8 @@
<<: !include common.yaml
esp32_ble_tracker:
esp32_ble:
max_connections: 9
bluetooth_proxy:
@@ -3,6 +3,7 @@ esp32_hosted:
slot: 1
active_high: true
reset_pin: GPIO15
use_psram: true
cmd_pin: GPIO13
clk_pin: GPIO12
d0_pin: GPIO11
+2
View File
@@ -2,6 +2,8 @@ esphome:
debug_scheduler: true
platformio_options:
board_build.flash_mode: dio
build_flags:
- "-DESPHOME_TEST_BUILD_FLAG"
environment_variables:
TEST_ENV_VAR: "test_value"
BUILD_NUMBER: "12345"
@@ -1,13 +1,3 @@
substitutions:
i2s_bclk_pin: GPIO27
i2s_lrclk_pin: GPIO26
i2s_mclk_pin: GPIO25
i2s_dout_pin: GPIO12
spdif_data_pin: GPIO4
packages:
i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
i2s_audio:
- id: i2s_output
@@ -20,6 +10,5 @@ speaker:
use_apll: true
timeout: 2s
sample_rate: 48000
bits_per_sample: 16bit
channel: stereo
i2s_mode: primary
@@ -0,0 +1,8 @@
substitutions:
i2s_bclk_pin: GPIO27
i2s_lrclk_pin: GPIO26
i2s_mclk_pin: GPIO25
i2s_dout_pin: GPIO12
spdif_data_pin: GPIO4
<<: !include common-spdif_mode.yaml
+10
View File
@@ -103,6 +103,16 @@ esphome:
- light.turn_on:
id: test_monochromatic_light
effect: !lambda 'return iteration > 1 ? "Strobe" : "none";'
# Cycle through configured effects (skip "None")
- light.effect.next: test_monochromatic_light
- light.effect.previous: test_monochromatic_light
# Cycle through effects including "None"
- light.effect.next:
id: test_monochromatic_light
include_none: true
- light.effect.previous:
id: test_monochromatic_light
include_none: true
- light.dim_relative:
id: test_monochromatic_light
relative_brightness: 5%
@@ -1,3 +1,6 @@
psram:
mode: quad
i2s_audio:
i2s_lrclk_pin: GPIO18
i2s_bclk_pin: GPIO19
@@ -12,6 +15,7 @@ microphone:
micro_wake_word:
microphone: echo_microphone
task_stack_in_psram: true
on_wake_word_detected:
- logger.log: "Wake word detected"
- micro_wake_word.stop:
@@ -0,0 +1,165 @@
#include "../common.h"
namespace esphome::mitsubishi_cn105::testing {
TEST(MitsubishiCN105ClimateTests, SupportedSwingModeOffLeavesTraitsEmpty) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_OFF);
EXPECT_FALSE(sut.traits().get_supports_swing_modes());
}
TEST(MitsubishiCN105ClimateTests, SupportedSwingModeVerticalExposesOffAndVertical) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
}
TEST(MitsubishiCN105ClimateTests, SupportedSwingModeHorizontalExposesOffAndHorizontal) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
EXPECT_FALSE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
}
TEST(MitsubishiCN105ClimateTests, SupportedSwingModeBothExposesAllExpectedModes) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_OFF));
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_VERTICAL));
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_HORIZONTAL));
EXPECT_TRUE(sut.traits().supports_swing_mode(climate::CLIMATE_SWING_BOTH));
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsVerticalSwingWhenSupported) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_VERTICAL);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsHorizontalSwingWhenSupported) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
sut.status().vane_mode = MitsubishiCN105::VaneMode::AUTO;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_HORIZONTAL);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsBothSwingWhenSupported) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_BOTH);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesMapsSwingOffWhenNoSwingActive) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
sut.status().vane_mode = MitsubishiCN105::VaneMode::POSITION_3;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesRemembersLastNonSwingPositions) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
sut.status().vane_mode = MitsubishiCN105::VaneMode::POSITION_4;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::RIGHT;
sut.apply_values_();
EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_4);
EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::RIGHT);
sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
sut.apply_values_();
EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_4);
EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::RIGHT);
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_BOTH);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesDoesNotOverwriteRememberedPositionWithUnknownValues) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_BOTH);
sut.last_non_swing_vane_mode_ = MitsubishiCN105::VaneMode::POSITION_2;
sut.last_non_swing_wide_vane_mode_ = MitsubishiCN105::WideVaneMode::LEFT;
sut.status().vane_mode = MitsubishiCN105::VaneMode::UNKNOWN;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::UNKNOWN;
sut.apply_values_();
EXPECT_EQ(sut.last_non_swing_vane_mode_, MitsubishiCN105::VaneMode::POSITION_2);
EXPECT_EQ(sut.last_non_swing_wide_vane_mode_, MitsubishiCN105::WideVaneMode::LEFT);
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesIgnoresUnsupportedVerticalSwingState) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_HORIZONTAL);
sut.status().vane_mode = MitsubishiCN105::VaneMode::SWING;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::CENTER;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
}
TEST(MitsubishiCN105ClimateTests, ApplyValuesIgnoresUnsupportedHorizontalSwingState) {
TestableMitsubishiCN105Climate sut;
sut.set_supported_swing_mode(climate::CLIMATE_SWING_VERTICAL);
sut.status().vane_mode = MitsubishiCN105::VaneMode::AUTO;
sut.status().wide_vane_mode = MitsubishiCN105::WideVaneMode::SWING;
sut.apply_values_();
EXPECT_EQ(sut.swing_mode, climate::CLIMATE_SWING_OFF);
}
} // namespace esphome::mitsubishi_cn105::testing
@@ -8,6 +8,7 @@
#include <vector>
#include "esphome/components/uart/uart_component.h"
#include "esphome/components/mitsubishi_cn105/mitsubishi_cn105.h"
#include "esphome/components/mitsubishi_cn105/mitsubishi_cn105_climate.h"
namespace esphome::mitsubishi_cn105::testing {
@@ -44,6 +45,7 @@ class TestableMitsubishiCN105 : public MitsubishiCN105 {
using MitsubishiCN105::State;
using MitsubishiCN105::UpdateFlag;
using MitsubishiCN105::state_;
using MitsubishiCN105::status_;
using MitsubishiCN105::operation_start_ms_;
using MitsubishiCN105::use_temperature_encoding_b_;
using MitsubishiCN105::set_wide_vane_high_bit_;
@@ -58,4 +60,13 @@ class TestableMitsubishiCN105 : public MitsubishiCN105 {
void set_current_time(uint32_t ms) { test_loop_time_ms = ms; }
};
class TestableMitsubishiCN105Climate : public MitsubishiCN105Climate {
public:
using MitsubishiCN105Climate::apply_values_;
using MitsubishiCN105Climate::last_non_swing_vane_mode_;
using MitsubishiCN105Climate::last_non_swing_wide_vane_mode_;
MitsubishiCN105::Status &status() { return static_cast<TestableMitsubishiCN105 &>(this->hp_).status_; }
};
} // namespace esphome::mitsubishi_cn105::testing
@@ -3,6 +3,9 @@ climate:
id: ac
name: "AC Test"
uart_id: uart_bus
update_interval: 30s
current_temperature_min_interval: 120s
supported_swing_modes: BOTH
esphome:
on_boot:
+4
View File
@@ -16,8 +16,12 @@ speaker:
id: speaker_id
dac_type: external
i2s_dout_pin: ${dout_pin}
bits_per_sample: 32bit
channel: stereo
- platform: mixer
output_speaker: speaker_id
bits_per_sample: 32
num_channels: 2
source_speakers:
- id: source_speaker_1_id
- id: source_speaker_2_id
+40
View File
@@ -0,0 +1,40 @@
esphome:
on_boot:
then:
- router.speaker.switch_output:
id: router_id
target_speaker: speaker_b_id
# id omitted: auto-resolved since there's a single router instance
- router.speaker.switch_output:
target_speaker: !lambda return id(speaker_a_id);
i2s_audio:
i2s_lrclk_pin: ${a_lrclk_pin}
i2s_bclk_pin: ${a_bclk_pin}
speaker:
- platform: i2s_audio
id: speaker_a_id
dac_type: external
i2s_dout_pin: ${a_dout_pin}
sample_rate: 48000
bits_per_sample: 16bit
channel: stereo
- platform: i2s_audio
id: speaker_b_id
dac_type: external
i2s_dout_pin: ${b_dout_pin}
spdif_mode: true
use_apll: true
sample_rate: 48000
bits_per_sample: 16bit
channel: stereo
i2s_mode: primary
- platform: router
id: router_id
output_speakers:
- speaker_a_id
- speaker_b_id
sample_rate: 48000
bits_per_sample: 16
num_channels: 2
@@ -0,0 +1,7 @@
substitutions:
a_lrclk_pin: GPIO4
a_bclk_pin: GPIO5
a_dout_pin: GPIO14
b_dout_pin: GPIO19
<<: !include common.yaml
@@ -1,4 +1,5 @@
rp2040:
variant: rp2040
enable_full_printf: false
logger:
@@ -0,0 +1,6 @@
rp2040:
variant: rp2350
enable_full_printf: false
logger:
level: VERBOSE
+8 -3
View File
@@ -1503,13 +1503,18 @@ async def test_websocket_refresh_command(
) -> None:
"""Test WebSocket refresh command triggers dashboard update."""
with patch("esphome.dashboard.web_server.DASHBOARD_SUBSCRIBER") as mock_subscriber:
mock_subscriber.request_refresh = Mock()
# Signal an asyncio.Event when request_refresh is invoked so the
# test can deterministically wait for the server-side handler to run
# instead of relying on a fixed sleep (flaky on Windows CI under load).
called = asyncio.Event()
mock_subscriber.request_refresh = Mock(side_effect=called.set)
# Send refresh command
await websocket_client.write_message(json.dumps({"event": "refresh"}))
# Give it a moment to process
await asyncio.sleep(0.01)
# Wait for the server to process the message and invoke request_refresh
async with asyncio.timeout(5):
await called.wait()
# Verify request_refresh was called
mock_subscriber.request_refresh.assert_called_once()
+3 -7
View File
@@ -34,9 +34,7 @@ def test_get_base_frontend_path_dev_mode() -> None:
# The function uses Path.resolve() which resolves symlinks
# The actual function adds "/" to the path, so we simulate that
test_path_with_slash = test_path if test_path.endswith("/") else test_path + "/"
expected = (
Path(os.getcwd()) / test_path_with_slash / "esphome_dashboard"
).resolve()
expected = (Path.cwd() / test_path_with_slash / "esphome_dashboard").resolve()
assert result == expected
@@ -62,9 +60,7 @@ def test_get_base_frontend_path_dev_mode_relative_path() -> None:
# The function uses Path.resolve() which resolves symlinks
# The actual function adds "/" to the path, so we simulate that
test_path_with_slash = test_path if test_path.endswith("/") else test_path + "/"
expected = (
Path(os.getcwd()) / test_path_with_slash / "esphome_dashboard"
).resolve()
expected = (Path.cwd() / test_path_with_slash / "esphome_dashboard").resolve()
assert result == expected
assert result.is_absolute()
@@ -157,7 +153,7 @@ def test_load_file_path(tmp_path: Path) -> None:
test_file = tmp_path / "test.txt"
test_file.write_bytes(b"test content")
with open(test_file, "rb") as f:
with test_file.open("rb") as f:
content = f.read()
assert content == b"test content"
+5 -3
View File
@@ -79,7 +79,7 @@ def shared_platformio_cache() -> Generator[Path]:
lock_file = Path.home() / ".esphome-integration-tests-init.lock"
# Always acquire the lock to ensure cache is ready before proceeding
with open(lock_file, "w") as lock_fd:
with lock_file.open("w") as lock_fd:
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX)
# Check if the native platform is installed (the actual indicator of a populated cache)
@@ -407,8 +407,10 @@ async def wait_and_connect_api_client(
# Wait for connection with timeout
try:
await asyncio.wait_for(connected_future, timeout=timeout)
except TimeoutError:
raise TimeoutError(f"Failed to connect to API after {timeout} seconds")
except TimeoutError as err:
raise TimeoutError(
f"Failed to connect to API after {timeout} seconds"
) from err
if return_disconnect_event:
yield client, disconnect_event
@@ -43,7 +43,7 @@ async def test_gpio_expander_cache(
# ensure logs are in the expected order
log_order = [
(digital_read_hw_pattern, 0),
[(digital_read_cache_pattern, i) for i in range(0, 8)],
[(digital_read_cache_pattern, i) for i in range(8)],
(digital_read_hw_pattern, 8),
[(digital_read_cache_pattern, i) for i in range(8, 16)],
(digital_read_hw_pattern, 16),
@@ -68,7 +68,7 @@ async def test_gpio_expander_cache(
# uint16_t component tests (single bank of 16 pins)
(uint16_read_hw_pattern, 0), # First pin triggers hw read
[
(uint16_read_cache_pattern, i) for i in range(0, 16)
(uint16_read_cache_pattern, i) for i in range(16)
], # All 16 pins return via cache
# After cache reset
(uint16_read_hw_pattern, 5), # First read after reset triggers hw
+2 -5
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
import importlib.util
import json
import os
from pathlib import Path
import sys
from unittest.mock import patch
@@ -13,12 +12,10 @@ import pytest
# Load the script-under-test as `check_import_time` (it's a hyphenated path
# inside `script/` that mirrors the existing `determine_jobs` pattern).
script_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "script")
)
script_dir = str((Path(__file__).parent / ".." / ".." / "script").resolve())
sys.path.insert(0, script_dir)
spec = importlib.util.spec_from_file_location(
"check_import_time", os.path.join(script_dir, "check_import_time.py")
"check_import_time", str(Path(script_dir) / "check_import_time.py")
)
check_import_time = importlib.util.module_from_spec(spec)
spec.loader.exec_module(check_import_time)
+101 -5
View File
@@ -3,7 +3,6 @@
from collections.abc import Generator
import importlib.util
import json
import os
from pathlib import Path
import sys
from unittest.mock import Mock, call, patch
@@ -11,9 +10,7 @@ from unittest.mock import Mock, call, patch
import pytest
# Add the script directory to Python path so we can import the module
script_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "script")
)
script_dir = str((Path(__file__).parent / ".." / ".." / "script").resolve())
sys.path.insert(0, script_dir)
# Import helpers module for patching
@@ -22,7 +19,7 @@ import helpers # noqa: E402
import script.helpers # noqa: E402
spec = importlib.util.spec_from_file_location(
"determine_jobs", os.path.join(script_dir, "determine-jobs.py")
"determine_jobs", str(Path(script_dir) / "determine-jobs.py")
)
determine_jobs = importlib.util.module_from_spec(spec)
spec.loader.exec_module(determine_jobs)
@@ -775,6 +772,88 @@ def test_should_run_import_time_with_branch() -> None:
mock_changed.assert_called_once_with("release")
@pytest.mark.parametrize(
("path", "expected_result"),
[
# Exact-file matches in the CI-irrelevant set.
(".yamllint", True),
(".github/dependabot.yml", True),
# Other top-level workflow files are irrelevant; ci.yml itself is not.
(".github/workflows/codeql.yml", True),
(".github/workflows/release.yml", True),
(".github/workflows/ci.yml", False),
# Nested files under workflows/ are not matched by the single-star glob.
(".github/workflows/matchers/gcc.json", False),
# build-image action: direct children only (single-star glob).
(".github/actions/build-image/action.yml", True),
(".github/actions/build-image/nested/file.yml", False),
# Other actions are CI-relevant.
(".github/actions/restore-python/action.yml", False),
# docker/** covers everything under docker/.
("docker/Dockerfile", True),
("docker/scripts/run.sh", True),
# Regular source files are CI-relevant.
("esphome/__main__.py", False),
("esphome/components/wifi/wifi_component.cpp", False),
("README.md", False),
("tests/script/test_determine_jobs.py", False),
],
)
def test_is_ci_irrelevant_path(path: str, expected_result: bool) -> None:
"""Test _is_ci_irrelevant_path mirrors the historic ci.yml path filter."""
assert determine_jobs._is_ci_irrelevant_path(path) == expected_result
@pytest.mark.parametrize(
("changed_files", "expected_result"),
[
# Empty diffs default to True — don't accidentally skip CI on a
# broken probe.
([], True),
# Any CI-relevant file flips the result to True.
(["esphome/__main__.py"], True),
(["esphome/components/wifi/wifi_component.cpp"], True),
(["README.md"], True),
# All-irrelevant diffs return False.
([".github/workflows/codeql.yml"], False),
(
[".github/workflows/codeql.yml", ".github/workflows/release.yml"],
False,
),
([".yamllint"], False),
([".github/dependabot.yml"], False),
(["docker/Dockerfile"], False),
(
[
".github/workflows/codeql.yml",
".github/dependabot.yml",
"docker/Dockerfile",
],
False,
),
# Mixed diffs always trigger CI.
(
[".github/workflows/codeql.yml", "esphome/__main__.py"],
True,
),
# ci.yml itself is treated as CI-relevant.
([".github/workflows/ci.yml"], True),
],
)
def test_should_run_core_ci(changed_files: list[str], expected_result: bool) -> None:
"""Test should_run_core_ci function."""
with patch.object(determine_jobs, "changed_files", return_value=changed_files):
assert determine_jobs.should_run_core_ci() == expected_result
def test_should_run_core_ci_with_branch() -> None:
"""Test should_run_core_ci passes the branch through to changed_files."""
with patch.object(determine_jobs, "changed_files") as mock_changed:
mock_changed.return_value = []
determine_jobs.should_run_core_ci("release")
mock_changed.assert_called_once_with("release")
@pytest.mark.parametrize(
("changed_files", "expected_result"),
[
@@ -1518,6 +1597,7 @@ def test_clang_tidy_mode_full_scan(
mock_should_run_clang_tidy: Mock,
mock_should_run_clang_format: Mock,
mock_should_run_python_linters: Mock,
mock_determine_cpp_unit_tests: Mock,
mock_changed_files: Mock,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
@@ -1529,6 +1609,9 @@ def test_clang_tidy_mode_full_scan(
mock_should_run_clang_tidy.return_value = True
mock_should_run_clang_format.return_value = False
mock_should_run_python_linters.return_value = False
# Without this mock, main() runs the real determine_cpp_unit_tests
# which loads the full component graph (~5s import of every component).
mock_determine_cpp_unit_tests.return_value = (False, [])
# Mock changed_files to return no component files
mock_changed_files.return_value = []
@@ -1584,6 +1667,7 @@ def test_clang_tidy_mode_targeted_scan(
mock_should_run_clang_tidy: Mock,
mock_should_run_clang_format: Mock,
mock_should_run_python_linters: Mock,
mock_determine_cpp_unit_tests: Mock,
mock_changed_files: Mock,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
@@ -1595,6 +1679,9 @@ def test_clang_tidy_mode_targeted_scan(
mock_should_run_clang_tidy.return_value = True
mock_should_run_clang_format.return_value = False
mock_should_run_python_linters.return_value = False
# Without this mock, main() runs the real determine_cpp_unit_tests
# which loads the full component graph (~5s import of every component).
mock_determine_cpp_unit_tests.return_value = (False, [])
# Create component names
components = [f"comp{i}" for i in range(component_count)]
@@ -2651,6 +2738,15 @@ def test_main_force_all_overrides_detection(
return_value={"should_run": "false"},
),
patch.object(determine_jobs, "should_run_benchmarks", return_value=False),
# create_intelligent_batches scans every tests/components/<name>/*.yaml
# under --force-all (~2500 YAML loads, ~10s in CI). This test only
# asserts that main() routes to it and returns non-empty -- the
# batching logic itself has its own dedicated tests.
patch.object(
determine_jobs,
"create_intelligent_batches",
return_value=([["fake_batch"]], None),
),
):
determine_jobs.main()
+1 -3
View File
@@ -12,9 +12,7 @@ import pytest
from pytest import MonkeyPatch
# Add the script directory to Python path so we can import helpers
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "script"))
)
sys.path.insert(0, str((Path(__file__).parent / ".." / ".." / "script").resolve()))
import helpers # noqa: E402
+1 -4
View File
@@ -1,6 +1,5 @@
"""Unit tests for script/build_helpers.py manifest override and build helpers."""
import os
from pathlib import Path
import sys
import textwrap
@@ -9,9 +8,7 @@ from unittest.mock import MagicMock, patch
import pytest
# Add the script directory to Python path so we can import build_helpers
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "script"))
)
sys.path.insert(0, str((Path(__file__).parent / ".." / ".." / "script").resolve()))
import build_helpers # noqa: E402
@@ -4,6 +4,7 @@ esphome:
esp32:
board: esp32-c6-devkitc-1
flash_size: 8MB
framework:
type: esp-idf
@@ -7,6 +7,9 @@ esp32:
variant: ESP32S3
framework:
type: esp-idf
# Use custom partition table with larger app partition (3MB)
# Default IDF partitions only allow 1.75MB which is too small for grouped tests
partitions: ../partitions_testing.csv
logger:
level: VERY_VERBOSE
+65 -2
View File
@@ -1,6 +1,11 @@
"""Tests for RP2040 component public helpers."""
"""Tests for RP2040 component public helpers and variant detection."""
from esphome.components.rp2040 import board_id_has_wifi
import pytest
from esphome.components.rp2040 import _detect_variant, board_id_has_wifi
from esphome.components.rp2040.const import VARIANT_RP2040, VARIANT_RP2350
import esphome.config_validation as cv
from esphome.const import CONF_BOARD, CONF_VARIANT
def test_board_id_has_wifi_for_known_wifi_board() -> None:
@@ -27,3 +32,61 @@ def test_board_id_has_wifi_for_unknown_board_returns_true() -> None:
"no CYW43" guard at compile time.
"""
assert board_id_has_wifi("not-a-real-board-id") is True
def test_detect_variant_derives_variant_from_board() -> None:
"""Board alone resolves to the matching variant."""
result = _detect_variant({CONF_BOARD: "rpipicow"})
assert result[CONF_BOARD] == "rpipicow"
assert result[CONF_VARIANT] == VARIANT_RP2040
def test_detect_variant_derives_variant_from_rp2350_board() -> None:
"""An RP2350 board resolves to ``RP2350``."""
result = _detect_variant({CONF_BOARD: "rpipico2"})
assert result[CONF_BOARD] == "rpipico2"
assert result[CONF_VARIANT] == VARIANT_RP2350
def test_detect_variant_only_picks_default_board_rp2040() -> None:
"""Variant alone picks Pico W as the canonical RP2040 board."""
result = _detect_variant({CONF_VARIANT: VARIANT_RP2040})
assert result[CONF_BOARD] == "rpipicow"
assert result[CONF_VARIANT] == VARIANT_RP2040
def test_detect_variant_only_picks_default_board_rp2350() -> None:
"""Variant alone picks Pico 2 W as the canonical RP2350 board."""
result = _detect_variant({CONF_VARIANT: VARIANT_RP2350})
assert result[CONF_BOARD] == "rpipico2w"
assert result[CONF_VARIANT] == VARIANT_RP2350
def test_detect_variant_matching_explicit_variant_passes() -> None:
"""Specifying both a board and the matching variant is allowed."""
result = _detect_variant({CONF_BOARD: "rpipico2", CONF_VARIANT: VARIANT_RP2350})
assert result[CONF_BOARD] == "rpipico2"
assert result[CONF_VARIANT] == VARIANT_RP2350
def test_detect_variant_mismatched_variant_raises() -> None:
"""Board/variant mismatch must be rejected and name the offending board."""
with pytest.raises(
cv.Invalid, match=r"does not match the selected board 'rpipicow'"
):
_detect_variant({CONF_BOARD: "rpipicow", CONF_VARIANT: VARIANT_RP2350})
def test_detect_variant_unknown_board_without_variant_raises() -> None:
"""Unknown board with no variant tells the user how to recover."""
with pytest.raises(cv.Invalid, match="please specify the chip variant"):
_detect_variant({CONF_BOARD: "not-a-real-board"})
def test_detect_variant_unknown_board_with_variant_passes() -> None:
"""Unknown board + explicit variant is accepted (with a warning)."""
result = _detect_variant(
{CONF_BOARD: "not-a-real-board", CONF_VARIANT: VARIANT_RP2040}
)
assert result[CONF_BOARD] == "not-a-real-board"
assert result[CONF_VARIANT] == VARIANT_RP2040
+2 -2
View File
@@ -70,11 +70,11 @@ def test_numeric_offset_slash() -> None:
def test_star() -> None:
assert _parse_cron_part("*", 0, 59, {}) == set(range(0, 60))
assert _parse_cron_part("*", 0, 59, {}) == set(range(60))
def test_question() -> None:
assert _parse_cron_part("?", 0, 59, {}) == set(range(0, 60))
assert _parse_cron_part("?", 0, 59, {}) == set(range(60))
def test_range() -> None:
+76 -2
View File
@@ -3,8 +3,20 @@
import pytest
from esphome.components.esp32 import const
from esphome.components.wifi import has_native_wifi, variant_has_wifi
from esphome.const import Platform
from esphome.components.wifi import (
check_placeholder_credentials,
has_native_wifi,
variant_has_wifi,
)
from esphome.const import (
CONF_AP,
CONF_NETWORKS,
CONF_SSID,
CONF_WIFI,
PLACEHOLDER_WIFI_SSID,
Platform,
)
from esphome.core import EsphomeError, Lambda
@pytest.mark.parametrize(
@@ -123,3 +135,65 @@ def test_has_native_wifi_esp32_without_variant_assumes_wifi() -> None:
def test_has_native_wifi_rp2040_without_board_assumes_wifi() -> None:
"""RP2040 without a board id falls open to True (custom-board default)."""
assert has_native_wifi(platform=Platform.RP2040) is True
def _wifi_config(
*,
networks: list[dict] | None = None,
ap: dict | None = None,
) -> dict:
"""Build a minimal config dict matching the post-validation shape."""
wifi: dict = {}
if networks is not None:
wifi[CONF_NETWORKS] = networks
if ap is not None:
wifi[CONF_AP] = ap
return {CONF_WIFI: wifi}
def test_check_placeholder_credentials_passes_with_real_ssid() -> None:
"""A real SSID compiles without complaint."""
config = _wifi_config(networks=[{CONF_SSID: "home_network"}])
assert check_placeholder_credentials(config) is None
def test_check_placeholder_credentials_refuses_placeholder_ssid() -> None:
"""The placeholder SSID is rejected with an actionable message."""
config = _wifi_config(networks=[{CONF_SSID: PLACEHOLDER_WIFI_SSID}])
with pytest.raises(EsphomeError) as exc_info:
check_placeholder_credentials(config)
message = str(exc_info.value)
assert "wifi.networks[0].ssid" in message
assert "secrets.yaml" in message
def test_check_placeholder_credentials_refuses_placeholder_in_second_network() -> None:
"""Index reporting picks the placeholder out of a mixed network list."""
config = _wifi_config(
networks=[
{CONF_SSID: "home_network"},
{CONF_SSID: PLACEHOLDER_WIFI_SSID},
],
)
with pytest.raises(EsphomeError) as exc_info:
check_placeholder_credentials(config)
assert "wifi.networks[1].ssid" in str(exc_info.value)
def test_check_placeholder_credentials_refuses_placeholder_ap_ssid() -> None:
"""An AP using the placeholder broadcast name is also refused."""
config = _wifi_config(ap={CONF_SSID: PLACEHOLDER_WIFI_SSID})
with pytest.raises(EsphomeError) as exc_info:
check_placeholder_credentials(config)
assert "wifi.ap.ssid" in str(exc_info.value)
def test_check_placeholder_credentials_no_wifi_passes() -> None:
"""Ethernet-only / wifi-less configs skip the check entirely."""
assert check_placeholder_credentials({}) is None
def test_check_placeholder_credentials_skips_template_ssid() -> None:
"""A templated (Lambda) SSID is not a string and is skipped."""
config = _wifi_config(networks=[{CONF_SSID: Lambda('return "x";')}])
assert check_placeholder_credentials(config) is None
+10 -10
View File
@@ -486,7 +486,7 @@ def test_preload_core_config_basic(setup_core: Path) -> None:
assert CONF_BUILD_PATH in config[CONF_ESPHOME]
# Verify default build path is "build/<device_name>"
build_path = config[CONF_ESPHOME][CONF_BUILD_PATH]
assert build_path.endswith(os.path.join("build", "test_device"))
assert build_path.endswith(str(Path("build") / "test_device"))
def test_preload_core_config_with_build_path(setup_core: Path) -> None:
@@ -523,7 +523,7 @@ def test_preload_core_config_env_build_path(setup_core: Path) -> None:
assert "test_device" in config[CONF_ESPHOME][CONF_BUILD_PATH]
# Verify it uses the env var path with device name appended
build_path = config[CONF_ESPHOME][CONF_BUILD_PATH]
expected_path = os.path.join("/env/build", "test_device")
expected_path = str(Path("/env/build") / "test_device")
assert build_path == expected_path or build_path == expected_path.replace(
"/", os.sep
)
@@ -739,7 +739,7 @@ async def test_add_includes_with_single_file(
"""Test add_includes copies a single header file to build directory."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create include file
include_file = tmp_path / "my_header.h"
@@ -769,7 +769,7 @@ async def test_add_includes_with_directory_unix(
"""Test add_includes copies all files from a directory on Unix."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create include directory with files
include_dir = tmp_path / "includes"
@@ -814,7 +814,7 @@ async def test_add_includes_with_directory_windows(
"""Test add_includes copies all files from a directory on Windows."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create include directory with files
include_dir = tmp_path / "includes"
@@ -856,7 +856,7 @@ async def test_add_includes_with_multiple_sources(
"""Test add_includes with multiple files and directories."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create various include sources
single_file = tmp_path / "single.h"
@@ -884,7 +884,7 @@ async def test_add_includes_empty_directory(
"""Test add_includes with an empty directory doesn't fail."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create empty directory
empty_dir = tmp_path / "empty"
@@ -906,7 +906,7 @@ async def test_add_includes_preserves_directory_structure_unix(
"""Test that add_includes preserves relative directory structure on Unix."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create nested directory structure
lib_dir = tmp_path / "lib"
@@ -940,7 +940,7 @@ async def test_add_includes_preserves_directory_structure_windows(
"""Test that add_includes preserves relative directory structure on Windows."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create nested directory structure
lib_dir = tmp_path / "lib"
@@ -973,7 +973,7 @@ async def test_add_includes_overwrites_existing_files(
"""Test that add_includes overwrites existing files in build directory."""
CORE.config_path = tmp_path / "config.yaml"
CORE.build_path = tmp_path / "build"
os.makedirs(CORE.build_path, exist_ok=True)
CORE.build_path.mkdir(parents=True, exist_ok=True)
# Create include file
include_file = tmp_path / "header.h"
+153 -3
View File
@@ -22,6 +22,7 @@ from esphome.const import (
KEY_CORE,
KEY_TARGET_FRAMEWORK,
KEY_TARGET_PLATFORM,
KEY_VARIANT,
)
from esphome.core import CORE
@@ -47,7 +48,12 @@ wifi:
"""
def _write_storage(storage_path: Path) -> None:
def _write_storage(
storage_path: Path,
*,
esp_platform: str = "ESP32",
core_platform: str | None = "esp32",
) -> None:
"""Write a vanilla StorageJSON sidecar for the cache tests."""
storage_path.parent.mkdir(parents=True, exist_ok=True)
data = {
@@ -59,14 +65,14 @@ def _write_storage(storage_path: Path) -> None:
"src_version": 1,
"address": "192.168.1.42",
"web_port": None,
"esp_platform": "ESP32",
"esp_platform": esp_platform,
"build_path": "/build/lite_test",
"firmware_bin_path": "/build/lite_test/firmware.bin",
"loaded_integrations": ["api", "logger", "ota", "wifi"],
"loaded_platforms": [],
"no_mdns": False,
"framework": "arduino",
"core_platform": "esp32",
"core_platform": core_platform,
}
storage_path.write_text(json.dumps(data))
@@ -123,6 +129,50 @@ def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None:
assert CORE.build_path == Path("/build/lite_test")
assert CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] == "esp32"
assert CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] == "arduino"
# upload_using_esptool reads get_esp32_variant() off CORE.data[KEY_ESP32].
from esphome.components.esp32.const import KEY_ESP32
assert CORE.data[KEY_ESP32][KEY_VARIANT] == "ESP32"
def test_load_compiled_config_populates_esp32_variant(tmp_path: Path) -> None:
"""ESP32 variants survive the cache fast path so esptool gets the right --chip."""
from esphome.components.esp32.const import KEY_ESP32
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
storage_dir = tmp_path / ".esphome" / "storage"
_write_storage(storage_dir / "lite_test.yaml.json", esp_platform="ESP32S3")
cache = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
_set_cache_mtime(cache, yaml_path, offset=5)
assert load_compiled_config(yaml_path) is not None
assert CORE.data[KEY_ESP32][KEY_VARIANT] == "ESP32S3"
def test_load_compiled_config_skips_esp32_block_for_other_platforms(
tmp_path: Path,
) -> None:
"""Non-esp32 targets shouldn't fabricate an esp32 data block."""
from esphome.components.esp32.const import KEY_ESP32
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
storage_dir = tmp_path / ".esphome" / "storage"
_write_storage(
storage_dir / "lite_test.yaml.json",
esp_platform="ESP8266",
core_platform="esp8266",
)
cache = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
_set_cache_mtime(cache, yaml_path, offset=5)
assert load_compiled_config(yaml_path) is not None
assert KEY_ESP32 not in CORE.data
@pytest.mark.parametrize(
@@ -203,6 +253,106 @@ def test_run_esphome_upload_and_logs_fall_back_when_no_cache(
mock_read.assert_called_once()
def test_run_esphome_upload_does_not_refresh_cache_without_sidecar(
tmp_path: Path,
) -> None:
"""Without a StorageJSON sidecar (no compile has run), the fallback
skips the cache write -- load_compiled_config requires the sidecar,
so writing the rendered (secret-resolved) YAML would be inert and
leak secrets to disk for nothing."""
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
with (
patch(
"esphome.__main__.read_config",
return_value={"esphome": {"name": "lite_test"}},
),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{"upload": lambda args, config: 0},
),
):
run_esphome(["esphome", "upload", str(yaml_path)])
mock_save.assert_not_called()
@pytest.mark.parametrize("command", ["upload", "logs"])
def test_run_esphome_upload_and_logs_refresh_cache_on_fallback(
tmp_path: Path, command: str
) -> None:
"""A stale-cache fallback rewrites the cache so the next call hits
the fast path. Without this, every upload/logs after a YAML edit
pays for read_config() until the next compile rewrites the cache."""
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
storage_dir = tmp_path / ".esphome" / "storage"
_write_storage(storage_dir / "lite_test.yaml.json")
cache = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
_set_cache_mtime(cache, yaml_path, offset=-60) # stale
fresh_config = {"esphome": {"name": "lite_test"}, "logger": {}}
with (
patch("esphome.__main__.read_config", return_value=fresh_config),
patch(
"esphome.compiled_config.save_compiled_config", wraps=save_compiled_config
) as mock_save,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{command: lambda args, config: 0},
),
):
assert run_esphome(["esphome", command, str(yaml_path)]) == 0
mock_save.assert_called_once_with(fresh_config)
# mtime is now newer than the source YAML, so a follow-up call hits
# the fast path instead of repeating read_config.
assert cache.stat().st_mtime >= yaml_path.stat().st_mtime
def test_run_esphome_upload_with_substitution_does_not_refresh_cache(
fresh_cache_files: Path,
) -> None:
"""`-s` substitutions skip the cache on both read and write -- saving
here would clobber the cache with a substitution-specific config."""
with (
patch("esphome.__main__.read_config", return_value={"esphome": {}}),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{"upload": lambda args, config: 0},
),
):
run_esphome(["esphome", "-s", "var", "val", "upload", str(fresh_cache_files)])
mock_save.assert_not_called()
def test_run_esphome_compile_does_not_refresh_cache_via_fallback(
fresh_cache_files: Path,
) -> None:
"""Compile writes the cache through update_storage_json, not via the
upload/logs fallback path -- the fallback save would skip the
storage_should_clean check."""
with (
patch("esphome.__main__.read_config", return_value={"esphome": {}}),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{"compile": lambda args, config: 0},
),
):
run_esphome(["esphome", "compile", str(fresh_cache_files)])
mock_save.assert_not_called()
def test_run_esphome_upload_with_substitution_skips_cache(
fresh_cache_files: Path,
) -> None:
@@ -67,7 +67,7 @@ def test_iter_component_configs_with_multi_conf(mock_get_component: Mock) -> Non
configs = list(config.iter_component_configs(test_config))
assert len(configs) == 2
for domain, component, conf in configs:
for domain, _component, conf in configs:
assert domain == "switch"
assert "name" in conf
+150 -15
View File
@@ -203,7 +203,7 @@ def test_generate_idf_component_yml_basic(tmp_component):
tmp_component.data = {"description": "test", "repository": {"url": "http://aaa"}}
result = generate_idf_component_yml(tmp_component)
assert result == "description: test\nversion: 1.0.0\nrepository: http://aaa\n"
assert result == "description: test\nrepository: http://aaa\n"
def test_generate_idf_component_yml_with_dependencies(tmp_component, tmp_path):
@@ -217,18 +217,16 @@ def test_generate_idf_component_yml_with_dependencies(tmp_component, tmp_path):
assert (
result
== f"""version: 1.0.0
dependencies:
== f"""dependencies:
dep:
version: '1.0'
override_path: {dep.path}
"""
)
def test_generate_idf_component_yml_missing_path_reraises(tmp_component):
# A dep without a path and without a recognised source should re-raise
# the underlying RuntimeError instead of silently producing a bad manifest.
def test_generate_idf_component_yml_missing_path_raises(tmp_component):
# A dep without a path is a contract violation — every dep is expected
# to have been downloaded before YAML generation. Raise loudly.
dep = IDFComponent("foo/bar", "1.0", source=None)
tmp_component.dependencies = [dep]
@@ -263,9 +261,14 @@ def test_check_library_data_invalid_platform(esp32_idf_core):
_check_library_data({"platforms": ["other"], "frameworks": "*"})
def test_check_library_data_invalid_framework(esp32_idf_core):
with pytest.raises(InvalidIDFComponent):
_check_library_data({"platforms": "*", "frameworks": ["other"]})
def test_check_library_data_invalid_framework(
esp32_idf_core: None, caplog: pytest.LogCaptureFixture
) -> None:
# Framework mismatch is a warning, not a hard skip: the library is still
# included so that PIO manifests that only list "arduino" (but actually
# compile under IDF) can be used without forking them.
_check_library_data({"name": "lib", "platforms": "*", "frameworks": ["other"]})
assert "do not include 'espidf'" in caplog.text
def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
@@ -290,7 +293,7 @@ def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
result = run_extra_script(script, library_dir=tmp_path, idf_target="esp32")
assert result.libpath == [os.path.join("src", "esp32")]
assert result.libpath == [str(Path("src") / "esp32")]
assert result.libs == ["algobsec"]
assert ("BAR", "1") in result.cppdefines
assert "FOO" in result.cppdefines
@@ -422,15 +425,37 @@ def test_convert_library_with_repository():
result = _convert_library_to_component(lib)
assert result.name == "foo/bar"
assert result.version == "1.2.3"
assert result.version == "*"
assert isinstance(result.source, GitSource)
assert result.source.ref == "v1.2.3"
def test_convert_library_missing_ref():
def test_convert_library_with_branch_ref():
lib = Library("name", None, "https://github.com/foo/bar.git#some-branch")
result = _convert_library_to_component(lib)
assert result.name == "foo/bar"
assert result.version == "*"
assert isinstance(result.source, GitSource)
assert result.source.ref == "some-branch"
def test_convert_library_missing_ref_uses_default_branch():
"""A bare URL with no #ref clones the remote's default branch.
Matches PIO's lib_deps behavior and external_components handling --
git.clone_or_update with ref=None leaves the depth-1 clone on
whatever branch the remote HEAD points at.
"""
lib = Library("name", None, "https://github.com/foo/bar.git")
with pytest.raises(ValueError):
_convert_library_to_component(lib)
result = _convert_library_to_component(lib)
assert result.name == "foo/bar"
assert result.version == "*"
assert isinstance(result.source, GitSource)
assert result.source.ref is None
def test_convert_library_registry(monkeypatch):
@@ -485,3 +510,113 @@ def test_process_dependencies_skips_invalid(tmp_component):
_process_dependencies(tmp_component)
assert tmp_component.dependencies == []
def test_process_dependencies_dict_form(tmp_component, monkeypatch):
"""PIO library.json shorthand ``{"owner/Name": "version"}`` is honored.
Iterating a dict gives string keys, which would silently fail the
``"name" in dependency`` substring check. Normalize to list-of-dicts
first so the dict form (used by e.g. tesla-ble for its nanopb dep)
is treated the same as the verbose list form.
"""
captured: list[Library] = []
def fake_generate(library):
captured.append(library)
return IDFComponent(
library.name, library.version, source=URLSource("http://dummy.com")
)
tmp_component.data = {
"dependencies": {
"nanopb/Nanopb": "^0.4.91",
"BareName": "1.2.3",
}
}
monkeypatch.setattr(
esphome.espidf.component, "_generate_idf_component", fake_generate
)
monkeypatch.setattr(esphome.espidf.component, "_check_library_data", lambda x: None)
_process_dependencies(tmp_component)
assert len(tmp_component.dependencies) == 2
names = sorted(lib.name for lib in captured)
versions = sorted(lib.version for lib in captured)
assert names == ["BareName", "nanopb/Nanopb"]
assert versions == ["1.2.3", "^0.4.91"]
def test_process_dependencies_dict_form_with_url_value(tmp_component, monkeypatch):
"""A dict-value that's a URL gets routed to ``repository`` like the list form."""
captured: list[Library] = []
def fake_generate(library):
captured.append(library)
return IDFComponent(library.name, "*", source=URLSource("http://dummy.com"))
tmp_component.data = {
"dependencies": {
"foo/Bar": "https://github.com/foo/bar.git#main",
}
}
monkeypatch.setattr(
esphome.espidf.component, "_generate_idf_component", fake_generate
)
monkeypatch.setattr(esphome.espidf.component, "_check_library_data", lambda x: None)
_process_dependencies(tmp_component)
assert len(captured) == 1
assert captured[0].name == "foo/Bar"
assert captured[0].version is None
assert captured[0].repository == "https://github.com/foo/bar.git#main"
def test_process_dependencies_dict_form_with_nested_spec(tmp_component, monkeypatch):
"""A dict-value that's itself a dict is merged into the entry.
PIO's library.json allows ``{"owner/Name": {"version": "...", ...}}``
for entries that need fields beyond just a version (platforms,
frameworks, etc.). The extra fields flow into _check_library_data
via the entry merge.
"""
captured: list[Library] = []
checked: list[dict] = []
def fake_generate(library):
captured.append(library)
return IDFComponent(
library.name, library.version, source=URLSource("http://dummy.com")
)
tmp_component.data = {
"dependencies": {
"nanopb/Nanopb": {"version": "^0.4.91", "platforms": "espidf"},
}
}
monkeypatch.setattr(
esphome.espidf.component, "_generate_idf_component", fake_generate
)
monkeypatch.setattr(
esphome.espidf.component,
"_check_library_data",
checked.append,
)
_process_dependencies(tmp_component)
assert len(captured) == 1
assert captured[0].name == "nanopb/Nanopb"
assert captured[0].version == "^0.4.91"
# Extra spec fields reach _check_library_data so platform/framework
# gating still applies.
assert checked == [
{
"name": "Nanopb",
"owner": "nanopb",
"version": "^0.4.91",
"platforms": "espidf",
}
]
+156
View File
@@ -0,0 +1,156 @@
"""Tests for esphome.espidf.framework helpers."""
# pylint: disable=protected-access
from pathlib import Path
from unittest.mock import patch
import pytest
from esphome.espidf.framework import _clone_idf_with_submodules, _parse_git_source
@pytest.mark.parametrize(
("source", "expected"),
[
# github:// shorthand
(
"github://espressif/esp-idf",
("https://github.com/espressif/esp-idf.git", None),
),
(
"github://espressif/esp-idf@master",
("https://github.com/espressif/esp-idf.git", "master"),
),
(
"github://espressif/esp-idf@release/v6.0",
("https://github.com/espressif/esp-idf.git", "release/v6.0"),
),
# explicit https://github.com/...git URL
(
"https://github.com/espressif/esp-idf.git",
("https://github.com/espressif/esp-idf.git", None),
),
(
"https://github.com/espressif/esp-idf.git@master",
("https://github.com/espressif/esp-idf.git", "master"),
),
(
"https://github.com/espressif/esp-idf.git@v6.0.1",
("https://github.com/espressif/esp-idf.git", "v6.0.1"),
),
# Tolerate a trailing ".git" on the shorthand so the user doesn't
# silently end up with a doubled "...esp-idf.git.git" URL.
(
"github://espressif/esp-idf.git",
("https://github.com/espressif/esp-idf.git", None),
),
(
"github://espressif/esp-idf.git@master",
("https://github.com/espressif/esp-idf.git", "master"),
),
],
)
def test_parse_git_source_recognized(
source: str, expected: tuple[str, str | None]
) -> None:
assert _parse_git_source(source) == expected
@pytest.mark.parametrize(
"source",
[
# archive URLs fall through to the existing download path
"https://github.com/espressif/esp-idf/archive/refs/heads/master.zip",
"https://dl.espressif.com/dl/esp-idf/v6.0.1/esp-idf-v6.0.1.zip",
"https://github.com/esphome-libs/esp-idf/releases/download/v5.5.4/esp-idf-v5.5.4.tar.xz",
# SSH and other git protocols are intentionally rejected — match
# external_components, which only recognizes github:// + structured
# dicts for these.
"git@github.com:espressif/esp-idf.git",
"ssh://git@github.com/espressif/esp-idf.git",
"git://github.com/espressif/esp-idf.git",
# non-GitHub .git URLs are intentionally rejected for the same reason
"https://gitlab.com/foo/bar.git",
"https://github.example.com/foo/bar.git",
],
)
def test_parse_git_source_rejected(source: str) -> None:
assert _parse_git_source(source) is None
def _make_idf_tree(framework_path: Path) -> None:
"""Create the minimum tree _clone_idf_with_submodules sanity-checks for."""
(framework_path / "tools").mkdir(parents=True)
(framework_path / "tools" / "idf_tools.py").write_text("# stub\n")
def test_clone_idf_with_submodules_without_ref(tmp_path: Path) -> None:
framework_path = tmp_path / "idf"
framework_path.mkdir()
_make_idf_tree(framework_path)
with patch("esphome.git.run_git_command", return_value="") as run_git_command_mock:
_clone_idf_with_submodules(
framework_path, "https://github.com/espressif/esp-idf.git", None
)
# No ref -> just clone + submodule update, no fetch/reset.
calls = [c.args[0] for c in run_git_command_mock.call_args_list]
assert calls[0] == [
"git",
"clone",
"--depth=1",
"--",
"https://github.com/espressif/esp-idf.git",
str(framework_path),
]
assert calls[-1][:5] == ["git", "submodule", "update", "--init", "--recursive"]
assert not any(c[1] == "fetch" for c in calls)
assert not any(c[1] == "reset" for c in calls)
def test_clone_idf_with_submodules_with_ref(tmp_path: Path) -> None:
framework_path = tmp_path / "idf"
framework_path.mkdir()
_make_idf_tree(framework_path)
with patch("esphome.git.run_git_command", return_value="") as run_git_command_mock:
_clone_idf_with_submodules(
framework_path,
"https://github.com/espressif/esp-idf.git",
"master",
)
calls = [c.args[0] for c in run_git_command_mock.call_args_list]
# clone, fetch ref, reset hard, submodule update
assert calls[0][:2] == ["git", "clone"]
assert calls[1] == [
"git",
"fetch",
"--depth=1",
"--",
"origin",
"master",
]
assert calls[2] == ["git", "reset", "--hard", "FETCH_HEAD"]
assert calls[3][:5] == ["git", "submodule", "update", "--init", "--recursive"]
def test_clone_idf_with_submodules_raises_when_tree_missing(
tmp_path: Path,
) -> None:
framework_path = tmp_path / "idf"
framework_path.mkdir()
# Deliberately do NOT call _make_idf_tree — simulate a clone that
# returned 0 but produced no tools/idf_tools.py.
with (
patch("esphome.git.run_git_command", return_value=""),
pytest.raises(RuntimeError, match="no usable ESP-IDF tree"),
):
_clone_idf_with_submodules(
framework_path,
"https://github.com/espressif/esp-idf.git",
None,
)
+58
View File
@@ -0,0 +1,58 @@
"""Tests for esphome.espidf.toolchain helpers."""
# pylint: disable=protected-access
from unittest.mock import patch
from esphome.const import CONF_FRAMEWORK, CONF_SOURCE
from esphome.core import CORE
from esphome.espidf import toolchain
def test_get_framework_source_override_no_config():
"""When CORE.config hasn't been set, no override is returned."""
CORE.config = None
assert toolchain._get_framework_source_override() is None
def test_get_framework_source_override_no_esp32_section():
"""A config without an esp32 section yields no override."""
CORE.config = {}
assert toolchain._get_framework_source_override() is None
def test_get_framework_source_override_no_framework_source():
"""An esp32 section without framework.source yields no override."""
CORE.config = {"esp32": {CONF_FRAMEWORK: {}}}
assert toolchain._get_framework_source_override() is None
def test_get_framework_source_override_returns_value():
"""A user-supplied framework source is returned verbatim."""
url = "https://example.com/esp-idf-v{VERSION}.tar.xz"
CORE.config = {"esp32": {CONF_FRAMEWORK: {CONF_SOURCE: url}}}
assert toolchain._get_framework_source_override() == url
def test_get_esphome_esp_idf_paths_forwards_source_override():
"""_get_esphome_esp_idf_paths threads the override into check_esp_idf_install."""
url = "https://my-mirror/esp-idf-v{VERSION}.tar.xz"
CORE.config = {"esp32": {CONF_FRAMEWORK: {CONF_SOURCE: url}}}
# Hit a fresh cache key so check_esp_idf_install is actually called.
toolchain._cache().paths.clear()
with patch.object(
toolchain, "check_esp_idf_install", return_value=("/fw", "/penv")
) as mock_install:
toolchain._get_esphome_esp_idf_paths("5.5.4")
mock_install.assert_called_once_with("5.5.4", source_url=url)
def test_get_esphome_esp_idf_paths_no_override():
"""When no source override is configured, source_url=None is passed."""
CORE.config = {}
toolchain._cache().paths.clear()
with patch.object(
toolchain, "check_esp_idf_install", return_value=("/fw", "/penv")
) as mock_install:
toolchain._get_esphome_esp_idf_paths("5.5.4")
mock_install.assert_called_once_with("5.5.4", source_url=None)
+302 -1
View File
@@ -4,7 +4,7 @@ from datetime import datetime, timedelta
import os
from pathlib import Path
from typing import Any
from unittest.mock import Mock
from unittest.mock import Mock, patch
import pytest
@@ -1001,3 +1001,304 @@ def test_refresh_picks_up_new_remote_commits(
"--hard",
"old_sha",
]
def test_resolve_symlink_stub_returns_none_on_non_windows(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""On non-Windows, resolve_symlink_stub returns None without calling git."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "file.yaml"
stub.write_text("static/file.yaml")
with patch("esphome.git.sys.platform", "linux"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
mock_run_git_command.assert_not_called()
def test_resolve_symlink_stub_returns_target_for_mode_120000(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A mode-120000 file is recognised as a stub; its target Path is returned."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
(repo_dir / "static").mkdir()
target = repo_dir / "static" / "real.yaml"
target.write_text("esphome:\n name: real\n")
stub = repo_dir / "real.yaml"
stub.write_text("static/real.yaml")
mock_run_git_command.return_value = "120000 abc123 0\treal.yaml"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result == target.resolve()
# Stub file itself was not modified — only inspected.
assert stub.read_text() == "static/real.yaml"
def test_resolve_symlink_stub_resolves_relative_parent_paths(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""Symlink targets with ``..`` segments resolve correctly within the repo."""
repo_dir = tmp_path / "repo"
(repo_dir / "subdir").mkdir(parents=True)
(repo_dir / "static").mkdir()
target = repo_dir / "static" / "shared.yaml"
target.write_text("shared content")
stub = repo_dir / "subdir" / "shared.yaml"
stub.write_text("../static/shared.yaml")
mock_run_git_command.return_value = "120000 abc123 0\tsubdir/shared.yaml"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result == target.resolve()
def test_resolve_symlink_stub_refuses_escape_outside_repo(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A symlink pointing outside the repository is not followed."""
outside = tmp_path / "outside.yaml"
outside.write_text("sensitive")
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "escape.yaml"
stub.write_text("../outside.yaml")
mock_run_git_command.return_value = "120000 abc123 0\tescape.yaml"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_returns_none_for_real_symlink(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A real symlink already opens transparently, so the helper short-circuits.
Skipped on Windows where symlink creation requires
SeCreateSymbolicLinkPrivilege.
"""
if os.name == "nt":
pytest.skip("Requires symlink-creation privilege on Windows")
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
target = repo_dir / "real.yaml"
target.write_text("real content")
real_link = repo_dir / "link.yaml"
real_link.symlink_to("real.yaml")
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, real_link)
assert result is None
# No git call needed for real symlinks.
mock_run_git_command.assert_not_called()
def test_resolve_symlink_stub_returns_none_for_regular_file(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A regular file (mode 100644) whose content looks path-shaped is not
followed."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
regular = repo_dir / "looks_like_path.txt"
regular.write_text("static/something.yaml")
mock_run_git_command.return_value = "100644 abc123 0\tlooks_like_path.txt"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, regular)
assert result is None
def test_resolve_symlink_stub_returns_none_when_git_fails(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""If ``git ls-files`` fails (e.g. not a repo), the helper returns None."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "real.yaml"
stub.write_text("static/real.yaml")
mock_run_git_command.side_effect = GitCommandError("ls-files exploded")
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_returns_none_for_non_utf8_content(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A file whose bytes are not valid UTF-8 must not raise — return None."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "binary.bin"
stub.write_bytes(b"\xff\xfe\x00\xff")
mock_run_git_command.return_value = "120000 abc123 0\tbinary.bin"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_preserves_whitespace_in_target(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""Only trailing CR/LF is stripped — internal whitespace is preserved."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
target_dir = repo_dir / "dir with spaces"
target_dir.mkdir()
target = target_dir / "real.yaml"
target.write_text("hello")
stub = repo_dir / "link.yaml"
# Trailing newline (as git's checkout may append) is stripped, but
# whitespace inside the target path itself must survive.
stub.write_bytes(b"dir with spaces/real.yaml\n")
mock_run_git_command.return_value = "120000 abc123 0\tlink.yaml"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result == target.resolve()
def test_resolve_symlink_stub_returns_none_for_directory_target(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A symlink pointing at a directory has no file content to load."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
(repo_dir / "dir_target").mkdir()
stub = repo_dir / "link_to_dir"
stub.write_text("dir_target")
mock_run_git_command.return_value = "120000 abc123 0\tlink_to_dir"
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_returns_none_when_resolve_raises(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""Path.resolve() raising (e.g. on a malformed target) must not propagate."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "broken.yaml"
stub.write_text("ignored")
mock_run_git_command.return_value = "120000 abc123 0\tbroken.yaml"
with (
patch("esphome.git.sys.platform", "win32"),
patch.object(Path, "resolve", side_effect=OSError("bad path")),
):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_returns_none_when_file_missing(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A file path that doesn't exist is rejected before git is consulted."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
missing = repo_dir / "ghost.yaml" # not created
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, missing)
assert result is None
mock_run_git_command.assert_not_called()
def test_resolve_symlink_stub_returns_none_when_path_outside_repo(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""A file path that isn't under repo_dir is rejected (ValueError from relative_to)."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
outside = tmp_path / "stray.yaml"
outside.write_text("something")
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, outside)
assert result is None
mock_run_git_command.assert_not_called()
def test_resolve_symlink_stub_returns_none_when_untracked(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""Empty `git ls-files` output (untracked file) makes the helper return None."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "untracked.yaml"
stub.write_text("static/foo.yaml")
mock_run_git_command.return_value = ""
with patch("esphome.git.sys.platform", "win32"):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
def test_resolve_symlink_stub_returns_none_when_read_bytes_raises(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""An OSError from read_bytes() (e.g. file vanished mid-call) must not propagate."""
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
stub = repo_dir / "racy.yaml"
stub.write_text("static/racy.yaml")
mock_run_git_command.return_value = "120000 abc123 0\tracy.yaml"
with (
patch("esphome.git.sys.platform", "win32"),
patch.object(Path, "read_bytes", side_effect=OSError("vanished")),
):
result = git.resolve_symlink_stub(repo_dir, stub)
assert result is None
+2 -1
View File
@@ -7,7 +7,7 @@ import stat
from unittest.mock import MagicMock, patch
from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
from hypothesis import given
from hypothesis import given, settings
from hypothesis.strategies import ip_addresses
import pytest
@@ -151,6 +151,7 @@ def test_is_ip_address__invalid(host):
assert actual is False
@settings(deadline=None)
@given(value=ip_addresses(v=4).map(str))
def test_is_ip_address__valid(value):
actual = helpers.is_ip_address(value)
+3 -3
View File
@@ -11,7 +11,7 @@ from pathlib import Path
import re
import sys
import time
from typing import Any
from typing import Any, Self
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
@@ -5110,11 +5110,11 @@ class MockSerial:
self.timeout = 0.1
self._is_open = False
def __enter__(self) -> MockSerial:
def __enter__(self) -> Self:
self._is_open = True
return self
def __exit__(self, *args: Any) -> None:
def __exit__(self, *args: object) -> None:
self._is_open = False
@property
+128
View File
@@ -0,0 +1,128 @@
"""Tests for esphome.espidf.size_summary.print_summary."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from esphome.espidf.size_summary import print_summary
def _write_size_json(tmp_path: Path, data: dict) -> Path:
"""Drop a fake esp_idf_size.json under ``tmp_path`` and return the path."""
out = tmp_path / "esp_idf_size.json"
out.write_text(json.dumps(data))
return out
def _esp32_size_data() -> dict:
"""Synthetic esp_idf_size.json for the original ESP32 (split IRAM/DRAM)."""
return {
"image_size": 827455,
"memory_types": {
"DRAM": {
"size": 180736,
"used": 47332,
"sections": {
".dram0.bss": {"abbrev_name": ".bss", "size": 30616},
".dram0.data": {"abbrev_name": ".data", "size": 16716},
},
},
"IRAM": {
"size": 131072,
"used": 80351,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 79323},
".iram0.vectors": {"abbrev_name": ".vectors", "size": 1028},
},
},
},
}
def _s3_size_data() -> dict:
"""Synthetic esp_idf_size.json for ESP32-S3 (unified DIRAM)."""
return {
"image_size": 724215,
"memory_types": {
"DIRAM": {
"size": 341760,
"used": 104999,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 58051},
".dram0.bss": {"abbrev_name": ".bss", "size": 27088},
".dram0.data": {"abbrev_name": ".data", "size": 19708},
".noinit": {"abbrev_name": ".noinit", "size": 152},
},
},
"IRAM": {
"size": 16384,
"used": 16384,
"sections": {
".iram0.text": {"abbrev_name": ".text", "size": 15356},
".iram0.vectors": {"abbrev_name": ".vectors", "size": 1028},
},
},
},
}
def test_print_summary_esp32_uses_dram(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Original ESP32: DRAM has no ``.text``, so RAM = DRAM.used / DRAM.size unchanged."""
size_json = _write_size_json(tmp_path, _esp32_size_data())
print_summary(size_json, partitions_csv=None)
out = capsys.readouterr().out
assert "RAM:" in out
assert "used 47332 bytes from 180736 bytes" in out
def test_print_summary_s3_falls_back_to_diram(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""ESP32-S3 with no DRAM key falls back to DIRAM and reports raw region usage."""
size_json = _write_size_json(tmp_path, _s3_size_data())
print_summary(size_json, partitions_csv=None)
out = capsys.readouterr().out
assert "used 104999 bytes from 341760 bytes" in out
def test_print_summary_skips_when_diram_total_collapses(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A zero-size region drops the RAM line rather than divide by zero."""
size_json = _write_size_json(
tmp_path,
{
"memory_types": {
"DIRAM": {
"size": 0,
"used": 0,
"sections": {},
},
},
},
)
print_summary(size_json, partitions_csv=None)
out = capsys.readouterr().out
assert "RAM:" not in out
def test_print_summary_handles_missing_json(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Missing size json is non-fatal and prints nothing."""
print_summary(tmp_path / "does_not_exist.json", partitions_csv=None)
assert capsys.readouterr().out == ""
def test_print_summary_handles_no_memory_types(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""A size json without ``memory_types`` still doesn't crash."""
size_json = _write_size_json(tmp_path, {"image_size": 0})
print_summary(size_json, partitions_csv=None)
assert capsys.readouterr().out == ""
+72 -1
View File
@@ -9,7 +9,7 @@ from unittest.mock import MagicMock, Mock, patch
import pytest
from esphome import storage_json
from esphome.const import CONF_DISABLED, CONF_MDNS
from esphome.const import CONF_DISABLED, CONF_MDNS, Toolchain
from esphome.core import CORE
@@ -308,6 +308,7 @@ def test_storage_json_from_esphome_core(setup_core: Path) -> None:
mock_core.loaded_platforms = {"sensor"}
mock_core.config = {CONF_MDNS: {CONF_DISABLED: True}}
mock_core.target_framework = "esp-idf"
mock_core.toolchain = Toolchain.ESP_IDF
with patch("esphome.components.esp32.get_esp32_variant") as mock_variant:
mock_variant.return_value = "ESP32-C3"
@@ -327,6 +328,7 @@ def test_storage_json_from_esphome_core(setup_core: Path) -> None:
assert result.no_mdns is True
assert result.framework == "esp-idf"
assert result.core_platform == "esp32"
assert result.toolchain == "esp-idf"
def test_storage_json_from_esphome_core_mdns_enabled(setup_core: Path) -> None:
@@ -345,10 +347,12 @@ def test_storage_json_from_esphome_core_mdns_enabled(setup_core: Path) -> None:
mock_core.loaded_platforms = set()
mock_core.config = {} # No MDNS config means enabled
mock_core.target_framework = "arduino"
mock_core.toolchain = None
result = storage_json.StorageJSON.from_esphome_core(mock_core, old=None)
assert result.no_mdns is False
assert result.toolchain is None
def test_storage_json_load_valid_file(tmp_path: Path) -> None:
@@ -470,6 +474,73 @@ def test_storage_json_equality() -> None:
assert storage1 != "not a storage object"
def _make_storage_with_toolchain(
toolchain: str | None,
) -> storage_json.StorageJSON:
return storage_json.StorageJSON(
storage_version=1,
name="dev",
friendly_name=None,
comment=None,
esphome_version="2024.1.0",
src_version=1,
address="dev.local",
web_port=None,
target_platform="ESP32",
build_path=Path("/build"),
firmware_bin_path=Path("/build/firmware.bin"),
loaded_integrations=set(),
loaded_platforms=set(),
no_mdns=False,
framework="esp-idf",
core_platform="esp32",
toolchain=toolchain,
)
def test_storage_json_toolchain_round_trip(setup_core: Path) -> None:
"""Sidecar toolchain survives save -> load -> apply_to_core."""
storage = _make_storage_with_toolchain("esp-idf")
path = setup_core / "storage.json"
path.write_text(storage.to_json())
# Serialization key is stable -- device-builder relies on it.
assert json.loads(path.read_text())["toolchain"] == "esp-idf"
loaded = storage_json.StorageJSON.load(path)
assert loaded is not None
assert loaded.toolchain == "esp-idf"
CORE.toolchain = None
with patch("esphome.components.esp32.get_esp32_variant"):
loaded.apply_to_core()
assert CORE.toolchain == Toolchain.ESP_IDF
def test_storage_json_apply_to_core_preserves_cli_toolchain(
setup_core: Path,
) -> None:
"""A CLI-set CORE.toolchain wins over the sidecar value."""
loaded = _make_storage_with_toolchain("esp-idf")
CORE.toolchain = Toolchain.PLATFORMIO
with patch("esphome.components.esp32.get_esp32_variant"):
loaded.apply_to_core()
assert CORE.toolchain == Toolchain.PLATFORMIO
def test_storage_json_apply_to_core_ignores_unknown_toolchain(
setup_core: Path,
) -> None:
"""Unknown enum values (corrupt sidecar / newer ESPHome) fall through to None."""
loaded = _make_storage_with_toolchain("gcc")
CORE.toolchain = None
with patch("esphome.components.esp32.get_esp32_variant"):
loaded.apply_to_core()
assert CORE.toolchain is None
def test_esphome_storage_json_as_dict() -> None:
"""Test EsphomeStorageJSON.as_dict returns correct dictionary."""
storage = storage_json.EsphomeStorageJSON(
+84 -2
View File
@@ -1,4 +1,3 @@
import glob
import logging
from pathlib import Path
from typing import Any
@@ -106,7 +105,7 @@ REMOTES = {
# Collect all input YAML files for test_substitutions_fixtures parametrized tests:
HERE = Path(__file__).parent
BASE_DIR = HERE / "fixtures" / "substitutions"
SOURCES = sorted(glob.glob(str(BASE_DIR / "*.input.yaml")))
SOURCES = sorted(str(p) for p in BASE_DIR.glob("*.input.yaml"))
assert SOURCES, f"test_substitutions_fixtures: No input YAML files found in {BASE_DIR}"
@@ -838,3 +837,86 @@ def test_include_vars_applied_to_lambda_value(tmp_path: Path) -> None:
assert isinstance(result["value"], Lambda)
assert result["value"].value == 'return "bar";'
@patch("esphome.git.resolve_symlink_stub")
@patch("esphome.git.clone_or_update")
def test_remote_package_symlink_stub_is_followed(
mock_clone_or_update: MagicMock,
mock_resolve_symlink_stub: MagicMock,
tmp_path: Path,
) -> None:
"""When a package YAML is a scalar (symlink stub) and resolve_symlink_stub
returns a target, the loader follows the target and uses its content."""
CORE.config_path = tmp_path / "test.yaml"
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
(repo_dir / "static").mkdir()
# Stub file: content is the target path string (simulating Windows behavior).
stub = repo_dir / "file1.yaml"
stub.write_text("static/file1.yaml")
# Real target with valid YAML mapping.
target = repo_dir / "static" / "file1.yaml"
target.write_text("substitutions:\n hello: world\n")
mock_clone_or_update.return_value = (repo_dir, None)
mock_resolve_symlink_stub.return_value = target
config: dict[str, Any] = {
"packages": {
"test_package": {
"url": "https://github.com/esphome/repo1",
"ref": "main",
"files": ["file1.yaml"],
}
}
}
# Must succeed (does not raise the helpful cv.Invalid) because the stub
# was followed and a valid mapping was loaded from the target.
do_packages_pass(config)
assert mock_resolve_symlink_stub.called
@patch("esphome.git.clone_or_update")
def test_remote_package_scalar_yaml_raises_helpful_error(
mock_clone_or_update: MagicMock, tmp_path: Path
) -> None:
"""A remote package YAML that is a top-level scalar (e.g. an unmaterialized
git symlink on Windows) raises a clear cv.Invalid, not AttributeError.
Regression test for the case where a repo containing a YAML symlink,
checked out on Windows without symlink privilege, lands as a short text
file containing the symlink target path. PyYAML parses that as a bare
string scalar; the package loader must reject it with a human-readable
error instead of dying inside ``.get()``.
"""
CORE.config_path = tmp_path / "test.yaml"
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
# Simulate the broken-symlink state: a YAML file whose entire content is
# the symlink target string. PyYAML parses this as a top-level scalar.
(repo_dir / "file1.yaml").write_text("static/file1.yaml")
mock_clone_or_update.return_value = (repo_dir, None)
config: dict[str, Any] = {
"packages": {
"test_package": {
"url": "https://github.com/esphome/repo1",
"ref": "main",
"files": ["file1.yaml"],
}
}
}
with pytest.raises(cv.Invalid) as exc_info:
do_packages_pass(config)
msg = str(exc_info.value)
assert "mapping at the top level" in msg
assert "file1.yaml" in msg
+4 -4
View File
@@ -1358,7 +1358,7 @@ def test_clean_build_handles_readonly_files(
# Create a read-only file (simulating git pack files on Windows)
readonly_file = git_dir / "pack-abc123.pack"
readonly_file.write_text("pack data")
os.chmod(readonly_file, stat.S_IRUSR) # Read-only
readonly_file.chmod(stat.S_IRUSR) # Read-only
# Setup mocks
mock_core.relative_pioenvs_path.return_value = pioenvs_dir
@@ -1393,7 +1393,7 @@ def test_clean_all_handles_readonly_files(
subdir.mkdir()
readonly_file = subdir / "readonly.txt"
readonly_file.write_text("content")
os.chmod(readonly_file, stat.S_IRUSR) # Read-only
readonly_file.chmod(stat.S_IRUSR) # Read-only
# Verify file is read-only
assert not os.access(readonly_file, os.W_OK)
@@ -1422,7 +1422,7 @@ def test_clean_build_reraises_for_other_errors(
test_file.write_text("content")
# Make subdir read-only so files inside can't be deleted
os.chmod(subdir, stat.S_IRUSR | stat.S_IXUSR)
subdir.chmod(stat.S_IRUSR | stat.S_IXUSR)
# Setup mocks
mock_core.relative_pioenvs_path.return_value = pioenvs_dir
@@ -1440,7 +1440,7 @@ def test_clean_build_reraises_for_other_errors(
clean_build()
finally:
# Cleanup - restore write permission so tmp_path cleanup works
os.chmod(subdir, stat.S_IRWXU)
subdir.chmod(stat.S_IRWXU)
# Tests for get_build_info()
+159 -1
View File
@@ -34,6 +34,14 @@ def clear_secrets_cache() -> None:
yaml_util._SECRET_CACHE.clear()
@pytest.fixture(autouse=True)
def clear_core_frontmatter() -> None:
"""Reset CORE.frontmatter between tests."""
core.CORE.frontmatter = {}
yield
core.CORE.frontmatter = {}
def test_include_with_vars(fixture_path: Path) -> None:
yaml_file = fixture_path / "yaml_util" / "includetest.yaml"
@@ -899,7 +907,7 @@ def test_format_path_current_obj_without_location_falls_back_to_key():
"""An ESPHomeDataBase current_obj with no esp_range falls back to the key's location."""
class _NoRange(ESPHomeDataBase, str):
pass
__slots__ = ()
obj = _NoRange.__new__(_NoRange, "value")
str.__init__(obj)
@@ -1182,3 +1190,153 @@ def test_track_yaml_loads_records_resolved_paths(tmp_path: Path) -> None:
with track_yaml_loads() as loaded:
yaml_util.load_yaml(link)
assert target.resolve() in loaded
# ---------------------------------------------------------------------------
# YAML frontmatter
# ---------------------------------------------------------------------------
def test_frontmatter_parsed_and_stored_on_core(tmp_path: Path) -> None:
"""A leading `---`-separated YAML document is stored as frontmatter and
stripped from the returned config."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text(
"author: Jesse\nlabels: [office, climate]\n---\nesphome:\n name: my_node\n"
)
config = yaml_util.load_yaml(yaml_file)
# Config does not contain frontmatter keys
assert "author" not in config
assert "labels" not in config
assert config["esphome"]["name"] == "my_node"
# Frontmatter is stored on CORE keyed by resolved path
frontmatter = core.CORE.frontmatter[yaml_file.resolve()]
assert frontmatter["author"] == "Jesse"
assert frontmatter["labels"] == ["office", "climate"]
def test_frontmatter_absent_when_single_document(tmp_path: Path) -> None:
"""A YAML file with a single document does not populate CORE.frontmatter."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text("esphome:\n name: my_node\n")
yaml_util.load_yaml(yaml_file)
assert yaml_file.resolve() not in core.CORE.frontmatter
def test_frontmatter_absent_when_leading_doc_separator(tmp_path: Path) -> None:
"""A leading `---` with no content above it is just a document start marker,
not frontmatter, and must not populate CORE.frontmatter."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text("---\nesphome:\n name: my_node\n")
config = yaml_util.load_yaml(yaml_file)
assert config["esphome"]["name"] == "my_node"
assert yaml_file.resolve() not in core.CORE.frontmatter
def test_frontmatter_supports_arbitrary_keys(tmp_path: Path) -> None:
"""Frontmatter keys are not validated — any structure is accepted."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text(
"any_key: any_value\n"
"nested:\n"
" count: 42\n"
" items:\n"
" - a\n"
" - b\n"
"---\n"
"esphome:\n"
" name: t\n"
)
yaml_util.load_yaml(yaml_file)
frontmatter = core.CORE.frontmatter[yaml_file.resolve()]
assert frontmatter["any_key"] == "any_value"
assert frontmatter["nested"]["count"] == 42
assert frontmatter["nested"]["items"] == ["a", "b"]
def test_frontmatter_supports_deeply_nested_paths(tmp_path: Path) -> None:
"""Frontmatter preserves deeply nested dict/list structures intact."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text(
"device:\n"
" metadata:\n"
" location:\n"
" building: HQ\n"
" floor: 3\n"
" room:\n"
" number: 302\n"
" occupants:\n"
" - name: Jesse\n"
" role:\n"
" title: maintainer\n"
" since: 2021\n"
" - name: Alice\n"
" role:\n"
" title: contributor\n"
" since: 2024\n"
"---\n"
"esphome:\n"
" name: t\n"
)
yaml_util.load_yaml(yaml_file)
fm = core.CORE.frontmatter[yaml_file.resolve()]
room = fm["device"]["metadata"]["location"]["room"]
assert room["number"] == 302
assert room["occupants"][0]["name"] == "Jesse"
assert room["occupants"][0]["role"]["title"] == "maintainer"
assert room["occupants"][0]["role"]["since"] == 2021
assert room["occupants"][1]["role"]["title"] == "contributor"
def test_frontmatter_more_than_two_documents_raises(tmp_path: Path) -> None:
"""Three or more YAML documents is unsupported and must raise."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text("a: 1\n---\nb: 2\n---\nc: 3\n")
with pytest.raises(EsphomeError, match="at most two are supported"):
yaml_util.load_yaml(yaml_file)
def test_frontmatter_empty_frontmatter_doc_not_stored(tmp_path: Path) -> None:
"""An empty (null) frontmatter document is treated as no frontmatter."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text("---\n---\nesphome:\n name: t\n")
config = yaml_util.load_yaml(yaml_file)
assert config["esphome"]["name"] == "t"
assert yaml_file.resolve() not in core.CORE.frontmatter
def test_frontmatter_empty_config_doc(tmp_path: Path) -> None:
"""An empty config document after a frontmatter document yields an empty config."""
yaml_file = tmp_path / "main.yaml"
yaml_file.write_text("only: frontmatter\n---\n")
config = yaml_util.load_yaml(yaml_file)
assert config == {}
assert core.CORE.frontmatter[yaml_file.resolve()]["only"] == "frontmatter"
def test_frontmatter_included_file_stored(tmp_path: Path) -> None:
"""Frontmatter on an !include'd file is also captured on CORE, keyed by
that file's resolved path."""
inc = tmp_path / "child.yaml"
inc.write_text("child_meta: hello\n---\nchild_key: value\n")
main = tmp_path / "main.yaml"
main.write_text("esphome:\n name: t\nchild: !include child.yaml\n")
config = yaml_util.load_yaml(main)
# !include is deferred; force resolution so the child file actually loads
force_load_include_files(config)
assert config["child"].load()["child_key"] == "value"
# Main file has no frontmatter
assert main.resolve() not in core.CORE.frontmatter
# Included file's frontmatter is captured
assert core.CORE.frontmatter[inc.resolve()]["child_meta"] == "hello"