github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-06 10:56:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

[core] Extract cache helpers into esphome/compiled_config.py

Browse Source 
storage_json.py is for the JSON sidecar; the validated-config cache
is YAML. Move it out into its own module.

Also consolidate the dispatcher's open-coded fast path (load cache +
load sidecar + apply_to_core) into a single load_compiled_config()
entry point, and collapse save_compiled_config()'s two try blocks
into one. mtime comparison gets its own _cache_is_fresh() helper so
the loader reads top to bottom.
This commit is contained in:
J. Nick Koston
2026-05-12 15:45:47 -05:00
parent 26d2d4f65d
commit aee2fc5761
5 changed files with 176 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit_tests/test_main_validated_config_cache.py → tests/unit_tests/test_compiled_config.py
+63 -116
View File
@@ -1,4 +1,4 @@
"""Tests for the validated-config cache fast path used by upload/logs."""
"""Tests for the validated-config cache used by upload/logs."""
from __future__ import annotations
@@ -10,6 +10,7 @@ from unittest.mock import patch
import pytest
from esphome.__main__ import run_esphome
from esphome.compiled_config import compiled_config_path, load_compiled_config
from esphome.const import (
CONF_API,
CONF_ESPHOME,
@@ -19,7 +20,6 @@ from esphome.const import (
KEY_TARGET_PLATFORM,
)
from esphome.core import CORE
from esphome.storage_json import StorageJSON, compiled_config_path, load_compiled_config
_VALIDATED_CONFIG_YAML = """\
esphome:
@@ -67,36 +67,46 @@ def _write_storage(storage_path: Path) -> None:
storage_path.write_text(json.dumps(data))
def _write_cache(cache_path: Path, body: str = _VALIDATED_CONFIG_YAML) -> Path:
"""Write the cache file and return it."""
cache_path.parent.mkdir(parents=True, exist_ok=True)
cache_path.write_text(body)
return cache_path
def _set_cache_mtime(cache_path: Path, yaml_path: Path, *, offset: int) -> None:
"""Force the cache file's mtime relative to the source YAML.
Positive offset → cache is fresh. Negative → cache is stale.
"""
yaml_stat = yaml_path.stat()
os.utime(cache_path, (yaml_stat.st_atime, yaml_stat.st_mtime + offset))
@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 + StorageJSON + cache, all consistent and 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))
cache = _write_cache(storage_dir / "lite_test.yaml.validated.yaml")
_set_cache_mtime(cache, yaml_path, offset=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")
assert str(compiled_config_path("device.yaml")).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."""
"""Fresh cache + sidecar → returns config and populates CORE."""
config = load_compiled_config(fresh_cache_files)
assert config is not None
@@ -104,78 +114,50 @@ def test_load_compiled_config_happy_path(fresh_cache_files: Path) -> None:
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()
# apply_to_core ran as part of the orchestration.
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(
"scenario",
["missing_cache", "stale_cache", "corrupt_cache", "missing_sidecar"],
)
def test_load_compiled_config_falls_back(tmp_path: Path, scenario: str) -> None:
"""All non-happy cases return None so the 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"
cache_path = storage_dir / "lite_test.yaml.validated.yaml"
sidecar_path = storage_dir / "lite_test.yaml.json"
if scenario == "missing_cache":
pass # no cache, no sidecar
elif scenario == "stale_cache":
_write_storage(sidecar_path)
_set_cache_mtime(_write_cache(cache_path), yaml_path, offset=-60)
elif scenario == "corrupt_cache":
_write_storage(sidecar_path)
_set_cache_mtime(
_write_cache(cache_path, "not: valid: yaml: ["), yaml_path, offset=5
)
elif scenario == "missing_sidecar":
# Cache fresh + parseable, but no StorageJSON → can't populate CORE.
_set_cache_mtime(_write_cache(cache_path), yaml_path, offset=5)
assert load_compiled_config(yaml_path) is None
@pytest.mark.parametrize("command", ["upload", "logs"])
def test_run_esphome_upload_and_logs_use_cache_when_fresh(
command: str, fresh_cache_files: Path
) -> None:
"""When the cache is fresh, upload/logs skip read_config() entirely."""
yaml_path = fresh_cache_files
captured = {}
"""upload/logs skip read_config() when the cache is fresh."""
captured: dict = {}
def _stub(_args, config):
captured["config"] = config
@@ -185,11 +167,9 @@ def test_run_esphome_upload_and_logs_use_cache_when_fresh(
patch("esphome.__main__.read_config") as mock_read,
patch.dict("esphome.__main__.POST_CONFIG_ACTIONS", {command: _stub}),
):
result = run_esphome(["esphome", command, str(yaml_path)])
assert run_esphome(["esphome", command, str(fresh_cache_files)]) == 0
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="
@@ -198,7 +178,7 @@ def test_run_esphome_upload_and_logs_use_cache_when_fresh(
def test_run_esphome_upload_and_logs_fall_back_when_no_cache(
tmp_path: Path, command: str
) -> None:
"""With no cache on disk, the dispatcher falls back to read_config()."""
"""Without a cache, the dispatcher falls back to read_config()."""
yaml_path = tmp_path / "lite_test.yaml"
yaml_path.write_text("esphome:\n name: lite_test\n")
@@ -209,46 +189,13 @@ def test_run_esphome_upload_and_logs_fall_back_when_no_cache(
{command: lambda args, config: 0},
),
):
result = run_esphome(["esphome", command, str(yaml_path)])
mock_read.assert_called_once()
assert result == 2
def test_run_esphome_upload_falls_back_when_cache_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", str(yaml_path)])
assert run_esphome(["esphome", command, str(yaml_path)]) == 2
mock_read.assert_called_once()
def test_run_esphome_compile_does_not_use_cache(fresh_cache_files: Path) -> None:
"""`compile` always re-validates, even with a fresh cache on disk.
The fast path is only for upload / logs -- compile is what writes
the cache in the first place, and it needs a fully validated
config to drive code generation.
"""
yaml_path = fresh_cache_files
"""compile always re-validates -- it's what writes the cache."""
with (
patch("esphome.__main__.read_config", return_value=None) as mock_read,
patch.dict(
@@ -256,6 +203,6 @@ def test_run_esphome_compile_does_not_use_cache(fresh_cache_files: Path) -> None
{"compile": lambda args, config: 0},
),
):
run_esphome(["esphome", "compile", str(yaml_path)])
run_esphome(["esphome", "compile", str(fresh_cache_files)])
mock_read.assert_called_once()