This commit is contained in:
J. Nick Koston
2026-02-26 06:40:38 -10:00
parent 0bfb294f32
commit da93485ae2
@@ -1,3 +1,5 @@
import pytest
from esphome.components import socket
from esphome.const import (
KEY_CORE,
@@ -93,9 +95,15 @@ def test_require_wake_loop_threadsafe__no_networking_does_not_consume_socket() -
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)
@pytest.mark.parametrize(
"platform",
[PLATFORM_ESP32, PLATFORM_BK72XX, PLATFORM_RTL87XX, PLATFORM_LN882X],
)
def test_require_wake_loop_threadsafe__fast_select_no_udp_socket(
platform: str,
) -> None:
"""Test that fast select platforms use task notifications instead of UDP socket."""
_setup_platform(platform)
CORE.config = {"wifi": True}
socket.require_wake_loop_threadsafe()
@@ -103,13 +111,13 @@ def test_require_wake_loop_threadsafe__esp32_no_udp_socket() -> None:
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)
# Verify no UDP socket was consumed (fast select platforms use 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."""
def test_require_wake_loop_threadsafe__non_fast_select_consumes_udp_socket() -> None:
"""Test that platforms without fast select consume a UDP socket for wake notifications."""
_setup_platform(PLATFORM_ESP8266)
CORE.config = {"wifi": True}
socket.require_wake_loop_threadsafe()
@@ -117,48 +125,3 @@ def test_require_wake_loop_threadsafe__non_esp32_consumes_udp_socket() -> None:
# Verify UDP socket was consumed
udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
assert udp_consumers.get("socket.wake_loop_threadsafe") == 1
def test_require_wake_loop_threadsafe__bk72xx_no_udp_socket() -> None:
"""Test that BK72xx (LibreTiny) uses task notifications instead of UDP socket."""
_setup_platform(PLATFORM_BK72XX)
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 (LibreTiny 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__rtl87xx_no_udp_socket() -> None:
"""Test that RTL87xx (LibreTiny) uses task notifications instead of UDP socket."""
_setup_platform(PLATFORM_RTL87XX)
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 (LibreTiny 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__ln882x_no_udp_socket() -> None:
"""Test that LN882H (LibreTiny) uses task notifications instead of UDP socket."""
_setup_platform(PLATFORM_LN882X)
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 (LibreTiny uses FreeRTOS task notifications)
udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
assert "socket.wake_loop_threadsafe" not in udp_consumers