evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add more SSDP discovery data and constants (#39984)

Browse source
This commit is contained in:
Ville Skyttä 2020-09-14 10:18:09 +03:00 • committed by GitHub
parent 2e1dbe51a4
commit eb0af3752c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 37 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
homeassistant/components/ssdp/__init__.py
View file

@ -17,16 +17,22 @@ SCAN_INTERVAL = timedelta(seconds=60)
# Attributes for accessing info from SSDP response
ATTR_SSDP_LOCATION = "ssdp_location"
ATTR_SSDP_ST = "ssdp_st"
ATTR_SSDP_USN = "ssdp_usn"
ATTR_SSDP_EXT = "ssdp_ext"
ATTR_SSDP_SERVER = "ssdp_server"
# Attributes for accessing info from retrieved UPnP device description
ATTR_UPNP_DEVICE_TYPE = "deviceType"
ATTR_UPNP_FRIENDLY_NAME = "friendlyName"
ATTR_UPNP_MANUFACTURER = "manufacturer"
ATTR_UPNP_MANUFACTURER_URL = "manufacturerURL"
ATTR_UPNP_MODEL_DESCRIPTION = "modelDescription"
ATTR_UPNP_MODEL_NAME = "modelName"
ATTR_UPNP_MODEL_NUMBER = "modelNumber"
ATTR_UPNP_PRESENTATION_URL = "presentationURL"
ATTR_UPNP_MODEL_URL = "modelURL"
ATTR_UPNP_SERIAL = "serialNumber"
ATTR_UPNP_UDN = "UDN"
ATTR_UPNP_UPC = "UPC"
ATTR_UPNP_PRESENTATION_URL = "presentationURL"
_LOGGER = logging.getLogger(__name__)
@ -107,6 +113,9 @@ class Scanner:
"""Process a single entry."""
info = {"st": entry.st}
for key in "usn", "ext", "server":
if key in entry.values:
info[key] = entry.values[key]
if entry.location:
@ -165,5 +174,12 @@ def info_from_entry(entry, device_info):
}
if device_info:
info.update(device_info)
info.pop("st", None)
if "usn" in info:
info[ATTR_SSDP_USN] = info.pop("usn")
if "ext" in info:
info[ATTR_SSDP_EXT] = info.pop("ext")
if "server" in info:
info[ATTR_SSDP_SERVER] = info.pop("server")
return info

26
tests/components/ssdp/test_init.py
View file

@ -15,7 +15,14 @@ async def test_scan_match_st(hass):
scanner = ssdp.Scanner(hass, {"mock-domain": [{"st": "mock-st"}]})
with patch(
"netdisco.ssdp.scan", return_value=[Mock(st="mock-st", location=None)]
"netdisco.ssdp.scan",
return_value=[
Mock(
st="mock-st",
location=None,
values={"usn": "mock-usn", "server": "mock-server", "ext": ""},
)
],
), patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
@ -24,6 +31,13 @@ async def test_scan_match_st(hass):
assert len(mock_init.mock_calls) == 1
assert mock_init.mock_calls[0][1][0] == "mock-domain"
assert mock_init.mock_calls[0][2]["context"] == {"source": "ssdp"}
assert mock_init.mock_calls[0][2]["data"] == {
ssdp.ATTR_SSDP_ST: "mock-st",
ssdp.ATTR_SSDP_LOCATION: None,
ssdp.ATTR_SSDP_USN: "mock-usn",
ssdp.ATTR_SSDP_SERVER: "mock-server",
ssdp.ATTR_SSDP_EXT: "",
}
@pytest.mark.parametrize(
@ -45,7 +59,7 @@ async def test_scan_match_upnp_devicedesc(hass, aioclient_mock, key):
with patch(
"netdisco.ssdp.scan",
return_value=[Mock(st="mock-st", location="http://1.1.1.1")],
return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
), patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
@ -82,7 +96,7 @@ async def test_scan_not_all_present(hass, aioclient_mock):
with patch(
"netdisco.ssdp.scan",
return_value=[Mock(st="mock-st", location="http://1.1.1.1")],
return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
), patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
@ -118,7 +132,7 @@ async def test_scan_not_all_match(hass, aioclient_mock):
with patch(
"netdisco.ssdp.scan",
return_value=[Mock(st="mock-st", location="http://1.1.1.1")],
return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
), patch.object(
hass.config_entries.flow, "async_init", return_value=mock_coro()
) as mock_init:
@ -135,7 +149,7 @@ async def test_scan_description_fetch_fail(hass, aioclient_mock, exc):
with patch(
"netdisco.ssdp.scan",
return_value=[Mock(st="mock-st", location="http://1.1.1.1")],
return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
):
await scanner.async_scan(None)
@ -152,6 +166,6 @@ async def test_scan_description_parse_fail(hass, aioclient_mock):
with patch(
"netdisco.ssdp.scan",
return_value=[Mock(st="mock-st", location="http://1.1.1.1")],
return_value=[Mock(st="mock-st", location="http://1.1.1.1", values={})],
):
await scanner.async_scan(None)