|
|
|
@@ -1,9 +1,21 @@
|
|
|
|
|
from esphome.components import socket
|
|
|
|
|
from esphome.const import (
|
|
|
|
|
KEY_CORE,
|
|
|
|
|
KEY_TARGET_PLATFORM,
|
|
|
|
|
PLATFORM_ESP32,
|
|
|
|
|
PLATFORM_ESP8266,
|
|
|
|
|
)
|
|
|
|
|
from esphome.core import CORE
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
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_platform()
|
|
|
|
|
CORE.config = {"wifi": True}
|
|
|
|
|
socket.require_wake_loop_threadsafe()
|
|
|
|
|
|
|
|
|
@@ -32,6 +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_platform()
|
|
|
|
|
# Call three times
|
|
|
|
|
CORE.config = {"openthread": True}
|
|
|
|
|
socket.require_wake_loop_threadsafe()
|
|
|
|
@@ -75,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
|
|
|
|
|