comments

tests/components/socket/test_wake_loop_threadsafe.py

@@ -1,16 +1,21 @@
 from esphome.components import socket
-from esphome.const import KEY_CORE, KEY_TARGET_PLATFORM, PLATFORM_ESP8266
+from esphome.const import (
+    KEY_CORE,
+    KEY_TARGET_PLATFORM,
+    PLATFORM_ESP32,
+    PLATFORM_ESP8266,
+)
 from esphome.core import CORE
 
 
-def _setup_non_esp32_platform() -> None:
-    """Set up CORE.data with a non-ESP32 platform for testing."""
-    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP8266}
+def _setup_platform(platform=PLATFORM_ESP8266) -> None:
+    """Set up CORE.data with a platform for testing."""
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: platform}
 
 
 def test_require_wake_loop_threadsafe__first_call() -> None:
     """Test that first call sets up define and consumes socket."""
-    _setup_non_esp32_platform()
+    _setup_platform()
     CORE.config = {"wifi": True}
     socket.require_wake_loop_threadsafe()
 
@@ -39,7 +44,7 @@ def test_require_wake_loop_threadsafe__idempotent() -> None:
 
 def test_require_wake_loop_threadsafe__multiple_calls() -> None:
     """Test that multiple calls only set up once."""
-    _setup_non_esp32_platform()
+    _setup_platform()
     # Call three times
     CORE.config = {"openthread": True}
     socket.require_wake_loop_threadsafe()
@@ -83,3 +88,29 @@ def test_require_wake_loop_threadsafe__no_networking_does_not_consume_socket() -
     udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
     assert "socket.wake_loop_threadsafe" not in udp_consumers
     assert udp_consumers == initial_udp
+
+
+def test_require_wake_loop_threadsafe__esp32_no_udp_socket() -> None:
+    """Test that ESP32 uses task notifications instead of UDP socket."""
+    _setup_platform(PLATFORM_ESP32)
+    CORE.config = {"wifi": True}
+    socket.require_wake_loop_threadsafe()
+
+    # Verify the define was added
+    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
+    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
+
+    # Verify no UDP socket was consumed (ESP32 uses FreeRTOS task notifications)
+    udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
+    assert "socket.wake_loop_threadsafe" not in udp_consumers
+
+
+def test_require_wake_loop_threadsafe__non_esp32_consumes_udp_socket() -> None:
+    """Test that non-ESP32 platforms consume a UDP socket for wake notifications."""
+    _setup_platform(PLATFORM_ESP8266)
+    CORE.config = {"wifi": True}
+    socket.require_wake_loop_threadsafe()
+
+    # Verify UDP socket was consumed
+    udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
+    assert udp_consumers.get("socket.wake_loop_threadsafe") == 1