core/tests/components/upnp/test_config_flow.py

"""Test UPnP/IGD config flow."""

from datetime import timedelta

from homeassistant import config_entries, data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.upnp.const import (
    CONFIG_ENTRY_SCAN_INTERVAL,
    CONFIG_ENTRY_ST,
    CONFIG_ENTRY_UDN,
    DEFAULT_SCAN_INTERVAL,
    DISCOVERY_LOCATION,
    DISCOVERY_ST,
    DISCOVERY_UDN,
    DISCOVERY_USN,
    DOMAIN,
)
from homeassistant.components.upnp.device import Device
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component

from .mock_device import MockDevice

from tests.async_mock import AsyncMock, patch
from tests.common import MockConfigEntry


async def test_flow_ssdp_discovery(hass: HomeAssistantType):
    """Test config flow: discovered + configured through ssdp."""
    udn = "uuid:device_1"
    mock_device = MockDevice(udn)
    discovery_infos = [
        {
            DISCOVERY_ST: mock_device.device_type,
            DISCOVERY_UDN: mock_device.udn,
            DISCOVERY_LOCATION: "dummy",
        }
    ]
    with patch.object(
        Device, "async_create_device", AsyncMock(return_value=mock_device)
    ), patch.object(Device, "async_discover", AsyncMock(return_value=discovery_infos)):
        # Discovered via step ssdp.
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": config_entries.SOURCE_SSDP},
            data={
                ssdp.ATTR_SSDP_ST: mock_device.device_type,
                ssdp.ATTR_UPNP_UDN: mock_device.udn,
                "friendlyName": mock_device.name,
            },
        )
        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
        assert result["step_id"] == "ssdp_confirm"

        # Confirm via step ssdp_confirm.
        result = await hass.config_entries.flow.async_configure(
            result["flow_id"],
            user_input={},
        )

        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
        assert result["title"] == mock_device.name
        assert result["data"] == {
            CONFIG_ENTRY_ST: mock_device.device_type,
            CONFIG_ENTRY_UDN: mock_device.udn,
        }


async def test_flow_ssdp_discovery_incomplete(hass: HomeAssistantType):
    """Test config flow: incomplete discovery through ssdp."""
    udn = "uuid:device_1"
    mock_device = MockDevice(udn)
    discovery_infos = [
        {
            DISCOVERY_ST: mock_device.device_type,
            DISCOVERY_UDN: mock_device.udn,
            DISCOVERY_LOCATION: "dummy",
        }
    ]
    with patch.object(
        Device, "async_create_device", AsyncMock(return_value=mock_device)
    ), patch.object(Device, "async_discover", AsyncMock(return_value=discovery_infos)):
        # Discovered via step ssdp.
        result = await hass.config_entries.flow.async_init(
            DOMAIN,
            context={"source": config_entries.SOURCE_SSDP},
            data={
                ssdp.ATTR_SSDP_ST: mock_device.device_type,
                # ssdp.ATTR_UPNP_UDN: mock_device.udn,  # Not provided.
                "friendlyName": mock_device.name,
            },
        )
        assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
        assert result["reason"] == "incomplete_discovery"


async def test_flow_user(hass: HomeAssistantType):
    """Test config flow: discovered + configured through user."""
    udn = "uuid:device_1"
    mock_device = MockDevice(udn)
    usn = f"{mock_device.udn}::{mock_device.device_type}"
    discovery_infos = [
        {
            DISCOVERY_USN: usn,
            DISCOVERY_ST: mock_device.device_type,
            DISCOVERY_UDN: mock_device.udn,
            DISCOVERY_LOCATION: "dummy",
        }
    ]

    with patch.object(
        Device, "async_create_device", AsyncMock(return_value=mock_device)
    ), patch.object(Device, "async_discover", AsyncMock(return_value=discovery_infos)):
        # Discovered via step user.
        result = await hass.config_entries.flow.async_init(
            DOMAIN, context={"source": config_entries.SOURCE_USER}
        )
        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
        assert result["step_id"] == "user"

        # Confirmed via step user.
        result = await hass.config_entries.flow.async_configure(
            result["flow_id"],
            user_input={"usn": usn},
        )

        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
        assert result["title"] == mock_device.name
        assert result["data"] == {
            CONFIG_ENTRY_ST: mock_device.device_type,
            CONFIG_ENTRY_UDN: mock_device.udn,
        }


async def test_flow_config(hass: HomeAssistantType):
    """Test config flow: discovered + configured through configuration.yaml."""
    udn = "uuid:device_1"
    mock_device = MockDevice(udn)
    usn = f"{mock_device.udn}::{mock_device.device_type}"
    discovery_infos = [
        {
            DISCOVERY_USN: usn,
            DISCOVERY_ST: mock_device.device_type,
            DISCOVERY_UDN: mock_device.udn,
            DISCOVERY_LOCATION: "dummy",
        }
    ]

    with patch.object(
        Device, "async_create_device", AsyncMock(return_value=mock_device)
    ), patch.object(Device, "async_discover", AsyncMock(return_value=discovery_infos)):
        # Discovered via step import.
        result = await hass.config_entries.flow.async_init(
            DOMAIN, context={"source": config_entries.SOURCE_IMPORT}
        )

        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
        assert result["title"] == mock_device.name
        assert result["data"] == {
            CONFIG_ENTRY_ST: mock_device.device_type,
            CONFIG_ENTRY_UDN: mock_device.udn,
        }


async def test_options_flow(hass: HomeAssistantType):
    """Test options flow."""
    # Set up config entry.
    udn = "uuid:device_1"
    mock_device = MockDevice(udn)
    discovery_infos = [
        {
            DISCOVERY_UDN: mock_device.udn,
            DISCOVERY_ST: mock_device.device_type,
            DISCOVERY_LOCATION: "http://192.168.1.1/desc.xml",
        }
    ]
    config_entry = MockConfigEntry(
        domain=DOMAIN,
        data={
            CONFIG_ENTRY_UDN: mock_device.udn,
            CONFIG_ENTRY_ST: mock_device.device_type,
        },
        options={CONFIG_ENTRY_SCAN_INTERVAL: DEFAULT_SCAN_INTERVAL},
    )
    config_entry.add_to_hass(hass)

    config = {
        # no upnp, ensures no import-flow is started.
    }
    async_discover = AsyncMock(return_value=discovery_infos)
    with patch.object(
        Device, "async_create_device", AsyncMock(return_value=mock_device)
    ), patch.object(Device, "async_discover", async_discover):
        # Initialisation of component.
        await async_setup_component(hass, "upnp", config)
        await hass.async_block_till_done()

        # DataUpdateCoordinator gets a default of 30 seconds for updates.
        coordinator = hass.data[DOMAIN]["coordinators"][mock_device.udn]
        assert coordinator.update_interval == timedelta(seconds=DEFAULT_SCAN_INTERVAL)

        # Options flow with no input results in form.
        result = await hass.config_entries.options.async_init(
            config_entry.entry_id,
        )
        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM

        # Options flow with input results in update to entry.
        result2 = await hass.config_entries.options.async_configure(
            result["flow_id"],
            user_input={CONFIG_ENTRY_SCAN_INTERVAL: 60},
        )
        assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
        assert config_entry.options == {
            CONFIG_ENTRY_SCAN_INTERVAL: 60,
        }

        # Also updates DataUpdateCoordinator.
        assert coordinator.update_interval == timedelta(seconds=60)