jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 22:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use SsdpServiceInfo for SOURCE_SSDP tests (part 4) (#60339)

Browse Source 
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-11-26 00:13:52 +01:00 committed by GitHub
parent 25f8d4a189
commit 22bdd38583
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 265 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/components/ssdp/__init__.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable from collections.abc import Awaitable, Iterator
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import timedelta from datetime import timedelta
from enum import Enum from enum import Enum
@ -144,7 +144,7 @@ class SsdpServiceInfo(
# Use a property if it is available, fallback to upnp data # Use a property if it is available, fallback to upnp data
if hasattr(self, name): if hasattr(self, name):
return getattr(self, name) return getattr(self, name)
return self.upnp.get(name) return self.upnp[name]
def get(self, name: str, default: Any = None) -> Any: def get(self, name: str, default: Any = None) -> Any:
""" """
@ -164,6 +164,22 @@ class SsdpServiceInfo(
return getattr(self, name) return getattr(self, name)
return self.upnp.get(name, default) return self.upnp.get(name, default)
def __iter__(self) -> Iterator[str]:
"""
Implement iter(self) on upnp data.
Deprecated, and will be removed in version 2022.6.
"""
if not self._warning_logged:
report(
"accessed discovery_info.__iter__() instead of discovery_info.upnp.__iter__(); this will fail in version 2022.6",
exclude_integrations={"ssdp"},
error_if_core=False,
level=logging.DEBUG,
)
self._warning_logged = True
return self.upnp.__iter__()
@bind_hass @bind_hass
async def async_register_callback( async def async_register_callback(

60
tests/components/denonavr/test_config_flow.py
View File

@ -300,12 +300,16 @@ async def test_config_flow_ssdp(hass):
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_UPNP_MANUFACTURER: TEST_MANUFACTURER, ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: TEST_MODEL, ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER, ssdp_location=TEST_SSDP_LOCATION,
ssdp.ATTR_SSDP_LOCATION: TEST_SSDP_LOCATION, upnp={
}, ssdp.ATTR_UPNP_MANUFACTURER: TEST_MANUFACTURER,
ssdp.ATTR_UPNP_MODEL_NAME: TEST_MODEL,
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER,
},
),
) )
assert result["type"] == "form" assert result["type"] == "form"
@ -336,12 +340,16 @@ async def test_config_flow_ssdp_not_denon(hass):
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_UPNP_MANUFACTURER: "NotSupported", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: TEST_MODEL, ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER, ssdp_location=TEST_SSDP_LOCATION,
ssdp.ATTR_SSDP_LOCATION: TEST_SSDP_LOCATION, upnp={
}, ssdp.ATTR_UPNP_MANUFACTURER: "NotSupported",
ssdp.ATTR_UPNP_MODEL_NAME: TEST_MODEL,
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -357,10 +365,14 @@ async def test_config_flow_ssdp_missing_info(hass):
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_UPNP_MANUFACTURER: TEST_MANUFACTURER, ssdp_usn="mock_usn",
ssdp.ATTR_SSDP_LOCATION: TEST_SSDP_LOCATION, ssdp_st="mock_st",
}, ssdp_location=TEST_SSDP_LOCATION,
upnp={
ssdp.ATTR_UPNP_MANUFACTURER: TEST_MANUFACTURER,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -376,12 +388,16 @@ async def test_config_flow_ssdp_ignored_model(hass):
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_UPNP_MANUFACTURER: TEST_MANUFACTURER, ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MODEL_NAME: TEST_IGNORED_MODEL, ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER, ssdp_location=TEST_SSDP_LOCATION,
ssdp.ATTR_SSDP_LOCATION: TEST_SSDP_LOCATION, upnp={
}, ssdp.ATTR_UPNP_MANUFACTURER: TEST_MANUFACTURER,
ssdp.ATTR_UPNP_MODEL_NAME: TEST_IGNORED_MODEL,
ssdp.ATTR_UPNP_SERIAL: TEST_SERIALNUMBER,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"

104
tests/components/hue/test_config_flow.py
View File

@ -333,11 +333,15 @@ async def test_bridge_ssdp(hass, mf_url, aioclient_mock):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://0.0.0.0/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MANUFACTURER_URL: mf_url, ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "1234", ssdp_location="http://0.0.0.0/",
}, upnp={
ssdp.ATTR_UPNP_MANUFACTURER_URL: mf_url,
ssdp.ATTR_UPNP_SERIAL: "1234",
},
),
) )
assert result["type"] == "form" assert result["type"] == "form"
@ -349,7 +353,11 @@ async def test_bridge_ssdp_discover_other_bridge(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ssdp.ATTR_UPNP_MANUFACTURER_URL: "http://www.notphilips.com"}, data=ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
upnp={ssdp.ATTR_UPNP_MANUFACTURER_URL: "http://www.notphilips.com"},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -361,12 +369,16 @@ async def test_bridge_ssdp_emulated_hue(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://0.0.0.0/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Home Assistant Bridge", ssdp_st="mock_st",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_location="http://0.0.0.0/",
ssdp.ATTR_UPNP_SERIAL: "1234", upnp={
}, ssdp.ATTR_UPNP_FRIENDLY_NAME: "Home Assistant Bridge",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
ssdp.ATTR_UPNP_SERIAL: "1234",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -378,10 +390,14 @@ async def test_bridge_ssdp_missing_location(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_SERIAL: "1234", ssdp_st="mock_st",
}, upnp={
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
ssdp.ATTR_UPNP_SERIAL: "1234",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -393,10 +409,14 @@ async def test_bridge_ssdp_missing_serial(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://0.0.0.0/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_st="mock_st",
}, ssdp_location="http://0.0.0.0/",
upnp={
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -408,12 +428,16 @@ async def test_bridge_ssdp_espalexa(hass):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://0.0.0.0/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Espalexa (0.0.0.0)", ssdp_st="mock_st",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_location="http://0.0.0.0/",
ssdp.ATTR_UPNP_SERIAL: "1234", upnp={
}, ssdp.ATTR_UPNP_FRIENDLY_NAME: "Espalexa (0.0.0.0)",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
ssdp.ATTR_UPNP_SERIAL: "1234",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -430,11 +454,15 @@ async def test_bridge_ssdp_already_configured(hass, aioclient_mock):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://0.0.0.0/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "1234", ssdp_location="http://0.0.0.0/",
}, upnp={
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
ssdp.ATTR_UPNP_SERIAL: "1234",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -587,11 +615,15 @@ async def test_ssdp_discovery_update_configuration(hass, aioclient_mock):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
const.DOMAIN, const.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: "http://1.1.1.1/", ssdp_usn="mock_usn",
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0], ssdp_st="mock_st",
ssdp.ATTR_UPNP_SERIAL: "aabbccddeeff", ssdp_location="http://1.1.1.1/",
}, upnp={
ssdp.ATTR_UPNP_MANUFACTURER_URL: config_flow.HUE_MANUFACTURERURL[0],
ssdp.ATTR_UPNP_SERIAL: "aabbccddeeff",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"

123
tests/components/konnected/test_config_flow.py
View File

@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest import pytest
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.components import konnected from homeassistant.components import konnected, ssdp
from homeassistant.components.konnected import config_flow from homeassistant.components.konnected import config_flow
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -113,11 +113,15 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL, ssdp_location="http://1.2.3.4:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL,
},
),
) )
assert result["type"] == "form" assert result["type"] == "form"
@ -134,11 +138,15 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL, ssdp_location="http://1.2.3.4:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -149,9 +157,12 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
}, ssdp_st="mock_st",
ssdp_location="http://1.2.3.4:1234/Device.xml",
upnp={},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -161,11 +172,15 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": "SHOULD_FAIL", ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL, ssdp_location="http://1.2.3.4:1234/Device.xml",
}, upnp={
"manufacturer": "SHOULD_FAIL",
"modelName": config_flow.KONN_MODEL,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -175,11 +190,15 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": "SHOULD_FAIL", ssdp_location="http://1.2.3.4:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": "SHOULD_FAIL",
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -195,11 +214,15 @@ async def test_ssdp(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.2.3.4:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL, ssdp_location="http://1.2.3.4:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
@ -317,11 +340,15 @@ async def test_import_ssdp_host_user_finish(hass, mock_panel):
ssdp_result = await hass.config_entries.flow.async_init( ssdp_result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://0.0.0.0:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL_PRO, ssdp_location="http://0.0.0.0:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL_PRO,
},
),
) )
assert ssdp_result["type"] == "abort" assert ssdp_result["type"] == "abort"
assert ssdp_result["reason"] == "already_in_progress" assert ssdp_result["reason"] == "already_in_progress"
@ -360,11 +387,15 @@ async def test_ssdp_already_configured(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://0.0.0.0:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL_PRO, ssdp_location="http://0.0.0.0:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL_PRO,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"
assert result["reason"] == "already_configured" assert result["reason"] == "already_configured"
@ -436,11 +467,15 @@ async def test_ssdp_host_update(hass, mock_panel):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, config_flow.DOMAIN,
context={"source": config_entries.SOURCE_SSDP}, context={"source": config_entries.SOURCE_SSDP},
data={ data=ssdp.SsdpServiceInfo(
"ssdp_location": "http://1.1.1.1:1234/Device.xml", ssdp_usn="mock_usn",
"manufacturer": config_flow.KONN_MANUFACTURER, ssdp_st="mock_st",
"modelName": config_flow.KONN_MODEL_PRO, ssdp_location="http://1.1.1.1:1234/Device.xml",
}, upnp={
"manufacturer": config_flow.KONN_MANUFACTURER,
"modelName": config_flow.KONN_MODEL_PRO,
},
),
) )
assert result["type"] == "abort" assert result["type"] == "abort"

