mirror of
https://github.com/home-assistant/core.git
synced 2025-11-09 10:59:40 +00:00
2fb85aab8e
* Incorporate GroupManager * Update quality scale * Fix group params * Revert quality scale change * Rename varaible * Move group action implementaton out of coordinator * Fix get_group_members hass access * entity -> entity_id
1345 lines
45 KiB
Python
1345 lines
45 KiB
Python
"""Tests for the Heos Media Player platform."""
|
|
|
|
from datetime import timedelta
|
|
import re
|
|
from typing import Any
|
|
|
|
from freezegun.api import FrozenDateTimeFactory
|
|
from pyheos import (
|
|
AddCriteriaType,
|
|
CommandFailedError,
|
|
Heos,
|
|
HeosError,
|
|
MediaItem,
|
|
MediaType as HeosMediaType,
|
|
PlayerUpdateResult,
|
|
PlayState,
|
|
RepeatType,
|
|
SignalHeosEvent,
|
|
SignalType,
|
|
const,
|
|
)
|
|
import pytest
|
|
from syrupy.assertion import SnapshotAssertion
|
|
from syrupy.filters import props
|
|
|
|
from homeassistant.components.heos.const import DOMAIN
|
|
from homeassistant.components.media_player import (
|
|
ATTR_GROUP_MEMBERS,
|
|
ATTR_INPUT_SOURCE,
|
|
ATTR_INPUT_SOURCE_LIST,
|
|
ATTR_MEDIA_CONTENT_ID,
|
|
ATTR_MEDIA_CONTENT_TYPE,
|
|
ATTR_MEDIA_DURATION,
|
|
ATTR_MEDIA_ENQUEUE,
|
|
ATTR_MEDIA_POSITION,
|
|
ATTR_MEDIA_POSITION_UPDATED_AT,
|
|
ATTR_MEDIA_REPEAT,
|
|
ATTR_MEDIA_SHUFFLE,
|
|
ATTR_MEDIA_VOLUME_LEVEL,
|
|
ATTR_MEDIA_VOLUME_MUTED,
|
|
DOMAIN as MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_CLEAR_PLAYLIST,
|
|
SERVICE_JOIN,
|
|
SERVICE_PLAY_MEDIA,
|
|
SERVICE_SELECT_SOURCE,
|
|
SERVICE_UNJOIN,
|
|
MediaType,
|
|
RepeatMode,
|
|
)
|
|
from homeassistant.const import (
|
|
ATTR_ENTITY_ID,
|
|
SERVICE_MEDIA_NEXT_TRACK,
|
|
SERVICE_MEDIA_PAUSE,
|
|
SERVICE_MEDIA_PLAY,
|
|
SERVICE_MEDIA_PREVIOUS_TRACK,
|
|
SERVICE_MEDIA_STOP,
|
|
SERVICE_REPEAT_SET,
|
|
SERVICE_SHUFFLE_SET,
|
|
SERVICE_VOLUME_MUTE,
|
|
SERVICE_VOLUME_SET,
|
|
STATE_IDLE,
|
|
STATE_PLAYING,
|
|
STATE_UNAVAILABLE,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
|
|
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
|
|
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
|
|
|
|
|
async def test_state_attributes(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, snapshot: SnapshotAssertion
|
|
) -> None:
|
|
"""Tests the state attributes."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state == snapshot(
|
|
exclude=props(
|
|
"entity_picture_local",
|
|
"context",
|
|
"last_changed",
|
|
"last_reported",
|
|
"last_updated",
|
|
)
|
|
)
|
|
|
|
|
|
async def test_updates_from_signals(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Tests dispatched signals update player."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
# Test player does not update for other players
|
|
player.state = PlayState.PLAY
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT, 2, const.EVENT_PLAYER_STATE_CHANGED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.state == STATE_IDLE
|
|
|
|
# Test player_update standard events
|
|
player.state = PlayState.PLAY
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.state == STATE_PLAYING
|
|
|
|
# Test player_update progress events
|
|
player.now_playing_media.duration = 360000
|
|
player.now_playing_media.current_position = 1000
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT,
|
|
player.player_id,
|
|
const.EVENT_PLAYER_NOW_PLAYING_PROGRESS,
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_MEDIA_POSITION_UPDATED_AT] is not None
|
|
assert state.attributes[ATTR_MEDIA_DURATION] == 360
|
|
assert state.attributes[ATTR_MEDIA_POSITION] == 1
|
|
|
|
|
|
async def test_updates_from_connection_event(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Tests player updates from connection event after connection failure."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
# Connected
|
|
player.available = True
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.state == STATE_IDLE
|
|
assert controller.load_players.call_count == 1
|
|
|
|
# Disconnected
|
|
controller.load_players.reset_mock()
|
|
player.available = False
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.HEOS_EVENT, SignalHeosEvent.DISCONNECTED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.state == STATE_UNAVAILABLE
|
|
assert controller.load_players.call_count == 0
|
|
|
|
# Connected handles refresh failure
|
|
controller.load_players.reset_mock()
|
|
controller.load_players.side_effect = CommandFailedError(None, "Failure", 1)
|
|
player.available = True
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.state == STATE_IDLE
|
|
assert controller.load_players.call_count == 1
|
|
assert "Unable to refresh players" in caplog.text
|
|
|
|
|
|
async def test_updates_from_connection_event_new_player_ids(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
device_registry: dr.DeviceRegistry,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
change_data_mapped_ids: PlayerUpdateResult,
|
|
) -> None:
|
|
"""Test player ids changed after reconnection updates ids."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
|
|
# Assert current IDs
|
|
assert device_registry.async_get_device(identifiers={(DOMAIN, "1")})
|
|
assert entity_registry.async_get_entity_id(MEDIA_PLAYER_DOMAIN, DOMAIN, "1")
|
|
|
|
# Send event which will result in updated IDs.
|
|
controller.load_players.return_value = change_data_mapped_ids
|
|
await controller.dispatcher.wait_send(
|
|
SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
# Assert updated IDs and previous don't exist
|
|
assert not device_registry.async_get_device(identifiers={(DOMAIN, "1")})
|
|
assert device_registry.async_get_device(identifiers={(DOMAIN, "101")})
|
|
assert not entity_registry.async_get_entity_id(MEDIA_PLAYER_DOMAIN, DOMAIN, "1")
|
|
assert entity_registry.async_get_entity_id(MEDIA_PLAYER_DOMAIN, DOMAIN, "101")
|
|
|
|
|
|
async def test_updates_from_sources_updated(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""Tests player updates from changes in sources list."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
controller.get_input_sources.return_value = []
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {}
|
|
)
|
|
freezer.tick(timedelta(seconds=1))
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == [
|
|
"Today's Hits Radio",
|
|
"Classical MPR (Classical Music)",
|
|
]
|
|
|
|
|
|
async def test_updates_from_players_changed(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
change_data: PlayerUpdateResult,
|
|
) -> None:
|
|
"""Test player updates from changes to available players."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
assert hass.states.get("media_player.test_player").state == STATE_IDLE
|
|
player.state = PlayState.PLAY
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.CONTROLLER_EVENT, const.EVENT_PLAYERS_CHANGED, change_data
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert hass.states.get("media_player.test_player").state == STATE_PLAYING
|
|
|
|
|
|
async def test_updates_from_players_changed_new_ids(
|
|
hass: HomeAssistant,
|
|
entity_registry: er.EntityRegistry,
|
|
device_registry: dr.DeviceRegistry,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
change_data_mapped_ids: PlayerUpdateResult,
|
|
) -> None:
|
|
"""Test player updates from changes to available players."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
# Assert device registry matches current id
|
|
assert device_registry.async_get_device(identifiers={(DOMAIN, "1")})
|
|
# Assert entity registry matches current id
|
|
assert (
|
|
entity_registry.async_get_entity_id(MEDIA_PLAYER_DOMAIN, DOMAIN, "1")
|
|
== "media_player.test_player"
|
|
)
|
|
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.CONTROLLER_EVENT,
|
|
const.EVENT_PLAYERS_CHANGED,
|
|
change_data_mapped_ids,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
# Assert device registry identifiers were updated
|
|
assert len(device_registry.devices) == 2
|
|
assert device_registry.async_get_device(identifiers={(DOMAIN, "101")})
|
|
# Assert entity registry unique id was updated
|
|
assert len(entity_registry.entities) == 2
|
|
assert (
|
|
entity_registry.async_get_entity_id(MEDIA_PLAYER_DOMAIN, DOMAIN, "101")
|
|
== "media_player.test_player"
|
|
)
|
|
|
|
|
|
async def test_updates_from_user_changed(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""Tests player updates from changes in user."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
controller._signed_in_username = None
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.CONTROLLER_EVENT, const.EVENT_USER_CHANGED, None
|
|
)
|
|
freezer.tick(timedelta(seconds=1))
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == [
|
|
"HEOS Drive - Line In 1",
|
|
"Speaker - Line In 1",
|
|
]
|
|
|
|
|
|
async def test_updates_from_groups_changed(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test player updates from changes to groups."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
# Assert current state
|
|
assert hass.states.get("media_player.test_player").attributes[
|
|
ATTR_GROUP_MEMBERS
|
|
] == ["media_player.test_player", "media_player.test_player_2"]
|
|
assert hass.states.get("media_player.test_player_2").attributes[
|
|
ATTR_GROUP_MEMBERS
|
|
] == ["media_player.test_player", "media_player.test_player_2"]
|
|
|
|
# Clear group information
|
|
controller._groups = {}
|
|
for player in controller.players.values():
|
|
player.group_id = None
|
|
await controller.dispatcher.wait_send(
|
|
SignalType.CONTROLLER_EVENT, const.EVENT_GROUPS_CHANGED, None
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
# Assert groups changed
|
|
assert (
|
|
hass.states.get("media_player.test_player").attributes[ATTR_GROUP_MEMBERS]
|
|
is None
|
|
)
|
|
assert (
|
|
hass.states.get("media_player.test_player_2").attributes[ATTR_GROUP_MEMBERS]
|
|
is None
|
|
)
|
|
|
|
|
|
async def test_clear_playlist(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the clear playlist service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the pause service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the play service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the previous track service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the next track service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the stop service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the volume mute service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the shuffle set service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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_repeat_set(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the repeat set service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_REPEAT_SET,
|
|
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_MEDIA_REPEAT: RepeatMode.ONE},
|
|
blocking=True,
|
|
)
|
|
player.set_play_mode.assert_called_once_with(RepeatType.ON_ONE, player.shuffle)
|
|
|
|
|
|
async def test_repeat_set_error(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the repeat 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)
|
|
with pytest.raises(
|
|
HomeAssistantError,
|
|
match=re.escape("Unable to set repeat: Failure (1)"),
|
|
):
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_REPEAT_SET,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_REPEAT: RepeatMode.ALL,
|
|
},
|
|
blocking=True,
|
|
)
|
|
player.set_play_mode.assert_called_once_with(RepeatType.ON_ALL, player.shuffle)
|
|
|
|
|
|
async def test_volume_set(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
|
|
) -> None:
|
|
"""Test the volume set service."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[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_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)
|
|
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(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
favorites: dict[int, MediaItem],
|
|
) -> None:
|
|
"""Tests selecting a music service favorite and state."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
# Test set music service preset
|
|
favorite = favorites[1]
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_SELECT_SOURCE,
|
|
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_INPUT_SOURCE: favorite.name},
|
|
blocking=True,
|
|
)
|
|
player.play_preset_station.assert_called_once_with(1)
|
|
# Test state is matched by station name
|
|
player.now_playing_media.type = HeosMediaType.STATION
|
|
player.now_playing_media.station = favorite.name
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_INPUT_SOURCE] == favorite.name
|
|
|
|
|
|
async def test_select_radio_favorite(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
favorites: dict[int, MediaItem],
|
|
) -> None:
|
|
"""Tests selecting a radio favorite and state."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
# Test set radio preset
|
|
favorite = favorites[2]
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_SELECT_SOURCE,
|
|
{ATTR_ENTITY_ID: "media_player.test_player", ATTR_INPUT_SOURCE: favorite.name},
|
|
blocking=True,
|
|
)
|
|
player.play_preset_station.assert_called_once_with(2)
|
|
# Test state is matched by album id
|
|
player.now_playing_media.type = HeosMediaType.STATION
|
|
player.now_playing_media.station = "Classical"
|
|
player.now_playing_media.album_id = favorite.media_id
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_INPUT_SOURCE] == favorite.name
|
|
|
|
|
|
async def test_select_radio_favorite_command_error(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
favorites: dict[int, MediaItem],
|
|
) -> None:
|
|
"""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,
|
|
},
|
|
blocking=True,
|
|
)
|
|
player.play_preset_station.assert_called_once_with(2)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source_name", "station"),
|
|
[
|
|
("HEOS Drive - Line In 1", "Line In 1"),
|
|
("Speaker - Line In 1", "Speaker - Line In 1"),
|
|
],
|
|
)
|
|
async def test_select_input_source(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
input_sources: list[MediaItem],
|
|
source_name: str,
|
|
station: str,
|
|
) -> None:
|
|
"""Tests selecting input source and state."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_SELECT_SOURCE,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_INPUT_SOURCE: source_name,
|
|
},
|
|
blocking=True,
|
|
)
|
|
input_sources = next(
|
|
input_sources
|
|
for input_sources in input_sources
|
|
if input_sources.name == source_name
|
|
)
|
|
player.play_media.assert_called_once_with(input_sources)
|
|
# Update the now_playing_media to reflect play_media
|
|
player.now_playing_media.source_id = const.MUSIC_SOURCE_AUX_INPUT
|
|
player.now_playing_media.station = station
|
|
player.now_playing_media.media_id = const.INPUT_AUX_IN_1
|
|
await player.heos.dispatcher.wait_send(
|
|
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
|
|
)
|
|
await hass.async_block_till_done()
|
|
state = hass.states.get("media_player.test_player")
|
|
assert state.attributes[ATTR_INPUT_SOURCE] == source_name
|
|
|
|
|
|
async def test_select_input_unknown_raises(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry
|
|
) -> None:
|
|
"""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,
|
|
)
|
|
|
|
|
|
async def test_select_input_command_error(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
input_sources: list[MediaItem],
|
|
) -> None:
|
|
"""Tests selecting an unknown input."""
|
|
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_media.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: input_source.name,
|
|
},
|
|
blocking=True,
|
|
)
|
|
player.play_media.assert_called_once_with(input_source)
|
|
|
|
|
|
async def test_unload_config_entry(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry
|
|
) -> None:
|
|
"""Test the player is set unavailable when the config entry is unloaded."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
assert await hass.config_entries.async_unload(config_entry.entry_id)
|
|
assert hass.states.get("media_player.test_player").state == STATE_UNAVAILABLE
|
|
|
|
|
|
@pytest.mark.parametrize("media_type", [MediaType.URL, MediaType.MUSIC])
|
|
async def test_play_media(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
media_type: MediaType,
|
|
) -> None:
|
|
"""Test the play media service with type url."""
|
|
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"
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_PLAY_MEDIA,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: media_type,
|
|
ATTR_MEDIA_CONTENT_ID: url,
|
|
},
|
|
blocking=True,
|
|
)
|
|
player.play_url.assert_called_once_with(url)
|
|
|
|
|
|
@pytest.mark.parametrize("media_type", [MediaType.URL, MediaType.MUSIC])
|
|
async def test_play_media_error(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
media_type: MediaType,
|
|
) -> None:
|
|
"""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"
|
|
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: media_type,
|
|
ATTR_MEDIA_CONTENT_ID: url,
|
|
},
|
|
blocking=True,
|
|
)
|
|
player.play_url.assert_called_once_with(url)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("content_id", "expected_index"), [("1", 1), ("Quick Select 2", 2)]
|
|
)
|
|
async def test_play_media_quick_select(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
content_id: str,
|
|
expected_index: int,
|
|
) -> None:
|
|
"""Test the play media service with type quick_select."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
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: content_id,
|
|
},
|
|
blocking=True,
|
|
)
|
|
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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: "quick_select",
|
|
ATTR_MEDIA_CONTENT_ID: "Invalid",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert player.play_quick_select.call_count == 0
|
|
|
|
|
|
@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: MockConfigEntry,
|
|
controller: Heos,
|
|
playlists: list[MediaItem],
|
|
enqueue: Any,
|
|
criteria: AddCriteriaType,
|
|
) -> None:
|
|
"""Test the play media service with type playlist."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
playlist = playlists[0]
|
|
service_data = {
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
|
|
ATTR_MEDIA_CONTENT_ID: playlist.name,
|
|
}
|
|
if enqueue is not None:
|
|
service_data[ATTR_MEDIA_ENQUEUE] = enqueue
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_PLAY_MEDIA,
|
|
service_data,
|
|
blocking=True,
|
|
)
|
|
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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: MediaType.PLAYLIST,
|
|
ATTR_MEDIA_CONTENT_ID: "Invalid",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert player.add_to_queue.call_count == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("content_id", "expected_index"), [("1", 1), ("Classical MPR (Classical Music)", 2)]
|
|
)
|
|
async def test_play_media_favorite(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
content_id: str,
|
|
expected_index: int,
|
|
) -> None:
|
|
"""Test the play media service with type favorite."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player = controller.players[1]
|
|
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: content_id,
|
|
},
|
|
blocking=True,
|
|
)
|
|
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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: "favorite",
|
|
ATTR_MEDIA_CONTENT_ID: "Invalid",
|
|
},
|
|
blocking=True,
|
|
)
|
|
assert player.play_preset_station.call_count == 0
|
|
|
|
|
|
async def test_play_media_invalid_type(
|
|
hass: HomeAssistant, config_entry: MockConfigEntry
|
|
) -> None:
|
|
"""Test the play media service with an invalid type."""
|
|
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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_MEDIA_CONTENT_TYPE: "Other",
|
|
ATTR_MEDIA_CONTENT_ID: "",
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("members", "expected"),
|
|
[
|
|
(["media_player.test_player_2"], [1, 2]),
|
|
(["media_player.test_player_2", "media_player.test_player"], [1, 2]),
|
|
(["media_player.test_player"], [1]),
|
|
],
|
|
)
|
|
async def test_media_player_join_group(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
members: list[str],
|
|
expected: tuple[int, list[int]],
|
|
) -> None:
|
|
"""Test grouping of media players through the join 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_JOIN,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_GROUP_MEMBERS: members,
|
|
},
|
|
blocking=True,
|
|
)
|
|
controller.set_group.assert_called_once_with(expected)
|
|
|
|
|
|
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.set_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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_GROUP_MEMBERS: ["media_player.test_player_2"],
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
async def test_media_player_group_members(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test group_members attribute."""
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
player_entity = hass.states.get("media_player.test_player")
|
|
assert player_entity.attributes[ATTR_GROUP_MEMBERS] == [
|
|
"media_player.test_player",
|
|
"media_player.test_player_2",
|
|
]
|
|
controller.get_groups.assert_called_once()
|
|
assert "Unable to get HEOS group info" not in caplog.text
|
|
|
|
|
|
async def test_media_player_group_members_error(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test error in HEOS API."""
|
|
controller.get_groups.side_effect = HeosError("error")
|
|
controller._groups = {}
|
|
config_entry.add_to_hass(hass)
|
|
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
|
assert "Unable to retrieve groups" in caplog.text
|
|
player_entity = hass.states.get("media_player.test_player")
|
|
assert player_entity.attributes[ATTR_GROUP_MEMBERS] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("entity_id", "expected_args"),
|
|
[("media_player.test_player", [1]), ("media_player.test_player_2", [1])],
|
|
)
|
|
async def test_media_player_unjoin_group(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
entity_id: str,
|
|
expected_args: list[int],
|
|
) -> None:
|
|
"""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,
|
|
{
|
|
ATTR_ENTITY_ID: entity_id,
|
|
},
|
|
blocking=True,
|
|
)
|
|
controller.set_group.assert_called_once_with(expected_args)
|
|
|
|
|
|
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.set_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,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
},
|
|
blocking=True,
|
|
)
|
|
|
|
|
|
async def test_media_player_group_fails_when_entity_removed(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
entity_registry: er.EntityRegistry,
|
|
) -> None:
|
|
"""Test grouping fails when entity removed."""
|
|
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")
|
|
|
|
# Attempt to group
|
|
with pytest.raises(ServiceValidationError, match="was not found"):
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_JOIN,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_GROUP_MEMBERS: ["media_player.test_player_2"],
|
|
},
|
|
blocking=True,
|
|
)
|
|
controller.set_group.assert_not_called()
|
|
|
|
|
|
async def test_media_player_group_fails_wrong_integration(
|
|
hass: HomeAssistant,
|
|
config_entry: MockConfigEntry,
|
|
controller: Heos,
|
|
entity_registry: er.EntityRegistry,
|
|
) -> None:
|
|
"""Test grouping fails when trying to join from the wrong integration."""
|
|
config_entry.add_to_hass(hass)
|
|
await hass.config_entries.async_setup(config_entry.entry_id)
|
|
|
|
# Create an entity in another integration
|
|
entry = entity_registry.async_get_or_create(
|
|
"media_player", "Other", "test_player_2"
|
|
)
|
|
|
|
# Attempt to group
|
|
with pytest.raises(
|
|
ServiceValidationError, match="is not a HEOS media player entity"
|
|
):
|
|
await hass.services.async_call(
|
|
MEDIA_PLAYER_DOMAIN,
|
|
SERVICE_JOIN,
|
|
{
|
|
ATTR_ENTITY_ID: "media_player.test_player",
|
|
ATTR_GROUP_MEMBERS: [entry.entity_id],
|
|
},
|
|
blocking=True,
|
|
)
|
|
controller.set_group.assert_not_called()
|