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

Bump pyheos to 1.0.0 (#135415)

Browse Source
This commit is contained in:
Andrew Sayre 2025-01-11 23:06:06 -06:00 committed by GitHub
parent 52c57eb2e5
commit 11fa6b2e4e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 217 additions and 227 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,6 +6,7 @@ import asyncio
from dataclasses import dataclass from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
from pyheos import ( from pyheos import (
Credentials, Credentials,
@ -13,6 +14,8 @@ from pyheos import (
HeosError, HeosError,
HeosOptions, HeosOptions,
HeosPlayer, HeosPlayer,
PlayerUpdateResult,
SignalHeosEvent,
const as heos_const, const as heos_const,
) )
@ -98,14 +101,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: HeosConfigEntry) -> bool
# Auth failure handler must be added before connecting to the host, otherwise # Auth failure handler must be added before connecting to the host, otherwise
# the event will be missed when login fails during connection. # the event will be missed when login fails during connection.
async def auth_failure(event: str) -> None: async def auth_failure() -> None:
"""Handle authentication failure.""" """Handle authentication failure."""
if event == heos_const.EVENT_USER_CREDENTIALS_INVALID: entry.async_start_reauth(hass)
entry.async_start_reauth(hass)
entry.async_on_unload( entry.async_on_unload(controller.add_on_user_credentials_invalid(auth_failure))
controller.dispatcher.connect(heos_const.SIGNAL_HEOS_EVENT, auth_failure)
)
try: try:
# Auto reconnect only operates if initial connection was successful. # Auto reconnect only operates if initial connection was successful.
@ -168,11 +168,11 @@ async def async_unload_entry(hass: HomeAssistant, entry: HeosConfigEntry) -> boo
class ControllerManager: class ControllerManager:
"""Class that manages events of the controller.""" """Class that manages events of the controller."""
def __init__(self, hass, controller): def __init__(self, hass: HomeAssistant, controller: Heos) -> None:
"""Init the controller manager.""" """Init the controller manager."""
self._hass = hass self._hass = hass
self._device_registry = None self._device_registry: dr.DeviceRegistry | None = None
self._entity_registry = None self._entity_registry: er.EntityRegistry | None = None
self.controller = controller self.controller = controller
async def connect_listeners(self): async def connect_listeners(self):
@ -181,56 +181,59 @@ class ControllerManager:
self._entity_registry = er.async_get(self._hass) self._entity_registry = er.async_get(self._hass)
# Handle controller events # Handle controller events
self.controller.dispatcher.connect( self.controller.add_on_controller_event(self._controller_event)
heos_const.SIGNAL_CONTROLLER_EVENT, self._controller_event
)
# Handle connection-related events # Handle connection-related events
self.controller.dispatcher.connect( self.controller.add_on_heos_event(self._heos_event)
heos_const.SIGNAL_HEOS_EVENT, self._heos_event
)
async def disconnect(self): async def disconnect(self):
"""Disconnect subscriptions.""" """Disconnect subscriptions."""
self.controller.dispatcher.disconnect_all() self.controller.dispatcher.disconnect_all()
await self.controller.disconnect() await self.controller.disconnect()
async def _controller_event(self, event, data): async def _controller_event(
self, event: str, data: PlayerUpdateResult | None
) -> None:
"""Handle controller event.""" """Handle controller event."""
if event == heos_const.EVENT_PLAYERS_CHANGED: if event == heos_const.EVENT_PLAYERS_CHANGED:
self.update_ids(data[heos_const.DATA_MAPPED_IDS]) assert data is not None
self.update_ids(data.updated_player_ids)
# Update players # Update players
async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED) async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED)
async def _heos_event(self, event): async def _heos_event(self, event):
"""Handle connection event.""" """Handle connection event."""
if event == heos_const.EVENT_CONNECTED: if event == SignalHeosEvent.CONNECTED:
try: try:
# Retrieve latest players and refresh status # Retrieve latest players and refresh status
data = await self.controller.load_players() data = await self.controller.load_players()
self.update_ids(data[heos_const.DATA_MAPPED_IDS]) self.update_ids(data.updated_player_ids)
except HeosError as ex: except HeosError as ex:
_LOGGER.error("Unable to refresh players: %s", ex) _LOGGER.error("Unable to refresh players: %s", ex)
# Update players # Update players
_LOGGER.debug("HEOS Controller event called, calling dispatcher")
async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED) async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED)
def update_ids(self, mapped_ids: dict[int, int]): def update_ids(self, mapped_ids: dict[int, int]):
"""Update the IDs in the device and entity registry.""" """Update the IDs in the device and entity registry."""
# mapped_ids contains the mapped IDs (new:old) # mapped_ids contains the mapped IDs (new:old)
for new_id, old_id in mapped_ids.items(): for old_id, new_id in mapped_ids.items():
# update device registry # update device registry
assert self._device_registry is not None
entry = self._device_registry.async_get_device( entry = self._device_registry.async_get_device(
identifiers={(DOMAIN, old_id)} identifiers={(DOMAIN, old_id)} # type: ignore[arg-type] # Fix in the future
) )
new_identifiers = {(DOMAIN, new_id)} new_identifiers = {(DOMAIN, new_id)}
if entry: if entry:
self._device_registry.async_update_device( self._device_registry.async_update_device(
entry.id, new_identifiers=new_identifiers entry.id,
new_identifiers=new_identifiers, # type: ignore[arg-type] # Fix in the future
) )
_LOGGER.debug( _LOGGER.debug(
"Updated device %s identifiers to %s", entry.id, new_identifiers "Updated device %s identifiers to %s", entry.id, new_identifiers
) )
# update entity registry # update entity registry
assert self._entity_registry is not None
entity_id = self._entity_registry.async_get_entity_id( entity_id = self._entity_registry.async_get_entity_id(
Platform.MEDIA_PLAYER, DOMAIN, str(old_id) Platform.MEDIA_PLAYER, DOMAIN, str(old_id)
) )
@ -249,7 +252,7 @@ class GroupManager:
) -> None: ) -> None:
"""Init group manager.""" """Init group manager."""
self._hass = hass self._hass = hass
self._group_membership: dict[str, str] = {} self._group_membership: dict[str, list[str]] = {}
self._disconnect_player_added = None self._disconnect_player_added = None
self._initialized = False self._initialized = False
self.controller = controller self.controller = controller
@ -268,7 +271,7 @@ class GroupManager:
} }
try: try:
groups = await self.controller.get_groups(refresh=True) groups = await self.controller.get_groups()
except HeosError as err: except HeosError as err:
_LOGGER.error("Unable to get HEOS group info: %s", err) _LOGGER.error("Unable to get HEOS group info: %s", err)
return group_info_by_entity_id return group_info_by_entity_id
@ -326,29 +329,26 @@ class GroupManager:
err, err,
) )
async def async_update_groups(self, event, data=None): async def async_update_groups(self) -> None:
"""Update the group membership from the controller.""" """Update the group membership from the controller."""
if event in ( if groups := await self.async_get_group_membership():
heos_const.EVENT_GROUPS_CHANGED, self._group_membership = groups
heos_const.EVENT_CONNECTED, _LOGGER.debug("Groups updated due to change event")
SIGNAL_HEOS_PLAYER_ADDED, # Let players know to update
): async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED)
if groups := await self.async_get_group_membership(): else:
self._group_membership = groups _LOGGER.debug("Groups empty")
_LOGGER.debug("Groups updated due to change event")
# Let players know to update
async_dispatcher_send(self._hass, SIGNAL_HEOS_UPDATED)
else:
_LOGGER.debug("Groups empty")
@callback
def connect_update(self): def connect_update(self):
"""Connect listener for when groups change and signal player update.""" """Connect listener for when groups change and signal player update."""
self.controller.dispatcher.connect(
heos_const.SIGNAL_CONTROLLER_EVENT, self.async_update_groups async def _on_controller_event(event: str, data: Any | None) -> None:
) if event == heos_const.EVENT_GROUPS_CHANGED:
self.controller.dispatcher.connect( await self.async_update_groups()
heos_const.SIGNAL_HEOS_EVENT, self.async_update_groups
) self.controller.add_on_controller_event(_on_controller_event)
self.controller.add_on_connected(self.async_update_groups)
# When adding a new HEOS player we need to update the groups. # When adding a new HEOS player we need to update the groups.
async def _async_handle_player_added(): async def _async_handle_player_added():
@ -356,7 +356,7 @@ class GroupManager:
# fully populated yet. This may only happen during early startup. # fully populated yet. This may only happen during early startup.
if len(self.players) <= len(self.entity_id_map) and not self._initialized: if len(self.players) <= len(self.entity_id_map) and not self._initialized:
self._initialized = True self._initialized = True
await self.async_update_groups(SIGNAL_HEOS_PLAYER_ADDED) await self.async_update_groups()
self._disconnect_player_added = async_dispatcher_connect( self._disconnect_player_added = async_dispatcher_connect(
self._hass, SIGNAL_HEOS_PLAYER_ADDED, _async_handle_player_added self._hass, SIGNAL_HEOS_PLAYER_ADDED, _async_handle_player_added
@ -462,7 +462,8 @@ class SourceManager:
None, None,
) )
def connect_update(self, hass, controller): @callback
def connect_update(self, hass: HomeAssistant, controller: Heos) -> None:
"""Connect listener for when sources change and signal player update. """Connect listener for when sources change and signal player update.
EVENT_SOURCES_CHANGED is often raised multiple times in response to a EVENT_SOURCES_CHANGED is often raised multiple times in response to a
@ -492,21 +493,22 @@ class SourceManager:
else: else:
return favorites, inputs return favorites, inputs
async def update_sources(event, data=None): async def _update_sources() -> None:
# If throttled, it will return None
if sources := await get_sources():
self.favorites, self.inputs = sources
self.source_list = self._build_source_list()
_LOGGER.debug("Sources updated due to changed event")
# Let players know to update
async_dispatcher_send(hass, SIGNAL_HEOS_UPDATED)
async def _on_controller_event(event: str, data: Any | None) -> None:
if event in ( if event in (
heos_const.EVENT_SOURCES_CHANGED, heos_const.EVENT_SOURCES_CHANGED,
heos_const.EVENT_USER_CHANGED, heos_const.EVENT_USER_CHANGED,
heos_const.EVENT_CONNECTED,
): ):
# If throttled, it will return None await _update_sources()
if sources := await get_sources():
self.favorites, self.inputs = sources
self.source_list = self._build_source_list()
_LOGGER.debug("Sources updated due to changed event")
# Let players know to update
async_dispatcher_send(hass, SIGNAL_HEOS_UPDATED)
controller.dispatcher.connect( controller.add_on_connected(_update_sources)
heos_const.SIGNAL_CONTROLLER_EVENT, update_sources controller.add_on_user_credentials_invalid(_update_sources)
) controller.add_on_controller_event(_on_controller_event)
controller.dispatcher.connect(heos_const.SIGNAL_HEOS_EVENT, update_sources)

