Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add config flow for Bose SoundTouch () 

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
pull/74292/head
Stefan Rado 2022-07-01 07:20:00 +02:00 committed by GitHub
parent 7655b84494
commit 273e9b287f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 584 additions and 236 deletions

2
CODEOWNERS
View File

@ -979,6 +979,8 @@ build.json @home-assistant/supervisor
/tests/components/songpal/ @rytilahti @shenxn
/homeassistant/components/sonos/ @cgtobi @jjlawren
/tests/components/sonos/ @cgtobi @jjlawren
/homeassistant/components/soundtouch/ @kroimon
/tests/components/soundtouch/ @kroimon
/homeassistant/components/spaceapi/ @fabaff
/tests/components/spaceapi/ @fabaff
/homeassistant/components/speedtestdotnet/ @rohankapoorcom @engrbm87

2
homeassistant/components/discovery/__init__.py
View File

@ -61,7 +61,6 @@ SERVICE_HANDLERS = {
"yamaha": ServiceDetails("media_player", "yamaha"),
"frontier_silicon": ServiceDetails("media_player", "frontier_silicon"),
"openhome": ServiceDetails("media_player", "openhome"),
"bose_soundtouch": ServiceDetails("media_player", "soundtouch"),
"bluesound": ServiceDetails("media_player", "bluesound"),
}
@ -70,6 +69,7 @@ OPTIONAL_SERVICE_HANDLERS: dict[str, tuple[str, str | None]] = {}
MIGRATED_SERVICE_HANDLERS = [
SERVICE_APPLE_TV,
"axis",
"bose_soundtouch",
"deconz",
SERVICE_DAIKIN,
"denonavr",

141
homeassistant/components/soundtouch/__init__.py
View File

@ -1 +1,142 @@
"""The soundtouch component."""
import logging
from libsoundtouch import soundtouch_device
from libsoundtouch.device import SoundTouchDevice
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant, ServiceCall
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
DOMAIN,
SERVICE_ADD_ZONE_SLAVE,
SERVICE_CREATE_ZONE,
SERVICE_PLAY_EVERYWHERE,
SERVICE_REMOVE_ZONE_SLAVE,
)
_LOGGER = logging.getLogger(__name__)
SERVICE_PLAY_EVERYWHERE_SCHEMA = vol.Schema({vol.Required("master"): cv.entity_id})
SERVICE_CREATE_ZONE_SCHEMA = vol.Schema(
{
vol.Required("master"): cv.entity_id,
vol.Required("slaves"): cv.entity_ids,
}
)
SERVICE_ADD_ZONE_SCHEMA = vol.Schema(
{
vol.Required("master"): cv.entity_id,
vol.Required("slaves"): cv.entity_ids,
}
)
SERVICE_REMOVE_ZONE_SCHEMA = vol.Schema(
{
vol.Required("master"): cv.entity_id,
vol.Required("slaves"): cv.entity_ids,
}
)
PLATFORMS = [Platform.MEDIA_PLAYER]
class SoundTouchData:
"""SoundTouch data stored in the Home Assistant data object."""
def __init__(self, device: SoundTouchDevice) -> None:
"""Initialize the SoundTouch data object for a device."""
self.device = device
self.media_player = None
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Bose SoundTouch component."""
async def service_handle(service: ServiceCall) -> None:
"""Handle the applying of a service."""
master_id = service.data.get("master")
slaves_ids = service.data.get("slaves")
slaves = []
if slaves_ids:
slaves = [
data.media_player
for data in hass.data[DOMAIN].values()
if data.media_player.entity_id in slaves_ids
]
master = next(
iter(
[
data.media_player
for data in hass.data[DOMAIN].values()
if data.media_player.entity_id == master_id
]
),
None,
)
if master is None:
_LOGGER.warning("Unable to find master with entity_id: %s", str(master_id))
return
if service.service == SERVICE_PLAY_EVERYWHERE:
slaves = [
data.media_player
for data in hass.data[DOMAIN].values()
if data.media_player.entity_id != master_id
]
await hass.async_add_executor_job(master.create_zone, slaves)
elif service.service == SERVICE_CREATE_ZONE:
await hass.async_add_executor_job(master.create_zone, slaves)
elif service.service == SERVICE_REMOVE_ZONE_SLAVE:
await hass.async_add_executor_job(master.remove_zone_slave, slaves)
elif service.service == SERVICE_ADD_ZONE_SLAVE:
await hass.async_add_executor_job(master.add_zone_slave, slaves)
hass.services.async_register(
DOMAIN,
SERVICE_PLAY_EVERYWHERE,
service_handle,
schema=SERVICE_PLAY_EVERYWHERE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_CREATE_ZONE,
service_handle,
schema=SERVICE_CREATE_ZONE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_REMOVE_ZONE_SLAVE,
service_handle,
schema=SERVICE_REMOVE_ZONE_SCHEMA,
)
hass.services.async_register(
DOMAIN,
SERVICE_ADD_ZONE_SLAVE,
service_handle,
schema=SERVICE_ADD_ZONE_SCHEMA,
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Bose SoundTouch from a config entry."""
device = await hass.async_add_executor_job(soundtouch_device, entry.data[CONF_HOST])
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = SoundTouchData(device)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
del hass.data[DOMAIN][entry.entry_id]
return unload_ok

