[esphome] Warn when a YAML merge (<<:) drops a key (#17246)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

tests/unit_tests/test_config_normalization.py

@@ -1,6 +1,7 @@
 """Unit tests for esphome.config module."""
 
 from collections.abc import Generator
+import logging
 from pathlib import Path
 from unittest.mock import MagicMock, Mock, patch
 
@@ -113,3 +114,57 @@ def test_ota_with_platform_list_and_captive_portal(fixtures_dir: Path) -> None:
     platforms = {p.get("platform") for p in result["ota"]}
     assert "esphome" in platforms, f"Expected esphome platform in {platforms}"
     assert "web_server" in platforms, f"Expected web_server platform in {platforms}"
+
+
+def _write_merge_conflict_config(tmp_path: Path, *, suppress: bool) -> Path:
+    """Create a config where two `<<` includes both define `logger:`.
+
+    The second `logger:` is dropped by the shallow merge. Returns the main file.
+    """
+    (tmp_path / "a.yaml").write_text("logger:\n  level: DEBUG\n")
+    (tmp_path / "b.yaml").write_text("logger:\n  level: INFO\n")
+    esphome_section = "esphome:\n  name: test\n"
+    if suppress:
+        esphome_section += "  merge_warnings: false\n"
+    main = tmp_path / "main.yaml"
+    main.write_text(f"{esphome_section}<<: !include a.yaml\n<<: !include b.yaml\n")
+    return main
+
+
+def test_validate_config_warns_on_dropped_merge_key(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """By default, a `<<` merge that drops a key logs a warning."""
+    main = _write_merge_conflict_config(tmp_path, suppress=False)
+    CORE.config_path = main
+    raw_config = yaml_util.load_yaml(main)
+
+    with caplog.at_level(logging.WARNING, logger="esphome.config"):
+        config.validate_config(raw_config, {})
+
+    assert any(
+        "was dropped while processing a '<<' merge" in record.message
+        and "logger" in record.message
+        for record in caplog.records
+    )
+    # The queue is drained so the warning cannot leak into a later run.
+    assert yaml_util.take_dropped_merge_keys() == []
+
+
+def test_validate_config_suppresses_merge_warning(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """`esphome: merge_warnings: false` hides the warning but still drains the queue."""
+    main = _write_merge_conflict_config(tmp_path, suppress=True)
+    CORE.config_path = main
+    raw_config = yaml_util.load_yaml(main)
+
+    with caplog.at_level(logging.WARNING, logger="esphome.config"):
+        config.validate_config(raw_config, {})
+
+    assert not any(
+        "was dropped while processing a '<<' merge" in record.message
+        for record in caplog.records
+    )
+    # The queue is drained even when the warning is suppressed.
+    assert yaml_util.take_dropped_merge_keys() == []

tests/unit_tests/test_yaml_util.py

@@ -1395,3 +1395,48 @@ def test_dump__redaction_flag_does_not_leak_between_calls() -> None:
     assert "\\033[8m" in redacted
     assert "\\033[8m" not in raw
     assert "\\033[8m" in redacted_again
+
+
+@pytest.fixture(autouse=True)
+def clear_dropped_merge_keys() -> None:
+    """Reset the dropped-merge-key queue between tests."""
+    core.CORE.data.pop(yaml_util._MERGE_WARNINGS_KEY, None)
+    yield
+    core.CORE.data.pop(yaml_util._MERGE_WARNINGS_KEY, None)
+
+
+def test_merge_include_records_dropped_keys(tmp_path: Path) -> None:
+    """A `<<` merge that overlaps an existing key records it (shallow first-wins)."""
+    (tmp_path / "a.yaml").write_text("api:\n  reboot_timeout: 5min\n")
+    (tmp_path / "b.yaml").write_text("api:\n  password: secret\n")
+    test_yaml = tmp_path / "test.yaml"
+    test_yaml.write_text("<<: !include a.yaml\n<<: !include b.yaml\n")
+
+    with patch.object(core.CORE, "config_path", test_yaml):
+        result = yaml_util.load_yaml(test_yaml)
+
+    # First definition wins; the second `api` block is dropped entirely.
+    assert result["api"] == {"reboot_timeout": "5min"}
+
+    dropped = yaml_util.take_dropped_merge_keys()
+    assert len(dropped) == 1
+    key, location = dropped[0]
+    assert key == "api"
+    assert "b.yaml" in location
+    # Queue is drained after being taken.
+    assert yaml_util.take_dropped_merge_keys() == []
+
+
+def test_merge_include_no_overlap_records_nothing(tmp_path: Path) -> None:
+    """A `<<` merge with distinct top-level keys drops nothing."""
+    (tmp_path / "a.yaml").write_text("api:\n  reboot_timeout: 5min\n")
+    (tmp_path / "b.yaml").write_text("logger:\n  level: DEBUG\n")
+    test_yaml = tmp_path / "test.yaml"
+    test_yaml.write_text("<<: !include a.yaml\n<<: !include b.yaml\n")
+
+    with patch.object(core.CORE, "config_path", test_yaml):
+        result = yaml_util.load_yaml(test_yaml)
+
+    assert result["api"] == {"reboot_timeout": "5min"}
+    assert result["logger"] == {"level": "DEBUG"}
+    assert yaml_util.take_dropped_merge_keys() == []