[sntp] Merge multiple instances to fix crash and undefined behavior

esphome/components/sntp/time.py

@@ -1,9 +1,14 @@
+import logging
+
 import esphome.codegen as cg
 from esphome.components import time as time_
+from esphome.config_helpers import merge_config
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
+    CONF_PLATFORM,
     CONF_SERVERS,
+    CONF_TIME,
     PLATFORM_BK72XX,
     PLATFORM_ESP32,
     PLATFORM_ESP8266,
@@ -12,13 +17,74 @@ from esphome.const import (
     PLATFORM_RTL87XX,
 )
 from esphome.core import CORE
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+_LOGGER = logging.getLogger(__name__)
 
 DEPENDENCIES = ["network"]
+
+CONF_SNTP = "sntp"
+
 sntp_ns = cg.esphome_ns.namespace("sntp")
 SNTPComponent = sntp_ns.class_("SNTPComponent", time_.RealTimeClock)
 
 DEFAULT_SERVERS = ["0.pool.ntp.org", "1.pool.ntp.org", "2.pool.ntp.org"]
 
+
+def _sntp_final_validate(config: ConfigType) -> None:
+    """Merge multiple SNTP instances into one, similar to OTA merging behavior."""
+    full_conf = fv.full_config.get()
+    time_confs = full_conf.get(CONF_TIME, [])
+
+    sntp_configs: list[ConfigType] = []
+    other_time_configs: list[ConfigType] = []
+
+    for time_conf in time_confs:
+        if time_conf.get(CONF_PLATFORM) == CONF_SNTP:
+            sntp_configs.append(time_conf)
+        else:
+            other_time_configs.append(time_conf)
+
+    if len(sntp_configs) <= 1:
+        return
+
+    # Merge all SNTP configs into the first one
+    merged = sntp_configs[0]
+    for sntp_conf in sntp_configs[1:]:
+        # Validate that IDs are consistent if manually specified
+        if merged[CONF_ID].is_manual and sntp_conf[CONF_ID].is_manual:
+            raise cv.Invalid(
+                f"Found multiple SNTP configurations but {CONF_ID} is inconsistent"
+            )
+        merged = merge_config(merged, sntp_conf)
+
+    # Deduplicate servers while preserving order
+    servers = merged[CONF_SERVERS]
+    unique_servers = list(dict.fromkeys(servers))
+
+    # Warn if we're dropping servers due to 3-server limit
+    if len(unique_servers) > 3:
+        dropped = unique_servers[3:]
+        unique_servers = unique_servers[:3]
+        _LOGGER.warning(
+            "SNTP supports maximum 3 servers. Dropped excess server(s): %s",
+            dropped,
+        )
+
+    merged[CONF_SERVERS] = unique_servers
+
+    _LOGGER.warning(
+        "Found and merged %d SNTP time configurations into one instance",
+        len(sntp_configs),
+    )
+
+    # Replace time configs with merged SNTP + other time platforms
+    other_time_configs.append(merged)
+    full_conf[CONF_TIME] = other_time_configs
+    fv.full_config.set(full_conf)
+
+
 CONFIG_SCHEMA = cv.All(
     time_.TIME_SCHEMA.extend(
         {
@@ -40,6 +106,8 @@ CONFIG_SCHEMA = cv.All(
     ),
 )
 
+FINAL_VALIDATE_SCHEMA = _sntp_final_validate
+
 
 async def to_code(config):
     servers = config[CONF_SERVERS]

esphome/components/web_server/ota/__init__.py

@@ -1,17 +1,10 @@
-import logging
-
 import esphome.codegen as cg
 from esphome.components.esp32 import add_idf_component
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
-from esphome.config_helpers import merge_config
 import esphome.config_validation as cv
-from esphome.const import CONF_ID, CONF_OTA, CONF_PLATFORM, CONF_WEB_SERVER
+from esphome.const import CONF_ID
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
-import esphome.final_validate as fv
-from esphome.types import ConfigType
-
-_LOGGER = logging.getLogger(__name__)
 
 CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["network", "web_server_base"]
@@ -19,53 +12,6 @@ DEPENDENCIES = ["network", "web_server_base"]
 web_server_ns = cg.esphome_ns.namespace("web_server")
 WebServerOTAComponent = web_server_ns.class_("WebServerOTAComponent", OTAComponent)
 
-
-def _web_server_ota_final_validate(config: ConfigType) -> None:
-    """Merge multiple web_server OTA instances into one.
-
-    Multiple web_server OTA instances register duplicate HTTP handlers for /update,
-    causing undefined behavior. Merge them into a single instance.
-    """
-    full_conf = fv.full_config.get()
-    ota_confs = full_conf.get(CONF_OTA, [])
-
-    web_server_ota_configs: list[ConfigType] = []
-    other_ota_configs: list[ConfigType] = []
-
-    for ota_conf in ota_confs:
-        if ota_conf.get(CONF_PLATFORM) == CONF_WEB_SERVER:
-            web_server_ota_configs.append(ota_conf)
-        else:
-            other_ota_configs.append(ota_conf)
-
-    if len(web_server_ota_configs) <= 1:
-        return
-
-    # Merge all web_server OTA configs into the first one
-    merged = web_server_ota_configs[0]
-    for ota_conf in web_server_ota_configs[1:]:
-        # Validate that IDs are consistent if manually specified
-        if (
-            merged[CONF_ID].is_manual
-            and ota_conf[CONF_ID].is_manual
-            and merged[CONF_ID] != ota_conf[CONF_ID]
-        ):
-            raise cv.Invalid(
-                f"Found multiple web_server OTA configurations but {CONF_ID} is inconsistent"
-            )
-        merged = merge_config(merged, ota_conf)
-
-    _LOGGER.warning(
-        "Found and merged %d web_server OTA configurations into one instance",
-        len(web_server_ota_configs),
-    )
-
-    # Replace OTA configs with merged web_server + other OTA platforms
-    other_ota_configs.append(merged)
-    full_conf[CONF_OTA] = other_ota_configs
-    fv.full_config.set(full_conf)
-
-
 CONFIG_SCHEMA = (
     cv.Schema(
         {
@@ -76,8 +22,6 @@ CONFIG_SCHEMA = (
     .extend(cv.COMPONENT_SCHEMA)
 )
 
-FINAL_VALIDATE_SCHEMA = _web_server_ota_final_validate
-
 
 @coroutine_with_priority(CoroPriority.WEB_SERVER_OTA)
 async def to_code(config):

esphome/components/wifi/__init__.py

@@ -12,6 +12,7 @@ from esphome.components.network import (
 from esphome.components.psram import is_guaranteed as psram_is_guaranteed
 from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
+from esphome.config_validation import only_with_esp_idf
 from esphome.const import (
     CONF_AP,
     CONF_BSSID,
@@ -351,7 +352,7 @@ CONFIG_SCHEMA = cv.All(
                 single=True
             ),
             cv.Optional(CONF_USE_PSRAM): cv.All(
-                cv.only_on_esp32, cv.requires_component("psram"), cv.boolean
+                only_with_esp_idf, cv.requires_component("psram"), cv.boolean
             ),
         }
     ),

tests/component_tests/ota/test_web_server_ota.py

@@ -1,18 +1,6 @@
 """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:
@@ -112,144 +100,3 @@ 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)

tests/component_tests/sntp/__init__.py

@@ -0,0 +1 @@
+"""Tests for SNTP component."""

tests/component_tests/sntp/config/sntp_test.yaml

@@ -0,0 +1,22 @@
+esphome:
+  name: sntp-test
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: "testssid"
+  password: "testpassword"
+
+# Test multiple SNTP instances that should be merged
+time:
+  - platform: sntp
+    servers:
+      - 192.168.1.1
+      - pool.ntp.org
+  - platform: sntp
+    servers:
+      - pool.ntp.org
+      - 192.168.1.2

tests/component_tests/sntp/test_init.py

@@ -0,0 +1,238 @@
+"""Tests for SNTP time configuration validation."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.sntp.time import CONF_SNTP, _sntp_final_validate
+from esphome.const import CONF_ID, CONF_PLATFORM, CONF_SERVERS, CONF_TIME
+from esphome.core import ID
+import esphome.final_validate as fv
+
+
+@pytest.mark.parametrize(
+    ("time_configs", "expected_count", "expected_servers", "warning_messages"),
+    [
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
+                }
+            ],
+            1,
+            ["192.168.1.1", "pool.ntp.org"],
+            [],
+            id="single_instance_no_merge",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_1", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
+                },
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_2", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.2"],
+                },
+            ],
+            1,
+            ["192.168.1.1", "pool.ntp.org", "192.168.1.2"],
+            ["Found and merged 2 SNTP time configurations into one instance"],
+            id="two_instances_merged",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_1", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
+                },
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_2", is_manual=False),
+                    CONF_SERVERS: ["pool.ntp.org", "192.168.1.2"],
+                },
+            ],
+            1,
+            ["192.168.1.1", "pool.ntp.org", "192.168.1.2"],
+            ["Found and merged 2 SNTP time configurations into one instance"],
+            id="deduplication_preserves_order",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_1", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.1", "pool.ntp.org"],
+                },
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_2", is_manual=False),
+                    CONF_SERVERS: ["192.168.1.2", "pool2.ntp.org"],
+                },
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_3", is_manual=False),
+                    CONF_SERVERS: ["pool3.ntp.org"],
+                },
+            ],
+            1,
+            ["192.168.1.1", "pool.ntp.org", "192.168.1.2"],
+            [
+                "SNTP supports maximum 3 servers. Dropped excess server(s): ['pool2.ntp.org', 'pool3.ntp.org']",
+                "Found and merged 3 SNTP time configurations into one instance",
+            ],
+            id="three_instances_drops_excess_servers",
+        ),
+        pytest.param(
+            [
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_1", is_manual=False),
+                    CONF_SERVERS: [
+                        "192.168.1.1",
+                        "pool.ntp.org",
+                        "pool.ntp.org",
+                        "192.168.1.1",
+                    ],
+                },
+                {
+                    CONF_PLATFORM: CONF_SNTP,
+                    CONF_ID: ID("sntp_time_2", is_manual=False),
+                    CONF_SERVERS: ["pool.ntp.org", "192.168.1.2"],
+                },
+            ],
+            1,
+            ["192.168.1.1", "pool.ntp.org", "192.168.1.2"],
+            ["Found and merged 2 SNTP time configurations into one instance"],
+            id="deduplication_multiple_duplicates",
+        ),
+    ],
+)
+def test_sntp_instance_merging(
+    time_configs: list[dict[str, Any]],
+    expected_count: int,
+    expected_servers: list[str],
+    warning_messages: list[str],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test SNTP instance merging behavior."""
+    # Create a mock full config with time configs
+    full_conf = {CONF_TIME: time_configs.copy()}
+
+    # Set the context var
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            _sntp_final_validate({})
+
+        # Get the updated config
+        updated_conf = fv.full_config.get()
+
+        # Check if merging occurred
+        if len(time_configs) > 1:
+            # Verify only one SNTP instance remains
+            sntp_instances = [
+                tc
+                for tc in updated_conf[CONF_TIME]
+                if tc.get(CONF_PLATFORM) == CONF_SNTP
+            ]
+            assert len(sntp_instances) == expected_count
+
+            # Verify server list
+            assert sntp_instances[0][CONF_SERVERS] == expected_servers
+
+            # Verify warnings
+            for expected_msg in warning_messages:
+                assert any(
+                    expected_msg in record.message for record in caplog.records
+                ), f"Expected warning message '{expected_msg}' not found in log"
+        else:
+            # Single instance should not trigger merging or warnings
+            assert len(caplog.records) == 0
+            # Config should be unchanged
+            assert updated_conf[CONF_TIME] == time_configs
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_sntp_inconsistent_manual_ids() -> None:
+    """Test that inconsistent manual IDs raise an error."""
+    # Create configs with manual IDs that are inconsistent
+    time_configs = [
+        {
+            CONF_PLATFORM: CONF_SNTP,
+            CONF_ID: ID("sntp_time_1", is_manual=True),
+            CONF_SERVERS: ["192.168.1.1"],
+        },
+        {
+            CONF_PLATFORM: CONF_SNTP,
+            CONF_ID: ID("sntp_time_2", is_manual=True),
+            CONF_SERVERS: ["192.168.1.2"],
+        },
+    ]
+
+    full_conf = {CONF_TIME: time_configs}
+
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(
+            cv.Invalid,
+            match="Found multiple SNTP configurations but id is inconsistent",
+        ):
+            _sntp_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_sntp_with_other_time_platforms(caplog: pytest.LogCaptureFixture) -> None:
+    """Test that SNTP merging doesn't affect other time platforms."""
+    time_configs = [
+        {
+            CONF_PLATFORM: CONF_SNTP,
+            CONF_ID: ID("sntp_time_1", is_manual=False),
+            CONF_SERVERS: ["192.168.1.1"],
+        },
+        {
+            CONF_PLATFORM: "homeassistant",
+            CONF_ID: ID("homeassistant_time", is_manual=False),
+        },
+        {
+            CONF_PLATFORM: CONF_SNTP,
+            CONF_ID: ID("sntp_time_2", is_manual=False),
+            CONF_SERVERS: ["192.168.1.2"],
+        },
+    ]
+
+    full_conf = {CONF_TIME: time_configs.copy()}
+
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            _sntp_final_validate({})
+
+        updated_conf = fv.full_config.get()
+
+        # Should have 2 time platforms: 1 merged SNTP + 1 homeassistant
+        assert len(updated_conf[CONF_TIME]) == 2
+
+        # Find the platforms
+        platforms = {tc[CONF_PLATFORM] for tc in updated_conf[CONF_TIME]}
+        assert platforms == {CONF_SNTP, "homeassistant"}
+
+        # Verify SNTP was merged
+        sntp_instances = [
+            tc for tc in updated_conf[CONF_TIME] if tc[CONF_PLATFORM] == CONF_SNTP
+        ]
+        assert len(sntp_instances) == 1
+        assert sntp_instances[0][CONF_SERVERS] == ["192.168.1.1", "192.168.1.2"]
+    finally:
+        fv.full_config.reset(token)