Use SsdpServiceInfo for ssdp tests (part 1) (#60320)

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/60336/head
epenet 2021-11-25 14:35:19 +01:00 committed by GitHub
parent 18a82e43a4
commit 9eed18f121
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 215 additions and 131 deletions

View File

@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import aiohttp
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.harmony.config_flow import CannotConnect
from homeassistant.components.harmony.const import DOMAIN, PREVIOUS_ACTIVE_ACTIVITY
from homeassistant.const import CONF_HOST, CONF_NAME
@ -58,10 +59,14 @@ async def test_form_ssdp(hass):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "Harmony Hub",
"ssdp_location": "http://192.168.1.12:8088/description",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.12:8088/description",
upnp={
"friendlyName": "Harmony Hub",
},
),
)
assert result["type"] == "form"
assert result["step_id"] == "link"
@ -106,10 +111,14 @@ async def test_form_ssdp_fails_to_get_remote_id(hass):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "Harmony Hub",
"ssdp_location": "http://192.168.1.12:8088/description",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.12:8088/description",
upnp={
"friendlyName": "Harmony Hub",
},
),
)
assert result["type"] == "abort"
assert result["reason"] == "cannot_connect"
@ -139,10 +148,14 @@ async def test_form_ssdp_aborts_before_checking_remoteid_if_host_known(hass):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "Harmony Hub",
"ssdp_location": "http://2.2.2.2:8088/description",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://2.2.2.2:8088/description",
upnp={
"friendlyName": "Harmony Hub",
},
),
)
assert result["type"] == "abort"

View File

@ -146,16 +146,20 @@ def dispatcher_fixture() -> Dispatcher:
@pytest.fixture(name="discovery_data")
def discovery_data_fixture() -> dict:
"""Return mock discovery data for testing."""
return {
ssdp.ATTR_SSDP_LOCATION: "http://127.0.0.1:60006/upnp/desc/aios_device/aios_device.xml",
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-denon-com:device:AiosDevice:1",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Office",
ssdp.ATTR_UPNP_MANUFACTURER: "Denon",
ssdp.ATTR_UPNP_MODEL_NAME: "HEOS Drive",
ssdp.ATTR_UPNP_MODEL_NUMBER: "DWSA-10 4.0",
ssdp.ATTR_UPNP_SERIAL: None,
ssdp.ATTR_UPNP_UDN: "uuid:e61de70c-2250-1c22-0080-0005cdf512be",
}
return ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://127.0.0.1:60006/upnp/desc/aios_device/aios_device.xml",
upnp={
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-denon-com:device:AiosDevice:1",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Office",
ssdp.ATTR_UPNP_MANUFACTURER: "Denon",
ssdp.ATTR_UPNP_MODEL_NAME: "HEOS Drive",
ssdp.ATTR_UPNP_MODEL_NUMBER: "DWSA-10 4.0",
ssdp.ATTR_UPNP_SERIAL: None,
ssdp.ATTR_UPNP_UDN: "uuid:e61de70c-2250-1c22-0080-0005cdf512be",
},
)
@pytest.fixture(name="quick_selects")

View File

