Use SsdpServiceInfo for ssdp tests (part 3) (#60334)

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-11-25 18:46:20 +01:00 committed by GitHub
parent 624d866239
commit f292691b7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 374 additions and 264 deletions

View File

@ -146,6 +146,24 @@ class SsdpServiceInfo(
return getattr(self, name) return getattr(self, name)
return self.upnp.get(name) return self.upnp.get(name)
def get(self, name: str, default: Any = None) -> Any:
"""
Allow property access by name for compatibility reason.
Deprecated, and will be removed in version 2022.6.
"""
if not self._warning_logged:
report(
f"accessed discovery_info.get('{name}') instead of discovery_info.{name}; this will fail in version 2022.6",
exclude_integrations={"ssdp"},
error_if_core=False,
level=logging.DEBUG,
)
self._warning_logged = True
if hasattr(self, name):
return getattr(self, name)
return self.upnp.get(name, default)
@bind_hass @bind_hass
async def async_register_callback( async def async_register_callback(

View File

@ -5,6 +5,7 @@ from unittest.mock import patch
import pydeconz import pydeconz
from homeassistant.components import ssdp
from homeassistant.components.deconz.config_flow import ( from homeassistant.components.deconz.config_flow import (
CONF_MANUAL_INPUT, CONF_MANUAL_INPUT,
CONF_SERIAL, CONF_SERIAL,
@ -17,11 +18,7 @@ from homeassistant.components.deconz.const import (
CONF_MASTER_GATEWAY, CONF_MASTER_GATEWAY,
DOMAIN as DECONZ_DOMAIN, DOMAIN as DECONZ_DOMAIN,
) )
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL
ATTR_SSDP_LOCATION,
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
)
from homeassistant.config_entries import ( from homeassistant.config_entries import (
SOURCE_HASSIO, SOURCE_HASSIO,
SOURCE_REAUTH, SOURCE_REAUTH,
@ -412,11 +409,15 @@ async def test_flow_ssdp_discovery(hass, aioclient_mock):
"""Test that config flow for one discovered bridge works.""" """Test that config flow for one discovered bridge works."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DECONZ_DOMAIN, DECONZ_DOMAIN,
data={ data=ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "http://1.2.3.4:80/", ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL, ssdp_st="mock_st",
ATTR_UPNP_SERIAL: BRIDGEID, ssdp_location="http://1.2.3.4:80/",
}, upnp={
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL,
ATTR_UPNP_SERIAL: BRIDGEID,
},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )
@ -446,7 +447,11 @@ async def test_flow_ssdp_bad_discovery(hass, aioclient_mock):
"""Test that SSDP discovery aborts if manufacturer URL is wrong.""" """Test that SSDP discovery aborts if manufacturer URL is wrong."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DECONZ_DOMAIN, DECONZ_DOMAIN,
data={ATTR_UPNP_MANUFACTURER_URL: "other"}, data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
upnp={ATTR_UPNP_MANUFACTURER_URL: "other"},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )
@ -464,11 +469,15 @@ async def test_ssdp_discovery_update_configuration(hass, aioclient_mock):
) as mock_setup_entry: ) as mock_setup_entry:
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DECONZ_DOMAIN, DECONZ_DOMAIN,
data={ data=ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "http://2.3.4.5:80/", ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL, ssdp_st="mock_st",
ATTR_UPNP_SERIAL: BRIDGEID, ssdp_location="http://2.3.4.5:80/",
}, upnp={
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL,
ATTR_UPNP_SERIAL: BRIDGEID,
},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
@ -485,11 +494,15 @@ async def test_ssdp_discovery_dont_update_configuration(hass, aioclient_mock):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DECONZ_DOMAIN, DECONZ_DOMAIN,
data={ data=ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "http://1.2.3.4:80/", ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL, ssdp_st="mock_st",
ATTR_UPNP_SERIAL: BRIDGEID, ssdp_location="http://1.2.3.4:80/",
}, upnp={
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL,
ATTR_UPNP_SERIAL: BRIDGEID,
},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )
@ -508,11 +521,15 @@ async def test_ssdp_discovery_dont_update_existing_hassio_configuration(
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DECONZ_DOMAIN, DECONZ_DOMAIN,
data={ data=ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "http://1.2.3.4:80/", ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL, ssdp_st="mock_st",
ATTR_UPNP_SERIAL: BRIDGEID, ssdp_location="http://1.2.3.4:80/",
}, upnp={
ATTR_UPNP_MANUFACTURER_URL: DECONZ_MANUFACTURERURL,
ATTR_UPNP_SERIAL: BRIDGEID,
},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )

View File

@ -1,8 +1,8 @@
"""Tests for the DirecTV component.""" """Tests for the DirecTV component."""
from http import HTTPStatus from http import HTTPStatus
from homeassistant.components import ssdp
from homeassistant.components.directv.const import CONF_RECEIVER_ID, DOMAIN from homeassistant.components.directv.const import CONF_RECEIVER_ID, DOMAIN
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION
from homeassistant.const import CONF_HOST, CONTENT_TYPE_JSON from homeassistant.const import CONF_HOST, CONTENT_TYPE_JSON
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -15,7 +15,12 @@ SSDP_LOCATION = "http://127.0.0.1/"
UPNP_SERIAL = "RID-028877455858" UPNP_SERIAL = "RID-028877455858"
MOCK_CONFIG = {DOMAIN: [{CONF_HOST: HOST}]} MOCK_CONFIG = {DOMAIN: [{CONF_HOST: HOST}]}
MOCK_SSDP_DISCOVERY_INFO = {ATTR_SSDP_LOCATION: SSDP_LOCATION} MOCK_SSDP_DISCOVERY_INFO = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=SSDP_LOCATION,
upnp={},
)
MOCK_USER_INPUT = {CONF_HOST: HOST} MOCK_USER_INPUT = {CONF_HOST: HOST}

View File

@ -1,4 +1,5 @@
"""Test the DirecTV config flow.""" """Test the DirecTV config flow."""
import dataclasses
from unittest.mock import patch from unittest.mock import patch
from aiohttp import ClientError as HTTPClientError from aiohttp import ClientError as HTTPClientError
@ -43,7 +44,7 @@ async def test_show_ssdp_form(
"""Test that the ssdp confirmation form is served.""" """Test that the ssdp confirmation form is served."""
mock_connection(aioclient_mock) 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( result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
) )
@ -77,7 +78,7 @@ async def test_ssdp_cannot_connect(
"""Test we abort SSDP flow on connection error.""" """Test we abort SSDP flow on connection error."""
aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError) aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP}, context={CONF_SOURCE: SOURCE_SSDP},
@ -94,7 +95,7 @@ async def test_ssdp_confirm_cannot_connect(
"""Test we abort SSDP flow on connection error.""" """Test we abort SSDP flow on connection error."""
aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError) aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP, CONF_HOST: HOST, CONF_NAME: HOST}, context={CONF_SOURCE: SOURCE_SSDP, CONF_HOST: HOST, CONF_NAME: HOST},
@ -128,7 +129,7 @@ async def test_ssdp_device_exists_abort(
"""Test we abort SSDP flow if DirecTV receiver already configured.""" """Test we abort SSDP flow if DirecTV receiver already configured."""
await setup_integration(hass, aioclient_mock, skip_entry_setup=True) await setup_integration(hass, aioclient_mock, skip_entry_setup=True)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP}, context={CONF_SOURCE: SOURCE_SSDP},
@ -145,8 +146,8 @@ async def test_ssdp_with_receiver_id_device_exists_abort(
"""Test we abort SSDP flow if DirecTV receiver already configured.""" """Test we abort SSDP flow if DirecTV receiver already configured."""
await setup_integration(hass, aioclient_mock, skip_entry_setup=True) await setup_integration(hass, aioclient_mock, skip_entry_setup=True)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
discovery_info[ATTR_UPNP_SERIAL] = UPNP_SERIAL discovery_info.upnp[ATTR_UPNP_SERIAL] = UPNP_SERIAL
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP}, context={CONF_SOURCE: SOURCE_SSDP},
@ -180,7 +181,7 @@ async def test_ssdp_unknown_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None: ) -> None:
"""Test we abort SSDP flow on unknown error.""" """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( with patch(
"homeassistant.components.directv.config_flow.DIRECTV.update", "homeassistant.components.directv.config_flow.DIRECTV.update",
side_effect=Exception, side_effect=Exception,
@ -199,7 +200,7 @@ async def test_ssdp_confirm_unknown_error(
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
) -> None: ) -> None:
"""Test we abort SSDP flow on unknown error.""" """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( with patch(
"homeassistant.components.directv.config_flow.DIRECTV.update", "homeassistant.components.directv.config_flow.DIRECTV.update",
side_effect=Exception, side_effect=Exception,
@ -249,7 +250,7 @@ async def test_full_ssdp_flow_implementation(
"""Test the full SSDP flow from start to finish.""" """Test the full SSDP flow from start to finish."""
mock_connection(aioclient_mock) 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( result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
) )

View File

@ -1,6 +1,7 @@
"""Test the DLNA config flow.""" """Test the DLNA config flow."""
from __future__ import annotations from __future__ import annotations
import dataclasses
from unittest.mock import Mock from unittest.mock import Mock
from async_upnp_client import UpnpDevice, UpnpError from async_upnp_client import UpnpDevice, UpnpError
@ -23,7 +24,6 @@ from homeassistant.const import (
CONF_URL, CONF_URL,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import DiscoveryInfoType
from .conftest import ( from .conftest import (
MOCK_DEVICE_LOCATION, MOCK_DEVICE_LOCATION,
@ -52,40 +52,43 @@ MOCK_CONFIG_IMPORT_DATA = {
MOCK_ROOT_DEVICE_UDN = "ROOT_DEVICE" MOCK_ROOT_DEVICE_UDN = "ROOT_DEVICE"
MOCK_DISCOVERY: DiscoveryInfoType = { MOCK_DISCOVERY = ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: MOCK_DEVICE_LOCATION, ssdp_usn="mock_usn",
ssdp.ATTR_SSDP_UDN: MOCK_DEVICE_UDN, ssdp_location=MOCK_DEVICE_LOCATION,
ssdp.ATTR_SSDP_ST: MOCK_DEVICE_TYPE, ssdp_udn=MOCK_DEVICE_UDN,
ssdp.ATTR_UPNP_UDN: MOCK_ROOT_DEVICE_UDN, ssdp_st=MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE, upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME, ssdp.ATTR_UPNP_UDN: MOCK_ROOT_DEVICE_UDN,
ssdp.ATTR_UPNP_SERVICE_LIST: { ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE,
"service": [ ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME,
{ ssdp.ATTR_UPNP_SERVICE_LIST: {
"SCPDURL": "/AVTransport/scpd.xml", "service": [
"controlURL": "/AVTransport/control.xml", {
"eventSubURL": "/AVTransport/event.xml", "SCPDURL": "/AVTransport/scpd.xml",
"serviceId": "urn:upnp-org:serviceId:AVTransport", "controlURL": "/AVTransport/control.xml",
"serviceType": "urn:schemas-upnp-org:service:AVTransport:1", "eventSubURL": "/AVTransport/event.xml",
}, "serviceId": "urn:upnp-org:serviceId:AVTransport",
{ "serviceType": "urn:schemas-upnp-org:service:AVTransport:1",
"SCPDURL": "/ConnectionManager/scpd.xml", },
"controlURL": "/ConnectionManager/control.xml", {
"eventSubURL": "/ConnectionManager/event.xml", "SCPDURL": "/ConnectionManager/scpd.xml",
"serviceId": "urn:upnp-org:serviceId:ConnectionManager", "controlURL": "/ConnectionManager/control.xml",
"serviceType": "urn:schemas-upnp-org:service:ConnectionManager:1", "eventSubURL": "/ConnectionManager/event.xml",
}, "serviceId": "urn:upnp-org:serviceId:ConnectionManager",
{ "serviceType": "urn:schemas-upnp-org:service:ConnectionManager:1",
"SCPDURL": "/RenderingControl/scpd.xml", },
"controlURL": "/RenderingControl/control.xml", {
"eventSubURL": "/RenderingControl/event.xml", "SCPDURL": "/RenderingControl/scpd.xml",
"serviceId": "urn:upnp-org:serviceId:RenderingControl", "controlURL": "/RenderingControl/control.xml",
"serviceType": "urn:schemas-upnp-org:service:RenderingControl:1", "eventSubURL": "/RenderingControl/event.xml",
}, "serviceId": "urn:upnp-org:serviceId:RenderingControl",
] "serviceType": "urn:schemas-upnp-org:service:RenderingControl:1",
},
]
},
}, },
ssdp.ATTR_HA_MATCHING_DOMAINS: {DLNA_DOMAIN}, x_homeassistant_matching_domains=(DLNA_DOMAIN,),
} )
async def test_user_flow_undiscovered_manual(hass: HomeAssistant) -> None: async def test_user_flow_undiscovered_manual(hass: HomeAssistant) -> None:
@ -545,13 +548,17 @@ async def test_ssdp_flow_existing(
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION, ssdp_usn="mock_usn",
ssdp.ATTR_SSDP_UDN: MOCK_DEVICE_UDN, ssdp_st="mock_st",
ssdp.ATTR_UPNP_UDN: MOCK_ROOT_DEVICE_UDN, ssdp_location=NEW_DEVICE_LOCATION,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE, ssdp_udn=MOCK_DEVICE_UDN,
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME, upnp={
}, ssdp.ATTR_UPNP_UDN: MOCK_ROOT_DEVICE_UDN,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME,
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
@ -581,14 +588,17 @@ async def test_ssdp_flow_upnp_udn(
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: NEW_DEVICE_LOCATION, ssdp_usn="mock_usn",
ssdp.ATTR_SSDP_UDN: MOCK_DEVICE_UDN, ssdp_location=NEW_DEVICE_LOCATION,
ssdp.ATTR_SSDP_ST: MOCK_DEVICE_TYPE, ssdp_udn=MOCK_DEVICE_UDN,
ssdp.ATTR_UPNP_UDN: "DIFFERENT_ROOT_DEVICE", ssdp_st=MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE, upnp={
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME, ssdp.ATTR_UPNP_UDN: "DIFFERENT_ROOT_DEVICE",
}, ssdp.ATTR_UPNP_DEVICE_TYPE: MOCK_DEVICE_TYPE,
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_DEVICE_NAME,
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
@ -598,8 +608,9 @@ async def test_ssdp_flow_upnp_udn(
async def test_ssdp_missing_services(hass: HomeAssistant) -> None: async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
"""Test SSDP ignores devices that are missing required services.""" """Test SSDP ignores devices that are missing required services."""
# No services defined at all # No services defined at all
discovery = dict(MOCK_DISCOVERY) discovery = dataclasses.replace(MOCK_DISCOVERY)
del discovery[ssdp.ATTR_UPNP_SERVICE_LIST] discovery.upnp = discovery.upnp.copy()
del discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST]
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
@ -609,11 +620,12 @@ async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
assert result["reason"] == "not_dmr" assert result["reason"] == "not_dmr"
# AVTransport service is missing # AVTransport service is missing
discovery = dict(MOCK_DISCOVERY) discovery = dataclasses.replace(MOCK_DISCOVERY)
discovery[ssdp.ATTR_UPNP_SERVICE_LIST] = { discovery.upnp = discovery.upnp.copy()
discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST] = {
"service": [ "service": [
service service
for service in discovery[ssdp.ATTR_UPNP_SERVICE_LIST]["service"] for service in discovery.upnp[ssdp.ATTR_UPNP_SERVICE_LIST]["service"]
if service.get("serviceId") != "urn:upnp-org:serviceId:AVTransport" if service.get("serviceId") != "urn:upnp-org:serviceId:AVTransport"
] ]
} }
@ -626,8 +638,9 @@ async def test_ssdp_missing_services(hass: HomeAssistant) -> None:
async def test_ssdp_ignore_device(hass: HomeAssistant) -> None: async def test_ssdp_ignore_device(hass: HomeAssistant) -> None:
"""Test SSDP discovery ignores certain devices.""" """Test SSDP discovery ignores certain devices."""
discovery = dict(MOCK_DISCOVERY) discovery = dataclasses.replace(MOCK_DISCOVERY)
discovery[ssdp.ATTR_HA_MATCHING_DOMAINS] = {DLNA_DOMAIN, "other_domain"} discovery.x_homeassistant_matching_domains = {DLNA_DOMAIN, "other_domain"}
assert discovery.x_homeassistant_matching_domains
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
@ -636,8 +649,11 @@ async def test_ssdp_ignore_device(hass: HomeAssistant) -> None:
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "alternative_integration" assert result["reason"] == "alternative_integration"
discovery = dict(MOCK_DISCOVERY) discovery = dataclasses.replace(MOCK_DISCOVERY)
discovery[ssdp.ATTR_UPNP_DEVICE_TYPE] = "urn:schemas-upnp-org:device:ZonePlayer:1" discovery.upnp = discovery.upnp.copy()
discovery.upnp[
ssdp.ATTR_UPNP_DEVICE_TYPE
] = "urn:schemas-upnp-org:device:ZonePlayer:1"
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
@ -652,9 +668,10 @@ async def test_ssdp_ignore_device(hass: HomeAssistant) -> None:
("LG Electronics.", "LG TV"), ("LG Electronics.", "LG TV"),
("Royal Philips Electronics", "Philips TV DMR"), ("Royal Philips Electronics", "Philips TV DMR"),
]: ]:
discovery = dict(MOCK_DISCOVERY) discovery = dataclasses.replace(MOCK_DISCOVERY)
discovery[ssdp.ATTR_UPNP_MANUFACTURER] = manufacturer discovery.upnp = discovery.upnp.copy()
discovery[ssdp.ATTR_UPNP_MODEL_NAME] = model discovery.upnp[ssdp.ATTR_UPNP_MANUFACTURER] = manufacturer
discovery.upnp[ssdp.ATTR_UPNP_MODEL_NAME] = model
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DLNA_DOMAIN, DLNA_DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},

View File

@ -1,9 +1,11 @@
"""Tests for AVM Fritz!Box config flow.""" """Tests for AVM Fritz!Box config flow."""
import dataclasses
from unittest.mock import patch from unittest.mock import patch
from fritzconnection.core.exceptions import FritzConnectionException, FritzSecurityError from fritzconnection.core.exceptions import FritzConnectionException, FritzSecurityError
import pytest import pytest
from homeassistant.components import ssdp
from homeassistant.components.device_tracker.const import ( from homeassistant.components.device_tracker.const import (
CONF_CONSIDER_HOME, CONF_CONSIDER_HOME,
DEFAULT_CONSIDER_HOME, DEFAULT_CONSIDER_HOME,
@ -14,11 +16,7 @@ from homeassistant.components.fritz.const import (
ERROR_CANNOT_CONNECT, ERROR_CANNOT_CONNECT,
ERROR_UNKNOWN, ERROR_UNKNOWN,
) )
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_UDN
ATTR_SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
)
from homeassistant.config_entries import ( from homeassistant.config_entries import (
SOURCE_IMPORT, SOURCE_IMPORT,
SOURCE_REAUTH, SOURCE_REAUTH,
@ -51,11 +49,15 @@ MOCK_DEVICE_INFO = {
ATTR_NEW_SERIAL_NUMBER: MOCK_SERIAL_NUMBER, ATTR_NEW_SERIAL_NUMBER: MOCK_SERIAL_NUMBER,
} }
MOCK_IMPORT_CONFIG = {CONF_HOST: MOCK_HOST, CONF_USERNAME: "username"} MOCK_IMPORT_CONFIG = {CONF_HOST: MOCK_HOST, CONF_USERNAME: "username"}
MOCK_SSDP_DATA = { MOCK_SSDP_DATA = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: f"https://{MOCK_IP}:12345/test", ssdp_usn="mock_usn",
ATTR_UPNP_FRIENDLY_NAME: "fake_name", ssdp_st="mock_st",
ATTR_UPNP_UDN: "uuid:only-a-test", ssdp_location=f"https://{MOCK_IP}:12345/test",
} upnp={
ATTR_UPNP_FRIENDLY_NAME: "fake_name",
ATTR_UPNP_UDN: "uuid:only-a-test",
},
)
MOCK_REQUEST = b'<?xml version="1.0" encoding="utf-8"?><SessionInfo><SID>xxxxxxxxxxxxxxxx</SID><Challenge>xxxxxxxx</Challenge><BlockTime>0</BlockTime><Rights><Name>Dial</Name><Access>2</Access><Name>App</Name><Access>2</Access><Name>HomeAuto</Name><Access>2</Access><Name>BoxAdmin</Name><Access>2</Access><Name>Phone</Name><Access>2</Access><Name>NAS</Name><Access>2</Access></Rights><Users><User last="1">FakeFritzUser</User></Users></SessionInfo>\n' MOCK_REQUEST = b'<?xml version="1.0" encoding="utf-8"?><SessionInfo><SID>xxxxxxxxxxxxxxxx</SID><Challenge>xxxxxxxx</Challenge><BlockTime>0</BlockTime><Rights><Name>Dial</Name><Access>2</Access><Name>App</Name><Access>2</Access><Name>HomeAuto</Name><Access>2</Access><Name>BoxAdmin</Name><Access>2</Access><Name>Phone</Name><Access>2</Access><Name>NAS</Name><Access>2</Access></Rights><Users><User last="1">FakeFritzUser</User></Users></SessionInfo>\n'
@ -407,8 +409,9 @@ async def test_ssdp_already_in_progress_host(
assert result["type"] == RESULT_TYPE_FORM assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm" assert result["step_id"] == "confirm"
MOCK_NO_UNIQUE_ID = MOCK_SSDP_DATA.copy() MOCK_NO_UNIQUE_ID = dataclasses.replace(MOCK_SSDP_DATA)
del MOCK_NO_UNIQUE_ID[ATTR_UPNP_UDN] MOCK_NO_UNIQUE_ID.upnp = MOCK_NO_UNIQUE_ID.upnp.copy()
del MOCK_NO_UNIQUE_ID.upnp[ATTR_UPNP_UDN]
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID
) )

View File

@ -1,4 +1,5 @@
"""Tests for AVM Fritz!Box config flow.""" """Tests for AVM Fritz!Box config flow."""
import dataclasses
from unittest import mock from unittest import mock
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
@ -6,12 +7,9 @@ from pyfritzhome import LoginError
import pytest import pytest
from requests.exceptions import HTTPError from requests.exceptions import HTTPError
from homeassistant.components import ssdp
from homeassistant.components.fritzbox.const import DOMAIN from homeassistant.components.fritzbox.const import DOMAIN
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_UDN
ATTR_SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
)
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_SSDP, SOURCE_USER from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -26,11 +24,15 @@ from .const import CONF_FAKE_NAME, MOCK_CONFIG
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0] MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
MOCK_SSDP_DATA = { MOCK_SSDP_DATA = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "https://fake_host:12345/test", ssdp_usn="mock_usn",
ATTR_UPNP_FRIENDLY_NAME: CONF_FAKE_NAME, ssdp_st="mock_st",
ATTR_UPNP_UDN: "uuid:only-a-test", ssdp_location="https://fake_host:12345/test",
} upnp={
ATTR_UPNP_FRIENDLY_NAME: CONF_FAKE_NAME,
ATTR_UPNP_UDN: "uuid:only-a-test",
},
)
@pytest.fixture(name="fritz") @pytest.fixture(name="fritz")
@ -201,8 +203,9 @@ async def test_ssdp(hass: HomeAssistant, fritz: Mock):
async def test_ssdp_no_friendly_name(hass: HomeAssistant, fritz: Mock): async def test_ssdp_no_friendly_name(hass: HomeAssistant, fritz: Mock):
"""Test starting a flow from discovery without friendly name.""" """Test starting a flow from discovery without friendly name."""
MOCK_NO_NAME = MOCK_SSDP_DATA.copy() MOCK_NO_NAME = dataclasses.replace(MOCK_SSDP_DATA)
del MOCK_NO_NAME[ATTR_UPNP_FRIENDLY_NAME] MOCK_NO_NAME.upnp = MOCK_NO_NAME.upnp.copy()
del MOCK_NO_NAME.upnp[ATTR_UPNP_FRIENDLY_NAME]
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_NAME DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_NAME
) )
@ -300,8 +303,9 @@ async def test_ssdp_already_in_progress_host(hass: HomeAssistant, fritz: Mock):
assert result["type"] == RESULT_TYPE_FORM assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm" assert result["step_id"] == "confirm"
MOCK_NO_UNIQUE_ID = MOCK_SSDP_DATA.copy() MOCK_NO_UNIQUE_ID = dataclasses.replace(MOCK_SSDP_DATA)
del MOCK_NO_UNIQUE_ID[ATTR_UPNP_UDN] MOCK_NO_UNIQUE_ID.upnp = MOCK_NO_UNIQUE_ID.upnp.copy()
del MOCK_NO_UNIQUE_ID.upnp[ATTR_UPNP_UDN]
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID
) )