104
homeassistant/components/soundtouch/config_flow.py Normal file
View File

@ -0,0 +1,104 @@
"""Config flow for Bose SoundTouch integration."""
import logging
from libsoundtouch import soundtouch_device
from requests import RequestException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST
from homeassistant.helpers import config_validation as cv
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class SoundtouchConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Bose SoundTouch."""
VERSION = 1
def __init__(self):
"""Initialize a new SoundTouch config flow."""
self.host = None
self.name = None
async def async_step_import(self, import_data):
"""Handle a flow initiated by configuration file."""
self.host = import_data[CONF_HOST]
try:
await self._async_get_device_id()
except RequestException:
return self.async_abort(reason="cannot_connect")
return await self._async_create_soundtouch_entry()
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
errors = {}
if user_input is not None:
self.host = user_input[CONF_HOST]
try:
await self._async_get_device_id(raise_on_progress=False)
except RequestException:
errors["base"] = "cannot_connect"
else:
return await self._async_create_soundtouch_entry()
return self.async_show_form(
step_id="user",
last_step=True,
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): cv.string,
}
),
errors=errors,
)
async def async_step_zeroconf(self, discovery_info):
"""Handle a flow initiated by a zeroconf discovery."""
self.host = discovery_info.host
try:
await self._async_get_device_id()
except RequestException:
return self.async_abort(reason="cannot_connect")
self.context["title_placeholders"] = {"name": self.name}
return await self.async_step_zeroconf_confirm()
async def async_step_zeroconf_confirm(self, user_input=None):
"""Handle user-confirmation of discovered node."""
if user_input is not None:
return await self._async_create_soundtouch_entry()
return self.async_show_form(
step_id="zeroconf_confirm",
last_step=True,
description_placeholders={"name": self.name},
)
async def _async_get_device_id(self, raise_on_progress: bool = True) -> None:
"""Get device ID from SoundTouch device."""
device = await self.hass.async_add_executor_job(soundtouch_device, self.host)
# Check if already configured
await self.async_set_unique_id(
device.config.device_id, raise_on_progress=raise_on_progress
)
self._abort_if_unique_id_configured(updates={CONF_HOST: self.host})
self.name = device.config.name
async def _async_create_soundtouch_entry(self):
"""Finish config flow and create a SoundTouch config entry."""
return self.async_create_entry(
title=self.name,
data={
CONF_HOST: self.host,
},
)

2
homeassistant/components/soundtouch/const.py
View File

@ -1,4 +1,4 @@
"""Constants for the Bose Soundtouch component."""
"""Constants for the Bose SoundTouch component."""
DOMAIN = "soundtouch"
SERVICE_PLAY_EVERYWHERE = "play_everywhere"
SERVICE_CREATE_ZONE = "create_zone"

9
homeassistant/components/soundtouch/manifest.json
View File

@ -1,10 +1,11 @@
{
"domain": "soundtouch",
"name": "Bose Soundtouch",
"name": "Bose SoundTouch",
"documentation": "https://www.home-assistant.io/integrations/soundtouch",
"requirements": ["libsoundtouch==0.8"],
"after_dependencies": ["zeroconf"],
"codeowners": [],
"zeroconf": ["_soundtouch._tcp.local."],
"codeowners": ["@kroimon"],
"iot_class": "local_polling",
"loggers": ["libsoundtouch"]
"loggers": ["libsoundtouch"],
"config_flow": true
}

238
homeassistant/components/soundtouch/media_player.py
View File

@ -1,23 +1,25 @@
"""Support for interface with a Bose Soundtouch."""
"""Support for interface with a Bose SoundTouch."""
from __future__ import annotations
from functools import partial
import logging
import re
from libsoundtouch import soundtouch_device
from libsoundtouch.device import SoundTouchDevice
from libsoundtouch.utils import Source
import voluptuous as vol
from homeassistant.components import media_source
from homeassistant.components.media_player import (
PLATFORM_SCHEMA,
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
)
from homeassistant.components.media_player.browse_media import (
async_process_play_media_url,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
@ -27,19 +29,16 @@ from homeassistant.const import (
STATE_PAUSED,
STATE_PLAYING,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, format_mac
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import (
DOMAIN,
SERVICE_ADD_ZONE_SLAVE,
SERVICE_CREATE_ZONE,
SERVICE_PLAY_EVERYWHERE,
SERVICE_REMOVE_ZONE_SLAVE,
)
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -50,137 +49,57 @@ MAP_STATUS = {
"STOP_STATE": STATE_OFF,
}
DATA_SOUNDTOUCH = "soundtouch"
ATTR_SOUNDTOUCH_GROUP = "soundtouch_group"
ATTR_SOUNDTOUCH_ZONE = "soundtouch_zone"
SOUNDTOUCH_PLAY_EVERYWHERE = vol.Schema({vol.Required("master"): cv.entity_id})
SOUNDTOUCH_CREATE_ZONE_SCHEMA = vol.Schema(
{vol.Required("master"): cv.entity_id, vol.Required("slaves"): cv.entity_ids}
)
SOUNDTOUCH_ADD_ZONE_SCHEMA = vol.Schema(
{vol.Required("master"): cv.entity_id, vol.Required("slaves"): cv.entity_ids}
)
SOUNDTOUCH_REMOVE_ZONE_SCHEMA = vol.Schema(
{vol.Required("master"): cv.entity_id, vol.Required("slaves"): cv.entity_ids}
)
DEFAULT_NAME = "Bose Soundtouch"
DEFAULT_PORT = 8090
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
}
PLATFORM_SCHEMA = vol.All(
PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_PORT): cv.port,
vol.Optional(CONF_NAME, default=""): cv.string,
}
),
)
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Bose Soundtouch platform."""
if DATA_SOUNDTOUCH not in hass.data:
hass.data[DATA_SOUNDTOUCH] = []
if discovery_info:
host = discovery_info["host"]
port = int(discovery_info["port"])
# if device already exists by config
if host in [device.config["host"] for device in hass.data[DATA_SOUNDTOUCH]]:
return
remote_config = {"id": "ha.component.soundtouch", "host": host, "port": port}
bose_soundtouch_entity = SoundTouchDevice(None, remote_config)
hass.data[DATA_SOUNDTOUCH].append(bose_soundtouch_entity)
add_entities([bose_soundtouch_entity], True)
else:
name = config.get(CONF_NAME)
remote_config = {
"id": "ha.component.soundtouch",
"port": config.get(CONF_PORT),
"host": config.get(CONF_HOST),
}
bose_soundtouch_entity = SoundTouchDevice(name, remote_config)
hass.data[DATA_SOUNDTOUCH].append(bose_soundtouch_entity)
add_entities([bose_soundtouch_entity], True)
def service_handle(service: ServiceCall) -> None:
"""Handle the applying of a service."""
master_device_id = service.data.get("master")
slaves_ids = service.data.get("slaves")
slaves = []
if slaves_ids:
slaves = [
device
for device in hass.data[DATA_SOUNDTOUCH]
if device.entity_id in slaves_ids
]
master = next(
iter(
[
device
for device in hass.data[DATA_SOUNDTOUCH]
if device.entity_id == master_device_id
]
),
None,
"""Set up the Bose SoundTouch platform."""
_LOGGER.warning(
"Configuration of the Bose SoundTouch platform in YAML is deprecated and will be "
"removed in a future release; Your existing configuration "
"has been imported into the UI automatically and can be safely removed "
"from your configuration.yaml file"
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
if master is None:
_LOGGER.warning(
"Unable to find master with entity_id: %s", str(master_device_id)
)
return
if service.service == SERVICE_PLAY_EVERYWHERE:
slaves = [
d for d in hass.data[DATA_SOUNDTOUCH] if d.entity_id != master_device_id
]
master.create_zone(slaves)
elif service.service == SERVICE_CREATE_ZONE:
master.create_zone(slaves)
elif service.service == SERVICE_REMOVE_ZONE_SLAVE:
master.remove_zone_slave(slaves)
elif service.service == SERVICE_ADD_ZONE_SLAVE:
master.add_zone_slave(slaves)
hass.services.register(
DOMAIN,
SERVICE_PLAY_EVERYWHERE,
service_handle,
schema=SOUNDTOUCH_PLAY_EVERYWHERE,
)
hass.services.register(
DOMAIN,
SERVICE_CREATE_ZONE,
service_handle,
schema=SOUNDTOUCH_CREATE_ZONE_SCHEMA,
)
hass.services.register(
DOMAIN,
SERVICE_REMOVE_ZONE_SLAVE,
service_handle,
schema=SOUNDTOUCH_REMOVE_ZONE_SCHEMA,
)
hass.services.register(
DOMAIN,
SERVICE_ADD_ZONE_SLAVE,
service_handle,
schema=SOUNDTOUCH_ADD_ZONE_SCHEMA,
)
class SoundTouchDevice(MediaPlayerEntity):
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Bose SoundTouch media player based on a config entry."""
device = hass.data[DOMAIN][entry.entry_id].device
media_player = SoundTouchMediaPlayer(device)
async_add_entities([media_player], True)
hass.data[DOMAIN][entry.entry_id].media_player = media_player
class SoundTouchMediaPlayer(MediaPlayerEntity):
"""Representation of a SoundTouch Bose device."""
_attr_supported_features = (
@ -197,28 +116,32 @@ class SoundTouchDevice(MediaPlayerEntity):
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.BROWSE_MEDIA
)
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
def __init__(self, name, config):
"""Create Soundtouch Entity."""
def __init__(self, device: SoundTouchDevice) -> None:
"""Create SoundTouch media player entity."""
self._device = device
self._attr_unique_id = self._device.config.device_id
self._attr_name = self._device.config.name
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._device.config.device_id)},
connections={
(CONNECTION_NETWORK_MAC, format_mac(self._device.config.mac_address))
},
manufacturer="Bose Corporation",
model=self._device.config.type,
name=self._device.config.name,
)
self._device = soundtouch_device(config["host"], config["port"])
if name is None:
self._name = self._device.config.name
else:
self._name = name
self._status = None
self._volume = None
self._config = config
self._zone = None
@property
def config(self):
"""Return specific soundtouch configuration."""
return self._config
@property
def device(self):
"""Return Soundtouch device."""
"""Return SoundTouch device."""
return self._device
def update(self):
@ -232,17 +155,15 @@ class SoundTouchDevice(MediaPlayerEntity):
"""Volume level of the media player (0..1)."""
return self._volume.actual / 100
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def state(self):
"""Return the state of the device."""
if self._status.source == "STANDBY":
if self._status is None or self._status.source == "STANDBY":
return STATE_OFF
if self._status.source == "INVALID_SOURCE":
return STATE_UNKNOWN
return MAP_STATUS.get(self._status.play_status, STATE_UNAVAILABLE)
@property
@ -478,15 +399,12 @@ class SoundTouchDevice(MediaPlayerEntity):
if not zone_status:
return None
# Due to a bug in the SoundTouch API itself client devices do NOT return their
# siblings as part of the "slaves" list. Only the master has the full list of
# slaves for some reason. To compensate for this shortcoming we have to fetch
# the zone info from the master when the current device is a slave until this is
# fixed in the SoundTouch API or libsoundtouch, or of course until somebody has a
# better idea on how to fix this.
# Client devices do NOT return their siblings as part of the "slaves" list.
# Only the master has the full list of slaves. To compensate for this shortcoming
# we have to fetch the zone info from the master when the current device is a slave.
# In addition to this shortcoming, libsoundtouch seems to report the "is_master"
# property wrong on some slaves, so the only reliable way to detect if the current
# devices is the master, is by comparing the master_id of the zone with the device_id
# devices is the master, is by comparing the master_id of the zone with the device_id.
if zone_status.master_id == self._device.config.device_id:
return self._build_zone_info(self.entity_id, zone_status.slaves)
@ -505,16 +423,16 @@ class SoundTouchDevice(MediaPlayerEntity):
def _get_instance_by_ip(self, ip_address):
"""Search and return a SoundTouchDevice instance by it's IP address."""
for instance in self.hass.data[DATA_SOUNDTOUCH]:
if instance and instance.config["host"] == ip_address:
return instance
for data in self.hass.data[DOMAIN].values():
if data.device.config.device_ip == ip_address:
return data.media_player
return None
def _get_instance_by_id(self, instance_id):
"""Search and return a SoundTouchDevice instance by it's ID (aka MAC address)."""
for instance in self.hass.data[DATA_SOUNDTOUCH]:
if instance and instance.device.config.device_id == instance_id:
return instance
for data in self.hass.data[DOMAIN].values():
if data.device.config.device_id == instance_id:
return data.media_player
return None
def _build_zone_info(self, master, zone_slaves):

8
homeassistant/components/soundtouch/services.yaml
View File

@ -1,6 +1,6 @@
play_everywhere:
name: Play everywhere
description: Play on all Bose Soundtouch devices.
description: Play on all Bose SoundTouch devices.
fields:
master:
name: Master
@ -13,7 +13,7 @@ play_everywhere:
create_zone:
name: Create zone
description: Create a Soundtouch multi-room zone.
description: Create a SoundTouch multi-room zone.
fields:
master:
name: Master
@ -35,7 +35,7 @@ create_zone:
add_zone_slave:
name: Add zone slave
description: Add a slave to a Soundtouch multi-room zone.
description: Add a slave to a SoundTouch multi-room zone.
fields:
master:
name: Master
@ -57,7 +57,7 @@ add_zone_slave:
remove_zone_slave:
name: Remove zone slave
description: Remove a slave from the Soundtouch multi-room zone.
description: Remove a slave from the SoundTouch multi-room zone.
fields:
master:
name: Master

21
homeassistant/components/soundtouch/strings.json Normal file
View File

@ -0,0 +1,21 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]"
}
},
"zeroconf_confirm": {
"title": "Confirm adding Bose SoundTouch device",
"description": "You are about to add the SoundTouch device named `{name}` to Home Assistant."
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

21
homeassistant/components/soundtouch/translations/en.json Normal file
View File

@ -0,0 +1,21 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"cannot_connect": "Failed to connect"
},
"step": {
"user": {
"data": {
"host": "Host"
}
},
"zeroconf_confirm": {
"description": "You are about to add the SoundTouch device named `{name}` to Home Assistant.",
"title": "Confirm adding Bose SoundTouch device"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -334,6 +334,7 @@ FLOWS = {
"sonarr",
"songpal",
"sonos",
"soundtouch",
"speedtestdotnet",
"spider",
"spotify",

5
homeassistant/generated/zeroconf.py
View File

@ -347,6 +347,11 @@ ZEROCONF = {
"domain": "sonos"
}
],
"_soundtouch._tcp.local.": [
{
"domain": "soundtouch"
}
],
"_spotify-connect._tcp.local.": [
{
"domain": "spotify"

34
tests/components/soundtouch/conftest.py
View File

@ -4,9 +4,9 @@ from requests_mock import Mocker
from homeassistant.components.media_player.const import DOMAIN as MEDIA_PLAYER_DOMAIN
from homeassistant.components.soundtouch.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PLATFORM
from homeassistant.const import CONF_HOST, CONF_NAME
from tests.common import load_fixture
from tests.common import MockConfigEntry, load_fixture
DEVICE_1_ID = "020000000001"
DEVICE_2_ID = "020000000002"
@ -14,8 +14,8 @@ DEVICE_1_IP = "192.168.42.1"
DEVICE_2_IP = "192.168.42.2"
DEVICE_1_URL = f"http://{DEVICE_1_IP}:8090"
DEVICE_2_URL = f"http://{DEVICE_2_IP}:8090"
DEVICE_1_NAME = "My Soundtouch 1"
DEVICE_2_NAME = "My Soundtouch 2"
DEVICE_1_NAME = "My SoundTouch 1"
DEVICE_2_NAME = "My SoundTouch 2"
DEVICE_1_ENTITY_ID = f"{MEDIA_PLAYER_DOMAIN}.my_soundtouch_1"
DEVICE_2_ENTITY_ID = f"{MEDIA_PLAYER_DOMAIN}.my_soundtouch_2"
@ -24,15 +24,29 @@ DEVICE_2_ENTITY_ID = f"{MEDIA_PLAYER_DOMAIN}.my_soundtouch_2"
@pytest.fixture
def device1_config() -> dict[str, str]:
"""Mock SoundTouch device 1 config."""
yield {CONF_PLATFORM: DOMAIN, CONF_HOST: DEVICE_1_IP, CONF_NAME: DEVICE_1_NAME}
def device1_config() -> MockConfigEntry:
"""Mock SoundTouch device 1 config entry."""
yield MockConfigEntry(
domain=DOMAIN,
unique_id=DEVICE_1_ID,
data={
CONF_HOST: DEVICE_1_IP,
CONF_NAME: "",
},
)
@pytest.fixture
def device2_config() -> dict[str, str]:
"""Mock SoundTouch device 2 config."""
yield {CONF_PLATFORM: DOMAIN, CONF_HOST: DEVICE_2_IP, CONF_NAME: DEVICE_2_NAME}
def device2_config() -> MockConfigEntry:
"""Mock SoundTouch device 2 config entry."""
yield MockConfigEntry(
domain=DOMAIN,
unique_id=DEVICE_2_ID,
data={
CONF_HOST: DEVICE_2_IP,
CONF_NAME: "",
},
)
@pytest.fixture(scope="session")

113
tests/components/soundtouch/test_config_flow.py Normal file
View File

@ -0,0 +1,113 @@
"""Test config flow."""
from unittest.mock import patch
from requests import RequestException
from requests_mock import ANY, Mocker
from homeassistant.components.soundtouch.const import DOMAIN
from homeassistant.components.zeroconf import ZeroconfServiceInfo
from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
from homeassistant.const import CONF_HOST, CONF_SOURCE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import DEVICE_1_ID, DEVICE_1_IP, DEVICE_1_NAME
async def test_user_flow_create_entry(
hass: HomeAssistant, device1_requests_mock_standby: Mocker
) -> None:
"""Test the full manual user flow from start to finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "user"
assert "flow_id" in result
with patch(
"homeassistant.components.soundtouch.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
CONF_HOST: DEVICE_1_IP,
},
)
assert len(mock_setup_entry.mock_calls) == 1
assert result.get("type") == FlowResultType.CREATE_ENTRY
assert result.get("title") == DEVICE_1_NAME
assert result.get("data") == {
CONF_HOST: DEVICE_1_IP,
}
assert "result" in result
assert result["result"].unique_id == DEVICE_1_ID
assert result["result"].title == DEVICE_1_NAME
async def test_user_flow_cannot_connect(
hass: HomeAssistant, requests_mock: Mocker
) -> None:
"""Test a manual user flow with an invalid host."""
requests_mock.get(ANY, exc=RequestException())
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data={
CONF_HOST: "invalid-hostname",
},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_zeroconf_flow_create_entry(
hass: HomeAssistant, device1_requests_mock_standby: Mocker
) -> None:
"""Test the zeroconf flow from start to finish."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_ZEROCONF},
data=ZeroconfServiceInfo(
host=DEVICE_1_IP,
addresses=[DEVICE_1_IP],
port=8090,
hostname="Bose-SM2-060000000001.local.",
type="_soundtouch._tcp.local.",
name=f"{DEVICE_1_NAME}._soundtouch._tcp.local.",
properties={
"DESCRIPTION": "SoundTouch",
"MAC": DEVICE_1_ID,
"MANUFACTURER": "Bose Corporation",
"MODEL": "SoundTouch",
},
),
)
assert result.get("type") == FlowResultType.FORM
assert result.get("step_id") == "zeroconf_confirm"
assert result.get("description_placeholders") == {"name": DEVICE_1_NAME}
with patch(
"homeassistant.components.soundtouch.async_setup_entry", return_value=True
) as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert len(mock_setup_entry.mock_calls) == 1
assert result.get("type") == FlowResultType.CREATE_ENTRY
assert result.get("title") == DEVICE_1_NAME
assert result.get("data") == {
CONF_HOST: DEVICE_1_IP,
}
assert "result" in result
assert result["result"].unique_id == DEVICE_1_ID
assert result["result"].title == DEVICE_1_NAME

119
tests/components/soundtouch/test_media_player.py
View File

@ -1,4 +1,5 @@
"""Test the SoundTouch component."""
from datetime import timedelta
from typing import Any
from requests_mock import Mocker
@ -25,22 +26,26 @@ from homeassistant.components.soundtouch.const import (
from homeassistant.components.soundtouch.media_player import (
ATTR_SOUNDTOUCH_GROUP,
ATTR_SOUNDTOUCH_ZONE,
DATA_SOUNDTOUCH,
)
from homeassistant.config_entries import RELOAD_AFTER_UPDATE_DELAY
from homeassistant.const import STATE_OFF, STATE_PAUSED, STATE_PLAYING
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.util import dt
from .conftest import DEVICE_1_ENTITY_ID, DEVICE_2_ENTITY_ID
from tests.common import MockConfigEntry, async_fire_time_changed
async def setup_soundtouch(hass: HomeAssistant, *configs: dict[str, str]):
async def setup_soundtouch(hass: HomeAssistant, *mock_entries: MockConfigEntry):
"""Initialize media_player for tests."""
assert await async_setup_component(
hass, MEDIA_PLAYER_DOMAIN, {MEDIA_PLAYER_DOMAIN: list(configs)}
)
assert await async_setup_component(hass, MEDIA_PLAYER_DOMAIN, {})
for mock_entry in mock_entries:
mock_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
await hass.async_start()
async def _test_key_service(
@ -59,7 +64,7 @@ async def _test_key_service(
async def test_playing_media(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
):
"""Test playing media info."""
@ -76,7 +81,7 @@ async def test_playing_media(
async def test_playing_radio(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_radio,
):
"""Test playing radio info."""
@ -89,7 +94,7 @@ async def test_playing_radio(
async def test_playing_aux(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_aux,
):
"""Test playing AUX info."""
@ -102,7 +107,7 @@ async def test_playing_aux(
async def test_playing_bluetooth(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_bluetooth,
):
"""Test playing Bluetooth info."""
@ -118,7 +123,7 @@ async def test_playing_bluetooth(
async def test_get_volume_level(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
):
"""Test volume level."""
@ -130,7 +135,7 @@ async def test_get_volume_level(
async def test_get_state_off(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
):
"""Test state device is off."""
@ -142,7 +147,7 @@ async def test_get_state_off(
async def test_get_state_pause(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp_paused,
):
"""Test state device is paused."""
@ -154,7 +159,7 @@ async def test_get_state_pause(
async def test_is_muted(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_volume_muted: str,
):
@ -170,7 +175,7 @@ async def test_is_muted(
async def test_should_turn_off(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -187,7 +192,7 @@ async def test_should_turn_off(
async def test_should_turn_on(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_key,
):
@ -204,7 +209,7 @@ async def test_should_turn_on(
async def test_volume_up(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -221,7 +226,7 @@ async def test_volume_up(
async def test_volume_down(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -238,7 +243,7 @@ async def test_volume_down(
async def test_set_volume_level(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_volume,
):
@ -258,7 +263,7 @@ async def test_set_volume_level(
async def test_mute(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -275,7 +280,7 @@ async def test_mute(
async def test_play(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp_paused,
device1_requests_mock_key,
):
@ -292,7 +297,7 @@ async def test_play(
async def test_pause(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -309,7 +314,7 @@ async def test_pause(
async def test_play_pause(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -326,7 +331,7 @@ async def test_play_pause(
async def test_next_previous_track(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_upnp,
device1_requests_mock_key,
):
@ -351,7 +356,7 @@ async def test_next_previous_track(
async def test_play_media(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_select,
):
@ -391,7 +396,7 @@ async def test_play_media(
async def test_play_media_url(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_dlna,
):
@ -415,7 +420,7 @@ async def test_play_media_url(
async def test_select_source_aux(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_select,
):
@ -435,7 +440,7 @@ async def test_select_source_aux(
async def test_select_source_bluetooth(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_select,
):
@ -455,7 +460,7 @@ async def test_select_source_bluetooth(
async def test_select_source_invalid_source(
hass: HomeAssistant,
device1_config: dict[str, str],
device1_config: MockConfigEntry,
device1_requests_mock_standby,
device1_requests_mock_select,
):
@ -477,14 +482,25 @@ async def test_select_source_invalid_source(
async def test_play_everywhere(
hass: HomeAssistant,
device1_config: dict[str, str],
device2_config: dict[str, str],
device1_config: MockConfigEntry,
device2_config: MockConfigEntry,
device1_requests_mock_standby,
device2_requests_mock_standby,
device1_requests_mock_set_zone,
):
"""Test play everywhere."""
await setup_soundtouch(hass, device1_config, device2_config)
await setup_soundtouch(hass, device1_config)
# no slaves, set zone must not be called
await hass.services.async_call(
DOMAIN,
SERVICE_PLAY_EVERYWHERE,
{"master": DEVICE_1_ENTITY_ID},
True,
)
assert device1_requests_mock_set_zone.call_count == 0
await setup_soundtouch(hass, device2_config)
# one master, one slave => set zone
await hass.services.async_call(
@ -504,27 +520,11 @@ async def test_play_everywhere(
)
assert device1_requests_mock_set_zone.call_count == 1
# remove second device
for entity in list(hass.data[DATA_SOUNDTOUCH]):
if entity.entity_id == DEVICE_1_ENTITY_ID:
continue
hass.data[DATA_SOUNDTOUCH].remove(entity)
await entity.async_remove()
# no slaves, set zone must not be called
await hass.services.async_call(
DOMAIN,
SERVICE_PLAY_EVERYWHERE,
{"master": DEVICE_1_ENTITY_ID},
True,
)
assert device1_requests_mock_set_zone.call_count == 1
async def test_create_zone(
hass: HomeAssistant,
device1_config: dict[str, str],
device2_config: dict[str, str],
device1_config: MockConfigEntry,
device2_config: MockConfigEntry,
device1_requests_mock_standby,
device2_requests_mock_standby,
device1_requests_mock_set_zone,
@ -567,8 +567,8 @@ async def test_create_zone(
async def test_remove_zone_slave(
hass: HomeAssistant,
device1_config: dict[str, str],
device2_config: dict[str, str],
device1_config: MockConfigEntry,
device2_config: MockConfigEntry,
device1_requests_mock_standby,
device2_requests_mock_standby,
device1_requests_mock_remove_zone_slave,
@ -609,8 +609,8 @@ async def test_remove_zone_slave(
async def test_add_zone_slave(
hass: HomeAssistant,
device1_config: dict[str, str],
device2_config: dict[str, str],
device1_config: MockConfigEntry,
device2_config: MockConfigEntry,
device1_requests_mock_standby,
device2_requests_mock_standby,
device1_requests_mock_add_zone_slave,
@ -651,14 +651,21 @@ async def test_add_zone_slave(
async def test_zone_attributes(
hass: HomeAssistant,
device1_config: dict[str, str],
device2_config: dict[str, str],
device1_config: MockConfigEntry,
device2_config: MockConfigEntry,
device1_requests_mock_standby,
device2_requests_mock_standby,
):
"""Test zone attributes."""
await setup_soundtouch(hass, device1_config, device2_config)
# Fast-forward time to allow all entities to be set up and updated again
async_fire_time_changed(
hass,
dt.utcnow() + timedelta(seconds=RELOAD_AFTER_UPDATE_DELAY + 1),
)
await hass.async_block_till_done()
entity_1_state = hass.states.get(DEVICE_1_ENTITY_ID)
assert entity_1_state.attributes[ATTR_SOUNDTOUCH_ZONE]["is_master"]
assert (