[core] Make the YAML load listener hook public (#20805)

esphome/yaml_util.py

@@ -54,8 +54,11 @@ _LOADED_SECRET_NAMES: set[str] = set()
 # Set while a secret_values_registered context is active; the dumper records
 # every emitted `!secret` name into it.
 _EMITTED_SECRET_NAMES: set[str] | None = None
-# Not thread-safe — config processing is single-threaded today.
-_load_listeners: list[Callable[[Path], None]] = []
+# Called with every file the YAML loader opens; ``track_yaml_loads`` builds on it.
+# Used by device-builder (esphome/device-builder), whose listener filters by
+# thread because the hook is process-global; keep the name and the per-file
+# ``Path`` signature. Not otherwise a supported API: use at your own risk.
+load_listeners: list[Callable[[Path], None]] = []
 
 DocumentPath = list[str | int]
 
@@ -127,11 +130,11 @@ def track_yaml_loads() -> Generator[list[Path]]:
     def _on_load(fname: Path) -> None:
         loaded.append(Path(fname).resolve())
 
-    _load_listeners.append(_on_load)
+    load_listeners.append(_on_load)
     try:
         yield loaded
     finally:
-        _load_listeners.remove(_on_load)
+        load_listeners.remove(_on_load)
 
 
 class ESPHomeDataBase:
@@ -667,7 +670,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
             secrets.add(Path(fname).resolve())
 
     with track_yaml_loads() as loaded:
-        _load_listeners.append(_capture_secret)
+        load_listeners.append(_capture_secret)
         try:
             try:
                 data = load_yaml(config_path)
@@ -679,7 +682,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
                 data, warn_on_unresolved=False, warn_on_load_error=False
             )
         finally:
-            _load_listeners.remove(_capture_secret)
+            load_listeners.remove(_capture_secret)
 
     # Deduplicate while preserving first-seen order.
     seen: set[Path] = set()
@@ -1120,7 +1123,7 @@ def load_yaml(
 
 def _load_yaml_internal(fname: Path, *, track_document_range: bool = True) -> Any:
     """Load a YAML file."""
-    for listener in _load_listeners:
+    for listener in tuple(load_listeners):
         listener(fname)
     try:
         with fname.open(encoding="utf-8") as f_handle:

tests/unit_tests/test_yaml_util.py

@@ -431,12 +431,32 @@ def test_track_yaml_loads_empty_outside_context(tmp_path: Path) -> None:
 
 def test_track_yaml_loads_cleanup_on_exception(tmp_path: Path) -> None:
     """Listener is removed even if the body raises."""
-    before = len(yaml_util._load_listeners)
+    before = len(yaml_util.load_listeners)
 
     with pytest.raises(RuntimeError), yaml_util.track_yaml_loads():
         raise RuntimeError("boom")
 
-    assert len(yaml_util._load_listeners) == before
+    assert len(yaml_util.load_listeners) == before
+
+
+def test_load_listener_removing_itself_does_not_skip_the_next(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A listener that removes itself while notified does not skip the one after it."""
+    yaml_file = tmp_path / "test.yaml"
+    yaml_file.write_text("key: value\n")
+    listeners: list[Callable[[Path], None]] = []
+    monkeypatch.setattr(yaml_util, "load_listeners", listeners)
+    seen: list[Path] = []
+
+    def first(fname: Path) -> None:
+        listeners.remove(first)
+
+    listeners.extend((first, seen.append))
+    yaml_util.load_yaml(yaml_file)
+
+    assert seen == [yaml_file.resolve()]
+    assert listeners == [seen.append]
 
 
 def test_track_yaml_loads_no_duplicate_load_on_top_level_include_failure(