mirror of
https://github.com/esphome/esphome.git
synced 2026-02-28 09:54:19 -07:00
482721d1ce
The Beken/RTL SDKs ship lwIP defaults tuned for a general-purpose WiFi SoC: TCP_SND_BUF=10*MSS (14.6KB), MEM_SIZE=32KB, 12 TCP + 22 UDP sockets. These waste significant RAM on memory-constrained chips. ESPAsyncWebServer allocates malloc(tcp_sndbuf()) per response chunk, and at 14.6KB this causes OOM on BK7231N. This tunes lwIP to match ESP32/ESP8266 ranges: - TCP_SND_BUF/TCP_WND: 4*MSS (was 10*MSS) - MEM_SIZE: 12KB BK72XX / 16KB RTL,LN (was 32KB) - Socket counts: dynamic from component registrations - All derived pool sizes adjusted accordingly Also splits socket consumer tracking into TCP/UDP, adds get_socket_counts() API, and updates ESP32 to use it. Closes https://github.com/esphome/esphome/issues/14095
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
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."""
|
|
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."""
|
|
# 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
|