View File

@ -177,19 +177,22 @@ async def test_ssdp(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context=context, context=context,
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://192.168.100.1:60957/rootDesc.xml", ssdp_usn="mock_usn",
ssdp.ATTR_SSDP_ST: "upnp:rootdevice", ssdp_st="upnp:rootdevice",
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:InternetGatewayDevice:1", ssdp_location="http://192.168.100.1:60957/rootDesc.xml",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi", upnp={
ssdp.ATTR_UPNP_MANUFACTURER: "Huawei", ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
ssdp.ATTR_UPNP_MANUFACTURER_URL: "http://www.huawei.com/", ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi",
ssdp.ATTR_UPNP_MODEL_NAME: "Huawei router", ssdp.ATTR_UPNP_MANUFACTURER: "Huawei",
ssdp.ATTR_UPNP_MODEL_NUMBER: "12345678", ssdp.ATTR_UPNP_MANUFACTURER_URL: "http://www.huawei.com/",
ssdp.ATTR_UPNP_PRESENTATION_URL: url, ssdp.ATTR_UPNP_MODEL_NAME: "Huawei router",
ssdp.ATTR_UPNP_SERIAL: "00000000", ssdp.ATTR_UPNP_MODEL_NUMBER: "12345678",
ssdp.ATTR_UPNP_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", ssdp.ATTR_UPNP_PRESENTATION_URL: url,
}, ssdp.ATTR_UPNP_SERIAL: "00000000",
ssdp.ATTR_UPNP_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM

