mirror of
https://github.com/esphome/esphome.git
synced 2026-11-06 13:15:57 +00:00
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
"""Integration test for the UDP component's deferred multicast join."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import socket
|
|
|
|
import pytest
|
|
|
|
from .log_utils import LineWaiter
|
|
from .state_utils import assert_device_name
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
MULTICAST_GROUP = "239.1.2.3"
|
|
|
|
|
|
def _get_free_udp_port() -> int:
|
|
"""Get a free UDP port by binding to port 0 and releasing."""
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
return port
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_udp_multicast_join(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
"""The group is joined from the poll after setup, once, and then receives."""
|
|
waiter = LineWaiter()
|
|
|
|
listen_port = _get_free_udp_port()
|
|
config = yaml_config.replace("UDP_LISTEN_PORT_PLACEHOLDER", str(listen_port))
|
|
|
|
async with (
|
|
run_compiled(config, line_callback=waiter.callback),
|
|
api_client_connected() as client,
|
|
):
|
|
await assert_device_name(client, "udp-multicast-test")
|
|
|
|
await waiter.wait_for(f"Joined multicast {MULTICAST_GROUP}", timeout=5.0)
|
|
|
|
# The poll cancels itself after the join, so a second poll period must not join again
|
|
await asyncio.sleep(2.5)
|
|
join_lines = [line for line in waiter.lines if "Joined multicast" in line]
|
|
assert len(join_lines) == 1, f"Expected one join, got: {join_lines}"
|
|
assert not any("IP_ADD_MEMBERSHIP" in line for line in waiter.lines), (
|
|
f"Join reported a failure. Log: {waiter.lines[-20:]}"
|
|
)
|
|
|
|
send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
send_sock.sendto(b"MULTICAST_PAYLOAD", (MULTICAST_GROUP, listen_port))
|
|
finally:
|
|
send_sock.close()
|
|
|
|
await waiter.wait_for("Received UDP:", timeout=5.0)
|