mirror of
https://github.com/esphome/esphome.git
synced 2026-09-13 00:07:32 +00:00
[core] Cache validated config for --from-storage-json fast path
Drop the StorageJSON `compiled_config` field and the `esphome/lite_config.py` module from the first iteration. The earlier shape stored the entire validated config inside the JSON sidecar, but configs can grow past a megabyte once packages and substitutions expand, and JSON can't round-trip the YAML-specific types (lambdas, ID instances, includes) the validation pipeline produces. Instead: dump the validated config to its own YAML file alongside the sidecar (`<file>.validated.yaml`), using `yaml_util.dump` so all esphome-specific tags survive the round trip. The fast path loads it back with `yaml_util.load_yaml(clear_secrets=False)` -- no schema validation, no final-validate, no external-component refresh. Staleness is gated by mtime: if the source YAML mtime is newer than the cache, the dispatcher falls back to a full `read_config()` with a warning. The cache is refreshed on every successful compile via `writer.update_storage_json`, so the dashboard's compile -> upload -> logs flow always sees a fresh cache. Drift surface shrinks to "what does yaml_util.dump emit" rather than "which top-level keys does upload/logs read" -- when those subcommands gain a new dependency, the cache already carries it.
This commit is contained in:
J. Nick Koston
+22
-9
@@ -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(
|
||||
|
||||
@@ -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
|
||||
+104
-1
@@ -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 (``<filename>.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()
|
||||
|
||||
|
||||
+9
-1
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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 ``<tmp>/lite_test.yaml`` so ``CORE.data_dir``
|
||||
points into ``<tmp>/.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()
|
||||
Reference in New Issue
Block a user