Introduce SsdpServiceInfo for ssdp discovery (#60284)

Co-authored-by: epenet <epenet@users.noreply.github.com>
pull/60315/head
epenet 2021-11-25 02:35:15 +01:00 committed by GitHub
parent 5b199bcc6d
commit 62bf42e421
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 76 additions and 10 deletions

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Awaitable
from dataclasses import dataclass, field
from datetime import timedelta
from enum import Enum
from ipaddress import IPv4Address, IPv6Address
@ -20,9 +21,11 @@ from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, MATCH_ALL
from homeassistant.core import HomeAssistant, callback as core_callback
from homeassistant.data_entry_flow import BaseServiceInfo
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.frame import report
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import async_get_ssdp, bind_hass
@ -85,6 +88,65 @@ SSDP_SOURCE_SSDP_CHANGE_MAPPING: Mapping[SsdpSource, SsdpChange] = {
_LOGGER = logging.getLogger(__name__)
@dataclass
class _HaServiceDescription:
"""Keys added by HA."""
x_homeassistant_matching_domains: set[str] = field(default_factory=set)
@dataclass
class _SsdpServiceDescription:
"""SSDP info with optional keys."""
ssdp_usn: str
ssdp_st: str
ssdp_location: str
ssdp_nt: str | None = None
ssdp_udn: str | None = None
ssdp_ext: str | None = None
ssdp_server: str | None = None
@dataclass
class _UpnpServiceDescription:
"""UPnP info."""
upnp: dict[str, Any]
@dataclass
class SsdpServiceInfo(
_HaServiceDescription,
_SsdpServiceDescription,
_UpnpServiceDescription,
BaseServiceInfo,
):
"""Prepared info from ssdp/upnp entries."""
# Used to prevent log flooding. To be removed in 2022.6
_warning_logged: bool = False
def __getitem__(self, name: str) -> 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['{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
# Use a property if it is available, fallback to upnp data
if hasattr(self, name):
return getattr(self, name)
return self.upnp.get(name)
@bind_hass
async def async_register_callback(
hass: HomeAssistant,

View File

@ -33,16 +33,20 @@ MOCK_UPNP_DEVICE = f"""
MOCK_UPNP_LOCATION = f"http://{MOCK_HOST}:8080/dd.xml"
MOCK_DISCOVER = {
ssdp.ATTR_UPNP_MANUFACTURER: "ARCAM",
ssdp.ATTR_UPNP_MODEL_NAME: " ",
ssdp.ATTR_UPNP_MODEL_NUMBER: "AVR450, AVR750",
ssdp.ATTR_UPNP_FRIENDLY_NAME: f"Arcam media client {MOCK_UUID}",
ssdp.ATTR_UPNP_SERIAL: "12343",
ssdp.ATTR_SSDP_LOCATION: f"http://{MOCK_HOST}:8080/dd.xml",
ssdp.ATTR_UPNP_UDN: MOCK_UDN,
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:MediaRenderer:1",
}
MOCK_DISCOVER = ssdp.SsdpServiceInfo(
ssdp_usn="mock_usn",
ssdp_st="mock_st",
ssdp_location=f"http://{MOCK_HOST}:8080/dd.xml",
upnp={
ssdp.ATTR_UPNP_MANUFACTURER: "ARCAM",
ssdp.ATTR_UPNP_MODEL_NAME: " ",
ssdp.ATTR_UPNP_MODEL_NUMBER: "AVR450, AVR750",
ssdp.ATTR_UPNP_FRIENDLY_NAME: f"Arcam media client {MOCK_UUID}",
ssdp.ATTR_UPNP_SERIAL: "12343",
ssdp.ATTR_UPNP_UDN: MOCK_UDN,
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:MediaRenderer:1",
},
)
@pytest.fixture(name="dummy_client", autouse=True)