12
homeassistant/components/heos/config_flow.py
View File

@ -5,7 +5,7 @@ import logging
from typing import TYPE_CHECKING, Any, cast from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlparse from urllib.parse import urlparse
from pyheos import CommandFailedError, Heos, HeosError, HeosOptions from pyheos import CommandAuthenticationError, Heos, HeosError, HeosOptions
import voluptuous as vol import voluptuous as vol
from homeassistant.components import ssdp from homeassistant.components import ssdp
@ -79,13 +79,9 @@ async def _validate_auth(
# Attempt to login (both username and password provided) # Attempt to login (both username and password provided)
try: try:
await heos.sign_in(user_input[CONF_USERNAME], user_input[CONF_PASSWORD]) await heos.sign_in(user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
except CommandFailedError as err: except CommandAuthenticationError as err:
if err.error_id in (6, 8, 10): # Auth-specific errors errors["base"] = "invalid_auth"
errors["base"] = "invalid_auth" _LOGGER.warning("Failed to sign-in to HEOS Account: %s", err)
_LOGGER.warning("Failed to sign-in to HEOS Account: %s", err)
else:
errors["base"] = "unknown"
_LOGGER.exception("Unexpected error occurred during sign-in")
return False return False
except HeosError: except HeosError:
errors["base"] = "unknown" errors["base"] = "unknown"

2
homeassistant/components/heos/manifest.json
View File

@ -6,7 +6,7 @@
"documentation": "https://www.home-assistant.io/integrations/heos", "documentation": "https://www.home-assistant.io/integrations/heos",
"iot_class": "local_push", "iot_class": "local_push",
"loggers": ["pyheos"], "loggers": ["pyheos"],
"requirements": ["pyheos==0.9.0"], "requirements": ["pyheos==1.0.0"],
"single_config_entry": true, "single_config_entry": true,
"ssdp": [ "ssdp": [
{ {

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

@ -8,7 +8,14 @@ import logging
from operator import ior from operator import ior
from typing import Any from typing import Any
from pyheos import HeosError, const as heos_const from pyheos import (
AddCriteriaType,
ControlType,
HeosError,
HeosPlayer,
PlayState,
const as heos_const,
)
from homeassistant.components import media_source from homeassistant.components import media_source
from homeassistant.components.media_player import ( from homeassistant.components.media_player import (
@ -47,25 +54,25 @@ BASE_SUPPORTED_FEATURES = (
) )
PLAY_STATE_TO_STATE = { PLAY_STATE_TO_STATE = {
heos_const.PlayState.PLAY: MediaPlayerState.PLAYING, PlayState.PLAY: MediaPlayerState.PLAYING,
heos_const.PlayState.STOP: MediaPlayerState.IDLE, PlayState.STOP: MediaPlayerState.IDLE,
heos_const.PlayState.PAUSE: MediaPlayerState.PAUSED, PlayState.PAUSE: MediaPlayerState.PAUSED,
} }
CONTROL_TO_SUPPORT = { CONTROL_TO_SUPPORT = {
heos_const.CONTROL_PLAY: MediaPlayerEntityFeature.PLAY, ControlType.PLAY: MediaPlayerEntityFeature.PLAY,
heos_const.CONTROL_PAUSE: MediaPlayerEntityFeature.PAUSE, ControlType.PAUSE: MediaPlayerEntityFeature.PAUSE,
heos_const.CONTROL_STOP: MediaPlayerEntityFeature.STOP, ControlType.STOP: MediaPlayerEntityFeature.STOP,
heos_const.CONTROL_PLAY_PREVIOUS: MediaPlayerEntityFeature.PREVIOUS_TRACK, ControlType.PLAY_PREVIOUS: MediaPlayerEntityFeature.PREVIOUS_TRACK,
heos_const.CONTROL_PLAY_NEXT: MediaPlayerEntityFeature.NEXT_TRACK, ControlType.PLAY_NEXT: MediaPlayerEntityFeature.NEXT_TRACK,
} }
HA_HEOS_ENQUEUE_MAP = { HA_HEOS_ENQUEUE_MAP = {
None: heos_const.AddCriteriaType.REPLACE_AND_PLAY, None: AddCriteriaType.REPLACE_AND_PLAY,
MediaPlayerEnqueue.ADD: heos_const.AddCriteriaType.ADD_TO_END, MediaPlayerEnqueue.ADD: AddCriteriaType.ADD_TO_END,
MediaPlayerEnqueue.REPLACE: heos_const.AddCriteriaType.REPLACE_AND_PLAY, MediaPlayerEnqueue.REPLACE: AddCriteriaType.REPLACE_AND_PLAY,
MediaPlayerEnqueue.NEXT: heos_const.AddCriteriaType.PLAY_NEXT, MediaPlayerEnqueue.NEXT: AddCriteriaType.PLAY_NEXT,
MediaPlayerEnqueue.PLAY: heos_const.AddCriteriaType.PLAY_NOW, MediaPlayerEnqueue.PLAY: AddCriteriaType.PLAY_NOW,
} }
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -118,11 +125,14 @@ class HeosMediaPlayer(MediaPlayerEntity):
_attr_name = None _attr_name = None
def __init__( def __init__(
self, player, source_manager: SourceManager, group_manager: GroupManager self,
player: HeosPlayer,
source_manager: SourceManager,
group_manager: GroupManager,
) -> None: ) -> None:
"""Initialize.""" """Initialize."""
self._media_position_updated_at = None self._media_position_updated_at = None
self._player = player self._player: HeosPlayer = player
self._source_manager = source_manager self._source_manager = source_manager
self._group_manager = group_manager self._group_manager = group_manager
self._attr_unique_id = str(player.player_id) self._attr_unique_id = str(player.player_id)
@ -134,10 +144,8 @@ class HeosMediaPlayer(MediaPlayerEntity):
sw_version=player.version, sw_version=player.version,
) )
async def _player_update(self, player_id, event): async def _player_update(self, event):
"""Handle player attribute updated.""" """Handle player attribute updated."""
if self._player.player_id != player_id:
return
if event == heos_const.EVENT_PLAYER_NOW_PLAYING_PROGRESS: if event == heos_const.EVENT_PLAYER_NOW_PLAYING_PROGRESS:
self._media_position_updated_at = utcnow() self._media_position_updated_at = utcnow()
await self.async_update_ha_state(True) await self.async_update_ha_state(True)
@ -149,11 +157,7 @@ class HeosMediaPlayer(MediaPlayerEntity):
async def async_added_to_hass(self) -> None: async def async_added_to_hass(self) -> None:
"""Device added to hass.""" """Device added to hass."""
# Update state when attributes of the player change # Update state when attributes of the player change
self.async_on_remove( self.async_on_remove(self._player.add_on_player_event(self._player_update))
self._player.heos.dispatcher.connect(
heos_const.SIGNAL_PLAYER_EVENT, self._player_update
)
)
# Update state when heos changes # Update state when heos changes
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect(self.hass, SIGNAL_HEOS_UPDATED, self._heos_updated) async_dispatcher_connect(self.hass, SIGNAL_HEOS_UPDATED, self._heos_updated)

11
homeassistant/components/heos/services.py
View File

@ -2,7 +2,7 @@
import logging import logging
from pyheos import CommandFailedError, Heos, HeosError, const from pyheos import CommandAuthenticationError, Heos, HeosError
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntryState from homeassistant.config_entries import ConfigEntryState
@ -69,16 +69,12 @@ def _get_controller(hass: HomeAssistant) -> Heos:
async def _sign_in_handler(service: ServiceCall) -> None: async def _sign_in_handler(service: ServiceCall) -> None:
"""Sign in to the HEOS account.""" """Sign in to the HEOS account."""
controller = _get_controller(service.hass) controller = _get_controller(service.hass)
if controller.connection_state != const.STATE_CONNECTED:
_LOGGER.error("Unable to sign in because HEOS is not connected")
return
username = service.data[ATTR_USERNAME] username = service.data[ATTR_USERNAME]
password = service.data[ATTR_PASSWORD] password = service.data[ATTR_PASSWORD]
try: try:
await controller.sign_in(username, password) await controller.sign_in(username, password)
except CommandFailedError as err: except CommandAuthenticationError as err:
_LOGGER.error("Sign in failed: %s", err) _LOGGER.error("Sign in failed: %s", err)
except HeosError as err: except HeosError as err:
_LOGGER.error("Unable to sign in: %s", err) _LOGGER.error("Unable to sign in: %s", err)
@ -88,9 +84,6 @@ async def _sign_out_handler(service: ServiceCall) -> None:
"""Sign out of the HEOS account.""" """Sign out of the HEOS account."""
controller = _get_controller(service.hass) controller = _get_controller(service.hass)
if controller.connection_state != const.STATE_CONNECTED:
_LOGGER.error("Unable to sign out because HEOS is not connected")
return
try: try:
await controller.sign_out() await controller.sign_out()
except HeosError as err: except HeosError as err:

2
requirements_all.txt generated
View File

@ -1980,7 +1980,7 @@ pygti==0.9.4
pyhaversion==22.8.0 pyhaversion==22.8.0
# homeassistant.components.heos # homeassistant.components.heos
pyheos==0.9.0 pyheos==1.0.0
# homeassistant.components.hive # homeassistant.components.hive
pyhiveapi==0.5.16 pyhiveapi==0.5.16

2
requirements_test_all.txt generated
View File

@ -1609,7 +1609,7 @@ pygti==0.9.4
pyhaversion==22.8.0 pyhaversion==22.8.0
# homeassistant.components.heos # homeassistant.components.heos
pyheos==0.9.0 pyheos==1.0.0
# homeassistant.components.hive # homeassistant.components.hive
pyhiveapi==0.5.16 pyhiveapi==0.5.16

122
tests/components/heos/conftest.py
View File

@ -3,9 +3,24 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from unittest.mock import Mock, patch from unittest.mock import AsyncMock, Mock, patch
from pyheos import Dispatcher, Heos, HeosGroup, HeosPlayer, MediaItem, const from pyheos import (
CONTROLS_ALL,
Dispatcher,
Heos,
HeosGroup,
HeosOptions,
HeosPlayer,
LineOutLevelType,
MediaItem,
MediaType,
NetworkType,
PlayerUpdateResult,
PlayState,
RepeatType,
const,
)
import pytest import pytest
import pytest_asyncio import pytest_asyncio
@ -71,26 +86,27 @@ def controller_fixture(
players, favorites, input_sources, playlists, change_data, dispatcher, group players, favorites, input_sources, playlists, change_data, dispatcher, group
): ):
"""Create a mock Heos controller fixture.""" """Create a mock Heos controller fixture."""
mock_heos = Mock(Heos) mock_heos = Heos(HeosOptions(host="127.0.0.1", dispatcher=dispatcher))
for player in players.values(): for player in players.values():
player.heos = mock_heos player.heos = mock_heos
mock_heos.return_value = mock_heos mock_heos.connect = AsyncMock()
mock_heos.dispatcher = dispatcher mock_heos.disconnect = AsyncMock()
mock_heos.get_players.return_value = players mock_heos.sign_in = AsyncMock()
mock_heos.players = players mock_heos.sign_out = AsyncMock()
mock_heos.get_favorites.return_value = favorites mock_heos.get_players = AsyncMock(return_value=players)
mock_heos.get_input_sources.return_value = input_sources mock_heos._players = players
mock_heos.get_playlists.return_value = playlists mock_heos.get_favorites = AsyncMock(return_value=favorites)
mock_heos.load_players.return_value = change_data mock_heos.get_input_sources = AsyncMock(return_value=input_sources)
mock_heos.is_signed_in = True mock_heos.get_playlists = AsyncMock(return_value=playlists)
mock_heos.signed_in_username = "user@user.com" mock_heos.load_players = AsyncMock(return_value=change_data)
mock_heos.connection_state = const.STATE_CONNECTED mock_heos._signed_in_username = "user@user.com"
mock_heos.get_groups.return_value = group mock_heos.get_groups = AsyncMock(return_value=group)
mock_heos.create_group.return_value = None mock_heos.create_group = AsyncMock(return_value=None)
new_mock = Mock(return_value=mock_heos)
mock_heos.new_mock = new_mock
with ( with (
patch("homeassistant.components.heos.Heos", new=mock_heos), patch("homeassistant.components.heos.Heos", new=new_mock),
patch("homeassistant.components.heos.config_flow.Heos", new=mock_heos), patch("homeassistant.components.heos.config_flow.Heos", new=new_mock),
): ):
yield mock_heos yield mock_heos
@ -106,24 +122,25 @@ def player_fixture(quick_selects):
"""Create two mock HeosPlayers.""" """Create two mock HeosPlayers."""
players = {} players = {}
for i in (1, 2): for i in (1, 2):
player = Mock(HeosPlayer) player = HeosPlayer(
player.player_id = i player_id=i,
if i > 1: name="Test Player" if i == 1 else f"Test Player {i}",
player.name = f"Test Player {i}" model="Test Model",
else: serial="",
player.name = "Test Player" version="1.0.0",
player.model = "Test Model" line_out=LineOutLevelType.VARIABLE,
player.version = "1.0.0" is_muted=False,
player.is_muted = False available=True,
player.available = True state=PlayState.STOP,
player.state = const.PlayState.STOP ip_address=f"127.0.0.{i}",
player.ip_address = f"127.0.0.{i}" network=NetworkType.WIRED,
player.network = "wired" shuffle=False,
player.shuffle = False repeat=RepeatType.OFF,
player.repeat = const.RepeatType.OFF volume=25,
player.volume = 25 heos=None,
)
player.now_playing_media = Mock() player.now_playing_media = Mock()
player.now_playing_media.supported_controls = const.CONTROLS_ALL player.now_playing_media.supported_controls = CONTROLS_ALL
player.now_playing_media.album_id = 1 player.now_playing_media.album_id = 1
player.now_playing_media.queue_id = 1 player.now_playing_media.queue_id = 1
player.now_playing_media.source_id = 1 player.now_playing_media.source_id = 1
@ -136,13 +153,30 @@ def player_fixture(quick_selects):
player.now_playing_media.current_position = None player.now_playing_media.current_position = None
player.now_playing_media.image_url = "http://" player.now_playing_media.image_url = "http://"
player.now_playing_media.song = "Song" player.now_playing_media.song = "Song"
player.get_quick_selects.return_value = quick_selects player.add_to_queue = AsyncMock()
player.clear_queue = AsyncMock()
player.get_quick_selects = AsyncMock(return_value=quick_selects)
player.mute = AsyncMock()
player.pause = AsyncMock()
player.play = AsyncMock()
player.play_input_source = AsyncMock()
player.play_next = AsyncMock()
player.play_previous = AsyncMock()
player.play_preset_station = AsyncMock()
player.play_quick_select = AsyncMock()
player.play_url = AsyncMock()
player.set_mute = AsyncMock()
player.set_play_mode = AsyncMock()
player.set_quick_select = AsyncMock()
player.set_volume = AsyncMock()
player.stop = AsyncMock()
player.unmute = AsyncMock()
players[player.player_id] = player players[player.player_id] = player
return players return players
@pytest.fixture(name="group") @pytest.fixture(name="group")
def group_fixture(players): def group_fixture():
"""Create a HEOS group consisting of two players.""" """Create a HEOS group consisting of two players."""
group = HeosGroup( group = HeosGroup(
name="Group", group_id=999, lead_player_id=1, member_player_ids=[2] name="Group", group_id=999, lead_player_id=1, member_player_ids=[2]
@ -158,7 +192,7 @@ def favorites_fixture() -> dict[int, MediaItem]:
source_id=const.MUSIC_SOURCE_PANDORA, source_id=const.MUSIC_SOURCE_PANDORA,
name="Today's Hits Radio", name="Today's Hits Radio",
media_id="123456789", media_id="123456789",
type=const.MediaType.STATION, type=MediaType.STATION,
playable=True, playable=True,
browsable=False, browsable=False,
image_url="", image_url="",
@ -168,7 +202,7 @@ def favorites_fixture() -> dict[int, MediaItem]:
source_id=const.MUSIC_SOURCE_TUNEIN, source_id=const.MUSIC_SOURCE_TUNEIN,
name="Classical MPR (Classical Music)", name="Classical MPR (Classical Music)",
media_id="s1234", media_id="s1234",
type=const.MediaType.STATION, type=MediaType.STATION,
playable=True, playable=True,
browsable=False, browsable=False,
image_url="", image_url="",
@ -184,7 +218,7 @@ def input_sources_fixture() -> Sequence[MediaItem]:
source_id=1, source_id=1,
name="HEOS Drive - Line In 1", name="HEOS Drive - Line In 1",
media_id=const.INPUT_AUX_IN_1, media_id=const.INPUT_AUX_IN_1,
type=const.MediaType.STATION, type=MediaType.STATION,
playable=True, playable=True,
browsable=False, browsable=False,
image_url="", image_url="",
@ -256,7 +290,7 @@ def playlists_fixture() -> Sequence[MediaItem]:
playlist = MediaItem( playlist = MediaItem(
source_id=const.MUSIC_SOURCE_PLAYLISTS, source_id=const.MUSIC_SOURCE_PLAYLISTS,
name="Awesome Music", name="Awesome Music",
type=const.MediaType.PLAYLIST, type=MediaType.PLAYLIST,
playable=True, playable=True,
browsable=True, browsable=True,
image_url="", image_url="",
@ -268,10 +302,10 @@ def playlists_fixture() -> Sequence[MediaItem]:
@pytest.fixture(name="change_data") @pytest.fixture(name="change_data")
def change_data_fixture() -> dict: def change_data_fixture() -> dict:
"""Create player change data for testing.""" """Create player change data for testing."""
return {const.DATA_MAPPED_IDS: {}, const.DATA_NEW: []} return PlayerUpdateResult()
@pytest.fixture(name="change_data_mapped_ids") @pytest.fixture(name="change_data_mapped_ids")
def change_data_mapped_ids_fixture() -> dict: def change_data_mapped_ids_fixture() -> dict:
"""Create player change data for testing.""" """Create player change data for testing."""
return {const.DATA_MAPPED_IDS: {101: 1}, const.DATA_NEW: []} return PlayerUpdateResult(updated_player_ids={1: 101})

16
tests/components/heos/test_config_flow.py
View File

@ -1,6 +1,6 @@
"""Tests for the Heos config flow module.""" """Tests for the Heos config flow module."""
from pyheos import CommandFailedError, HeosError from pyheos import CommandAuthenticationError, CommandFailedError, HeosError
import pytest import pytest
from homeassistant.components import heos, ssdp from homeassistant.components import heos, ssdp
@ -199,14 +199,9 @@ async def test_reconfigure_cannot_connect_recovers(
("error", "expected_error_key"), ("error", "expected_error_key"),
[ [
( (
CommandFailedError("sign_in", "Invalid credentials", 6), CommandAuthenticationError("sign_in", "Invalid credentials", 6),
"invalid_auth", "invalid_auth",
), ),
(
CommandFailedError("sign_in", "User not logged in", 8),
"invalid_auth",
),
(CommandFailedError("sign_in", "user not found", 10), "invalid_auth"),
(CommandFailedError("sign_in", "System error", 12), "unknown"), (CommandFailedError("sign_in", "System error", 12), "unknown"),
(HeosError(), "unknown"), (HeosError(), "unknown"),
], ],
@ -337,14 +332,9 @@ async def test_options_flow_missing_one_param_recovers(
("error", "expected_error_key"), ("error", "expected_error_key"),
[ [
( (
CommandFailedError("sign_in", "Invalid credentials", 6), CommandAuthenticationError("sign_in", "Invalid credentials", 6),
"invalid_auth", "invalid_auth",
), ),
(
CommandFailedError("sign_in", "User not logged in", 8),
"invalid_auth",
),
(CommandFailedError("sign_in", "user not found", 10), "invalid_auth"),
(CommandFailedError("sign_in", "System error", 12), "unknown"), (CommandFailedError("sign_in", "System error", 12), "unknown"),
(HeosError(), "unknown"), (HeosError(), "unknown"),
], ],

14
tests/components/heos/test_init.py
View File

@ -4,7 +4,7 @@ import asyncio
from typing import cast from typing import cast
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from pyheos import CommandFailedError, HeosError, const from pyheos import CommandFailedError, HeosError, SignalHeosEvent, SignalType, const
import pytest import pytest
from homeassistant.components.heos import ( from homeassistant.components.heos import (
@ -82,7 +82,7 @@ async def test_async_setup_entry_with_options_loads_platforms(
# Assert options passed and methods called # Assert options passed and methods called
assert config_entry_options.state is ConfigEntryState.LOADED assert config_entry_options.state is ConfigEntryState.LOADED
options = cast(HeosOptions, controller.call_args[0][0]) options = cast(HeosOptions, controller.new_mock.call_args[0][0])
assert options.host == config_entry_options.data[CONF_HOST] assert options.host == config_entry_options.data[CONF_HOST]
assert options.credentials.username == config_entry_options.options[CONF_USERNAME] assert options.credentials.username == config_entry_options.options[CONF_USERNAME]
assert options.credentials.password == config_entry_options.options[CONF_PASSWORD] assert options.credentials.password == config_entry_options.options[CONF_PASSWORD]
@ -103,10 +103,9 @@ async def test_async_setup_entry_auth_failure_starts_reauth(
# Simulates what happens when the controller can't sign-in during connection # Simulates what happens when the controller can't sign-in during connection
async def connect_send_auth_failure() -> None: async def connect_send_auth_failure() -> None:
controller.is_signed_in = False controller._signed_in_username = None
controller.signed_in_username = None
controller.dispatcher.send( controller.dispatcher.send(
const.SIGNAL_HEOS_EVENT, const.EVENT_USER_CREDENTIALS_INVALID SignalType.HEOS_EVENT, SignalHeosEvent.USER_CREDENTIALS_INVALID
) )
controller.connect.side_effect = connect_send_auth_failure controller.connect.side_effect = connect_send_auth_failure
@ -133,8 +132,7 @@ async def test_async_setup_entry_not_signed_in_loads_platforms(
) -> None: ) -> None:
"""Test setup does not retrieve favorites when not logged in.""" """Test setup does not retrieve favorites when not logged in."""
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)
controller.is_signed_in = False controller._signed_in_username = None
controller.signed_in_username = None
with patch.object( with patch.object(
hass.config_entries, "async_forward_entry_setups" hass.config_entries, "async_forward_entry_setups"
) as forward_mock: ) as forward_mock:
@ -213,7 +211,7 @@ async def test_update_sources_retry(
source_manager.max_retry_attempts = 1 source_manager.max_retry_attempts = 1
controller.get_favorites.side_effect = CommandFailedError("Test", "test", 0) controller.get_favorites.side_effect = CommandFailedError("Test", "test", 0)
controller.dispatcher.send( controller.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {} SignalType.CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {}
) )
# Wait until it's finished # Wait until it's finished
while "Unable to update sources" not in caplog.text: while "Unable to update sources" not in caplog.text:

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

@ -3,8 +3,15 @@
import asyncio import asyncio
from typing import Any from typing import Any
from pyheos import CommandFailedError, const from pyheos import (
from pyheos.error import HeosError AddCriteriaType,
CommandFailedError,
HeosError,
PlayState,
SignalHeosEvent,
SignalType,
const,
)
import pytest import pytest
from homeassistant.components.heos import media_player from homeassistant.components.heos import media_player
@ -115,18 +122,18 @@ async def test_updates_from_signals(
player = controller.players[1] player = controller.players[1]
# Test player does not update for other players # Test player does not update for other players
player.state = const.PlayState.PLAY player.state = PlayState.PLAY
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, 2, const.EVENT_PLAYER_STATE_CHANGED SignalType.PLAYER_EVENT, 2, const.EVENT_PLAYER_STATE_CHANGED
) )
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
assert state.state == STATE_IDLE assert state.state == STATE_IDLE
# Test player_update standard events # Test player_update standard events
player.state = const.PlayState.PLAY player.state = PlayState.PLAY
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
) )
await hass.async_block_till_done() await hass.async_block_till_done()
@ -137,7 +144,7 @@ async def test_updates_from_signals(
player.now_playing_media.duration = 360000 player.now_playing_media.duration = 360000
player.now_playing_media.current_position = 1000 player.now_playing_media.current_position = 1000
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, SignalType.PLAYER_EVENT,
player.player_id, player.player_id,
const.EVENT_PLAYER_NOW_PLAYING_PROGRESS, const.EVENT_PLAYER_NOW_PLAYING_PROGRESS,
) )
@ -167,7 +174,7 @@ async def test_updates_from_connection_event(
# Connected # Connected
player.available = True player.available = True
player.heos.dispatcher.send(const.SIGNAL_HEOS_EVENT, const.EVENT_CONNECTED) player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED)
await event.wait() await event.wait()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
assert state.state == STATE_IDLE assert state.state == STATE_IDLE
@ -175,10 +182,9 @@ async def test_updates_from_connection_event(
# Disconnected # Disconnected
event.clear() event.clear()
player.reset_mock()
controller.load_players.reset_mock() controller.load_players.reset_mock()
player.available = False player.available = False
player.heos.dispatcher.send(const.SIGNAL_HEOS_EVENT, const.EVENT_DISCONNECTED) player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.DISCONNECTED)
await event.wait() await event.wait()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
assert state.state == STATE_UNAVAILABLE assert state.state == STATE_UNAVAILABLE
@ -186,11 +192,10 @@ async def test_updates_from_connection_event(
# Connected handles refresh failure # Connected handles refresh failure
event.clear() event.clear()
player.reset_mock()
controller.load_players.reset_mock() controller.load_players.reset_mock()
controller.load_players.side_effect = CommandFailedError(None, "Failure", 1) controller.load_players.side_effect = CommandFailedError(None, "Failure", 1)
player.available = True player.available = True
player.heos.dispatcher.send(const.SIGNAL_HEOS_EVENT, const.EVENT_CONNECTED) player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED)
await event.wait() await event.wait()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
assert state.state == STATE_IDLE assert state.state == STATE_IDLE
@ -213,7 +218,7 @@ async def test_updates_from_sources_updated(
input_sources.clear() input_sources.clear()
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {} SignalType.CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {}
) )
await event.wait() await event.wait()
source_list = config_entry.runtime_data.source_manager.source_list source_list = config_entry.runtime_data.source_manager.source_list
@ -241,9 +246,9 @@ async def test_updates_from_players_changed(
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal) async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
assert hass.states.get("media_player.test_player").state == STATE_IDLE assert hass.states.get("media_player.test_player").state == STATE_IDLE
player.state = const.PlayState.PLAY player.state = PlayState.PLAY
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_PLAYERS_CHANGED, change_data SignalType.CONTROLLER_EVENT, const.EVENT_PLAYERS_CHANGED, change_data
) )
await event.wait() await event.wait()
await hass.async_block_till_done() await hass.async_block_till_done()
@ -279,7 +284,7 @@ async def test_updates_from_players_changed_new_ids(
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal) async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, SignalType.CONTROLLER_EVENT,
const.EVENT_PLAYERS_CHANGED, const.EVENT_PLAYERS_CHANGED,
change_data_mapped_ids, change_data_mapped_ids,
) )
@ -309,10 +314,9 @@ async def test_updates_from_user_changed(
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal) async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
controller.is_signed_in = False controller._signed_in_username = None
controller.signed_in_username = None
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_CONTROLLER_EVENT, const.EVENT_USER_CHANGED, None SignalType.CONTROLLER_EVENT, const.EVENT_USER_CHANGED, None
) )
await event.wait() await event.wait()
source_list = config_entry.runtime_data.source_manager.source_list source_list = config_entry.runtime_data.source_manager.source_list
@ -555,7 +559,7 @@ async def test_select_favorite(
# Test state is matched by station name # Test state is matched by station name
player.now_playing_media.station = favorite.name player.now_playing_media.station = favorite.name
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
) )
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
@ -581,7 +585,7 @@ async def test_select_radio_favorite(
player.now_playing_media.station = "Classical" player.now_playing_media.station = "Classical"
player.now_playing_media.album_id = favorite.media_id player.now_playing_media.album_id = favorite.media_id
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
) )
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
@ -634,7 +638,7 @@ async def test_select_input_source(
player.now_playing_media.source_id = const.MUSIC_SOURCE_AUX_INPUT player.now_playing_media.source_id = const.MUSIC_SOURCE_AUX_INPUT
player.now_playing_media.media_id = const.INPUT_AUX_IN_1 player.now_playing_media.media_id = const.INPUT_AUX_IN_1
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
) )
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get("media_player.test_player") state = hass.states.get("media_player.test_player")
@ -831,7 +835,7 @@ async def test_play_media_playlist(
blocking=True, blocking=True,
) )
player.add_to_queue.assert_called_once_with( player.add_to_queue.assert_called_once_with(
playlist, const.AddCriteriaType.REPLACE_AND_PLAY playlist, AddCriteriaType.REPLACE_AND_PLAY
) )
# Play with enqueuing # Play with enqueuing
player.add_to_queue.reset_mock() player.add_to_queue.reset_mock()
@ -846,9 +850,7 @@ async def test_play_media_playlist(
}, },
blocking=True, blocking=True,
) )
player.add_to_queue.assert_called_once_with( player.add_to_queue.assert_called_once_with(playlist, AddCriteriaType.ADD_TO_END)
playlist, const.AddCriteriaType.ADD_TO_END
)
# Invalid name # Invalid name
player.add_to_queue.reset_mock() player.add_to_queue.reset_mock()
await hass.services.async_call( await hass.services.async_call(
@ -1028,7 +1030,7 @@ async def test_media_player_unjoin_group(
player = controller.players[1] player = controller.players[1]
player.heos.dispatcher.send( player.heos.dispatcher.send(
const.SIGNAL_PLAYER_EVENT, SignalType.PLAYER_EVENT,
player.player_id, player.player_id,
const.EVENT_PLAYER_STATE_CHANGED, const.EVENT_PLAYER_STATE_CHANGED,
) )

37
tests/components/heos/test_services.py
View File

@ -1,6 +1,6 @@
"""Tests for the services module.""" """Tests for the services module."""
from pyheos import CommandFailedError, HeosError, const from pyheos import CommandAuthenticationError, HeosError
import pytest import pytest
from homeassistant.components.heos.const import ( from homeassistant.components.heos.const import (
@ -38,30 +38,14 @@ async def test_sign_in(hass: HomeAssistant, config_entry, controller) -> None:
controller.sign_in.assert_called_once_with("test@test.com", "password") controller.sign_in.assert_called_once_with("test@test.com", "password")
async def test_sign_in_not_connected(
hass: HomeAssistant, config_entry, controller, caplog: pytest.LogCaptureFixture
) -> None:
"""Test sign-in service logs error when not connected."""
await setup_component(hass, config_entry)
controller.connection_state = const.STATE_RECONNECTING
await hass.services.async_call(
DOMAIN,
SERVICE_SIGN_IN,
{ATTR_USERNAME: "test@test.com", ATTR_PASSWORD: "password"},
blocking=True,
)
assert controller.sign_in.call_count == 0
assert "Unable to sign in because HEOS is not connected" in caplog.text
async def test_sign_in_failed( async def test_sign_in_failed(
hass: HomeAssistant, config_entry, controller, caplog: pytest.LogCaptureFixture hass: HomeAssistant, config_entry, controller, caplog: pytest.LogCaptureFixture
) -> None: ) -> None:
"""Test sign-in service logs error when not connected.""" """Test sign-in service logs error when not connected."""
await setup_component(hass, config_entry) await setup_component(hass, config_entry)
controller.sign_in.side_effect = CommandFailedError("", "Invalid credentials", 6) controller.sign_in.side_effect = CommandAuthenticationError(
"", "Invalid credentials", 6
)
await hass.services.async_call( await hass.services.async_call(
DOMAIN, DOMAIN,
@ -115,19 +99,6 @@ async def test_sign_out(hass: HomeAssistant, config_entry, controller) -> None:
assert controller.sign_out.call_count == 1 assert controller.sign_out.call_count == 1
async def test_sign_out_not_connected(
hass: HomeAssistant, config_entry, controller, caplog: pytest.LogCaptureFixture
) -> None:
"""Test the sign-out service."""
await setup_component(hass, config_entry)
controller.connection_state = const.STATE_RECONNECTING
await hass.services.async_call(DOMAIN, SERVICE_SIGN_OUT, {}, blocking=True)
assert controller.sign_out.call_count == 0
assert "Unable to sign out because HEOS is not connected" in caplog.text
async def test_sign_out_not_loaded_raises(hass: HomeAssistant, config_entry) -> None: async def test_sign_out_not_loaded_raises(hass: HomeAssistant, config_entry) -> None:
"""Test the sign-out service when entry not loaded raises exception.""" """Test the sign-out service when entry not loaded raises exception."""
await setup_component(hass, config_entry) await setup_component(hass, config_entry)