diff --git a/esphome/__main__.py b/esphome/__main__.py index 33f1727696..4358cdf1a3 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -2445,25 +2445,38 @@ def run_esphome(argv): # `read_config()` pipeline (parse + schema validate + final-validate + # external component refresh) for those two subcommands is dead work and # produces a wall of "Reading configuration ..." log lines for every - # remote install. Source the platform / build metadata from the - # StorageJSON sidecar instead; fall back to full validation if it's - # missing or stale so a cold cache never produces a worse outcome. + # remote install. Reload the validated config that the last compile + # cached alongside the StorageJSON sidecar; fall back to full + # validation if the cache is missing or older than the YAML so a + # cold cache never produces a worse outcome. config = None if getattr(args, "from_storage_json", False) and args.command in ( "upload", "logs", ): - from esphome.lite_config import load_lite_config_from_storage + from esphome.storage_json import ( + StorageJSON, + ext_storage_path, + load_compiled_config, + ) - config = load_lite_config_from_storage(conf_path, command_line_substitutions) + config = load_compiled_config(conf_path) + if config is not None: + storage = StorageJSON.load(ext_storage_path(conf_path.name)) + if storage is None: + config = None + else: + storage.apply_to_core() + _LOGGER.info( + "Loaded validated config cache for %s, skipping validation.", + conf_path.name, + ) if config is None: _LOGGER.warning( - "StorageJSON sidecar missing or stale for %s; falling back to " - "full config validation.", + "Validated config cache for %s is missing or older than the " + "YAML; falling back to full config validation.", conf_path, ) - else: - _LOGGER.info("Loaded device metadata from StorageJSON sidecar.") if config is None: config = read_config( diff --git a/esphome/lite_config.py b/esphome/lite_config.py deleted file mode 100644 index 89816a00c7..0000000000 --- a/esphome/lite_config.py +++ /dev/null @@ -1,197 +0,0 @@ -"""Lite-config loader for ``upload`` / ``logs`` fast paths. - -When the caller (typically a dashboard) already has a known-good -firmware binary on disk and only needs the CLI to ship bytes to a -device or stream logs back, running the full ``read_config()`` -pipeline is dead work. It re-parses every ``!include``, runs every -component's schema validator, executes all final-validate hooks, and -fetches/refreshes external components -- producing a fully validated -``Config`` object that the upload / logs subcommands only consult for a -handful of leaf keys. - -This module loads the per-config ``StorageJSON`` sidecar that was -written by the last successful ``compile`` run, populates ``CORE`` with -the platform / build metadata from it, and re-parses just enough of the -YAML head (substitutions + packages, no schema validation) to recover -``api:`` / ``logger:`` / ``ota:`` / ``mqtt:`` / network blocks for the -subcommand callers. - -If the sidecar is missing or older than the YAML, this returns ``None`` -so the dispatcher can fall back to the full ``read_config()`` path. -That makes the flag a pure optimisation: a cold cache never produces a -worse outcome than today. -""" - -from __future__ import annotations - -import logging -from pathlib import Path -from typing import Any - -from esphome.const import ( - CONF_API, - CONF_ESPHOME, - CONF_ETHERNET, - CONF_FRIENDLY_NAME, - CONF_LOGGER, - CONF_MQTT, - CONF_NAME, - CONF_OPENTHREAD, - CONF_OTA, - CONF_USE_ADDRESS, - CONF_WEB_SERVER, - CONF_WIFI, - KEY_CORE, - KEY_TARGET_FRAMEWORK, - KEY_TARGET_PLATFORM, -) -from esphome.core import CORE -from esphome.storage_json import StorageJSON, ext_storage_path -from esphome.types import ConfigType - -_LOGGER = logging.getLogger(__name__) - -# Top-level keys we copy from the raw YAML into the lite config dict. -# Anything outside this list is irrelevant to ``upload`` / ``logs``. -_LITE_TOP_LEVEL_KEYS: tuple[str, ...] = ( - CONF_ESPHOME, - CONF_API, - CONF_LOGGER, - CONF_OTA, - CONF_MQTT, - CONF_WIFI, - CONF_ETHERNET, - CONF_OPENTHREAD, - CONF_WEB_SERVER, -) - - -def _parse_yaml_head( - conf_path: Path, command_line_substitutions: dict[str, Any] | None -) -> dict[str, Any] | None: - """Load and lightly process the YAML, without schema validation. - - Resolves ``!include`` / ``!secret`` (via ``yaml_util``), - substitutions, and ``packages:`` -- the three passes whose output - the upload / logs subcommands need. Returns ``None`` on any error - so the caller can fall back to the full ``read_config()`` path. - """ - from esphome import yaml_util - from esphome.components.packages import resolve_packages - from esphome.components.substitutions import do_substitution_pass - - try: - config = yaml_util.load_yaml(conf_path) - except Exception: # pylint: disable=broad-except - return None - - try: - config = do_substitution_pass(config, command_line_substitutions) - except Exception: # pylint: disable=broad-except - return None - - try: - config = resolve_packages( - config, command_line_substitutions=command_line_substitutions - ) - except Exception: # pylint: disable=broad-except - return None - - return config - - -def _build_lite_config(raw_config: dict[str, Any]) -> ConfigType: - """Project the raw YAML down to the keys upload / logs read.""" - return {key: raw_config[key] for key in _LITE_TOP_LEVEL_KEYS if key in raw_config} - - -def _populate_core_from_storage(storage: StorageJSON) -> None: - """Populate ``CORE`` fields the subcommands read off the platform.""" - CORE.name = storage.name - CORE.friendly_name = storage.friendly_name - CORE.build_path = storage.build_path - CORE.loaded_integrations = set(storage.loaded_integrations) - CORE.loaded_platforms = set(storage.loaded_platforms) - - core_platform = storage.core_platform or ( - storage.target_platform.lower() if storage.target_platform else None - ) - CORE.data.setdefault(KEY_CORE, {}) - if core_platform is not None: - CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = core_platform - if storage.framework is not None: - CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = storage.framework - - -def _ensure_address_in_config(config: ConfigType, storage: StorageJSON) -> None: - """Make sure ``CORE.address`` resolves once the lite config is set. - - ``CORE.address`` reads ``use_address`` off the wifi / ethernet / - openthread block; if the dashboard rewrote the YAML between the - last compile and now those blocks may be missing from the lite - parse. As a backstop, lift the address from the StorageJSON when - we have one and the YAML doesn't supply one already. - """ - if storage.address is None: - return - for network_type in (CONF_WIFI, CONF_ETHERNET, CONF_OPENTHREAD): - if network_type in config and CONF_USE_ADDRESS in config[network_type]: - return - # Fall back to a synthetic wifi block so ``CORE.address`` resolves. - config.setdefault(CONF_WIFI, {})[CONF_USE_ADDRESS] = storage.address - - -def load_lite_config_from_storage( - conf_path: Path, command_line_substitutions: dict[str, Any] | None = None -) -> ConfigType | None: - """Build a minimal config + populate CORE from the StorageJSON sidecar. - - Returns the lite config dict on success, or ``None`` when the - dispatcher should fall back to ``read_config()`` (sidecar missing, - stale, unreadable, or required fields absent). - """ - storage_path = ext_storage_path(conf_path.name) - try: - yaml_stat = conf_path.stat() - except OSError: - return None - try: - storage_stat = storage_path.stat() - except OSError: - return None - - # The sidecar is written by `compile`; if the YAML has been edited - # since, the cached platform / loaded_integrations may no longer - # describe what the binary on disk was built from. Fall back to a - # full validation pass in that case. - if storage_stat.st_mtime < yaml_stat.st_mtime: - return None - - storage = StorageJSON.load(storage_path) - if storage is None: - return None - if not storage.target_platform and not storage.core_platform: - # An incomplete sidecar (e.g. from a wizard run that never - # compiled) can't drive the upload / logs subcommands. - return None - - raw_config = _parse_yaml_head(conf_path, command_line_substitutions) - if raw_config is None: - return None - if CONF_ESPHOME not in raw_config or CONF_NAME not in raw_config[CONF_ESPHOME]: - return None - - config = _build_lite_config(raw_config) - - # Backfill `esphome:` block fields from the sidecar so downstream - # callers that read `config["esphome"]["name"]` work even when the - # YAML uses substitutions that didn't survive the light parse. - esphome_block = config.setdefault(CONF_ESPHOME, {}) - esphome_block.setdefault(CONF_NAME, storage.name) - if storage.friendly_name is not None: - esphome_block.setdefault(CONF_FRIENDLY_NAME, storage.friendly_name) - - _populate_core_from_storage(storage) - _ensure_address_in_config(config, storage) - - return config diff --git a/esphome/storage_json.py b/esphome/storage_json.py index c6df16ce78..d760297150 100644 --- a/esphome/storage_json.py +++ b/esphome/storage_json.py @@ -11,7 +11,7 @@ from esphome import const from esphome.const import CONF_DISABLED, CONF_MDNS from esphome.core import CORE from esphome.helpers import write_file_if_changed -from esphome.types import CoreType +from esphome.types import ConfigType, CoreType _LOGGER = logging.getLogger(__name__) @@ -56,6 +56,82 @@ def archive_storage_path() -> Path: return CORE.relative_config_path("archive") +def compiled_config_path(config_filename: str) -> Path: + """Path to the cached validated config alongside the storage sidecar. + + Written after every successful compile from the validated config + dict (via ``yaml_util.dump``). Powers the dispatcher's + ``--from-storage-json`` fast path in ``esphome upload`` and + ``esphome logs`` so they can skip the full ``read_config()`` + validation pipeline. + + Lives next to the existing JSON sidecar (``.json``) but + in its own file so the small metadata sidecar isn't bloated by + configs that can reach a megabyte or more once packages and + substitutions are expanded. + """ + return CORE.data_dir / "storage" / f"{config_filename}.validated.yaml" + + +def save_compiled_config(config: ConfigType) -> None: + """Dump the validated config to its sidecar YAML file. + + Called by the writer at the end of ``compile`` so the next call + to ``esphome upload --from-storage-json`` / ``esphome logs + --from-storage-json`` for this YAML can skip validation. Failures + here are non-fatal: the worst case is that the fast path falls + back to a full ``read_config`` next time. + """ + from esphome import yaml_util + + try: + # show_secrets=True so the cache is self-contained; the file + # lives next to the binary it describes, in the same trust zone + # as the rest of .esphome/storage/. + rendered = yaml_util.dump(config, show_secrets=True) + except Exception as err: # pylint: disable=broad-except + _LOGGER.debug("Skipping compiled config cache write: %s", err) + return + try: + write_file_if_changed(compiled_config_path(CORE.config_filename), rendered) + except OSError as err: + _LOGGER.debug("Skipping compiled config cache write: %s", err) + + +def load_compiled_config(config_path: Path) -> ConfigType | None: + """Load the cached validated config for ``--from-storage-json``. + + Returns ``None`` (so the caller falls back to ``read_config``) if + the cache is missing or older than the source YAML. The mtime + check catches the common "user edited the YAML and forgot to + recompile" case; deeper drift (an edited ``!include`` whose + parent YAML mtime didn't change) is the user's responsibility — + this flag is opt-in and assumes the caller knows the binary on + disk matches the cache. + """ + cache_path = compiled_config_path(config_path.name) + try: + yaml_mtime = config_path.stat().st_mtime + except OSError: + return None + try: + cache_mtime = cache_path.stat().st_mtime + except OSError: + return None + if cache_mtime < yaml_mtime: + return None + + from esphome import yaml_util + + try: + # clear_secrets=False so we don't disturb any in-flight secret + # state; the cache is self-contained and resolves no !secret + # references. + return yaml_util.load_yaml(cache_path, clear_secrets=False) + except Exception: # pylint: disable=broad-except + return None + + def _to_path_if_not_none(value: str | None) -> Path | None: """Convert a string to Path if it's not None.""" return Path(value) if value is not None else None @@ -256,6 +332,33 @@ class StorageJSON: except Exception: # pylint: disable=broad-except return None + def apply_to_core(self) -> None: + """Populate ``CORE`` from this sidecar. + + Used by the ``--from-storage-json`` fast path in + ``esphome upload`` / ``esphome logs``: those subcommands read + a handful of ``CORE`` attributes (``target_platform``, + ``build_path``, ``name``, ``loaded_integrations``) that the + normal flow populates during ``read_config``. Lifting them off + the sidecar lets us skip the validation pass entirely. + """ + from esphome.const import KEY_CORE, KEY_TARGET_FRAMEWORK, KEY_TARGET_PLATFORM + + CORE.name = self.name + CORE.friendly_name = self.friendly_name + CORE.build_path = self.build_path + CORE.loaded_integrations = set(self.loaded_integrations) + CORE.loaded_platforms = set(self.loaded_platforms) + + core_platform = self.core_platform or ( + self.target_platform.lower() if self.target_platform else None + ) + CORE.data.setdefault(KEY_CORE, {}) + if core_platform is not None: + CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = core_platform + if self.framework is not None: + CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = self.framework + def __eq__(self, o) -> bool: return isinstance(o, StorageJSON) and self.as_dict() == o.as_dict() diff --git a/esphome/writer.py b/esphome/writer.py index 2fa43fa5eb..67df3d1f22 100644 --- a/esphome/writer.py +++ b/esphome/writer.py @@ -24,7 +24,7 @@ from esphome.helpers import ( walk_files, write_file_if_changed, ) -from esphome.storage_json import StorageJSON, storage_path +from esphome.storage_json import StorageJSON, save_compiled_config, storage_path _LOGGER = logging.getLogger(__name__) @@ -109,6 +109,14 @@ def update_storage_json() -> None: path = storage_path() old = StorageJSON.load(path) new = StorageJSON.from_esphome_core(CORE, old) + + # Always refresh the validated-config cache so `esphome upload + # --from-storage-json` and `esphome logs --from-storage-json` can + # skip re-validating after this compile. Lives in its own file + # next to the sidecar; mtime gates staleness on the read side. + if CORE.config is not None: + save_compiled_config(CORE.config) + if old == new: return diff --git a/tests/unit_tests/test_main_from_storage_json.py b/tests/unit_tests/test_main_from_storage_json.py new file mode 100644 index 0000000000..2433511f7f --- /dev/null +++ b/tests/unit_tests/test_main_from_storage_json.py @@ -0,0 +1,264 @@ +"""Tests for the `--from-storage-json` validated-config cache fast path.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from esphome.__main__ import run_esphome +from esphome.const import ( + CONF_API, + CONF_ESPHOME, + CONF_NAME, + KEY_CORE, + KEY_TARGET_FRAMEWORK, + KEY_TARGET_PLATFORM, +) +from esphome.core import CORE +from esphome.storage_json import StorageJSON, compiled_config_path, load_compiled_config + +_VALIDATED_CONFIG_YAML = """\ +esphome: + name: lite_test + friendly_name: Lite Test Device +esp32: + board: nodemcu-32s +logger: + baud_rate: 115200 +api: + port: 6053 + encryption: + key: 6dGhpcyBpcyBhIHRlc3Q= +ota: + - platform: esphome + port: 3232 + password: secret +wifi: + ssid: ssid + use_address: 192.168.1.42 +""" + + +def _write_storage(storage_path: Path) -> None: + """Write a vanilla StorageJSON sidecar for the cache tests.""" + storage_path.parent.mkdir(parents=True, exist_ok=True) + data = { + "storage_version": 1, + "name": "lite_test", + "friendly_name": "Lite Test Device", + "comment": None, + "esphome_version": "2026.1.0", + "src_version": 1, + "address": "192.168.1.42", + "web_port": None, + "esp_platform": "ESP32", + "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", + } + storage_path.write_text(json.dumps(data)) + + +@pytest.fixture +def fresh_cache_files(tmp_path: Path) -> Path: + """Set up a YAML + StorageJSON + validated-config cache. + + Cache mtime is bumped 5s past the YAML so the staleness check + treats it as fresh. + """ + 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_path = storage_dir / "lite_test.yaml.validated.yaml" + cache_path.write_text(_VALIDATED_CONFIG_YAML) + yaml_stat = yaml_path.stat() + os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) + + return yaml_path + + +def test_compiled_config_path_lives_alongside_sidecar(setup_core: Path) -> None: + """The cache file shape is predictable from the YAML filename.""" + path = compiled_config_path("device.yaml") + assert str(path).endswith("storage/device.yaml.validated.yaml") + + +def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None: + """Fresh cache → returns the validated config dict.""" + config = load_compiled_config(fresh_cache_files) + + assert config is not None + assert config[CONF_ESPHOME][CONF_NAME] == "lite_test" + assert config[CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q=" + assert config["ota"][0]["password"] == "secret" + + +def test_load_compiled_config_missing_cache(tmp_path: Path) -> None: + """No cache file on disk → None so caller falls back.""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text("esphome:\n name: lite_test\n") + CORE.config_path = yaml_path + + assert load_compiled_config(yaml_path) is None + + +def test_load_compiled_config_stale_cache(tmp_path: Path) -> None: + """Cache older than the YAML → None (the YAML was edited).""" + 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" + storage_dir.mkdir(parents=True, exist_ok=True) + cache_path = storage_dir / "lite_test.yaml.validated.yaml" + cache_path.write_text(_VALIDATED_CONFIG_YAML) + + yaml_stat = yaml_path.stat() + # Cache is one minute older than the YAML — the YAML's been edited + # since the last compile, so the cache no longer describes the + # binary on disk. + os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime - 60)) + + assert load_compiled_config(yaml_path) is None + + +def test_load_compiled_config_corrupt_cache(tmp_path: Path) -> None: + """Cache file is unparseable → None so caller falls back.""" + 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" + storage_dir.mkdir(parents=True, exist_ok=True) + cache_path = storage_dir / "lite_test.yaml.validated.yaml" + cache_path.write_text("not: valid: yaml: [") + + yaml_stat = yaml_path.stat() + os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) + + assert load_compiled_config(yaml_path) is None + + +def test_storage_json_apply_to_core_populates_target_platform(tmp_path: Path) -> None: + """apply_to_core sets the CORE attributes upload / logs read.""" + storage_path = tmp_path / "lite_test.yaml.json" + _write_storage(storage_path) + storage = StorageJSON.load(storage_path) + assert storage is not None + + storage.apply_to_core() + + assert CORE.name == "lite_test" + assert CORE.friendly_name == "Lite Test Device" + 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" + assert "api" in CORE.loaded_integrations + + +@pytest.mark.parametrize("command", ["upload", "logs"]) +def test_run_esphome_from_storage_json_skips_read_config( + command: str, fresh_cache_files: Path +) -> None: + """`--from-storage-json` makes the dispatcher skip read_config().""" + yaml_path = fresh_cache_files + + captured = {} + + def _stub(_args, config): + captured["config"] = config + return 0 + + with ( + patch("esphome.__main__.read_config") as mock_read, + patch.dict("esphome.__main__.POST_CONFIG_ACTIONS", {command: _stub}), + ): + result = run_esphome( + ["esphome", command, "--from-storage-json", str(yaml_path)] + ) + + mock_read.assert_not_called() + assert result == 0 + # The dispatcher hands the cached config dict through unchanged. + assert captured["config"][CONF_ESPHOME][CONF_NAME] == "lite_test" + assert captured["config"][CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q=" + + +@pytest.mark.parametrize("command", ["upload", "logs"]) +def test_run_esphome_from_storage_json_falls_back_when_missing( + tmp_path: Path, command: str +) -> None: + """With no cache on disk, the dispatcher falls back to read_config().""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text("esphome:\n name: lite_test\n") + + with ( + patch("esphome.__main__.read_config", return_value=None) as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {command: lambda args, config: 0}, + ), + ): + result = run_esphome( + ["esphome", command, "--from-storage-json", str(yaml_path)] + ) + + mock_read.assert_called_once() + assert result == 2 + + +def test_run_esphome_from_storage_json_falls_back_when_stale( + tmp_path: Path, +) -> None: + """If YAML mtime > cache mtime, dispatcher falls back to read_config().""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text("esphome:\n name: lite_test\n") + + storage_dir = tmp_path / ".esphome" / "storage" + _write_storage(storage_dir / "lite_test.yaml.json") + + cache_path = storage_dir / "lite_test.yaml.validated.yaml" + cache_path.write_text(_VALIDATED_CONFIG_YAML) + yaml_stat = yaml_path.stat() + os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime - 60)) + + with ( + patch("esphome.__main__.read_config", return_value=None) as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {"upload": lambda args, config: 0}, + ), + ): + run_esphome(["esphome", "upload", "--from-storage-json", str(yaml_path)]) + + mock_read.assert_called_once() + + +def test_run_esphome_without_flag_still_calls_read_config( + fresh_cache_files: Path, +) -> None: + """Sanity: omitting the flag preserves the current behaviour.""" + yaml_path = fresh_cache_files + + with ( + patch("esphome.__main__.read_config", return_value=None) as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {"upload": lambda args, config: 0}, + ), + ): + run_esphome(["esphome", "upload", str(yaml_path)]) + + mock_read.assert_called_once() diff --git a/tests/unit_tests/test_main_lite_config.py b/tests/unit_tests/test_main_lite_config.py deleted file mode 100644 index cf0fa3c086..0000000000 --- a/tests/unit_tests/test_main_lite_config.py +++ /dev/null @@ -1,280 +0,0 @@ -"""Tests for the StorageJSON-backed lite config fast path.""" - -from __future__ import annotations - -import json -import os -from pathlib import Path -from unittest.mock import patch - -import pytest - -from esphome.__main__ import run_esphome -from esphome.const import ( - CONF_API, - CONF_ESPHOME, - CONF_LOGGER, - CONF_NAME, - CONF_OTA, - CONF_USE_ADDRESS, - CONF_WIFI, - KEY_CORE, - KEY_TARGET_FRAMEWORK, - KEY_TARGET_PLATFORM, -) -from esphome.core import CORE -from esphome.lite_config import load_lite_config_from_storage - -# A YAML the lite parser can load. Contains every block ``upload`` / -# ``logs`` cares about (api, logger, ota, wifi) plus a substitution so -# the tests exercise the substitution pass too. -_SAMPLE_YAML = """\ -substitutions: - device_name: lite_test - encryption_key: 6dGhpcyBpcyBhIHRlc3Q= - -esphome: - name: ${device_name} - friendly_name: Lite Test Device - -esp32: - board: nodemcu-32s - -logger: - -api: - encryption: - key: ${encryption_key} - -ota: - - platform: esphome - -wifi: - ssid: "ssid" - password: "password" - use_address: 192.168.1.42 -""" - - -def _write_storage( - storage_path: Path, - *, - name: str = "lite_test", - friendly_name: str | None = "Lite Test Device", - address: str | None = "192.168.1.42", - target_platform: str = "ESP32", - core_platform: str | None = "esp32", - framework: str | None = "arduino", - build_path: str | None = "/build/lite_test", - firmware_bin_path: str | None = "/build/lite_test/firmware.bin", - loaded_integrations: list[str] | None = None, - loaded_platforms: list[str] | None = None, -) -> None: - """Write a StorageJSON sidecar to ``storage_path`` for the tests.""" - storage_path.parent.mkdir(parents=True, exist_ok=True) - data = { - "storage_version": 1, - "name": name, - "friendly_name": friendly_name, - "comment": None, - "esphome_version": "2026.1.0", - "src_version": 1, - "address": address, - "web_port": None, - "esp_platform": target_platform, - "build_path": build_path, - "firmware_bin_path": firmware_bin_path, - "loaded_integrations": loaded_integrations or ["api", "logger", "ota", "wifi"], - "loaded_platforms": loaded_platforms or [], - "no_mdns": False, - "framework": framework, - "core_platform": core_platform, - } - storage_path.write_text(json.dumps(data)) - - -@pytest.fixture -def lite_config_files(tmp_path: Path) -> tuple[Path, Path]: - """Create a YAML + matching StorageJSON for the lite-config tests. - - The YAML lives at ``/lite_test.yaml`` so ``CORE.data_dir`` - points into ``/.esphome``. The storage sidecar is written - after the YAML and then back-dated, so callers can rely on - ``storage.mtime >= yaml.mtime`` matching the production happy - path (compile writes the sidecar last). - """ - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text(_SAMPLE_YAML) - CORE.config_path = yaml_path - - storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" - _write_storage(storage_path) - # Storage is younger than the YAML by virtue of being created last, - # but make the relationship explicit for tests that depend on it. - yaml_stat = yaml_path.stat() - os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) - - return yaml_path, storage_path - - -def test_lite_config_happy_path(lite_config_files: tuple[Path, Path]) -> None: - """Storage + YAML in sync: lite config returns a populated dict and CORE.""" - yaml_path, _ = lite_config_files - - config = load_lite_config_from_storage(yaml_path, {}) - - assert config is not None - assert config[CONF_ESPHOME][CONF_NAME] == "lite_test" - assert CONF_LOGGER in config - assert CONF_API in config - assert config[CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q=" - assert CONF_OTA in config - assert config[CONF_OTA][0]["platform"] == "esphome" - # Lite config should NOT contain bulky non-essential keys like esp32. - assert "esp32" not in config - - # CORE populated from storage. - assert CORE.name == "lite_test" - assert CORE.friendly_name == "Lite Test Device" - assert CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] == "esp32" - assert CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] == "arduino" - assert CORE.build_path == Path("/build/lite_test") - assert "api" in CORE.loaded_integrations - - -def test_lite_config_missing_storage_returns_none(tmp_path: Path) -> None: - """No StorageJSON sidecar → return None so caller falls back.""" - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text(_SAMPLE_YAML) - CORE.config_path = yaml_path - - assert load_lite_config_from_storage(yaml_path, {}) is None - - -def test_lite_config_stale_storage_returns_none(tmp_path: Path) -> None: - """StorageJSON older than YAML → return None. - - The YAML may have grown a new ``api:`` key since the binary was - built; falling back to ``read_config()`` catches that. - """ - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text(_SAMPLE_YAML) - CORE.config_path = yaml_path - - storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" - _write_storage(storage_path) - - yaml_stat = yaml_path.stat() - # Storage is one minute older than the YAML. - os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime - 60)) - - assert load_lite_config_from_storage(yaml_path, {}) is None - - -def test_lite_config_address_backfilled_from_storage(tmp_path: Path) -> None: - """No wifi/ethernet block in YAML → address comes from storage. - - Some users gate their network blocks behind substitutions that - don't fire in the lite parse. The loader should fall back to the - sidecar's ``address`` so ``CORE.address`` still resolves. - """ - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text( - "esphome:\n name: lite_test\nesp32:\n board: nodemcu-32s\n" - "logger:\napi:\nota:\n - platform: esphome\n" - ) - CORE.config_path = yaml_path - - storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" - _write_storage(storage_path, address="10.0.0.5") - - yaml_stat = yaml_path.stat() - os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) - - config = load_lite_config_from_storage(yaml_path, {}) - assert config is not None - assert config[CONF_WIFI][CONF_USE_ADDRESS] == "10.0.0.5" - - -def test_lite_config_corrupt_storage_returns_none(tmp_path: Path) -> None: - """A malformed sidecar (StorageJSON.load returns None) → fall back.""" - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text(_SAMPLE_YAML) - CORE.config_path = yaml_path - - storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" - storage_path.parent.mkdir(parents=True, exist_ok=True) - storage_path.write_text("not valid json{") - - yaml_stat = yaml_path.stat() - os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) - - assert load_lite_config_from_storage(yaml_path, {}) is None - - -@pytest.mark.parametrize("command", ["upload", "logs"]) -def test_run_esphome_from_storage_json_skips_read_config( - tmp_path: Path, - command: str, - lite_config_files: tuple[Path, Path], -) -> None: - """`--from-storage-json` makes the dispatcher skip read_config().""" - yaml_path, _ = lite_config_files - - # Stub out the subcommand handler so the test doesn't try to open - # a network connection or run platform-specific upload logic. - with ( - patch("esphome.__main__.read_config") as mock_read, - patch.dict( - "esphome.__main__.POST_CONFIG_ACTIONS", - {command: lambda args, config: 0}, - ), - ): - result = run_esphome( - ["esphome", command, "--from-storage-json", str(yaml_path)] - ) - - mock_read.assert_not_called() - assert result == 0 - - -@pytest.mark.parametrize("command", ["upload", "logs"]) -def test_run_esphome_from_storage_json_falls_back_when_missing( - tmp_path: Path, command: str -) -> None: - """With no sidecar on disk, the dispatcher falls back to read_config().""" - yaml_path = tmp_path / "lite_test.yaml" - yaml_path.write_text(_SAMPLE_YAML) - - with ( - patch("esphome.__main__.read_config", return_value=None) as mock_read, - patch.dict( - "esphome.__main__.POST_CONFIG_ACTIONS", - {command: lambda args, config: 0}, - ), - ): - result = run_esphome( - ["esphome", command, "--from-storage-json", str(yaml_path)] - ) - - mock_read.assert_called_once() - # read_config returned None → dispatcher exits with 2. - assert result == 2 - - -def test_run_esphome_without_flag_still_calls_read_config( - tmp_path: Path, lite_config_files: tuple[Path, Path] -) -> None: - """Sanity: omitting the flag preserves the current behaviour.""" - yaml_path, _ = lite_config_files - - with ( - patch("esphome.__main__.read_config", return_value=None) as mock_read, - patch.dict( - "esphome.__main__.POST_CONFIG_ACTIONS", - {"upload": lambda args, config: 0}, - ), - ): - run_esphome(["esphome", "upload", str(yaml_path)]) - - mock_read.assert_called_once()