View File

@ -3,12 +3,14 @@ from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable from collections.abc import Awaitable
import dataclasses
from typing import Any from typing import Any
from unittest.mock import AsyncMock, Mock, patch from unittest.mock import AsyncMock, Mock, patch
from hyperion import const from hyperion import const
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.hyperion.const import ( from homeassistant.components.hyperion.const import (
CONF_AUTH_ID, CONF_AUTH_ID,
CONF_CREATE_TOKEN, CONF_CREATE_TOKEN,
@ -65,39 +67,41 @@ TEST_REQUEST_TOKEN_FAIL = {
"error": "Token request timeout or denied", "error": "Token request timeout or denied",
} }
TEST_SSDP_SERVICE_INFO = { TEST_SSDP_SERVICE_INFO = ssdp.SsdpServiceInfo(
"ssdp_location": f"http://{TEST_HOST}:{TEST_PORT_UI}/description.xml", ssdp_st="upnp:rootdevice",
"ssdp_st": "upnp:rootdevice", ssdp_location=f"http://{TEST_HOST}:{TEST_PORT_UI}/description.xml",
"deviceType": "urn:schemas-upnp-org:device:Basic:1", ssdp_usn=f"uuid:{TEST_SYSINFO_ID}",
"friendlyName": f"Hyperion ({TEST_HOST})", ssdp_ext="",
"manufacturer": "Hyperion Open Source Ambient Lighting", ssdp_server="Raspbian GNU/Linux 10 (buster)/10 UPnP/1.0 Hyperion/2.0.0-alpha.8",
"manufacturerURL": "https://www.hyperion-project.org", upnp={
"modelDescription": "Hyperion Open Source Ambient Light", "deviceType": "urn:schemas-upnp-org:device:Basic:1",
"modelName": "Hyperion", "friendlyName": f"Hyperion ({TEST_HOST})",
"modelNumber": "2.0.0-alpha.8", "manufacturer": "Hyperion Open Source Ambient Lighting",
"modelURL": "https://www.hyperion-project.org", "manufacturerURL": "https://www.hyperion-project.org",
"serialNumber": f"{TEST_SYSINFO_ID}", "modelDescription": "Hyperion Open Source Ambient Light",
"UDN": f"uuid:{TEST_SYSINFO_ID}", "modelName": "Hyperion",
"ports": { "modelNumber": "2.0.0-alpha.8",
"jsonServer": f"{TEST_PORT}", "modelURL": "https://www.hyperion-project.org",
"sslServer": "8092", "serialNumber": f"{TEST_SYSINFO_ID}",
"protoBuffer": "19445", "UDN": f"uuid:{TEST_SYSINFO_ID}",
"flatBuffer": "19400", "ports": {
"jsonServer": f"{TEST_PORT}",
"sslServer": "8092",
"protoBuffer": "19445",
"flatBuffer": "19400",
},
"presentationURL": "index.html",
"iconList": {
"icon": {
"mimetype": "image/png",
"height": "100",
"width": "100",
"depth": "32",
"url": "img/hyperion/ssdp_icon.png",
}
},
}, },
"presentationURL": "index.html", )
"iconList": {
"icon": {
"mimetype": "image/png",
"height": "100",
"width": "100",
"depth": "32",
"url": "img/hyperion/ssdp_icon.png",
}
},
"ssdp_usn": f"uuid:{TEST_SYSINFO_ID}",
"ssdp_ext": "",
"ssdp_server": "Raspbian GNU/Linux 10 (buster)/10 UPnP/1.0 Hyperion/2.0.0-alpha.8",
}
async def _create_mock_entry(hass: HomeAssistant) -> MockConfigEntry: async def _create_mock_entry(hass: HomeAssistant) -> MockConfigEntry:
@ -639,8 +643,9 @@ async def test_ssdp_missing_serial(hass: HomeAssistant) -> None:
"""Check an SSDP flow where no id is provided.""" """Check an SSDP flow where no id is provided."""
client = create_mock_client() client = create_mock_client()
bad_data = {**TEST_SSDP_SERVICE_INFO} bad_data = dataclasses.replace(TEST_SSDP_SERVICE_INFO)
del bad_data["serialNumber"] bad_data.upnp = bad_data.upnp.copy()
del bad_data.upnp["serialNumber"]
with patch( with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client "homeassistant.components.hyperion.client.HyperionClient", return_value=client
@ -656,8 +661,9 @@ async def test_ssdp_failure_bad_port_json(hass: HomeAssistant) -> None:
"""Check an SSDP flow with bad json port.""" """Check an SSDP flow with bad json port."""
client = create_mock_client() client = create_mock_client()
bad_data: dict[str, Any] = {**TEST_SSDP_SERVICE_INFO} bad_data = dataclasses.replace(TEST_SSDP_SERVICE_INFO)
bad_data["ports"]["jsonServer"] = "not_a_port" bad_data.upnp = bad_data.upnp.copy()
bad_data.upnp["ports"]["jsonServer"] = "not_a_port"
with patch( with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client "homeassistant.components.hyperion.client.HyperionClient", return_value=client
@ -676,8 +682,8 @@ async def test_ssdp_failure_bad_port_ui(hass: HomeAssistant) -> None:
client = create_mock_client() client = create_mock_client()
client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP) client.async_is_auth_required = AsyncMock(return_value=TEST_AUTH_REQUIRED_RESP)
bad_data = {**TEST_SSDP_SERVICE_INFO} bad_data = dataclasses.replace(TEST_SSDP_SERVICE_INFO)
bad_data["ssdp_location"] = f"http://{TEST_HOST}:not_a_port/description.xml" bad_data.ssdp_location = f"http://{TEST_HOST}:not_a_port/description.xml"
with patch( with patch(
"homeassistant.components.hyperion.client.HyperionClient", return_value=client "homeassistant.components.hyperion.client.HyperionClient", return_value=client

View File

@ -29,8 +29,12 @@ MOCK_OPTIONS = {
const.CONF_INTERFACES: ["Home", "VPS0"], const.CONF_INTERFACES: ["Home", "VPS0"],
} }
MOCK_SSDP_DISCOVERY_INFO = { MOCK_SSDP_DISCOVERY_INFO = ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: SSDP_LOCATION, ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_UDN: "uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx", ssdp_st="mock_st",
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_NAME, ssdp_location=SSDP_LOCATION,
} upnp={
ssdp.ATTR_UPNP_UDN: "uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_NAME,
},
)

View File

@ -1,5 +1,6 @@
"""Test Keenetic NDMS2 setup process.""" """Test Keenetic NDMS2 setup process."""
import dataclasses
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from ndms2_client import ConnectionException from ndms2_client import ConnectionException
@ -164,7 +165,7 @@ async def test_connection_error(hass: HomeAssistant, connect_error) -> None:
async def test_ssdp_works(hass: HomeAssistant, connect) -> None: async def test_ssdp_works(hass: HomeAssistant, connect) -> None:
"""Test host already configured and discovered.""" """Test host already configured and discovered."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP}, context={CONF_SOURCE: config_entries.SOURCE_SSDP},
@ -200,7 +201,7 @@ async def test_ssdp_already_configured(hass: HomeAssistant) -> None:
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP}, context={CONF_SOURCE: config_entries.SOURCE_SSDP},
@ -221,7 +222,7 @@ async def test_ssdp_ignored(hass: HomeAssistant) -> None:
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP}, context={CONF_SOURCE: config_entries.SOURCE_SSDP},
@ -245,10 +246,8 @@ async def test_ssdp_update_host(hass: HomeAssistant) -> None:
new_ip = "10.10.10.10" new_ip = "10.10.10.10"
discovery_info = { discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
**MOCK_SSDP_DISCOVERY_INFO, discovery_info.ssdp_location = f"http://{new_ip}/"
ssdp.ATTR_SSDP_LOCATION: f"http://{new_ip}/",
}
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
@ -264,10 +263,9 @@ async def test_ssdp_update_host(hass: HomeAssistant) -> None:
async def test_ssdp_reject_no_udn(hass: HomeAssistant) -> None: async def test_ssdp_reject_no_udn(hass: HomeAssistant) -> None:
"""Discovered device has no UDN.""" """Discovered device has no UDN."""
discovery_info = { discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
**MOCK_SSDP_DISCOVERY_INFO, discovery_info.upnp = {**discovery_info.upnp}
} discovery_info.upnp.pop(ssdp.ATTR_UPNP_UDN)
discovery_info.pop(ssdp.ATTR_UPNP_UDN)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
@ -282,10 +280,9 @@ async def test_ssdp_reject_no_udn(hass: HomeAssistant) -> None:
async def test_ssdp_reject_non_keenetic(hass: HomeAssistant) -> None: async def test_ssdp_reject_non_keenetic(hass: HomeAssistant) -> None:
"""Discovered device does not look like a keenetic router.""" """Discovered device does not look like a keenetic router."""
discovery_info = { discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO)
**MOCK_SSDP_DISCOVERY_INFO, discovery_info.upnp = {**discovery_info.upnp}
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Suspicious device", discovery_info.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME] = "Suspicious device"
}
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN, keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP}, context={CONF_SOURCE: config_entries.SOURCE_SSDP},

View File

@ -211,12 +211,16 @@ async def test_ssdp_already_configured(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: SSDP_URL_SLL, ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20", ssdp_st="mock_st",
ssdp.ATTR_UPNP_PRESENTATION_URL: URL, ssdp_location=SSDP_URL_SLL,
ssdp.ATTR_UPNP_SERIAL: SERIAL, upnp={
}, ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20",
ssdp.ATTR_UPNP_PRESENTATION_URL: URL,
ssdp.ATTR_UPNP_SERIAL: SERIAL,
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
@ -227,12 +231,16 @@ async def test_ssdp(hass, service):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: SSDP_URL, ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20", ssdp_st="mock_st",
ssdp.ATTR_UPNP_PRESENTATION_URL: URL, ssdp_location=SSDP_URL,
ssdp.ATTR_UPNP_SERIAL: SERIAL, upnp={
}, ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20",
ssdp.ATTR_UPNP_PRESENTATION_URL: URL,
ssdp.ATTR_UPNP_SERIAL: SERIAL,
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user" assert result["step_id"] == "user"

View File

@ -7,7 +7,7 @@ from samsungtvws.exceptions import ConnectionFailure, HttpApiError
from websocket import WebSocketException, WebSocketProtocolException from websocket import WebSocketException, WebSocketProtocolException
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.components import dhcp, zeroconf from homeassistant.components import dhcp, ssdp, zeroconf
from homeassistant.components.samsungtv.const import ( from homeassistant.components.samsungtv.const import (
CONF_MANUFACTURER, CONF_MANUFACTURER,
CONF_MODEL, CONF_MODEL,
@ -24,7 +24,6 @@ from homeassistant.components.samsungtv.const import (
TIMEOUT_WEBSOCKET, TIMEOUT_WEBSOCKET,
) )
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import (
ATTR_SSDP_LOCATION,
ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER, ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME, ATTR_UPNP_MODEL_NAME,
@ -63,27 +62,39 @@ MOCK_IMPORT_WSDATA = {
CONF_PORT: 8002, CONF_PORT: 8002,
} }
MOCK_USER_DATA = {CONF_HOST: "fake_host", CONF_NAME: "fake_name"} MOCK_USER_DATA = {CONF_HOST: "fake_host", CONF_NAME: "fake_name"}
MOCK_SSDP_DATA = { MOCK_SSDP_DATA = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "https://fake_host:12345/test", ssdp_usn="mock_usn",
ATTR_UPNP_FRIENDLY_NAME: "[TV] fake_name", ssdp_st="mock_st",
ATTR_UPNP_MANUFACTURER: "Samsung fake_manufacturer", ssdp_location="https://fake_host:12345/test",
ATTR_UPNP_MODEL_NAME: "fake_model", upnp={
ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172de", ATTR_UPNP_FRIENDLY_NAME: "[TV] fake_name",
} ATTR_UPNP_MANUFACTURER: "Samsung fake_manufacturer",
MOCK_SSDP_DATA_NOPREFIX = { ATTR_UPNP_MODEL_NAME: "fake_model",
ATTR_SSDP_LOCATION: "http://fake2_host:12345/test", ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172de",
ATTR_UPNP_FRIENDLY_NAME: "fake2_name", },
ATTR_UPNP_MANUFACTURER: "Samsung fake2_manufacturer", )
ATTR_UPNP_MODEL_NAME: "fake2_model", MOCK_SSDP_DATA_NOPREFIX = ssdp.SsdpServiceInfo(
ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df", ssdp_usn="mock_usn",
} ssdp_st="mock_st",
MOCK_SSDP_DATA_WRONGMODEL = { ssdp_location="http://fake2_host:12345/test",
ATTR_SSDP_LOCATION: "http://fake2_host:12345/test", upnp={
ATTR_UPNP_FRIENDLY_NAME: "fake2_name", ATTR_UPNP_FRIENDLY_NAME: "fake2_name",
ATTR_UPNP_MANUFACTURER: "fake2_manufacturer", ATTR_UPNP_MANUFACTURER: "Samsung fake2_manufacturer",
ATTR_UPNP_MODEL_NAME: "HW-Qfake", ATTR_UPNP_MODEL_NAME: "fake2_model",
ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df", ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df",
} },
)
MOCK_SSDP_DATA_WRONGMODEL = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location="http://fake2_host:12345/test",
upnp={
ATTR_UPNP_FRIENDLY_NAME: "fake2_name",
ATTR_UPNP_MANUFACTURER: "fake2_manufacturer",
ATTR_UPNP_MODEL_NAME: "HW-Qfake",
ATTR_UPNP_UDN: "uuid:0d1cef00-00dc-1000-9c80-4844f7b172df",
},
)
MOCK_DHCP_DATA = dhcp.DhcpServiceInfo( MOCK_DHCP_DATA = dhcp.DhcpServiceInfo(
ip="fake_host", macaddress="aa:bb:cc:dd:ee:ff", hostname="fake_hostname" ip="fake_host", macaddress="aa:bb:cc:dd:ee:ff", hostname="fake_hostname"
) )

View File

@ -130,14 +130,18 @@ async def test_ssdp(hass, aioclient_mock):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://192.168.1.2:5200/Printer.xml", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:Printer:1", ssdp_st="mock_st",
ssdp.ATTR_UPNP_MANUFACTURER: "Samsung Electronics", ssdp_location="http://192.168.1.2:5200/Printer.xml",
ssdp.ATTR_UPNP_PRESENTATION_URL: url, upnp={
ssdp.ATTR_UPNP_SERIAL: "00000000", ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:Printer:1",
ssdp.ATTR_UPNP_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX", ssdp.ATTR_UPNP_MANUFACTURER: "Samsung Electronics",
}, ssdp.ATTR_UPNP_PRESENTATION_URL: url,
ssdp.ATTR_UPNP_SERIAL: "00000000",
ssdp.ATTR_UPNP_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM

View File

@ -308,11 +308,15 @@ async def test_ssdp_discovery_failed(hass, mock_ssdp_no_yamaha, mock_get_source_
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://127.0.0.1/desc.xml", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: "MC20", ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "123456789", ssdp_location="http://127.0.0.1/desc.xml",
}, upnp={
ssdp.ATTR_UPNP_MODEL_NAME: "MC20",
ssdp.ATTR_UPNP_SERIAL: "123456789",
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
@ -326,11 +330,15 @@ async def test_ssdp_discovery_successful_add_device(
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://127.0.0.1/desc.xml", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: "MC20", ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "1234567890", ssdp_location="http://127.0.0.1/desc.xml",
}, upnp={
ssdp.ATTR_UPNP_MODEL_NAME: "MC20",
ssdp.ATTR_UPNP_SERIAL: "1234567890",
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -364,11 +372,15 @@ async def test_ssdp_discovery_existing_device_update(
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://127.0.0.1/desc.xml", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: "MC20", ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "1234567890", ssdp_location="http://127.0.0.1/desc.xml",
}, upnp={
ssdp.ATTR_UPNP_MODEL_NAME: "MC20",
ssdp.ATTR_UPNP_SERIAL: "1234567890",
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"

View File

@ -8,12 +8,8 @@ import zigpy.config
from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.components import usb, zeroconf from homeassistant.components import ssdp, usb, zeroconf
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL
ATTR_SSDP_LOCATION,
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
)
from homeassistant.components.zha import config_flow from homeassistant.components.zha import config_flow
from homeassistant.components.zha.core.const import ( from homeassistant.components.zha.core.const import (
CONF_BAUDRATE, CONF_BAUDRATE,
@ -281,11 +277,15 @@ async def test_discovery_via_usb_deconz_already_discovered(detect_mock, hass):
"""Test usb flow -- deconz discovered.""" """Test usb flow -- deconz discovered."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
"deconz", "deconz",
data={ data=ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: "http://1.2.3.4:80/", ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER_URL: "http://www.dresden-elektronik.de", ssdp_st="mock_st",
ATTR_UPNP_SERIAL: "0000000000000000", ssdp_location="http://1.2.3.4:80/",
}, upnp={
ATTR_UPNP_MANUFACTURER_URL: "http://www.dresden-elektronik.de",
ATTR_UPNP_SERIAL: "0000000000000000",
},
),
context={"source": SOURCE_SSDP}, context={"source": SOURCE_SSDP},
) )
await hass.async_block_till_done() await hass.async_block_till_done()