@ -79,7 +79,9 @@ async def test_create_entry_when_friendly_name_valid(hass, controller):
assert DATA_DISCOVERED_HOSTS not in hass.data
async def test_discovery_shows_create_form(hass, controller, discovery_data):
async def test_discovery_shows_create_form(
hass, controller, discovery_data: ssdp.SsdpServiceInfo
):
"""Test discovery shows form to confirm setup and subsequent abort."""
await hass.config_entries.flow.async_init(
@ -91,9 +93,9 @@ async def test_discovery_shows_create_form(hass, controller, discovery_data):
assert len(flows_in_progress) == 1
assert hass.data[DATA_DISCOVERED_HOSTS] == {"Office (127.0.0.1)": "127.0.0.1"}
port = urlparse(discovery_data[ssdp.ATTR_SSDP_LOCATION]).port
discovery_data[ssdp.ATTR_SSDP_LOCATION] = f"http://127.0.0.2:{port}/"
discovery_data[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
port = urlparse(discovery_data.ssdp_location).port
discovery_data.ssdp_location = f"http://127.0.0.2:{port}/"
discovery_data.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
await hass.config_entries.flow.async_init(
heos.DOMAIN, context={"source": SOURCE_SSDP}, data=discovery_data
@ -109,7 +111,7 @@ async def test_discovery_shows_create_form(hass, controller, discovery_data):
async def test_discovery_flow_aborts_already_setup(
hass, controller, discovery_data, config_entry
hass, controller, discovery_data: ssdp.SsdpServiceInfo, config_entry
):
"""Test discovery flow aborts when entry already setup."""
config_entry.add_to_hass(hass)
@ -120,12 +122,14 @@ async def test_discovery_flow_aborts_already_setup(
assert result["reason"] == "single_instance_allowed"
async def test_discovery_sets_the_unique_id(hass, controller, discovery_data):
async def test_discovery_sets_the_unique_id(
hass, controller, discovery_data: ssdp.SsdpServiceInfo
):
"""Test discovery sets the unique id."""
port = urlparse(discovery_data[ssdp.ATTR_SSDP_LOCATION]).port
discovery_data[ssdp.ATTR_SSDP_LOCATION] = f"http://127.0.0.2:{port}/"
discovery_data[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
port = urlparse(discovery_data.ssdp_location).port
discovery_data.ssdp_location = f"http://127.0.0.2:{port}/"
discovery_data.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Bedroom"
await hass.config_entries.flow.async_init(
heos.DOMAIN, context={"source": SOURCE_SSDP}, data=discovery_data

View File

@ -326,11 +326,15 @@ async def test_form_ssdp_already_configured(hass: HomeAssistant) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
@ -341,11 +345,15 @@ async def test_form_ssdp(hass: HomeAssistant):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://{MOCK_HOSTNAME}{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
@ -385,11 +393,15 @@ async def test_form_ssdp_existing_entry(hass: HomeAssistant):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://3.3.3.3{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
await hass.async_block_till_done()
@ -412,11 +424,15 @@ async def test_form_ssdp_existing_entry_with_no_port(hass: HomeAssistant):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3/{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://3.3.3.3/{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
await hass.async_block_till_done()
@ -439,11 +455,15 @@ async def test_form_ssdp_existing_entry_with_alternate_port(hass: HomeAssistant)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"http://3.3.3.3:1443/{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://3.3.3.3:1443/{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
await hass.async_block_till_done()
@ -466,11 +486,15 @@ async def test_form_ssdp_existing_entry_no_port_https(hass: HomeAssistant):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: f"https://3.3.3.3/{ISY_URL_POSTFIX}",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"https://3.3.3.3/{ISY_URL_POSTFIX}",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "myisy",
ssdp.ATTR_UPNP_UDN: f"{UDN_UUID_PREFIX}{MOCK_UUID}",
},
),
)
await hass.async_block_till_done()

View File

@ -3,13 +3,9 @@ from http import HTTPStatus
import re
from socket import gaierror as SocketGIAError
from homeassistant.components import zeroconf
from homeassistant.components import ssdp, zeroconf
from homeassistant.components.roku.const import DOMAIN
from homeassistant.components.ssdp import (
ATTR_SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
)
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_SERIAL
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
@ -24,11 +20,15 @@ SSDP_LOCATION = "http://192.168.1.160/"
UPNP_FRIENDLY_NAME = "My Roku 3"
UPNP_SERIAL = "1GU48T017973"
MOCK_SSDP_DISCOVERY_INFO = {
ATTR_SSDP_LOCATION: SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME: UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL: UPNP_SERIAL,
}
MOCK_SSDP_DISCOVERY_INFO = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=SSDP_LOCATION,
upnp={
ATTR_UPNP_FRIENDLY_NAME: UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL: UPNP_SERIAL,
},
)
HOMEKIT_HOST = "192.168.1.161"

View File

@ -48,7 +48,7 @@ async def test_duplicate_error(
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
)
@ -216,7 +216,7 @@ async def test_ssdp_cannot_connect(
"""Test we abort SSDP flow on connection error."""
mock_connection(aioclient_mock, error=True)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP},
@ -231,7 +231,7 @@ async def test_ssdp_unknown_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow on unknown error."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
with patch(
"homeassistant.components.roku.config_flow.Roku.update",
side_effect=Exception,
@ -252,7 +252,7 @@ async def test_ssdp_discovery(
"""Test the SSDP discovery flow."""
mock_connection(aioclient_mock)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
)

View File

@ -1,5 +1,6 @@
"""Test the songpal config flow."""
import copy
import dataclasses
from unittest.mock import patch
from homeassistant.components import ssdp
@ -26,17 +27,21 @@ from tests.common import MockConfigEntry
UDN = "uuid:1234"
SSDP_DATA = {
ssdp.ATTR_UPNP_UDN: UDN,
ssdp.ATTR_UPNP_FRIENDLY_NAME: FRIENDLY_NAME,
ssdp.ATTR_SSDP_LOCATION: f"http://{HOST}:52323/dmr.xml",
"X_ScalarWebAPI_DeviceInfo": {
"X_ScalarWebAPI_BaseURL": ENDPOINT,
"X_ScalarWebAPI_ServiceList": {
"X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
SSDP_DATA = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://{HOST}:52323/dmr.xml",
upnp={
ssdp.ATTR_UPNP_UDN: UDN,
ssdp.ATTR_UPNP_FRIENDLY_NAME: FRIENDLY_NAME,
"X_ScalarWebAPI_DeviceInfo": {
"X_ScalarWebAPI_BaseURL": ENDPOINT,
"X_ScalarWebAPI_ServiceList": {
"X_ScalarWebAPI_ServiceType": ["guide", "system", "audio", "avContent"],
},
},
},
}
)
def _flow_next(hass, flow_id):
@ -150,8 +155,9 @@ def _create_mock_config_entry(hass):
async def test_ssdp_bravia(hass):
"""Test discovering a bravia TV."""
ssdp_data = copy.deepcopy(SSDP_DATA)
ssdp_data["X_ScalarWebAPI_DeviceInfo"]["X_ScalarWebAPI_ServiceList"][
ssdp_data = dataclasses.replace(SSDP_DATA)
ssdp_data.upnp = copy.deepcopy(ssdp_data.upnp)
ssdp_data.upnp["X_ScalarWebAPI_DeviceInfo"]["X_ScalarWebAPI_ServiceList"][
"X_ScalarWebAPI_ServiceType"
].append("videoScreen")
result = await hass.config_entries.flow.async_init(

View File

@ -387,11 +387,15 @@ async def test_form_ssdp(hass: HomeAssistant, service: MagicMock):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.5:5000",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX99", # MAC address, but SSDP does not have `-`
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.5:5000",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX99", # MAC address, but SSDP does not have `-`
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "link"
@ -434,11 +438,15 @@ async def test_reconfig_ssdp(hass: HomeAssistant, service: MagicMock):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.5:5000",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.5:5000",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reconfigure_successful"
@ -462,11 +470,15 @@ async def test_skip_reconfig_ssdp(hass: HomeAssistant, service: MagicMock):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.5:5000",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.5:5000",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
@ -490,11 +502,15 @@ async def test_existing_ssdp(hass: HomeAssistant, service: MagicMock):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_SSDP},
data={
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.5:5000",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.1.5:5000",
upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: "mydsm",
ssdp.ATTR_UPNP_SERIAL: "001132XXXX59", # Existing in MACS[0], but SSDP does not have `-`
},
),
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"

View File

@ -6,6 +6,7 @@ from unittest.mock import patch
import aiounifi
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.unifi.config_flow import async_discover_unifi
from homeassistant.components.unifi.const import (
CONF_ALLOW_BANDWIDTH_SENSORS,
@ -547,12 +548,16 @@ async def test_form_ssdp(hass):
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"ssdp_location": "http://192.168.208.1:41417/rootDesc.xml",
"serialNumber": "e0:63:da:20:14:a9",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.208.1:41417/rootDesc.xml",
upnp={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"serialNumber": "e0:63:da:20:14:a9",
},
),
)
assert result["type"] == "form"
assert result["step_id"] == "user"
@ -579,12 +584,16 @@ async def test_form_ssdp_aborts_if_host_already_exists(hass):
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"ssdp_location": "http://192.168.208.1:41417/rootDesc.xml",
"serialNumber": "e0:63:da:20:14:a9",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.208.1:41417/rootDesc.xml",
upnp={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"serialNumber": "e0:63:da:20:14:a9",
},
),
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -602,12 +611,16 @@ async def test_form_ssdp_aborts_if_serial_already_exists(hass):
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"ssdp_location": "http://192.168.208.1:41417/rootDesc.xml",
"serialNumber": "e0:63:da:20:14:a9",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://192.168.208.1:41417/rootDesc.xml",
upnp={
"friendlyName": "UniFi Dream Machine",
"modelDescription": "UniFi Dream Machine Pro",
"serialNumber": "e0:63:da:20:14:a9",
},
),
)
assert result["type"] == "abort"
assert result["reason"] == "already_configured"
@ -625,12 +638,16 @@ async def test_form_ssdp_gets_form_with_ignored_entry(hass):
result = await hass.config_entries.flow.async_init(
UNIFI_DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data={
"friendlyName": "UniFi Dream Machine New",
"modelDescription": "UniFi Dream Machine Pro",
"ssdp_location": "http://1.2.3.4:41417/rootDesc.xml",
"serialNumber": "e0:63:da:20:14:a9",
},
data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://1.2.3.4:41417/rootDesc.xml",
upnp={
"friendlyName": "UniFi Dream Machine New",
"modelDescription": "UniFi Dream Machine Pro",
"serialNumber": "e0:63:da:20:14:a9",
},
),
)
assert result["type"] == "form"
assert result["step_id"] == "user"