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

Raise exceptions in HEOS service actions (#136049)

Browse Source 
* Raise errors instead of log

* Correct docstring typo
This commit is contained in:
Andrew Sayre 2025-01-20 13:29:57 -06:00 committed by GitHub
parent a4d2fe2d89
commit dde6dc0421
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 541 additions and 417 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
homeassistant/components/heos/__init__.py
View File

@ -28,7 +28,11 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.exceptions import (
ConfigEntryNotReady,
HomeAssistantError,
ServiceValidationError,
)
from homeassistant.helpers import device_registry as dr, entity_registry as er
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.dispatcher import (
@ -306,7 +310,7 @@ class GroupManager:
return group_info_by_entity_id
async def async_join_players(
self, leader_id: int, leader_entity_id: str, member_entity_ids: list[str]
self, leader_id: int, member_entity_ids: list[str]
) -> None:
"""Create a group a group leader and member players."""
# Resolve HEOS player_id for each member entity_id
@ -320,26 +324,11 @@ class GroupManager:
)
member_ids.append(member_id)
try:
await self.controller.create_group(leader_id, member_ids)
except HeosError as err:
_LOGGER.error(
"Failed to group %s with %s: %s",
leader_entity_id,
member_entity_ids,
err,
)
async def async_unjoin_player(self, player_id: int, player_entity_id: str):
async def async_unjoin_player(self, player_id: int):
"""Remove `player_entity_id` from any group."""
try:
await self.controller.create_group(player_id, [])
except HeosError as err:
_LOGGER.error(
"Failed to ungroup %s: %s",
player_entity_id,
err,
)
async def async_update_groups(self) -> None:
"""Update the group membership from the controller."""
@ -449,7 +438,11 @@ class SourceManager:
await player.play_input_source(input_source.media_id)
return
_LOGGER.error("Unknown source: %s", source)
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unknown_source",
translation_placeholders={"source": source},
)
def get_current_source(self, now_playing_media):
"""Determine current source from now playing media."""

45
homeassistant/components/heos/media_player.py
View File

