jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move SsdpServiceInfo to service_info helpers (#135661)

Browse Source 
* Move SsdpServiceInfo to service_info helpers

* docstring

* Move string constants

* Adjust components
This commit is contained in:
epenet 2025-01-15 14:00:27 +01:00 committed by GitHub
parent 4ccc686295
commit 8c13daf6d9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
23 changed files with 405 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
homeassistant/components/dlna_dms/config_flow.py
View File

@ -14,6 +14,11 @@ from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_DEVICE_ID, CONF_HOST, CONF_URL
from homeassistant.data_entry_flow import AbortFlow
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERVICE_LIST,
SsdpServiceInfo,
)
from .const import CONF_SOURCE_ID, CONFIG_VERSION, DEFAULT_NAME, DOMAIN
from .util import generate_source_id
@ -33,7 +38,7 @@ class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize flow."""
self._discoveries: dict[str, ssdp.SsdpServiceInfo] = {}
self._discoveries: dict[str, SsdpServiceInfo] = {}
self._location: str | None = None
self._usn: str | None = None
self._name: str | None = None
@ -60,14 +65,14 @@ class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
}
discovery_choices = {
host: f"{discovery.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)} ({host})"
host: f"{discovery.upnp.get(ATTR_UPNP_FRIENDLY_NAME)} ({host})"
for host, discovery in self._discoveries.items()
}
data_schema = vol.Schema({vol.Optional(CONF_HOST): vol.In(discovery_choices)})
return self.async_show_form(step_id="user", data_schema=data_schema)
async def async_step_ssdp(
self, discovery_info: ssdp.SsdpServiceInfo
self, discovery_info: SsdpServiceInfo
) -> ConfigFlowResult:
"""Handle a flow initialized by SSDP discovery."""
if LOGGER.isEnabledFor(logging.DEBUG):
@ -81,7 +86,7 @@ class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
# Abort if the device doesn't support all services required for a DmsDevice.
# Use the discovery_info instead of DmsDevice.is_profile_device to avoid
# contacting the device again.
discovery_service_list = discovery_info.upnp.get(ssdp.ATTR_UPNP_SERVICE_LIST)
discovery_service_list = discovery_info.upnp.get(ATTR_UPNP_SERVICE_LIST)
if not discovery_service_list:
return self.async_abort(reason="not_dms")
@ -135,7 +140,7 @@ class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=self._name, data=data)
async def _async_parse_discovery(
self, discovery_info: ssdp.SsdpServiceInfo, raise_on_progress: bool = True
self, discovery_info: SsdpServiceInfo, raise_on_progress: bool = True
) -> None:
"""Get required details from an SSDP discovery.
@ -162,15 +167,15 @@ class DlnaDmsFlowHandler(ConfigFlow, domain=DOMAIN):
)
self._name = (
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
discovery_info.upnp.get(ATTR_UPNP_FRIENDLY_NAME)
or urlparse(self._location).hostname
or DEFAULT_NAME
)
async def _async_get_discoveries(self) -> list[ssdp.SsdpServiceInfo]:
async def _async_get_discoveries(self) -> list[SsdpServiceInfo]:
"""Get list of unconfigured DLNA devices discovered by SSDP."""
# Get all compatible devices from ssdp's cache
discoveries: list[ssdp.SsdpServiceInfo] = []
discoveries: list[SsdpServiceInfo] = []
for udn_st in DmsDevice.DEVICE_TYPES:
st_discoveries = await ssdp.async_get_discovery_info_by_st(
self.hass, udn_st

10
homeassistant/components/lametric/config_flow.py
View File

@ -23,11 +23,6 @@ from demetriek import (
import voluptuous as vol
from yarl import URL
from homeassistant.components.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_DEVICE, CONF_HOST, CONF_MAC
from homeassistant.data_entry_flow import AbortFlow
@ -44,6 +39,11 @@ from homeassistant.helpers.selector import (
TextSelectorType,
)
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from homeassistant.util.network import is_link_local
from .const import DOMAIN, LOGGER

6
homeassistant/components/openhome/config_flow.py
View File

@ -3,13 +3,13 @@
import logging
from typing import Any
from homeassistant.components.ssdp import (
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
SsdpServiceInfo,
)
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_NAME
from .const import DOMAIN

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

@ -4,7 +4,6 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine, Mapping
from dataclasses import dataclass, field
from datetime import timedelta
from enum import Enum
from functools import partial
@ -44,13 +43,36 @@ from homeassistant.const import (
__version__ as current_version,
)
from homeassistant.core import Event, HassJob, HomeAssistant, callback as core_callback
from homeassistant.data_entry_flow import BaseServiceInfo
from homeassistant.helpers import config_validation as cv, discovery_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.deprecation import (
DeprecatedConstant,
all_with_deprecated_constants,
check_if_deprecated_constant,
dir_with_deprecated_constants,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.instance_id import async_get as async_get_instance_id
from homeassistant.helpers.network import NoURLAvailableError, get_url
from homeassistant.helpers.service_info.ssdp import (
ATTR_NT as _ATTR_NT,
ATTR_ST as _ATTR_ST,
ATTR_UPNP_DEVICE_TYPE as _ATTR_UPNP_DEVICE_TYPE,
ATTR_UPNP_FRIENDLY_NAME as _ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER as _ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MANUFACTURER_URL as _ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_MODEL_DESCRIPTION as _ATTR_UPNP_MODEL_DESCRIPTION,
ATTR_UPNP_MODEL_NAME as _ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_MODEL_NUMBER as _ATTR_UPNP_MODEL_NUMBER,
ATTR_UPNP_MODEL_URL as _ATTR_UPNP_MODEL_URL,
ATTR_UPNP_PRESENTATION_URL as _ATTR_UPNP_PRESENTATION_URL,
ATTR_UPNP_SERIAL as _ATTR_UPNP_SERIAL,
ATTR_UPNP_SERVICE_LIST as _ATTR_UPNP_SERVICE_LIST,
ATTR_UPNP_UDN as _ATTR_UPNP_UDN,
ATTR_UPNP_UPC as _ATTR_UPNP_UPC,
SsdpServiceInfo as _SsdpServiceInfo,
)
from homeassistant.helpers.system_info import async_get_system_info
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import async_get_ssdp, bind_hass
@ -77,30 +99,90 @@ ATTR_SSDP_SERVER = "ssdp_server"
ATTR_SSDP_BOOTID = "BOOTID.UPNP.ORG"
ATTR_SSDP_NEXTBOOTID = "NEXTBOOTID.UPNP.ORG"
# Attributes for accessing info from retrieved UPnP device description
ATTR_ST = "st"
ATTR_NT = "nt"
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_MODEL_URL = "modelURL"
ATTR_UPNP_SERIAL = "serialNumber"
ATTR_UPNP_SERVICE_LIST = "serviceList"
ATTR_UPNP_UDN = "UDN"
ATTR_UPNP_UPC = "UPC"
ATTR_UPNP_PRESENTATION_URL = "presentationURL"
_DEPRECATED_ATTR_ST = DeprecatedConstant(
_ATTR_ST,
"homeassistant.helpers.service_info.ssdp.ATTR_ST",
"2026.2",
)
_DEPRECATED_ATTR_NT = DeprecatedConstant(
_ATTR_NT,
"homeassistant.helpers.service_info.ssdp.ATTR_NT",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_DEVICE_TYPE = DeprecatedConstant(
_ATTR_UPNP_DEVICE_TYPE,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_DEVICE_TYPE",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_FRIENDLY_NAME = DeprecatedConstant(
_ATTR_UPNP_FRIENDLY_NAME,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_FRIENDLY_NAME",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MANUFACTURER = DeprecatedConstant(
_ATTR_UPNP_MANUFACTURER,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MANUFACTURER_URL = DeprecatedConstant(
_ATTR_UPNP_MANUFACTURER_URL,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER_URL",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MODEL_DESCRIPTION = DeprecatedConstant(
_ATTR_UPNP_MODEL_DESCRIPTION,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_DESCRIPTION",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MODEL_NAME = DeprecatedConstant(
_ATTR_UPNP_MODEL_NAME,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NAME",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MODEL_NUMBER = DeprecatedConstant(
_ATTR_UPNP_MODEL_NUMBER,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NUMBER",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_MODEL_URL = DeprecatedConstant(
_ATTR_UPNP_MODEL_URL,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_URL",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_SERIAL = DeprecatedConstant(
_ATTR_UPNP_SERIAL,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERIAL",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_SERVICE_LIST = DeprecatedConstant(
_ATTR_UPNP_SERVICE_LIST,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERVICE_LIST",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_UDN = DeprecatedConstant(
_ATTR_UPNP_UDN,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UDN",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_UPC = DeprecatedConstant(
_ATTR_UPNP_UPC,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UPC",
"2026.2",
)
_DEPRECATED_ATTR_UPNP_PRESENTATION_URL = DeprecatedConstant(
_ATTR_UPNP_PRESENTATION_URL,
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_PRESENTATION_URL",
"2026.2",
)
# Attributes for accessing info added by Home Assistant
ATTR_HA_MATCHING_DOMAINS = "x_homeassistant_matching_domains"
PRIMARY_MATCH_KEYS = [
ATTR_UPNP_MANUFACTURER,
ATTR_ST,
ATTR_UPNP_DEVICE_TYPE,
ATTR_NT,
ATTR_UPNP_MANUFACTURER_URL,
_ATTR_UPNP_MANUFACTURER,
_ATTR_ST,
_ATTR_UPNP_DEVICE_TYPE,
_ATTR_NT,
_ATTR_UPNP_MANUFACTURER_URL,
]
_LOGGER = logging.getLogger(__name__)
@ -108,27 +190,16 @@ _LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@dataclass(slots=True)
class SsdpServiceInfo(BaseServiceInfo):
"""Prepared info from ssdp/upnp entries."""
ssdp_usn: str
ssdp_st: str
upnp: Mapping[str, Any]
ssdp_location: str | None = None
ssdp_nt: str | None = None
ssdp_udn: str | None = None
ssdp_ext: str | None = None
ssdp_server: str | None = None
ssdp_headers: Mapping[str, Any] = field(default_factory=dict)
ssdp_all_locations: set[str] = field(default_factory=set)
x_homeassistant_matching_domains: set[str] = field(default_factory=set)
_DEPRECATED_SsdpServiceInfo = DeprecatedConstant(
_SsdpServiceInfo,
"homeassistant.helpers.service_info.ssdp.SsdpServiceInfo",
"2026.2",
)
SsdpChange = Enum("SsdpChange", "ALIVE BYEBYE UPDATE")
type SsdpHassJobCallback = HassJob[
[SsdpServiceInfo, SsdpChange], Coroutine[Any, Any, None] | None
[_SsdpServiceInfo, SsdpChange], Coroutine[Any, Any, None] | None
]
SSDP_SOURCE_SSDP_CHANGE_MAPPING: Mapping[SsdpSource, SsdpChange] = {
@ -148,7 +219,9 @@ def _format_err(name: str, *args: Any) -> str:
@bind_hass
async def async_register_callback(
hass: HomeAssistant,
callback: Callable[[SsdpServiceInfo, SsdpChange], Coroutine[Any, Any, None] | None],
callback: Callable[
[_SsdpServiceInfo, SsdpChange], Coroutine[Any, Any, None] | None
],
match_dict: dict[str, str] | None = None,
) -> Callable[[], None]:
"""Register to receive a callback on ssdp broadcast.
@ -169,7 +242,7 @@ async def async_register_callback(
@bind_hass
async def async_get_discovery_info_by_udn_st(
hass: HomeAssistant, udn: str, st: str
) -> SsdpServiceInfo | None:
) -> _SsdpServiceInfo | None:
"""Fetch the discovery info cache."""
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
return await scanner.async_get_discovery_info_by_udn_st(udn, st)
@ -178,7 +251,7 @@ async def async_get_discovery_info_by_udn_st(
@bind_hass
async def async_get_discovery_info_by_st(
hass: HomeAssistant, st: str
) -> list[SsdpServiceInfo]:
) -> list[_SsdpServiceInfo]:
"""Fetch all the entries matching the st."""
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
return await scanner.async_get_discovery_info_by_st(st)
@ -187,7 +260,7 @@ async def async_get_discovery_info_by_st(
@bind_hass
async def async_get_discovery_info_by_udn(
hass: HomeAssistant, udn: str
) -> list[SsdpServiceInfo]:
) -> list[_SsdpServiceInfo]:
"""Fetch all the entries matching the udn."""
scanner: Scanner = hass.data[DOMAIN][SSDP_SCANNER]
return await scanner.async_get_discovery_info_by_udn(udn)
@ -227,7 +300,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
def _async_process_callbacks(
hass: HomeAssistant,
callbacks: list[SsdpHassJobCallback],
discovery_info: SsdpServiceInfo,
discovery_info: _SsdpServiceInfo,
ssdp_change: SsdpChange,
) -> None:
for callback in callbacks:
@ -562,11 +635,11 @@ class Scanner:
)
def _async_dismiss_discoveries(
self, byebye_discovery_info: SsdpServiceInfo
self, byebye_discovery_info: _SsdpServiceInfo
) -> None:
"""Dismiss all discoveries for the given address."""
for flow in self.hass.config_entries.flow.async_progress_by_init_data_type(
SsdpServiceInfo,
_SsdpServiceInfo,
lambda service_info: bool(
service_info.ssdp_st == byebye_discovery_info.ssdp_st
and service_info.ssdp_location == byebye_discovery_info.ssdp_location
@ -589,7 +662,7 @@ class Scanner:
async def _async_headers_to_discovery_info(
self, ssdp_device: SsdpDevice, headers: CaseInsensitiveDict
) -> SsdpServiceInfo:
) -> _SsdpServiceInfo:
"""Combine the headers and description into discovery_info.
Building this is a bit expensive so we only do it on demand.
@ -602,7 +675,7 @@ class Scanner:
async def async_get_discovery_info_by_udn_st(
self, udn: str, st: str
) -> SsdpServiceInfo | None:
) -> _SsdpServiceInfo | None:
"""Return discovery_info for a udn and st."""
for ssdp_device in self._ssdp_devices:
if ssdp_device.udn == udn:
@ -612,7 +685,7 @@ class Scanner:
)
return None
async def async_get_discovery_info_by_st(self, st: str) -> list[SsdpServiceInfo]:
async def async_get_discovery_info_by_st(self, st: str) -> list[_SsdpServiceInfo]:
"""Return matching discovery_infos for a st."""
return [
await self._async_headers_to_discovery_info(ssdp_device, headers)
@ -620,7 +693,7 @@ class Scanner:
if (headers := ssdp_device.combined_headers(st))
]
async def async_get_discovery_info_by_udn(self, udn: str) -> list[SsdpServiceInfo]:
async def async_get_discovery_info_by_udn(self, udn: str) -> list[_SsdpServiceInfo]:
"""Return matching discovery_infos for a udn."""
return [
await self._async_headers_to_discovery_info(ssdp_device, headers)
@ -665,7 +738,7 @@ def discovery_info_from_headers_and_description(
ssdp_device: SsdpDevice,
combined_headers: CaseInsensitiveDict,
info_desc: Mapping[str, Any],
) -> SsdpServiceInfo:
) -> _SsdpServiceInfo:
"""Convert headers and description to discovery_info."""
ssdp_usn = combined_headers["usn"]
ssdp_st = combined_headers.get_lower("st")
@ -681,11 +754,11 @@ def discovery_info_from_headers_and_description(
ssdp_st = combined_headers["nt"]
# Ensure UPnP "udn" is set
if ATTR_UPNP_UDN not in upnp_info:
if _ATTR_UPNP_UDN not in upnp_info:
if udn := _udn_from_usn(ssdp_usn):
upnp_info[ATTR_UPNP_UDN] = udn
upnp_info[_ATTR_UPNP_UDN] = udn
return SsdpServiceInfo(
return _SsdpServiceInfo(
ssdp_usn=ssdp_usn,
ssdp_st=ssdp_st,
ssdp_ext=combined_headers.get_lower("ext"),
@ -887,3 +960,11 @@ class Server:
"""Stop UPnP/SSDP servers."""
for server in self._upnp_servers:
await server.async_stop()
# These can be removed if no deprecated constant are in this module anymore
__getattr__ = partial(check_if_deprecated_constant, module_globals=globals())
__dir__ = partial(
dir_with_deprecated_constants, module_globals_keys=[*globals().keys()]
)
__all__ = all_with_deprecated_constants(globals())

23
homeassistant/components/upnp/config_flow.py
View File

@ -9,7 +9,6 @@ from urllib.parse import urlparse
import voluptuous as vol
from homeassistant.components import ssdp
from homeassistant.components.ssdp import SsdpServiceInfo
from homeassistant.config_entries import (
SOURCE_IGNORE,
ConfigEntry,
@ -18,6 +17,12 @@ from homeassistant.config_entries import (
OptionsFlow,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_DEVICE_TYPE,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MODEL_NAME,
SsdpServiceInfo,
)
from .const import (
CONFIG_ENTRY_FORCE_POLL,
@ -37,17 +42,17 @@ from .const import (
from .device import async_get_mac_address_from_host, get_preferred_location
def _friendly_name_from_discovery(discovery_info: ssdp.SsdpServiceInfo) -> str:
def _friendly_name_from_discovery(discovery_info: SsdpServiceInfo) -> str:
"""Extract user-friendly name from discovery."""
return cast(
str,
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
or discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME)
discovery_info.upnp.get(ATTR_UPNP_FRIENDLY_NAME)
or discovery_info.upnp.get(ATTR_UPNP_MODEL_NAME)
or discovery_info.ssdp_headers.get("_host", ""),
)
def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool:
def _is_complete_discovery(discovery_info: SsdpServiceInfo) -> bool:
"""Test if discovery is complete and usable."""
return bool(
discovery_info.ssdp_udn
@ -59,7 +64,7 @@ def _is_complete_discovery(discovery_info: ssdp.SsdpServiceInfo) -> bool:
async def _async_discovered_igd_devices(
hass: HomeAssistant,
) -> list[ssdp.SsdpServiceInfo]:
) -> list[SsdpServiceInfo]:
"""Discovery IGD devices."""
return await ssdp.async_get_discovery_info_by_st(
hass, ST_IGD_V1
@ -76,10 +81,10 @@ async def _async_mac_address_from_discovery(
return await async_get_mac_address_from_host(hass, host)
def _is_igd_device(discovery_info: ssdp.SsdpServiceInfo) -> bool:
def _is_igd_device(discovery_info: SsdpServiceInfo) -> bool:
"""Test if discovery is a complete IGD device."""
root_device_info = discovery_info.upnp
return root_device_info.get(ssdp.ATTR_UPNP_DEVICE_TYPE) in {ST_IGD_V1, ST_IGD_V2}
return root_device_info.get(ATTR_UPNP_DEVICE_TYPE) in {ST_IGD_V1, ST_IGD_V2}
class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
@ -167,7 +172,7 @@ class UpnpFlowHandler(ConfigFlow, domain=DOMAIN):
)
async def async_step_ssdp(
self, discovery_info: ssdp.SsdpServiceInfo
self, discovery_info: SsdpServiceInfo
) -> ConfigFlowResult:
"""Handle a discovered UPnP/IGD device.

14
homeassistant/components/yamaha_musiccast/config_flow.py
View File

@ -10,10 +10,14 @@ from aiohttp import ClientConnectorError
from aiomusiccast import MusicCastConnectionException, MusicCastDevice
import voluptuous as vol
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from . import get_upnp_desc
from .const import CONF_SERIAL, CONF_UPNP_DESC, DOMAIN
@ -81,7 +85,7 @@ class MusicCastFlowHandler(ConfigFlow, domain=DOMAIN):
)
async def async_step_ssdp(
self, discovery_info: ssdp.SsdpServiceInfo
self, discovery_info: SsdpServiceInfo
) -> ConfigFlowResult:
"""Handle ssdp discoveries."""
if not await MusicCastDevice.check_yamaha_ssdp(
@ -89,7 +93,7 @@ class MusicCastFlowHandler(ConfigFlow, domain=DOMAIN):
):
return self.async_abort(reason="yxc_control_url_missing")
self.serial_number = discovery_info.upnp[ssdp.ATTR_UPNP_SERIAL]
self.serial_number = discovery_info.upnp[ATTR_UPNP_SERIAL]
self.upnp_description = discovery_info.ssdp_location
# ssdp_location and hostname have been checked in check_yamaha_ssdp so it is safe to ignore type assignment
@ -105,9 +109,7 @@ class MusicCastFlowHandler(ConfigFlow, domain=DOMAIN):
self.context.update(
{
"title_placeholders": {
"name": discovery_info.upnp.get(
ssdp.ATTR_UPNP_FRIENDLY_NAME, self.host
)
"name": discovery_info.upnp.get(ATTR_UPNP_FRIENDLY_NAME, self.host)
}
}
)

2
homeassistant/config_entries.py
View File

@ -87,11 +87,11 @@ from .util.enum import try_parse_enum
if TYPE_CHECKING:
from .components.bluetooth import BluetoothServiceInfoBleak
from .components.ssdp import SsdpServiceInfo
from .components.usb import UsbServiceInfo
from .helpers.service_info.dhcp import DhcpServiceInfo
from .helpers.service_info.hassio import HassioServiceInfo
from .helpers.service_info.mqtt import MqttServiceInfo
from .helpers.service_info.ssdp import SsdpServiceInfo
from .helpers.service_info.zeroconf import ZeroconfServiceInfo

2
homeassistant/helpers/config_entry_flow.py
View File

@ -16,10 +16,10 @@ if TYPE_CHECKING:
import asyncio
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.ssdp import SsdpServiceInfo
from .service_info.dhcp import DhcpServiceInfo
from .service_info.mqtt import MqttServiceInfo
from .service_info.ssdp import SsdpServiceInfo
from .service_info.zeroconf import ZeroconfServiceInfo
type DiscoveryFunctionType[_R] = Callable[[HomeAssistant], _R]

41
homeassistant/helpers/service_info/ssdp.py Normal file
View File

@ -0,0 +1,41 @@
"""SSDP discovery data."""
from collections.abc import Mapping
from dataclasses import dataclass, field
from typing import Any, Final
from homeassistant.data_entry_flow import BaseServiceInfo
# Attributes for accessing info from retrieved UPnP device description
ATTR_ST: Final = "st"
ATTR_NT: Final = "nt"
ATTR_UPNP_DEVICE_TYPE: Final = "deviceType"
ATTR_UPNP_FRIENDLY_NAME: Final = "friendlyName"
ATTR_UPNP_MANUFACTURER: Final = "manufacturer"
ATTR_UPNP_MANUFACTURER_URL: Final = "manufacturerURL"
ATTR_UPNP_MODEL_DESCRIPTION: Final = "modelDescription"
ATTR_UPNP_MODEL_NAME: Final = "modelName"
ATTR_UPNP_MODEL_NUMBER: Final = "modelNumber"
ATTR_UPNP_MODEL_URL: Final = "modelURL"
ATTR_UPNP_SERIAL: Final = "serialNumber"
ATTR_UPNP_SERVICE_LIST: Final = "serviceList"
ATTR_UPNP_UDN: Final = "UDN"
ATTR_UPNP_UPC: Final = "UPC"
ATTR_UPNP_PRESENTATION_URL: Final = "presentationURL"
@dataclass(slots=True)
class SsdpServiceInfo(BaseServiceInfo):
"""Prepared info from ssdp/upnp entries."""
ssdp_usn: str
ssdp_st: str
upnp: Mapping[str, Any]
ssdp_location: str | None = None
ssdp_nt: str | None = None
ssdp_udn: str | None = None
ssdp_ext: str | None = None
ssdp_server: str | None = None
ssdp_headers: Mapping[str, Any] = field(default_factory=dict)
ssdp_all_locations: set[str] = field(default_factory=set)
x_homeassistant_matching_domains: set[str] = field(default_factory=set)

5
tests/components/deconz/test_config_flow.py
View File

@ -20,12 +20,15 @@ from homeassistant.components.deconz.const import (
DOMAIN as DECONZ_DOMAIN,
HASSIO_CONFIGURATION_URL,
)
from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL
from homeassistant.config_entries import SOURCE_HASSIO, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT, CONTENT_TYPE_JSON
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
)
from .conftest import API_KEY, BRIDGE_ID

10
tests/components/deconz/test_hub.py
View File

@ -9,15 +9,15 @@ from syrupy import SnapshotAssertion
from homeassistant.components import ssdp
from homeassistant.components.deconz.config_flow import DECONZ_MANUFACTURERURL
from homeassistant.components.deconz.const import DOMAIN as DECONZ_DOMAIN
from homeassistant.components.ssdp import (
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
ATTR_UPNP_UDN,
)
from homeassistant.config_entries import SOURCE_SSDP
from homeassistant.const import STATE_OFF, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
ATTR_UPNP_UDN,
)
from .conftest import BRIDGE_ID

2
tests/components/directv/test_config_flow.py
View File

@ -6,11 +6,11 @@ from unittest.mock import patch
from aiohttp import ClientError as HTTPClientError
from homeassistant.components.directv.const import CONF_RECEIVER_ID, DOMAIN
from homeassistant.components.ssdp import ATTR_UPNP_SERIAL
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.ssdp import ATTR_UPNP_SERIAL
from . import (
HOST,

5
tests/components/fritz/const.py
View File

@ -2,7 +2,6 @@
from homeassistant.components import ssdp
from homeassistant.components.fritz.const import DOMAIN
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_UDN
from homeassistant.const import (
CONF_DEVICES,
CONF_HOST,
@ -11,6 +10,10 @@ from homeassistant.const import (
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
)
ATTR_HOST = "host"
ATTR_NEW_SERIAL_NUMBER = "NewSerialNumber"

5
tests/components/fritzbox/test_config_flow.py
View File

@ -11,11 +11,14 @@ from requests.exceptions import HTTPError
from homeassistant.components import ssdp
from homeassistant.components.fritzbox.const import DOMAIN
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_UDN
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_DEVICES, CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
)
from .const import CONF_FAKE_NAME, MOCK_CONFIG

5
tests/components/kaleidescape/__init__.py
View File

@ -1,7 +1,10 @@
"""Tests for Kaleidescape integration."""
from homeassistant.components import ssdp
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_SERIAL
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
)
MOCK_HOST = "127.0.0.1"
MOCK_SERIAL = "123456"

10
tests/components/lametric/test_config_flow.py
View File

@ -14,17 +14,17 @@ from demetriek import (
import pytest
from homeassistant.components.lametric.const import DOMAIN
from homeassistant.components.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_DEVICE, CONF_HOST, CONF_MAC
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker

13
tests/components/openhome/test_config_flow.py
View File

@ -1,12 +1,15 @@
"""Tests for the Openhome config flow module."""
from homeassistant.components import ssdp
from homeassistant.components.openhome.const import DOMAIN
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_UDN
from homeassistant.config_entries import SOURCE_SSDP
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_UDN,
SsdpServiceInfo,
)
from tests.common import MockConfigEntry
@ -14,7 +17,7 @@ MOCK_UDN = "uuid:4c494e4e-1234-ab12-abcd-01234567819f"
MOCK_FRIENDLY_NAME = "Test Client"
MOCK_SSDP_LOCATION = "http://device:12345/description.xml"
MOCK_DISCOVER = ssdp.SsdpServiceInfo(
MOCK_DISCOVER = SsdpServiceInfo(
ssdp_usn="usn",
ssdp_st="st",
ssdp_location=MOCK_SSDP_LOCATION,
@ -60,7 +63,7 @@ async def test_device_exists(hass: HomeAssistant) -> None:
async def test_missing_udn(hass: HomeAssistant) -> None:
"""Test a ssdp import where discovery is missing udn."""
broken_discovery = ssdp.SsdpServiceInfo(
broken_discovery = SsdpServiceInfo(
ssdp_usn="usn",
ssdp_st="st",
ssdp_location=MOCK_SSDP_LOCATION,
@ -79,7 +82,7 @@ async def test_missing_udn(hass: HomeAssistant) -> None:
async def test_missing_ssdp_location(hass: HomeAssistant) -> None:
"""Test a ssdp import where discovery is missing udn."""
broken_discovery = ssdp.SsdpServiceInfo(
broken_discovery = SsdpServiceInfo(
ssdp_usn="usn",
ssdp_st="st",
ssdp_location="",

5
tests/components/roku/__init__.py
View File

@ -3,7 +3,10 @@
from ipaddress import ip_address
from homeassistant.components import ssdp, zeroconf
from homeassistant.components.ssdp import ATTR_UPNP_FRIENDLY_NAME, ATTR_UPNP_SERIAL
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_SERIAL,
)
NAME = "Roku 3"
NAME_ROKUTV = '58" Onn Roku TV'

12
tests/components/samsungtv/const.py
View File

@ -8,12 +8,6 @@ from homeassistant.components.samsungtv.const import (
METHOD_LEGACY,
METHOD_WEBSOCKET,
)
from homeassistant.components.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_UDN,
)
from homeassistant.const import (
CONF_HOST,
CONF_IP_ADDRESS,
@ -24,6 +18,12 @@ from homeassistant.const import (
CONF_PORT,
CONF_TOKEN,
)
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_UDN,
)
MOCK_CONFIG = {
CONF_HOST: "fake_host",

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

@ -33,13 +33,6 @@ from homeassistant.components.samsungtv.const import (
TIMEOUT_REQUEST,
TIMEOUT_WEBSOCKET,
)
from homeassistant.components.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_UDN,
SsdpServiceInfo,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_HOST,
@ -54,6 +47,13 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import BaseServiceInfo, FlowResultType
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_UDN,
SsdpServiceInfo,
)
from homeassistant.setup import async_setup_component
from .const import (

122
tests/components/ssdp/test_init.py
View File

@ -2,6 +2,7 @@
from datetime import datetime
from ipaddress import IPv4Address
from typing import Any
from unittest.mock import ANY, AsyncMock, patch
from async_upnp_client.server import UpnpServer
@ -19,6 +20,24 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.discovery_flow import DiscoveryKey
from homeassistant.helpers.service_info.ssdp import (
ATTR_NT,
ATTR_ST,
ATTR_UPNP_DEVICE_TYPE,
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_MODEL_DESCRIPTION,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_MODEL_NUMBER,
ATTR_UPNP_MODEL_URL,
ATTR_UPNP_PRESENTATION_URL,
ATTR_UPNP_SERIAL,
ATTR_UPNP_SERVICE_LIST,
ATTR_UPNP_UDN,
ATTR_UPNP_UPC,
SsdpServiceInfo,
)
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
@ -26,6 +45,7 @@ from tests.common import (
MockConfigEntry,
MockModule,
async_fire_time_changed,
import_and_test_deprecated_constant,
mock_integration,
)
from tests.test_util.aiohttp import AiohttpClientMocker
@ -1094,3 +1114,105 @@ async def test_ssdp_rediscover_no_match(
await hass.async_block_till_done()
assert len(mock_flow_init.mock_calls) == 1
@pytest.mark.parametrize(
("constant_name", "replacement_name", "replacement"),
[
(
"SsdpServiceInfo",
"homeassistant.helpers.service_info.ssdp.SsdpServiceInfo",
SsdpServiceInfo,
),
(
"ATTR_ST",
"homeassistant.helpers.service_info.ssdp.ATTR_ST",
ATTR_ST,
),
(
"ATTR_NT",
"homeassistant.helpers.service_info.ssdp.ATTR_NT",
ATTR_NT,
),
(
"ATTR_UPNP_DEVICE_TYPE",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_DEVICE_TYPE",
ATTR_UPNP_DEVICE_TYPE,
),
(
"ATTR_UPNP_FRIENDLY_NAME",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_FRIENDLY_NAME",
ATTR_UPNP_FRIENDLY_NAME,
),
(
"ATTR_UPNP_MANUFACTURER",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER",
ATTR_UPNP_MANUFACTURER,
),
(
"ATTR_UPNP_MANUFACTURER_URL",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MANUFACTURER_URL",
ATTR_UPNP_MANUFACTURER_URL,
),
(
"ATTR_UPNP_MODEL_DESCRIPTION",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_DESCRIPTION",
ATTR_UPNP_MODEL_DESCRIPTION,
),
(
"ATTR_UPNP_MODEL_NAME",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NAME",
ATTR_UPNP_MODEL_NAME,
),
(
"ATTR_UPNP_MODEL_NUMBER",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_NUMBER",
ATTR_UPNP_MODEL_NUMBER,
),
(
"ATTR_UPNP_MODEL_URL",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_MODEL_URL",
ATTR_UPNP_MODEL_URL,
),
(
"ATTR_UPNP_SERIAL",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERIAL",
ATTR_UPNP_SERIAL,
),
(
"ATTR_UPNP_SERVICE_LIST",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_SERVICE_LIST",
ATTR_UPNP_SERVICE_LIST,
),
(
"ATTR_UPNP_UDN",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UDN",
ATTR_UPNP_UDN,
),
(
"ATTR_UPNP_UPC",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_UPC",
ATTR_UPNP_UPC,
),
(
"ATTR_UPNP_PRESENTATION_URL",
"homeassistant.helpers.service_info.ssdp.ATTR_UPNP_PRESENTATION_URL",
ATTR_UPNP_PRESENTATION_URL,
),
],
)
def test_deprecated_constants(
caplog: pytest.LogCaptureFixture,
constant_name: str,
replacement_name: str,
replacement: Any,
) -> None:
"""Test deprecated automation constants."""
import_and_test_deprecated_constant(
caplog,
ssdp,
constant_name,
replacement_name,
replacement,
"2026.2",
)

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

@ -3,18 +3,18 @@
from pywilight.const import DOMAIN
from homeassistant.components import ssdp
from homeassistant.components.ssdp import (
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_MODEL_NUMBER,
ATTR_UPNP_SERIAL,
)
from homeassistant.components.wilight.config_flow import (
CONF_MODEL_NAME,
CONF_SERIAL_NUMBER,
)
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_MANUFACTURER,
ATTR_UPNP_MODEL_NAME,
ATTR_UPNP_MODEL_NUMBER,
ATTR_UPNP_SERIAL,
)
from tests.common import MockConfigEntry

5
tests/components/zha/test_config_flow.py
View File

@ -22,7 +22,6 @@ import zigpy.types
from homeassistant import config_entries
from homeassistant.components import ssdp, usb, zeroconf
from homeassistant.components.hassio import AddonError, AddonState
from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL
from homeassistant.components.zha import config_flow, radio_manager
from homeassistant.components.zha.const import (
CONF_BAUDRATE,
@ -43,6 +42,10 @@ from homeassistant.config_entries import (
from homeassistant.const import CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_MANUFACTURER_URL,
ATTR_UPNP_SERIAL,
)
from tests.common import MockConfigEntry