Reload a symlinked file by the path the loader read it by

This commit is contained in:
J. Nick Koston
2026-10-13 09:30:56 -10:00
parent 3e6324445b
commit 50fdc8e85b
2 changed files with 35 additions and 5 deletions
+13 -5
View File
@@ -83,7 +83,9 @@ _DOMAIN = "yaml_edit"
def own_documents() -> set[Path]:
"""Every file the loader read for this configuration, found once per run."""
"""Every file the loader read for this configuration, found once per run.
A file behind an include path with a substitution that gave nothing but
substitutions is the one kind this misses."""
if (documents := CORE.data.get(_DOMAIN)) is None:
documents = CORE.data[_DOMAIN] = _own_documents()
return documents
@@ -216,8 +218,11 @@ def apply_line_edits(edits: list[LineEdit]) -> dict[Path, Snapshot]:
and after for restore_files. Nothing is written while any line is stale,
and a file that no longer loads is put back here."""
by_path: dict[Path, list[LineEdit]] = {}
seen_as: dict[Path, Path] = {} # the path the loader read a file by
for edit in edits:
by_path.setdefault(editable_file(edit.path), []).append(edit)
resolved = editable_file(edit.path)
by_path.setdefault(resolved, []).append(edit)
seen_as.setdefault(resolved, edit.path)
originals = {}
for path, own in by_path.items():
original = read_text(path)
@@ -225,9 +230,12 @@ def apply_line_edits(edits: list[LineEdit]) -> dict[Path, Snapshot]:
try:
for path, snapshot in originals.items():
write_keeping_mode(path, snapshot.written)
reloads = list(originals)
if (main := CORE.config_path.resolve()) not in originals:
reloads.append(main) # its !secret tags register a value written elsewhere
# Reloaded as the loader saw them: a relative !include inside a
# symlinked file resolves beside the link
reloads = [seen_as[path] for path in originals]
if CORE.config_path.resolve() not in originals:
# Its !secret tags register a value written elsewhere
reloads.append(CORE.config_path)
for path in reloads:
try:
# The secret registry stays: later output still redacts
+22
View File
@@ -299,6 +299,28 @@ def test_apply_writes_reloads_drops_the_cache_and_restores(tmp_path: Path) -> No
assert not cache.exists()
def test_apply_reloads_a_symlinked_file_by_the_link(tmp_path: Path) -> None:
"""A relative !include inside the file resolves beside the link, as the
loader read it, not beside the target."""
shared = tmp_path / "shared"
shared.mkdir()
(shared / "device.yaml").write_bytes(
b"name: kitchen\nfriendly_name: !include fn.yaml\n"
)
(tmp_path / "fn.yaml").write_bytes(b"Kitchen\n") # beside the link only
(tmp_path / "device.yaml").symlink_to(shared / "device.yaml")
_setup(tmp_path, "esphome: !include device.yaml\n")
doc, line_no = source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
assert doc == tmp_path / "device.yaml"
match = field_line_re(CONF_NAME, "kitchen").match(line_at(doc, line_no))
edit = LineEdit(doc, line_no, match.string, rewrite(match, "garage"))
originals = apply_line_edits([edit])
assert list(originals) == [(shared / "device.yaml").resolve()]
assert (shared / "device.yaml").read_text() == (
"name: garage\nfriendly_name: !include fn.yaml\n"
)
def test_apply_rolls_back_a_rewrite_the_loader_refuses(tmp_path: Path) -> None:
path = _setup(tmp_path, YAML)
edit = _name_edit("garage")