[web_server.ota] Merge multiple instances to prevent undefined behavior (#11905)

tests/component_tests/ota/test_web_server_ota.py

@@ -1,6 +1,18 @@
 """Tests for the web_server OTA platform."""
 
+from __future__ import annotations
+
 from collections.abc import Callable
+import logging
+from typing import Any
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.web_server.ota import _web_server_ota_final_validate
+from esphome.const import CONF_ID, CONF_OTA, CONF_PLATFORM, CONF_WEB_SERVER
+from esphome.core import ID
+import esphome.final_validate as fv
 
 
 def test_web_server_ota_generated(generate_main: Callable[[str], str]) -> None:
@@ -100,3 +112,144 @@ def test_web_server_ota_esp8266(generate_main: Callable[[str], str]) -> None:
     # Check web server OTA component is present
     assert "WebServerOTAComponent" in main_cpp
     assert "web_server::WebServerOTAComponent" in main_cpp
+
+
+@pytest.mark.parametrize(
+    ("ota_configs", "expected_count", "warning_expected"),
+    [
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_WEB_SERVER,
+                    CONF_ID: ID("ota_web", is_manual=False),
+                }
+            ],
+            1,
+            False,
+            id="single_instance_no_merge",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_WEB_SERVER,
+                    CONF_ID: ID("ota_web_1", is_manual=False),
+                },
+                {
+                    CONF_PLATFORM: CONF_WEB_SERVER,
+                    CONF_ID: ID("ota_web_2", is_manual=False),
+                },
+            ],
+            1,
+            True,
+            id="two_instances_merged",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_WEB_SERVER,
+                    CONF_ID: ID("ota_web_1", is_manual=False),
+                },
+                {
+                    CONF_PLATFORM: "esphome",
+                    CONF_ID: ID("ota_esphome", is_manual=False),
+                },
+                {
+                    CONF_PLATFORM: CONF_WEB_SERVER,
+                    CONF_ID: ID("ota_web_2", is_manual=False),
+                },
+            ],
+            2,
+            True,
+            id="mixed_platforms_web_server_merged",
+        ),
+    ],
+)
+def test_web_server_ota_instance_merging(
+    ota_configs: list[dict[str, Any]],
+    expected_count: int,
+    warning_expected: bool,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test web_server OTA instance merging behavior."""
+    full_conf = {CONF_OTA: ota_configs.copy()}
+
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            _web_server_ota_final_validate({})
+
+        updated_conf = fv.full_config.get()
+
+        # Verify total number of OTA platforms
+        assert len(updated_conf[CONF_OTA]) == expected_count
+
+        # Verify warning
+        if warning_expected:
+            assert any(
+                "Found and merged" in record.message
+                and "web_server OTA" in record.message
+                for record in caplog.records
+            ), "Expected merge warning not found in log"
+        else:
+            assert len(caplog.records) == 0, "Unexpected warnings logged"
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_web_server_ota_consistent_manual_ids(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test that consistent manual IDs can be merged successfully."""
+    ota_configs = [
+        {
+            CONF_PLATFORM: CONF_WEB_SERVER,
+            CONF_ID: ID("ota_web", is_manual=True),
+        },
+        {
+            CONF_PLATFORM: CONF_WEB_SERVER,
+            CONF_ID: ID("ota_web", is_manual=True),
+        },
+    ]
+
+    full_conf = {CONF_OTA: ota_configs}
+
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            _web_server_ota_final_validate({})
+
+        updated_conf = fv.full_config.get()
+        assert len(updated_conf[CONF_OTA]) == 1
+        assert updated_conf[CONF_OTA][0][CONF_ID].id == "ota_web"
+        assert any(
+            "Found and merged" in record.message and "web_server OTA" in record.message
+            for record in caplog.records
+        )
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_web_server_ota_inconsistent_manual_ids() -> None:
+    """Test that inconsistent manual IDs raise an error."""
+    ota_configs = [
+        {
+            CONF_PLATFORM: CONF_WEB_SERVER,
+            CONF_ID: ID("ota_web_1", is_manual=True),
+        },
+        {
+            CONF_PLATFORM: CONF_WEB_SERVER,
+            CONF_ID: ID("ota_web_2", is_manual=True),
+        },
+    ]
+
+    full_conf = {CONF_OTA: ota_configs}
+
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(
+            cv.Invalid,
+            match="Found multiple web_server OTA configurations but id is inconsistent",
+        ):
+            _web_server_ota_final_validate({})
+    finally:
+        fv.full_config.reset(token)