24
tests/components/upnp/conftest.py
View File

@ -30,17 +30,19 @@ TEST_USN = f"{TEST_UDN}::{TEST_ST}"
TEST_LOCATION = "http://192.168.1.1/desc.xml" TEST_LOCATION = "http://192.168.1.1/desc.xml"
TEST_HOSTNAME = urlparse(TEST_LOCATION).hostname TEST_HOSTNAME = urlparse(TEST_LOCATION).hostname
TEST_FRIENDLY_NAME = "friendly name" TEST_FRIENDLY_NAME = "friendly name"
TEST_DISCOVERY = { TEST_DISCOVERY = ssdp.SsdpServiceInfo(
ssdp.ATTR_SSDP_LOCATION: TEST_LOCATION, ssdp_usn=TEST_USN,
ssdp.ATTR_SSDP_ST: TEST_ST, ssdp_st=TEST_ST,
ssdp.ATTR_SSDP_USN: TEST_USN, ssdp_location=TEST_LOCATION,
ssdp.ATTR_UPNP_UDN: TEST_UDN, upnp={
"usn": TEST_USN, ssdp.ATTR_UPNP_UDN: TEST_UDN,
"location": TEST_LOCATION, "usn": TEST_USN,
"_host": TEST_HOSTNAME, "location": TEST_LOCATION,
"_udn": TEST_UDN, "_host": TEST_HOSTNAME,
"friendlyName": TEST_FRIENDLY_NAME, "_udn": TEST_UDN,
} "friendlyName": TEST_FRIENDLY_NAME,
},
)
class MockDevice: class MockDevice:

