[store_yaml] Simplify redaction pipeline, fix Windows newline test failure

This commit is contained in:
J. Nick Koston
2026-07-03 20:40:14 -05:00
parent a25b7a806b
commit 8f2c09e3c8
5 changed files with 89 additions and 75 deletions
@@ -25,6 +25,7 @@ except ImportError:
from backports import zstd # type: ignore[import-not-found, no-redef]
from esphome.components.store_yaml import unpack_envelope
from esphome.yaml_util import find_secret_references
from .types import RunCompiledFunction
@@ -212,7 +213,7 @@ async def test_store_yaml_recovery(
assert b"recoverme123" not in envelope, (
"inline sensitive value leaked into the recovery blob"
)
assert b"!secret 'ota_password'" in combined, (
assert "ota_password" in find_secret_references(combined.decode()), (
"expected the inline OTA password to be recovered as a !secret reference"
)
assert b'ota_password: ""' in files["secrets.yaml"], (
+21 -8
View File
@@ -13,6 +13,7 @@ from esphome.components.store_yaml import (
_gather_files,
_generate_redacted_files,
_pack_envelope,
_read_files_verbatim,
unpack_envelope,
)
from esphome.core import CORE, EsphomeError
@@ -35,8 +36,6 @@ def project(tmp_path: Path) -> Path:
@pytest.fixture(autouse=True)
def _clear_config() -> None:
CORE.config = {}
yield
yaml_util._SECRET_VALUES.clear()
def _sources(
@@ -58,16 +57,29 @@ def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]:
# ---------------------------------------------------------------------------
def test_gather_returns_verbatim_content_and_flags_secrets(project: Path) -> None:
def test_gather_maps_rel_paths_and_flags_secrets(project: Path) -> None:
discovered = _sources(
project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
)
files, secret_rels = _gather_files(discovered)
contents = dict(files)
assert contents["secrets.yaml"] == b"api_key: SUPER_SECRET\n"
entries, secret_rels = _gather_files(discovered)
assert dict(entries) == {
"entry.yaml": project / "entry.yaml",
"secrets.yaml": project / "secrets.yaml",
}
assert secret_rels == {"secrets.yaml"}
def test_read_files_verbatim_returns_exact_bytes(project: Path) -> None:
"""`include_secrets: true` embeds the on-disk bytes untouched."""
discovered = _sources(
project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
)
entries, _ = _gather_files(discovered)
contents = dict(_read_files_verbatim(entries))
assert contents["secrets.yaml"] == (project / "secrets.yaml").read_bytes()
assert contents["entry.yaml"] == (project / "entry.yaml").read_bytes()
def test_gather_flags_secret_symlinked_to_other_name(
project: Path, tmp_path: Path
) -> None:
@@ -117,12 +129,13 @@ def test_gather_raises_on_load_errors(project: Path) -> None:
_gather_files(discovered)
def test_gather_raises_on_unreadable_file(
def test_read_files_verbatim_raises_on_unreadable_file(
project: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An unreadable tracked file fails the build instead of producing a
silently partial recovery blob."""
discovered = _sources(project, "entry.yaml", "wifi.yaml")
entries, _ = _gather_files(discovered)
orig_read_bytes = Path.read_bytes
def fake_read_bytes(self: Path) -> bytes:
@@ -132,7 +145,7 @@ def test_gather_raises_on_unreadable_file(
monkeypatch.setattr(Path, "read_bytes", fake_read_bytes)
with pytest.raises(EsphomeError, match="wifi.yaml"):
_gather_files(discovered)
_read_files_verbatim(entries)
def test_gather_warns_on_unresolved_includes(
+11
View File
@@ -16,6 +16,7 @@ from unittest.mock import Mock, patch
import pytest
from esphome import yaml_util
from esphome.core import CORE
here = Path(__file__).parent
@@ -32,6 +33,16 @@ def reset_core():
CORE.reset()
@pytest.fixture(autouse=True)
def clear_yaml_secrets():
"""Isolate the yaml_util secrets registry between tests."""
yaml_util._SECRET_VALUES.clear()
yaml_util._SECRET_CACHE.clear()
yield
yaml_util._SECRET_VALUES.clear()
yaml_util._SECRET_CACHE.clear()
@pytest.fixture
def fixture_path() -> Path:
"""
+6 -22
View File
@@ -25,16 +25,6 @@ from esphome.yaml_util import (
)
@pytest.fixture(autouse=True)
def clear_secrets_cache() -> None:
"""Clear the secrets cache before each test."""
yaml_util._SECRET_VALUES.clear()
yaml_util._SECRET_CACHE.clear()
yield
yaml_util._SECRET_VALUES.clear()
yaml_util._SECRET_CACHE.clear()
@pytest.fixture(autouse=True)
def clear_core_frontmatter() -> None:
"""Reset CORE.frontmatter between tests."""
@@ -1446,22 +1436,16 @@ def test_secret_values_registered_swaps_scalars_in_dump() -> None:
def test_secret_values_registered_does_not_clobber_real_secrets() -> None:
"""A value already mapped by a real `!secret` keeps its original name."""
yaml_util._SECRET_VALUES["hunter2"] = "original_name"
try:
with yaml_util.secret_values_registered({"hunter2": "generated_name"}):
out = yaml_util.dump({"password": make_data_base("hunter2")})
assert "!secret 'original_name'" in out
# The pre-existing mapping survives the context exit.
assert yaml_util.is_secret("hunter2") == "original_name"
finally:
yaml_util._SECRET_VALUES.clear()
with yaml_util.secret_values_registered({"hunter2": "generated_name"}):
out = yaml_util.dump({"password": make_data_base("hunter2")})
assert "!secret 'original_name'" in out
# The pre-existing mapping survives the context exit.
assert yaml_util.is_secret("hunter2") == "original_name"
def test_registered_secret_names() -> None:
yaml_util._SECRET_VALUES["value_a"] = "name_a"
try:
assert "name_a" in yaml_util.registered_secret_names()
finally:
yaml_util._SECRET_VALUES.clear()
assert "name_a" in yaml_util.registered_secret_names()
@pytest.fixture(autouse=True)