move to socket

This commit is contained in:
J. Nick Koston
2025-11-02 21:57:57 -06:00
parent 2ac95abea7
commit acd26600dd
7 changed files with 87 additions and 75 deletions
+12
View File
@@ -0,0 +1,12 @@
"""Configuration file for socket component tests."""
import pytest
from esphome.core import CORE
@pytest.fixture(autouse=True)
def reset_core():
"""Reset CORE after each test."""
yield
CORE.reset()
+42
View File
@@ -0,0 +1,42 @@
from esphome.components import socket
from esphome.core import CORE
def test_require_wake_loop_threadsafe__first_call() -> None:
"""Test that first call sets up define and consumes socket."""
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
# 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."""
# Call three times
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)
-40
View File
@@ -70,43 +70,3 @@ async def test_register_component__with_setup_priority(monkeypatch):
assert add_mock.call_count == 4
app_mock.register_component.assert_called_with(var)
assert core_mock.component_ids == []
def test_require_wake_loop_threadsafe__first_call() -> None:
"""Test that first call sets up define and consumes socket."""
ch.require_wake_loop_threadsafe()
# Verify CORE.data was updated
assert ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
# Verify the define was added
assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in ch.CORE.defines)
def test_require_wake_loop_threadsafe__idempotent() -> None:
"""Test that subsequent calls are idempotent."""
# Set up initial state as if already called
ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
# Call again - should not raise or fail
ch.require_wake_loop_threadsafe()
# Verify state is still True
assert ch.CORE.data[ch.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 ch.CORE.defines)
def test_require_wake_loop_threadsafe__multiple_calls() -> None:
"""Test that multiple calls only set up once."""
# Call three times
ch.require_wake_loop_threadsafe()
ch.require_wake_loop_threadsafe()
ch.require_wake_loop_threadsafe()
# Verify CORE.data was set
assert ch.CORE.data[ch.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 ch.CORE.defines)