esphome/tests/component_tests/api/test_outgoing_connection.py

"""Tests for the api outgoing_connection option."""

from collections.abc import Callable
from pathlib import Path

import pytest

from esphome.components import socket
from esphome.components.api import (
    CONFIG_SCHEMA,
    _validate_outgoing_host_ipv6,
    _validate_outgoing_socket_implementation,
)
from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
import esphome.config_validation as cv
from esphome.const import PlatformFramework
from esphome.core import CORE
import esphome.final_validate as fv
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable

KEY = "bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU="
ESP32_PLATFORM_DATA = {KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32}


def _api_config(outgoing: ConfigType, *, encryption: bool = True) -> ConfigType:
    config: ConfigType = {"outgoing_connection": outgoing}
    if encryption:
        config["encryption"] = {"key": KEY}
    return config


def test_outgoing_connection_generates_defines(
    generate_main: Callable[[str | Path], str],
) -> None:
    """A valid config emits the compile-time defines with defaults applied."""
    generate_main("tests/component_tests/api/test_outgoing_connection.yaml")

    defines = {define.name: define.value for define in CORE.defines}
    assert "USE_API_OUTGOING_CONNECTION" in defines
    assert str(defines["API_OUTGOING_CONNECTION_HOST"]) == '"192.168.1.2"'
    assert str(defines["API_OUTGOING_CONNECTION_PORT"]) == "6054"
    assert str(defines["API_OUTGOING_CONNECTION_DELAY"]) == "60000"


def test_outgoing_connection_defaults(
    set_core_config: SetCoreConfigCallable,
) -> None:
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    config = CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}))
    outgoing = config["outgoing_connection"]
    assert outgoing["port"] == 6054
    assert outgoing["delay"].total_milliseconds == 60000


def test_outgoing_connection_bare_block(
    set_core_config: SetCoreConfigCallable,
) -> None:
    """A bare outgoing_connection: block is valid; the device dials the
    remembered last dial-back client."""
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    config = CONFIG_SCHEMA(_api_config(None))
    outgoing = config["outgoing_connection"]
    assert "host" not in outgoing
    assert outgoing["port"] == 6054


def test_outgoing_connection_delay_bounded(
    set_core_config: SetCoreConfigCallable,
) -> None:
    """A delay past half the uint32 millisecond range is rejected, not wrapped."""
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    with pytest.raises(cv.Invalid, match="value must be at most"):
        CONFIG_SCHEMA(_api_config({"delay": "60d"}))


def test_outgoing_connection_requires_encryption(
    set_core_config: SetCoreConfigCallable,
) -> None:
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    with pytest.raises(cv.Invalid, match="requires 'encryption'"):
        CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}, encryption=False))


@pytest.mark.parametrize(
    ("platform_framework", "platform_data", "socket_conf"),
    [
        # The platform default on these two, resolved like AUTO_LOAD does
        (PlatformFramework.ESP8266_ARDUINO, None, None),
        (PlatformFramework.RP2040_ARDUINO, None, None),
        # An explicit selection elsewhere
        (
            PlatformFramework.ESP32_IDF,
            ESP32_PLATFORM_DATA,
            {"implementation": "lwip_tcp"},
        ),
    ],
)
def test_outgoing_connection_rejects_lwip_tcp(
    set_core_config: SetCoreConfigCallable,
    platform_framework: PlatformFramework,
    platform_data: ConfigType | None,
    socket_conf: ConfigType | None,
) -> None:
    """The resolved lwip_tcp socket is rejected at final validate."""
    set_core_config(platform_framework, platform_data=platform_data)
    fv.full_config.set({"socket": socket_conf or socket.CONFIG_SCHEMA({})})
    config = CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}))
    with pytest.raises(cv.Invalid, match="lwip_tcp"):
        _validate_outgoing_socket_implementation(config)


def test_outgoing_connection_rejects_hostnames(
    set_core_config: SetCoreConfigCallable,
) -> None:
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    with pytest.raises(cv.Invalid, match="not a valid IP address"):
        CONFIG_SCHEMA(_api_config({"host": "homeassistant.local"}))


def test_outgoing_connection_ipv6_host_requires_ipv6(
    set_core_config: SetCoreConfigCallable,
) -> None:
    set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
    config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
    with pytest.raises(cv.Invalid, match="IPv6 is not"):
        _validate_outgoing_host_ipv6(config)


def test_outgoing_connection_ipv6_host_passes_with_ipv6_enabled(
    set_core_config: SetCoreConfigCallable,
) -> None:
    set_core_config(
        PlatformFramework.ESP32_IDF,
        platform_data=ESP32_PLATFORM_DATA,
        full_config={"network": {"enable_ipv6": True}},
    )
    config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
    assert _validate_outgoing_host_ipv6(config) is config


def test_outgoing_connection_ipv6_host_with_ipv6(
    generate_main: Callable[[str | Path], str],
) -> None:
    generate_main("tests/component_tests/api/test_outgoing_connection_ipv6.yaml")

    defines = {define.name: define.value for define in CORE.defines}
    assert str(defines["API_OUTGOING_CONNECTION_HOST"]) == '"fd00::1"'