[core] Move wake_loop out of socket component into core (#15446)

Co-authored-by: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com>
2026-05-31 07:57:40 +08:00 · 2026-04-06 11:01:03 -10:00
parent 62b4b250c7
commit ab45591507
38 changed files with 397 additions and 618 deletions
@@ -1,127 +0,0 @@
-import pytest
-
-from esphome.components import socket
-from esphome.const import (
-    KEY_CORE,
-    KEY_TARGET_PLATFORM,
-    PLATFORM_BK72XX,
-    PLATFORM_ESP32,
-    PLATFORM_ESP8266,
-    PLATFORM_LN882X,
-    PLATFORM_RTL87XX,
-)
-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()
-
-    # Verify CORE.data was updated
-    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Verify the define was added
-    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
-
-
-def test_require_wake_loop_threadsafe__idempotent() -> None:
-    """Test that subsequent calls are idempotent."""
-    # Set up initial state as if already called
-    CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
-    CORE.config = {"ethernet": True}
-
-    # Call again - should not raise or fail
-    socket.require_wake_loop_threadsafe()
-
-    # Verify state is still True
-    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Define should not be added since flag was already True
-    assert not any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
-
-
-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()
-    socket.require_wake_loop_threadsafe()
-    socket.require_wake_loop_threadsafe()
-
-    # Verify CORE.data was set
-    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Verify the define was added (only once, but we can just check it exists)
-    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
-
-
-def test_require_wake_loop_threadsafe__no_networking() -> None:
-    """Test that wake loop is NOT configured when no networking is configured."""
-    # Set up config without any networking components
-    CORE.config = {"esphome": {"name": "test"}, "logger": {}}
-
-    # Call require_wake_loop_threadsafe
-    socket.require_wake_loop_threadsafe()
-
-    # Verify CORE.data flag was NOT set (since has_networking returns False)
-    assert socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED not in CORE.data
-
-    # Verify the define was NOT added
-    assert not any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
-
-
-def test_require_wake_loop_threadsafe__no_networking_does_not_consume_socket() -> None:
-    """Test that no socket is consumed when no networking is configured."""
-    # Set up config without any networking components
-    CORE.config = {"logger": {}}
-
-    # Track initial socket consumer state
-    initial_udp = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
-
-    # Call require_wake_loop_threadsafe
-    socket.require_wake_loop_threadsafe()
-
-    # Verify no socket was consumed
-    udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
-    assert "socket.wake_loop_threadsafe" not in udp_consumers
-    assert udp_consumers == initial_udp
-
-
-@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()
-
-    # 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 (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_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()
-
-    # Verify UDP socket was consumed
-    udp_consumers = CORE.data.get(socket.KEY_SOCKET_CONSUMERS_UDP, {})
-    assert udp_consumers.get("socket.wake_loop_threadsafe") == 1