@ -29,6 +29,7 @@ from homeassistant.components.media_player import (
async_process_play_media_url,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
@ -96,10 +97,10 @@ type _FuncType[**_P] = Callable[_P, Awaitable[Any]]
type _ReturnFuncType[**_P] = Callable[_P, Coroutine[Any, Any, None]]
def log_command_error[**_P](
command: str,
def catch_action_error[**_P](
action: str,
) -> Callable[[_FuncType[_P]], _ReturnFuncType[_P]]:
"""Return decorator that logs command failure."""
"""Return decorator that catches errors and raises HomeAssistantError."""
def decorator(func: _FuncType[_P]) -> _ReturnFuncType[_P]:
@wraps(func)
@ -107,7 +108,11 @@ def log_command_error[**_P](
try:
await func(*args, **kwargs)
except (HeosError, ValueError) as ex:
_LOGGER.error("Unable to %s: %s", command, ex)
raise HomeAssistantError(
translation_domain=HEOS_DOMAIN,
translation_key="action_error",
translation_placeholders={"action": action, "error": str(ex)},
) from ex
return wrapper
@ -174,49 +179,49 @@ class HeosMediaPlayer(MediaPlayerEntity):
)
async_dispatcher_send(self.hass, SIGNAL_HEOS_PLAYER_ADDED)
@log_command_error("clear playlist")
@catch_action_error("clear playlist")
async def async_clear_playlist(self) -> None:
"""Clear players playlist."""
await self._player.clear_queue()
@log_command_error("join_players")
@catch_action_error("join players")
async def async_join_players(self, group_members: list[str]) -> None:
"""Join `group_members` as a player group with the current player."""
await self._group_manager.async_join_players(
self._player.player_id, self.entity_id, group_members
self._player.player_id, group_members
)
@log_command_error("pause")
@catch_action_error("pause")
async def async_media_pause(self) -> None:
"""Send pause command."""
await self._player.pause()
@log_command_error("play")
@catch_action_error("play")
async def async_media_play(self) -> None:
"""Send play command."""
await self._player.play()
@log_command_error("move to previous track")
@catch_action_error("move to previous track")
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._player.play_previous()
@log_command_error("move to next track")
@catch_action_error("move to next track")
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._player.play_next()
@log_command_error("stop")
@catch_action_error("stop")
async def async_media_stop(self) -> None:
"""Send stop command."""
await self._player.stop()
@log_command_error("set mute")
@catch_action_error("set mute")
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
await self._player.set_mute(mute)
@log_command_error("play media")
@catch_action_error("play media")
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
@ -281,17 +286,17 @@ class HeosMediaPlayer(MediaPlayerEntity):
raise ValueError(f"Unsupported media type '{media_type}'")
@log_command_error("select source")
@catch_action_error("select source")
async def async_select_source(self, source: str) -> None:
"""Select input source."""
await self._source_manager.play_source(source, self._player)
@log_command_error("set shuffle")
@catch_action_error("set shuffle")
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode."""
await self._player.set_play_mode(self._player.repeat, shuffle)
@log_command_error("set volume level")
@catch_action_error("set volume level")
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._player.set_volume(int(volume * 100))
@ -304,12 +309,10 @@ class HeosMediaPlayer(MediaPlayerEntity):
ior, current_support, BASE_SUPPORTED_FEATURES
)
@log_command_error("unjoin_player")
@catch_action_error("unjoin player")
async def async_unjoin_player(self) -> None:
"""Remove this player from any group."""
await self._group_manager.async_unjoin_player(
self._player.player_id, self.entity_id
)
await self._group_manager.async_unjoin_player(self._player.player_id)
@property
def available(self) -> bool:

4
homeassistant/components/heos/quality_scale.yaml
View File

@ -23,9 +23,7 @@ rules:
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: todo
comment: Actions currently only log and instead should raise exceptions.
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done

6
homeassistant/components/heos/strings.json
View File

@ -91,8 +91,14 @@
}
},
"exceptions": {
"action_error": {
"message": "Unable to {action}: {error}"
},
"integration_not_loaded": {
"message": "The HEOS integration is not loaded"
},
"unknown_source": {
"message": "Unknown source: {source}"
}
},
"issues": {

650
tests/components/heos/test_media_player.py
View File

@ -1,12 +1,16 @@
"""Tests for the Heos Media Player platform."""
import asyncio
from collections.abc import Sequence
import re
from typing import Any
from pyheos import (
AddCriteriaType,
CommandFailedError,
Heos,
HeosError,
MediaItem,
PlayState,
SignalHeosEvent,
SignalType,
@ -58,7 +62,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.setup import async_setup_component
@ -326,17 +330,12 @@ async def test_updates_from_user_changed(
async def test_clear_playlist(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the clear playlist service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_CLEAR_PLAYLIST,
@ -344,23 +343,35 @@ async def test_clear_playlist(
blocking=True,
)
assert player.clear_queue.call_count == 1
player.clear_queue.reset_mock()
async def test_clear_playlist_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test error raised when clear playlist fails."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.clear_queue.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to clear playlist: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError, match=re.escape("Unable to clear playlist: Failure (1)")
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_CLEAR_PLAYLIST,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.clear_queue.call_count == 1
async def test_pause(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the pause service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PAUSE,
@ -368,23 +379,35 @@ async def test_pause(
blocking=True,
)
assert player.pause.call_count == 1
player.pause.reset_mock()
async def test_pause_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the pause service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.pause.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to pause: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError, match=re.escape("Unable to pause: Failure (1)")
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PAUSE,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.pause.call_count == 1
async def test_play(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the play service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY,
@ -392,23 +415,35 @@ async def test_play(
blocking=True,
)
assert player.play.call_count == 1
player.play.reset_mock()
async def test_play_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the play service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.play.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to play: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError, match=re.escape("Unable to play: Failure (1)")
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PLAY,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.play.call_count == 1
async def test_previous_track(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the previous track service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PREVIOUS_TRACK,
@ -416,23 +451,36 @@ async def test_previous_track(
blocking=True,
)
assert player.play_previous.call_count == 1
player.play_previous.reset_mock()
async def test_previous_track_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the previous track service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.play_previous.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to move to previous track: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to move to previous track: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_PREVIOUS_TRACK,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.play_previous.call_count == 1
async def test_next_track(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the next track service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_NEXT_TRACK,
@ -440,23 +488,36 @@ async def test_next_track(
blocking=True,
)
assert player.play_next.call_count == 1
player.play_next.reset_mock()
async def test_next_track_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the next track service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.play_next.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to move to next track: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to move to next track: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_NEXT_TRACK,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.play_next.call_count == 1
async def test_stop(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the stop service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_STOP,
@ -464,23 +525,36 @@ async def test_stop(
blocking=True,
)
assert player.stop.call_count == 1
player.stop.reset_mock()
async def test_stop_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the stop service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.stop.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to stop: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to stop: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_MEDIA_STOP,
{ATTR_ENTITY_ID: "media_player.test_player"},
blocking=True,
)
assert player.stop.call_count == 1
async def test_volume_mute(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the volume mute service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_MUTE,
@ -488,23 +562,36 @@ async def test_volume_mute(
blocking=True,
)
assert player.set_mute.call_count == 1
player.set_mute.reset_mock()
async def test_volume_mute_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the volume mute service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.set_mute.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to set mute: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to set mute: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_VOLUME_MUTED: True},
blocking=True,
)
assert player.set_mute.call_count == 1
async def test_shuffle_set(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the shuffle set service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SHUFFLE_SET,
@ -512,23 +599,36 @@ async def test_shuffle_set(
blocking=True,
)
player.set_play_mode.assert_called_once_with(player.repeat, True)
player.set_play_mode.reset_mock()
async def test_shuffle_set_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the shuffle set service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.set_play_mode.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to set shuffle: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to set shuffle: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SHUFFLE_SET,
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_SHUFFLE: True},
blocking=True,
)
player.set_play_mode.assert_called_once_with(player.repeat, True)
async def test_volume_set(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the volume set service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_SET,
@ -536,9 +636,27 @@ async def test_volume_set(
blocking=True,
)
player.set_volume.assert_called_once_with(100)
player.set_volume.reset_mock()
async def test_volume_set_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the volume set service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.set_volume.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to set volume level: Failure (1)" in caplog.text
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to set volume level: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_VOLUME_LEVEL: 1},
blocking=True,
)
player.set_volume.assert_called_once_with(100)
async def test_select_favorite(
@ -594,26 +712,31 @@ async def test_select_radio_favorite(
async def test_select_radio_favorite_command_error(
hass: HomeAssistant,
config_entry,
config,
controller,
favorites,
caplog: pytest.LogCaptureFixture,
config_entry: MockConfigEntry,
controller: Heos,
favorites: dict[int, MediaItem],
) -> None:
"""Tests command error logged when playing favorite."""
await setup_platform(hass, config_entry, config)
"""Tests command error raises when playing favorite."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
# Test set radio preset
favorite = favorites[2]
player.play_preset_station.side_effect = CommandFailedError(None, "Failure", 1)
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to select source: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_INPUT_SOURCE: favorite.name},
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_INPUT_SOURCE: favorite.name,
},
blocking=True,
)
player.play_preset_station.assert_called_once_with(2)
assert "Unable to select source: Failure (1)" in caplog.text
async def test_select_input_source(
@ -645,37 +768,41 @@ async def test_select_input_source(
assert state.attributes[ATTR_INPUT_SOURCE] == input_source.name
async def test_select_input_unknown(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
@pytest.mark.usefixtures("controller")
async def test_select_input_unknown_raises(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Tests selecting an unknown input."""
await setup_platform(hass, config_entry, config)
"""Tests selecting an unknown input raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
with pytest.raises(
ServiceValidationError,
match=re.escape("Unknown source: Unknown"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_INPUT_SOURCE: "Unknown"},
blocking=True,
)
assert "Unknown source: Unknown" in caplog.text
async def test_select_input_command_error(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
input_sources,
config_entry: MockConfigEntry,
controller: Heos,
input_sources: Sequence[MediaItem],
) -> None:
"""Tests selecting an unknown input."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
input_source = input_sources[0]
player.play_input_source.side_effect = CommandFailedError(None, "Failure", 1)
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to select source: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
@ -686,7 +813,6 @@ async def test_select_input_command_error(
blocking=True,
)
player.play_input_source.assert_called_once_with(input_source.media_id)
assert "Unable to select source: Failure (1)" in caplog.text
async def test_unload_config_entry(
@ -698,105 +824,99 @@ async def test_unload_config_entry(
assert hass.states.get("media_player.test_player").state == STATE_UNAVAILABLE
async def test_play_media_url(
@pytest.mark.parametrize("media_type", [MediaType.URL, MediaType.MUSIC])
async def test_play_media(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
config_entry: MockConfigEntry,
controller: Heos,
media_type: MediaType,
) -> None:
"""Test the play media service with type url."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
url = "http://news/podcast.mp3"
# First pass completes successfully, second pass raises command error
for _ in range(2):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: MediaType.URL,
ATTR_MEDIA_CONTENT_TYPE: media_type,
ATTR_MEDIA_CONTENT_ID: url,
},
blocking=True,
)
player.play_url.assert_called_once_with(url)
player.play_url.reset_mock()
player.play_url.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to play media: Failure (1)" in caplog.text
async def test_play_media_music(
@pytest.mark.parametrize("media_type", [MediaType.URL, MediaType.MUSIC])
async def test_play_media_error(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
config_entry: MockConfigEntry,
controller: Heos,
media_type: MediaType,
) -> None:
"""Test the play media service with type music."""
await setup_platform(hass, config_entry, config)
"""Test the play media service with type url error raises."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
player.play_url.side_effect = CommandFailedError(None, "Failure", 1)
url = "http://news/podcast.mp3"
# First pass completes successfully, second pass raises command error
for _ in range(2):
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to play media: Failure (1)"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_TYPE: media_type,
ATTR_MEDIA_CONTENT_ID: url,
},
blocking=True,
)
player.play_url.assert_called_once_with(url)
player.play_url.reset_mock()
player.play_url.side_effect = CommandFailedError(None, "Failure", 1)
assert "Unable to play media: Failure (1)" in caplog.text
@pytest.mark.parametrize(
("content_id", "expected_index"), [("1", 1), ("Quick Select 2", 2)]
)
async def test_play_media_quick_select(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
quick_selects,
config_entry: MockConfigEntry,
controller: Heos,
content_id: str,
expected_index: int,
) -> None:
"""Test the play media service with type quick_select."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
quick_select = list(quick_selects.items())[0]
index = quick_select[0]
name = quick_select[1]
# Play by index
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: "quick_select",
ATTR_MEDIA_CONTENT_ID: str(index),
ATTR_MEDIA_CONTENT_ID: content_id,
},
blocking=True,
)
player.play_quick_select.assert_called_once_with(index)
# Play by name
player.play_quick_select.reset_mock()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: "quick_select",
ATTR_MEDIA_CONTENT_ID: name,
},
blocking=True,
)
player.play_quick_select.assert_called_once_with(index)
# Invalid name
player.play_quick_select.reset_mock()
player.play_quick_select.assert_called_once_with(expected_index)
async def test_play_media_quick_select_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the play media service with invalid quick_select raises."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to play media: Invalid quick select 'Invalid'"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
@ -808,51 +928,56 @@ async def test_play_media_quick_select(
blocking=True,
)
assert player.play_quick_select.call_count == 0
assert "Unable to play media: Invalid quick select 'Invalid'" in caplog.text
@pytest.mark.parametrize(
("enqueue", "criteria"),
[
(None, AddCriteriaType.REPLACE_AND_PLAY),
(True, AddCriteriaType.ADD_TO_END),
("next", AddCriteriaType.PLAY_NEXT),
],
)
async def test_play_media_playlist(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
playlists,
config_entry: MockConfigEntry,
controller: Heos,
playlists: Sequence[MediaItem],
enqueue: Any,
criteria: AddCriteriaType,
) -> None:
"""Test the play media service with type playlist."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
playlist = playlists[0]
# Play without enqueuing
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
service_data = {
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
ATTR_MEDIA_CONTENT_ID: playlist.name,
},
blocking=True,
)
player.add_to_queue.assert_called_once_with(
playlist, AddCriteriaType.REPLACE_AND_PLAY
)
# Play with enqueuing
player.add_to_queue.reset_mock()
}
if enqueue is not None:
service_data[ATTR_MEDIA_ENQUEUE] = enqueue
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
ATTR_MEDIA_CONTENT_ID: playlist.name,
ATTR_MEDIA_ENQUEUE: True,
},
service_data,
blocking=True,
)
player.add_to_queue.assert_called_once_with(playlist, AddCriteriaType.ADD_TO_END)
# Invalid name
player.add_to_queue.reset_mock()
player.add_to_queue.assert_called_once_with(playlist, criteria)
async def test_play_media_playlist_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the play media service with an invalid playlist name."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to play media: Invalid playlist 'Invalid'"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
@ -864,50 +989,46 @@ async def test_play_media_playlist(
blocking=True,
)
assert player.add_to_queue.call_count == 0
assert "Unable to play media: Invalid playlist 'Invalid'" in caplog.text
@pytest.mark.parametrize(
("content_id", "expected_index"), [("1", 1), ("Classical MPR (Classical Music)", 2)]
)
async def test_play_media_favorite(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
favorites,
config_entry: MockConfigEntry,
controller: Heos,
content_id: str,
expected_index: int,
) -> None:
"""Test the play media service with type favorite."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
quick_select = list(favorites.items())[0]
index = quick_select[0]
name = quick_select[1].name
# Play by index
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: "favorite",
ATTR_MEDIA_CONTENT_ID: str(index),
ATTR_MEDIA_CONTENT_ID: content_id,
},
blocking=True,
)
player.play_preset_station.assert_called_once_with(index)
# Play by name
player.play_preset_station.reset_mock()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.test_player",
ATTR_MEDIA_CONTENT_TYPE: "favorite",
ATTR_MEDIA_CONTENT_ID: name,
},
blocking=True,
)
player.play_preset_station.assert_called_once_with(index)
# Invalid name
player.play_preset_station.reset_mock()
player.play_preset_station.assert_called_once_with(expected_index)
async def test_play_media_favorite_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test the play media service with an invalid favorite raises."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to play media: Invalid favorite 'Invalid'"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
@ -919,18 +1040,19 @@ async def test_play_media_favorite(
blocking=True,
)
assert player.play_preset_station.call_count == 0
assert "Unable to play media: Invalid favorite 'Invalid'" in caplog.text
@pytest.mark.usefixtures("controller")
async def test_play_media_invalid_type(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
"""Test the play media service with an invalid type."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to play media: Unsupported media type 'Other'"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
@ -941,18 +1063,14 @@ async def test_play_media_invalid_type(
},
blocking=True,
)
assert "Unable to play media: Unsupported media type 'Other'" in caplog.text
async def test_media_player_join_group(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test grouping of media players through the join service."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_JOIN,
@ -968,9 +1086,19 @@ async def test_media_player_join_group(
2,
],
)
assert "Failed to group media_player.test_player with" not in caplog.text
async def test_media_player_join_group_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test grouping of media players through the join service raises error."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
controller.create_group.side_effect = HeosError("error")
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to join players: error"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_JOIN,
@ -980,7 +1108,6 @@ async def test_media_player_join_group(
},
blocking=True,
)
assert "Failed to group media_player.test_player with" in caplog.text
async def test_media_player_group_members(
@ -1019,22 +1146,11 @@ async def test_media_player_group_members_error(
async def test_media_player_unjoin_group(
hass: HomeAssistant,
config_entry,
config,
controller,
caplog: pytest.LogCaptureFixture,
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test ungrouping of media players through the join service."""
await setup_platform(hass, config_entry, config)
player = controller.players[1]
player.heos.dispatcher.send(
SignalType.PLAYER_EVENT,
player.player_id,
const.EVENT_PLAYER_STATE_CHANGED,
)
await hass.async_block_till_done()
"""Test ungrouping of media players through the unjoin service."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_UNJOIN,
@ -1044,9 +1160,19 @@ async def test_media_player_unjoin_group(
blocking=True,
)
controller.create_group.assert_called_once_with(1, [])
assert "Failed to ungroup media_player.test_player" not in caplog.text
async def test_media_player_unjoin_group_error(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
) -> None:
"""Test ungrouping of media players through the unjoin service error raises."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
controller.create_group.side_effect = HeosError("error")
with pytest.raises(
HomeAssistantError,
match=re.escape("Unable to unjoin player: error"),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_UNJOIN,
@ -1055,19 +1181,17 @@ async def test_media_player_unjoin_group(
},
blocking=True,
)
assert "Failed to ungroup media_player.test_player" in caplog.text
async def test_media_player_group_fails_when_entity_removed(
hass: HomeAssistant,
config_entry,
config,
controller,
config_entry: MockConfigEntry,
controller: Heos,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test grouping fails when entity removed."""
await setup_platform(hass, config_entry, config)
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
# Remove one of the players
entity_registry.async_remove("media_player.test_player_2")