[core] Save the validated config cache on the first upload or logs run (#18367)

This commit is contained in:
J. Nick Koston
2026-08-14 18:45:23 -07:00
committed by GitHub
parent e5224e22ae
commit b178f74e5d
12 changed files with 573 additions and 64 deletions
+12 -12
View File
@@ -2732,7 +2732,8 @@ def run_esphome(argv):
conf_path.name,
)
if config is None:
cache_missed = config is None
if cache_missed:
from esphome.config import read_config
config = read_config(
@@ -2741,26 +2742,25 @@ def run_esphome(argv):
# Snapshot only needed by `esphome config --no-defaults`.
snapshot_user_config=getattr(args, "no_defaults", False),
)
# Refresh the cache so the next upload/logs hits the fast path
# instead of re-running read_config. Skip when the storage
# sidecar is absent (no compile has run): the cache would
# never be loaded back, so writing secrets to disk is wasted.
if cache_eligible and config is not None:
from esphome.compiled_config import save_compiled_config
from esphome.storage_json import ext_storage_path
if ext_storage_path(conf_path.name).exists():
save_compiled_config(config)
if config is None:
return 2
CORE.config = config
# Fallback for platforms whose validators didn't set the toolchain
# (only the esp32 component reads esp32.framework.toolchain). All
# other platforms only support PlatformIO today.
# other platforms only support PlatformIO today. Must run before the
# cache refresh below so its sidecar records the same toolchain a
# compile would.
if CORE.toolchain is None:
CORE.toolchain = Toolchain.PLATFORMIO
# Refresh the cache so the next upload/logs hits the fast path
# instead of re-running read_config.
if cache_eligible and cache_missed:
from esphome.compiled_config import save_compiled_config_and_sidecar
save_compiled_config_and_sidecar(config)
if args.command not in POST_CONFIG_ACTIONS:
safe_print(f"Unknown command {args.command}")
return 1
+69 -8
View File
@@ -18,9 +18,9 @@ from pathlib import Path
from typing import Any
from esphome.const import __version__ as ESPHOME_VERSION
from esphome.core import CORE, Lambda
from esphome.core import CORE, EsphomeError, Lambda
from esphome.helpers import write_file
from esphome.storage_json import StorageJSON, ext_storage_path
from esphome.storage_json import StorageJSON, ext_storage_path, storage_path
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
@@ -65,7 +65,71 @@ def save_compiled_config(config: ConfigType) -> None:
# non-basic dict key), so every upload/logs pays the slow path.
_LOGGER.warning("Cannot cache the validated config: %s", err)
except Exception as err: # noqa: BLE001 # pylint: disable=broad-except
_LOGGER.debug("Skipping compiled config cache write: %s", err)
# Likely persistent (permissions, full disk): every upload/logs
# pays the slow path until it clears, so surface it.
_LOGGER.warning("Skipping compiled config cache write: %s", err)
def save_compiled_config_and_sidecar(config: ConfigType) -> None:
"""Refresh the cache from the upload/logs fallback (CORE.config must be set).
The cache is only written when a complete sidecar is on disk:
load_compiled_config can't use it otherwise, and it holds resolved
secrets.
"""
if _refresh_sidecar():
save_compiled_config(config)
def _refresh_sidecar() -> bool:
"""Ensure a complete sidecar is on disk; True when one is.
Writes one (without claiming a build) when missing or wizard-only.
Failures are non-fatal; the next upload/logs pays the slow path again.
"""
try:
path = storage_path()
try:
old = StorageJSON.load_strict(path)
except Exception as err: # noqa: BLE001 # pylint: disable=broad-except
# Present but unreadable: it may hold a real build's metadata,
# and a fresh rewrite would also stop the next compile from
# cleaning a possibly incoherent build tree.
_LOGGER.warning(
"Not caching: storage sidecar %s is unreadable (%s)", path, err
)
return False
if old is not None and old.can_apply_to_core():
# Compile-written; nothing to refresh.
return True
if CORE.build_path is not None and CORE.build_path.exists():
# An unvalidated build tree: its absent or mismatched sidecar
# is what makes the next compile wipe it, so don't vouch for
# a build this run never saw.
_LOGGER.warning(
"Not caching: build tree %s has no matching sidecar; "
"'esphome compile' will settle it",
CORE.build_path,
)
return False
new = StorageJSON.from_esphome_core(CORE, old, claim_build=False)
if not new.can_apply_to_core():
_LOGGER.warning("Not caching: rebuilt storage sidecar is still incomplete")
return False
new.save(path)
return True
except (OSError, EsphomeError) as err:
# write_file wraps OSError into EsphomeError. Persistent
# (unwritable storage dir), so surface that every upload/logs
# pays the slow path.
_LOGGER.warning("Could not refresh the storage sidecar: %s", err)
except Exception: # noqa: BLE001 # pylint: disable=broad-except
# A structural bug; keep the traceback so it isn't mistaken
# for the I/O failure above.
_LOGGER.warning(
"Unexpected error refreshing the storage sidecar", exc_info=True
)
return False
def load_compiled_config(conf_path: Path) -> ConfigType | None:
@@ -98,11 +162,8 @@ def load_compiled_config(conf_path: Path) -> ConfigType | None:
return None
storage = StorageJSON.load(ext_storage_path(conf_path.name))
if storage is None:
return None
# apply_to_core assumes a real compile wrote the sidecar; wizard-only
# sidecars leave both of these unset and can't drive upload/logs.
if not storage.core_platform and not storage.target_platform:
if storage is None or not storage.can_apply_to_core():
_LOGGER.debug("Ignoring compiled config cache: sidecar missing or incomplete")
return None
storage.apply_to_core()
return config
+3
View File
@@ -570,6 +570,9 @@ def get_download_types(storage_json):
the shape stable so the download panel
doesn't have to special-case per-platform schemas.
"""
# No recorded firmware path means nothing was built; no downloads.
if storage_json.firmware_bin_path is None:
return []
return [
{
"title": "Factory format (Previously Modern)",
+3
View File
@@ -113,6 +113,9 @@ def get_download_types(storage_json):
the shape stable so the download panel
doesn't have to special-case per-platform schemas.
"""
# No recorded firmware path means nothing was built; no downloads.
if storage_json.firmware_bin_path is None:
return []
return [
{
"title": "Standard format",
+3
View File
@@ -182,6 +182,9 @@ def get_download_types(storage_json: StorageJSON = None):
the shape stable so the download panel
doesn't have to special-case per-platform schemas.
"""
# No recorded firmware path means nothing was built; no downloads.
if storage_json.firmware_bin_path is None:
return []
types = [
{
"title": "UF2 package (recommended)",
+3
View File
@@ -473,6 +473,9 @@ def copy_files() -> None:
def get_download_types(storage_json: StorageJSON) -> list[dict[str, str]]:
"""Get the download types for the firmware."""
# No recorded firmware path means nothing was built; no downloads.
if storage_json.firmware_bin_path is None:
return []
types = []
UF2_PATH = "zephyr/zephyr.uf2"
DFU_PATH = "firmware.zip"
+3
View File
@@ -156,6 +156,9 @@ def get_download_types(storage_json):
the shape stable so the download panel
doesn't have to special-case per-platform schemas.
"""
# No recorded firmware path means nothing was built; no downloads.
if storage_json.firmware_bin_path is None:
return []
return [
{
"title": "UF2 factory format",
+48 -8
View File
@@ -71,8 +71,11 @@ def archive_storage_path() -> Path:
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
"""Convert a string to Path; None and the legacy "None" both map to None.
Sidecars written before as_dict skipped unset paths hold str(None).
"""
return Path(value) if value is not None and value != "None" else None
def _parse_framework_version(framework_version: str) -> Version:
@@ -170,8 +173,10 @@ class StorageJSON:
"address": self.address,
"web_port": self.web_port,
"esp_platform": self.target_platform,
"build_path": str(self.build_path),
"firmware_bin_path": str(self.firmware_bin_path),
"build_path": str(self.build_path) if self.build_path else None,
"firmware_bin_path": (
str(self.firmware_bin_path) if self.firmware_bin_path else None
),
"loaded_integrations": sorted(self.loaded_integrations),
"loaded_platforms": sorted(self.loaded_platforms),
"no_mdns": self.no_mdns,
@@ -189,7 +194,18 @@ class StorageJSON:
write_file_if_changed(path, self.to_json())
@staticmethod
def from_esphome_core(esph: CoreType, old: StorageJSON | None) -> StorageJSON:
def from_esphome_core(
esph: CoreType, old: StorageJSON | None, *, claim_build: bool = True
) -> StorageJSON:
"""Build a sidecar from post-validation CORE state.
claim_build=False (the upload/logs fallback, which runs no build)
carries the build-artifact fields (esphome_version,
firmware_bin_path) from *old* instead of asserting this run built
firmware. Validation-derived fields (platform, framework_version,
toolchain, build_path) always stamp; storage_should_clean compares
them against the next compile.
"""
hardware = esph.target_platform.upper()
framework_version: str | None = None
if esph.is_esp32:
@@ -204,13 +220,21 @@ class StorageJSON:
name=esph.name,
friendly_name=esph.friendly_name,
comment=esph.comment,
esphome_version=const.__version__,
esphome_version=(
const.__version__
if claim_build
else (old.esphome_version if old else None)
),
src_version=1,
address=esph.address,
web_port=esph.web_port,
target_platform=hardware,
build_path=esph.build_path,
firmware_bin_path=esph.firmware_bin,
firmware_bin_path=(
esph.firmware_bin
if claim_build
else (old.firmware_bin_path if old else None)
),
loaded_integrations=esph.loaded_integrations,
loaded_platforms=esph.loaded_platforms,
no_mdns=(
@@ -302,11 +326,27 @@ class StorageJSON:
except Exception: # noqa: BLE001 # pylint: disable=broad-except
return None
@staticmethod
def load_strict(path: Path) -> StorageJSON | None:
"""Like load, but None only means missing; an unreadable file raises."""
if not path.is_file():
return None
return StorageJSON._load_impl(path)
def can_apply_to_core(self) -> bool:
"""True when the sidecar carries everything apply_to_core hands CORE.
Wizard-written sidecars leave build_path unset (older wizards also
the platform fields) and can't drive upload/logs.
"""
return bool((self.core_platform or self.target_platform) and self.build_path)
def apply_to_core(self) -> None:
"""Populate CORE with the metadata upload/logs read.
Inverse of :meth:`from_esphome_core`. Keep paired -- a new
attribute upload/logs needs has to be captured there too.
attribute upload/logs needs has to be captured there too and
reflected in :meth:`can_apply_to_core`.
Validator-only fields (loaded_integrations/platforms,
friendly_name) are skipped; the fast path doesn't run
validation and CORE.__init__ defaults them.
@@ -1,10 +1,15 @@
"""Shared storage-sidecar factory for the lazy-import fixture scripts."""
from pathlib import Path
from esphome.storage_json import StorageJSON
def make_storage() -> StorageJSON:
"""A minimal post-compile esp32 sidecar the upload/logs fast path accepts."""
"""A minimal post-compile esp32 sidecar the upload/logs fast path accepts.
build_path must be set: the fast path rejects sidecars without one.
"""
return StorageJSON(
storage_version=1,
name="test",
@@ -15,8 +20,8 @@ def make_storage() -> StorageJSON:
address="1.2.3.4",
web_port=None,
target_platform="ESP32S3",
build_path=None,
firmware_bin_path=None,
build_path=Path("/build/test"),
firmware_bin_path=Path("/build/test/firmware.bin"),
loaded_integrations=set(),
loaded_platforms=set(),
no_mdns=False,
+268 -31
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from contextlib import contextmanager
from ipaddress import IPv4Address, IPv4Network
import json
import os
@@ -19,6 +20,7 @@ from esphome.compiled_config import (
compiled_config_path,
load_compiled_config,
save_compiled_config,
save_compiled_config_and_sidecar,
)
from esphome.const import (
CONF_API,
@@ -31,7 +33,16 @@ from esphome.const import (
KEY_VARIANT,
Toolchain,
)
from esphome.core import CORE, ID, HexInt, Lambda, MACAddress, TimePeriodMilliseconds
from esphome.core import (
CORE,
ID,
EsphomeError,
HexInt,
Lambda,
MACAddress,
TimePeriodMilliseconds,
)
from esphome.storage_json import StorageJSON
from esphome.util import OrderedDict
_VALIDATED_CONFIG = {
@@ -54,8 +65,9 @@ def _cache_body(config: dict | None = None) -> str:
def _write_storage(
storage_path: Path,
*,
esp_platform: str = "ESP32",
esp_platform: str | None = "ESP32",
core_platform: str | None = "esp32",
build_path: str | None = "/build/lite_test",
) -> None:
"""Write a vanilla StorageJSON sidecar for the cache tests."""
storage_path.parent.mkdir(parents=True, exist_ok=True)
@@ -69,7 +81,7 @@ def _write_storage(
"address": "192.168.1.42",
"web_port": None,
"esp_platform": esp_platform,
"build_path": "/build/lite_test",
"build_path": build_path,
"firmware_bin_path": "/build/lite_test/firmware.bin",
"loaded_integrations": ["api", "logger", "ota", "wifi"],
"loaded_platforms": [],
@@ -359,31 +371,262 @@ 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) config would be inert and
leak secrets to disk for nothing."""
def _storage_fixture(tmp_path: Path) -> StorageJSON:
"""A loaded StorageJSON instance matching _write_storage's contents."""
fixture = tmp_path / "fixture_storage.json"
_write_storage(fixture)
return StorageJSON.load(fixture)
def _bare_yaml(tmp_path: Path) -> Path:
"""A minimal YAML with CORE.config_path pointed at it."""
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
return yaml_path
@contextmanager
def _fallback_run(command: str = "upload", **from_core_kwargs) -> Any:
"""Patch the fallback path's collaborators for a run_esphome call.
Without kwargs, from_esphome_core stays real (yielded mock is None).
"""
with (
patch(
"esphome.config.read_config",
return_value={"esphome": {"name": "lite_test"}},
),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
) as mock_read,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{"upload": lambda args, config: 0},
{command: lambda args, config: 0},
),
):
run_esphome(["esphome", "upload", str(yaml_path)])
if not from_core_kwargs:
yield mock_read, None
return
with patch.object(
StorageJSON, "from_esphome_core", **from_core_kwargs
) as mock_from_core:
yield mock_read, mock_from_core
@pytest.mark.parametrize("command", ["upload", "logs"])
def test_run_esphome_fallback_writes_sidecar_and_cache_without_sidecar(
tmp_path: Path, command: str
) -> None:
"""A never-compiled config caches on its first upload/logs run: the
fallback writes the StorageJSON sidecar itself (load_compiled_config
needs it), so the second run hits the fast path."""
yaml_path = _bare_yaml(tmp_path)
storage_dir = tmp_path / ".esphome" / "storage"
with _fallback_run(command, return_value=_storage_fixture(tmp_path)) as (
mock_read,
mock_from_core,
):
assert run_esphome(["esphome", command, str(yaml_path)]) == 0
mock_from_core.assert_called_once()
assert (storage_dir / "lite_test.yaml.validated.json").exists()
storage = StorageJSON.load(storage_dir / "lite_test.yaml.json")
assert storage is not None
# No compile happened, so the sidecar must not claim one.
assert mock_from_core.call_args.kwargs == {"claim_build": False}
# The second run loads the cache instead of re-validating.
assert run_esphome(["esphome", command, str(yaml_path)]) == 0
mock_read.assert_called_once()
# as_dict serialized unset paths as str(None) until 2026.9; files
# written by those wizards are still on disk.
_WIZARD_SIDECAR_CASES = pytest.mark.parametrize(
"wizard_kwargs",
[
{"esp_platform": None, "core_platform": None, "build_path": None},
{"build_path": None},
{"build_path": "None"},
],
ids=["legacy_wizard", "modern_wizard", "none_string_wizard"],
)
def _prime_core(tmp_path: Path) -> None:
"""Set the post-validation CORE state from_esphome_core reads."""
CORE.name = "lite_test"
CORE.build_path = tmp_path / "build" / "lite_test"
CORE.data[KEY_CORE] = {
KEY_TARGET_PLATFORM: "esp8266",
KEY_TARGET_FRAMEWORK: "arduino",
}
@_WIZARD_SIDECAR_CASES
def test_run_esphome_fallback_completes_wizard_sidecar(
tmp_path: Path, wizard_kwargs: dict[str, Any]
) -> None:
"""A wizard-written sidecar can't drive the fast path (no build_path;
older wizards also no platform fields); the fallback rewrites it from
CORE so the cache loads on the next run."""
yaml_path = _bare_yaml(tmp_path)
storage_dir = tmp_path / ".esphome" / "storage"
_write_storage(storage_dir / "lite_test.yaml.json", **wizard_kwargs)
with _fallback_run(return_value=_storage_fixture(tmp_path)) as (_, mock_from_core):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
mock_from_core.assert_called_once()
storage = StorageJSON.load(storage_dir / "lite_test.yaml.json")
assert storage is not None and storage.core_platform == "esp32"
# What the wizard recorded about a build (nothing, or a real one)
# carries through instead of being stamped with this run's values.
assert storage.esphome_version == "2026.1.0"
assert load_compiled_config(yaml_path) is not None
def test_run_esphome_fallback_skips_cache_when_sidecar_write_fails(
tmp_path: Path,
) -> None:
"""A failed sidecar write is non-fatal and skips the cache save too:
without the sidecar the cache could never be loaded back, so writing
it would only leave resolved secrets on disk."""
yaml_path = _bare_yaml(tmp_path)
with (
_fallback_run(side_effect=RuntimeError("boom")),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
mock_save.assert_not_called()
assert not (tmp_path / ".esphome" / "storage" / "lite_test.yaml.json").exists()
def test_run_esphome_fallback_write_failure_takes_io_branch(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""StorageJSON.save raises EsphomeError (write_file wraps OSError into
it), which must land in the plain I/O warning, not the traceback
branch for structural bugs."""
yaml_path = _bare_yaml(tmp_path)
with (
_fallback_run(return_value=_storage_fixture(tmp_path)),
patch.object(StorageJSON, "save", side_effect=EsphomeError("boom")),
patch("esphome.compiled_config.save_compiled_config") as mock_save,
caplog.at_level("WARNING", logger="esphome.compiled_config"),
):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
mock_save.assert_not_called()
assert "Could not refresh the storage sidecar" in caplog.text
assert "Unexpected error" not in caplog.text
def test_run_esphome_fallback_leaves_unreadable_sidecar_alone(tmp_path: Path) -> None:
"""A present-but-corrupt sidecar is not overwritten: it may hold a real
build's metadata, and replacing it would suppress the next compile's
clean of a possibly incoherent build tree. The cache save is skipped."""
yaml_path = _bare_yaml(tmp_path)
storage_dir = tmp_path / ".esphome" / "storage"
sidecar = storage_dir / "lite_test.yaml.json"
sidecar.parent.mkdir(parents=True, exist_ok=True)
sidecar.write_text("{truncated", encoding="utf-8")
with _fallback_run(return_value=None) as (_, mock_from_core):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
mock_from_core.assert_not_called()
assert sidecar.read_text(encoding="utf-8") == "{truncated"
assert not (storage_dir / "lite_test.yaml.validated.json").exists()
def test_run_esphome_fallback_skips_cache_when_rebuilt_sidecar_incomplete(
tmp_path: Path,
) -> None:
"""If the rebuilt sidecar would still be incomplete, nothing is written:
the cache could never be loaded back, so saving it would only rewrite
resolved secrets on every run."""
yaml_path = _bare_yaml(tmp_path)
storage_dir = tmp_path / ".esphome" / "storage"
incomplete = tmp_path / "incomplete_storage.json"
_write_storage(incomplete, build_path=None)
with _fallback_run(return_value=StorageJSON.load(incomplete)):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
assert not (storage_dir / "lite_test.yaml.json").exists()
assert not (storage_dir / "lite_test.yaml.validated.json").exists()
def test_run_esphome_fallback_sidecar_records_platformio_toolchain(
tmp_path: Path,
) -> None:
"""The toolchain fallback runs before the sidecar write, so platforms
whose validators leave CORE.toolchain unset record the same
"platformio" a compile writes, not null."""
yaml_path = _bare_yaml(tmp_path)
_prime_core(tmp_path)
assert CORE.toolchain is None
with _fallback_run():
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
storage = StorageJSON.load(
tmp_path / ".esphome" / "storage" / "lite_test.yaml.json"
)
assert storage is not None
assert storage.toolchain == "platformio"
@pytest.mark.parametrize("existing_sidecar", [None, "wizard"])
def test_run_esphome_fallback_skips_sidecar_when_build_tree_exists(
tmp_path: Path, existing_sidecar: str | None
) -> None:
"""An existing build tree with a missing or wizard-only sidecar keeps
it that way: the mismatch is what makes the next compile wipe the
unknown tree, so the fallback writes nothing and skips the cache."""
yaml_path = _bare_yaml(tmp_path)
_prime_core(tmp_path)
CORE.build_path.mkdir(parents=True)
storage_dir = tmp_path / ".esphome" / "storage"
if existing_sidecar == "wizard":
_write_storage(storage_dir / "lite_test.yaml.json", build_path=None)
wizard_body = (storage_dir / "lite_test.yaml.json").read_text(encoding="utf-8")
with _fallback_run(return_value=_storage_fixture(tmp_path)) as (_, mock_from_core):
assert run_esphome(["esphome", "upload", str(yaml_path)]) == 0
mock_from_core.assert_not_called()
assert not (storage_dir / "lite_test.yaml.validated.json").exists()
if existing_sidecar == "wizard":
sidecar_body = (storage_dir / "lite_test.yaml.json").read_text(encoding="utf-8")
assert sidecar_body == wizard_body
else:
assert not (storage_dir / "lite_test.yaml.json").exists()
def test_save_compiled_config_and_sidecar_builds_real_sidecar(tmp_path: Path) -> None:
"""Drive the real from_esphome_core on the fallback path: the
post-validation CORE state yields a complete, loadable sidecar."""
yaml_path = _bare_yaml(tmp_path)
_prime_core(tmp_path)
CORE.config = {CONF_ESPHOME: {CONF_NAME: "lite_test"}}
CORE.toolchain = Toolchain.PLATFORMIO
save_compiled_config_and_sidecar(CORE.config)
storage = StorageJSON.load(
tmp_path / ".esphome" / "storage" / "lite_test.yaml.json"
)
assert storage is not None
assert storage.core_platform == "esp8266"
assert storage.build_path is not None
# No compile happened, so the sidecar must not claim one.
assert storage.esphome_version is None
assert storage.firmware_bin_path is None
assert load_compiled_config(yaml_path) is not None
@pytest.mark.parametrize("command", ["upload", "logs"])
@@ -409,6 +652,7 @@ def test_run_esphome_upload_and_logs_refresh_cache_on_fallback(
patch(
"esphome.compiled_config.save_compiled_config", wraps=save_compiled_config
) as mock_save,
patch.object(StorageJSON, "from_esphome_core") as mock_from_core,
patch.dict(
"esphome.__main__.POST_CONFIG_ACTIONS",
{command: lambda args, config: 0},
@@ -417,6 +661,8 @@ def test_run_esphome_upload_and_logs_refresh_cache_on_fallback(
assert run_esphome(["esphome", command, str(yaml_path)]) == 0
mock_save.assert_called_once_with(fresh_config)
# The compile-written sidecar is complete; the fallback leaves it alone.
mock_from_core.assert_not_called()
# 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
@@ -647,24 +893,15 @@ def test_int_keys_coerce_to_strings(primed_storage: Path) -> None:
assert config["table"] == {"1": "a", "2": "b"}
def test_load_compiled_config_rejects_wizard_only_sidecar(tmp_path: Path) -> None:
"""A wizard-only sidecar (no compile -- no core_platform / target_platform)
can't drive upload/logs, so the fast path falls back."""
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
CORE.config_path = yaml_path
@_WIZARD_SIDECAR_CASES
def test_load_compiled_config_rejects_wizard_only_sidecar(
tmp_path: Path, wizard_kwargs: dict[str, Any]
) -> None:
"""A wizard-written sidecar (no build_path; older wizards also no
platform fields) can't drive upload/logs, so the fast path falls back."""
yaml_path = _bare_yaml(tmp_path)
storage_dir = tmp_path / ".esphome" / "storage"
storage_dir.mkdir(parents=True, exist_ok=True)
# StorageJSON with both core_platform and target_platform unset.
(storage_dir / "lite_test.yaml.json").write_text(
'{"storage_version": 1, "name": "lite_test", "friendly_name": null, '
'"comment": null, "esphome_version": null, "src_version": 1, '
'"address": null, "web_port": null, "esp_platform": null, '
'"build_path": null, "firmware_bin_path": null, '
'"loaded_integrations": [], "loaded_platforms": [], "no_mdns": false, '
'"framework": null, "core_platform": null}'
)
_write_storage(storage_dir / "lite_test.yaml.json", **wizard_kwargs)
cache_path = _write_cache(storage_dir / "lite_test.yaml.validated.json")
_set_cache_mtime(cache_path, yaml_path, offset=5)
+52
View File
@@ -0,0 +1,52 @@
"""Platform get_download_types contract for never-built configs.
Wizard-written and upload/logs-fallback sidecars record no
firmware_bin_path; the download panel must get an empty list for them,
not entries pointing at files that were never built.
"""
from __future__ import annotations
from importlib import import_module
from pathlib import Path
from typing import Any
import pytest
from esphome.storage_json import StorageJSON
PLATFORMS = ["esp32", "esp8266", "rp2", "libretiny", "nrf52"]
def _download_types(platform: str, storage: StorageJSON) -> list[dict[str, Any]]:
return import_module(f"esphome.components.{platform}").get_download_types(storage)
def _wizard_storage() -> StorageJSON:
return StorageJSON.from_wizard(
name="test_device",
friendly_name="Test Device",
address="test_device.local",
platform="ESP32",
)
@pytest.mark.parametrize("platform", PLATFORMS)
def test_no_firmware_path_yields_no_downloads(platform: str) -> None:
"""No recorded firmware path means nothing was built; no downloads."""
assert _download_types(platform, _wizard_storage()) == []
@pytest.mark.parametrize("platform", PLATFORMS)
def test_recorded_firmware_path_yields_downloads(platform: str, tmp_path: Path) -> None:
"""With a firmware path recorded, every platform offers entries in
the documented title/description/file/download shape."""
storage = _wizard_storage()
storage.firmware_bin_path = tmp_path / "firmware.bin"
types = _download_types(platform, storage)
assert types
assert all(
{"title", "description", "file", "download"} <= entry.keys() for entry in types
)
+99
View File
@@ -915,3 +915,102 @@ def test_storage_json_load_area(tmp_path: Path) -> None:
legacy = storage_json.StorageJSON.load(legacy_path)
assert legacy is not None
assert legacy.area is None
def test_from_esphome_core_without_claiming_a_build(setup_core: Path) -> None:
"""claim_build=False carries the build artifact fields from the old
sidecar while validation-derived fields still stamp from CORE."""
mock_core = MagicMock()
mock_core.name = "my_device"
mock_core.friendly_name = "My Device"
mock_core.comment = None
mock_core.address = "my_device.local"
mock_core.web_port = None
mock_core.target_platform = "esp8266"
mock_core.is_esp32 = False
mock_core.is_nrf52 = False
mock_core.build_path = "/build/my_device"
mock_core.loaded_integrations = set()
mock_core.loaded_platforms = set()
mock_core.config = {}
mock_core.target_framework = "arduino"
mock_core.toolchain = Toolchain.PLATFORMIO
mock_core.area = None
old = storage_json.StorageJSON.from_wizard(
name="my_device",
friendly_name="My Device",
address="my_device.local",
platform="ESP8266",
)
old.esphome_version = "2025.1.0"
old.firmware_bin_path = Path("/old/firmware.bin")
result = storage_json.StorageJSON.from_esphome_core(
mock_core, old, claim_build=False
)
# Build artifact fields carry from the old sidecar, not this run.
assert result.esphome_version == "2025.1.0"
assert result.firmware_bin_path == Path("/old/firmware.bin")
# Validation-derived fields stamp from CORE.
assert result.build_path == "/build/my_device"
assert result.toolchain == "platformio"
assert result.core_platform == "esp8266"
# With no old sidecar, no build is claimed at all.
bare = storage_json.StorageJSON.from_esphome_core(
mock_core, None, claim_build=False
)
assert bare.esphome_version is None
assert bare.firmware_bin_path is None
def test_load_strict_distinguishes_missing_from_unreadable(tmp_path: Path) -> None:
"""load_strict returns None only for a missing file; corrupt raises."""
assert storage_json.StorageJSON.load_strict(tmp_path / "missing.json") is None
corrupt = tmp_path / "corrupt.json"
corrupt.write_text("{truncated")
with pytest.raises(ValueError):
storage_json.StorageJSON.load_strict(corrupt)
def test_as_dict_serializes_unset_paths_as_null(setup_core: Path) -> None:
"""Unset build/firmware paths serialize as JSON null, not str(None)."""
storage = storage_json.StorageJSON.from_wizard(
name="wiz",
friendly_name="Wiz",
address="wiz.local",
platform="ESP32",
)
result = storage.as_dict()
assert result["build_path"] is None
assert result["firmware_bin_path"] is None
def test_load_treats_legacy_none_string_paths_as_unset(tmp_path: Path) -> None:
"""Sidecars written before as_dict emitted null hold str(None); those
must load as unset, not as Path("None")."""
file_path = tmp_path / "legacy_none.json"
file_path.write_text(
json.dumps(
{
"storage_version": 1,
"name": "wiz",
"friendly_name": "Wiz",
"esp_platform": "ESP32",
"core_platform": "esp32",
"build_path": "None",
"firmware_bin_path": "None",
}
)
)
result = storage_json.StorageJSON.load(file_path)
assert result is not None
assert result.build_path is None
assert result.firmware_bin_path is None