14
tests/components/upnp/test_config_flow.py
View File

@ -66,12 +66,14 @@ async def test_flow_ssdp_incomplete_discovery(hass: HomeAssistant):
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: TEST_LOCATION, ssdp_usn=TEST_USN,
ssdp.ATTR_SSDP_ST: TEST_ST, ssdp_st=TEST_ST,
ssdp.ATTR_SSDP_USN: TEST_USN, ssdp_location=TEST_LOCATION,
# ssdp.ATTR_UPNP_UDN: TEST_UDN, # Not provided. upnp={
}, # ssdp.ATTR_UPNP_UDN: TEST_UDN, # Not provided.
},
),
) )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "incomplete_discovery" assert result["reason"] == "incomplete_discovery"

54
tests/components/wilight/__init__.py
View File

@ -2,8 +2,8 @@
from pywilight.const import DOMAIN from pywilight.const import DOMAIN
from homeassistant.components import ssdp
from homeassistant.components.ssdp import ( from homeassistant.components.ssdp import (
ATTR_SSDP_LOCATION,
ATTR_UPNP_MANUFACTURER, ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME, ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_MODEL_NUMBER, ATTR_UPNP_MODEL_NUMBER,
@ -33,28 +33,40 @@ UPNP_MAC_ADDRESS = "5C:CF:7F:8B:CA:56"
UPNP_MANUFACTURER_NOT_WILIGHT = "Test" UPNP_MANUFACTURER_NOT_WILIGHT = "Test"
CONF_COMPONENTS = "components" CONF_COMPONENTS = "components"
MOCK_SSDP_DISCOVERY_INFO_P_B = { MOCK_SSDP_DISCOVERY_INFO_P_B = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: SSDP_LOCATION, ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER: UPNP_MANUFACTURER, ssdp_st="mock_st",
ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B, ssdp_location=SSDP_LOCATION,
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER, upnp={
ATTR_UPNP_SERIAL: UPNP_SERIAL, ATTR_UPNP_MANUFACTURER: UPNP_MANUFACTURER,
} ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B,
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER,
ATTR_UPNP_SERIAL: UPNP_SERIAL,
},
)
MOCK_SSDP_DISCOVERY_INFO_WRONG_MANUFACTURER = { MOCK_SSDP_DISCOVERY_INFO_WRONG_MANUFACTURER = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: SSDP_LOCATION, ssdp_usn="mock_usn",
ATTR_UPNP_MANUFACTURER: UPNP_MANUFACTURER_NOT_WILIGHT, ssdp_st="mock_st",
ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B, ssdp_location=SSDP_LOCATION,
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER, upnp={
ATTR_UPNP_SERIAL: ATTR_UPNP_SERIAL, ATTR_UPNP_MANUFACTURER: UPNP_MANUFACTURER_NOT_WILIGHT,
} ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B,
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER,
ATTR_UPNP_SERIAL: ATTR_UPNP_SERIAL,
},
)
MOCK_SSDP_DISCOVERY_INFO_MISSING_MANUFACTURER = { MOCK_SSDP_DISCOVERY_INFO_MISSING_MANUFACTURER = ssdp.SsdpServiceInfo(
ATTR_SSDP_LOCATION: SSDP_LOCATION, ssdp_usn="mock_usn",
ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B, ssdp_st="mock_st",
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER, ssdp_location=SSDP_LOCATION,
ATTR_UPNP_SERIAL: ATTR_UPNP_SERIAL, upnp={
} ATTR_UPNP_MODEL_NAME: UPNP_MODEL_NAME_P_B,
ATTR_UPNP_MODEL_NUMBER: UPNP_MODEL_NUMBER,
ATTR_UPNP_SERIAL: ATTR_UPNP_SERIAL,
},
)
async def setup_integration( async def setup_integration(

15
tests/components/wilight/test_config_flow.py
View File

@ -1,4 +1,5 @@
"""Test the WiLight config flow.""" """Test the WiLight config flow."""
import dataclasses
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
@ -55,7 +56,7 @@ def mock_dummy_get_components_from_model_wrong():
async def test_show_ssdp_form(hass: HomeAssistant) -> None: async def test_show_ssdp_form(hass: HomeAssistant) -> None:
"""Test that the ssdp confirmation form is served.""" """Test that the ssdp confirmation form is served."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_P_B.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_P_B)
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
) )
@ -71,7 +72,7 @@ async def test_show_ssdp_form(hass: HomeAssistant) -> None:
async def test_ssdp_not_wilight_abort_1(hass: HomeAssistant) -> None: async def test_ssdp_not_wilight_abort_1(hass: HomeAssistant) -> None:
"""Test that the ssdp aborts not_wilight.""" """Test that the ssdp aborts not_wilight."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_WRONG_MANUFACTURER.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_WRONG_MANUFACTURER)
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
) )
@ -83,7 +84,7 @@ async def test_ssdp_not_wilight_abort_1(hass: HomeAssistant) -> None:
async def test_ssdp_not_wilight_abort_2(hass: HomeAssistant) -> None: async def test_ssdp_not_wilight_abort_2(hass: HomeAssistant) -> None:
"""Test that the ssdp aborts not_wilight.""" """Test that the ssdp aborts not_wilight."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_MISSING_MANUFACTURER.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_MISSING_MANUFACTURER)
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
) )
@ -97,7 +98,7 @@ async def test_ssdp_not_wilight_abort_3(
) -> None: ) -> None:
"""Test that the ssdp aborts not_wilight.""" """Test that the ssdp aborts not_wilight."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_P_B.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_P_B)
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
) )
@ -111,7 +112,7 @@ async def test_ssdp_not_supported_abort(
) -> None: ) -> None:
"""Test that the ssdp aborts not_supported.""" """Test that the ssdp aborts not_supported."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_P_B.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_P_B)
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
) )
@ -134,7 +135,7 @@ async def test_ssdp_device_exists_abort(hass: HomeAssistant) -> None:
entry.add_to_hass(hass) entry.add_to_hass(hass)
discovery_info = MOCK_SSDP_DISCOVERY_INFO_P_B.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_P_B)
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},
@ -148,7 +149,7 @@ async def test_ssdp_device_exists_abort(hass: HomeAssistant) -> None:
async def test_full_ssdp_flow_implementation(hass: HomeAssistant) -> None: async def test_full_ssdp_flow_implementation(hass: HomeAssistant) -> None:
"""Test the full SSDP flow from start to finish.""" """Test the full SSDP flow from start to finish."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO_P_B.copy() discovery_info = dataclasses.replace(MOCK_SSDP_DISCOVERY_INFO_P_B)
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
) )