diff --git a/homeassistant/components/heos/coordinator.py b/homeassistant/components/heos/coordinator.py index ee0aeb3f165..dd0e0a19d0b 100644 --- a/homeassistant/components/heos/coordinator.py +++ b/homeassistant/components/heos/coordinator.py @@ -68,6 +68,11 @@ class HeosCoordinator(DataUpdateCoordinator[None]): """Get input sources across all devices.""" return self._inputs + @property + def favorites(self) -> dict[int, MediaItem]: + """Get favorite stations.""" + return self._favorites + async def async_setup(self) -> None: """Set up the coordinator; connect to the host; and retrieve initial data.""" # Add before connect as it may occur during initial connection diff --git a/homeassistant/components/heos/diagnostics.py b/homeassistant/components/heos/diagnostics.py new file mode 100644 index 00000000000..bf33fc9bc15 --- /dev/null +++ b/homeassistant/components/heos/diagnostics.py @@ -0,0 +1,90 @@ +"""Define the HEOS integration diagnostics module.""" + +from collections.abc import Mapping, Sequence +import dataclasses +from typing import Any + +from pyheos import HeosError + +from homeassistant.components.diagnostics import async_redact_data +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.device_registry import DeviceEntry + +from .const import ATTR_PASSWORD, ATTR_USERNAME, DOMAIN +from .coordinator import HeosConfigEntry + +TO_REDACT = [ + ATTR_PASSWORD, + ATTR_USERNAME, + "signed_in_username", + "serial", + "serial_number", +] + + +def _as_dict( + data: Any, redact: bool = False +) -> Mapping[str, Any] | Sequence[Any] | Any: + """Convert dataclasses to dicts within various data structures.""" + if dataclasses.is_dataclass(data): + data_dict = dataclasses.asdict(data) # type: ignore[arg-type] + return data_dict if not redact else async_redact_data(data_dict, TO_REDACT) + if not isinstance(data, (Mapping, Sequence)): + return data + if isinstance(data, Sequence): + return [_as_dict(val) for val in data] + return {k: _as_dict(v) for k, v in data.items()} + + +async def async_get_config_entry_diagnostics( + hass: HomeAssistant, config_entry: HeosConfigEntry +) -> dict[str, Any]: + """Return diagnostics for a config entry.""" + coordinator = config_entry.runtime_data + diagnostics = { + "config_entry": async_redact_data(config_entry.as_dict(), TO_REDACT), + "heos": { + "connection_state": coordinator.heos.connection_state, + "current_credentials": _as_dict( + coordinator.heos.current_credentials, redact=True + ), + }, + "groups": _as_dict(coordinator.heos.groups), + "source_list": coordinator.async_get_source_list(), + "inputs": _as_dict(coordinator.inputs), + "favorites": _as_dict(coordinator.favorites), + } + # Try getting system information + try: + system_info = await coordinator.heos.get_system_info() + except HeosError as err: + diagnostics["system"] = {"error": str(err)} + else: + diagnostics["system"] = _as_dict(system_info, redact=True) + return diagnostics + + +async def async_get_device_diagnostics( + hass: HomeAssistant, config_entry: HeosConfigEntry, device: DeviceEntry +) -> dict[str, Any]: + """Return diagnostics for a device.""" + entity_registry = er.async_get(hass) + entities = entity_registry.entities.get_entries_for_device_id(device.id, True) + player_id = next( + int(value) for domain, value in device.identifiers if domain == DOMAIN + ) + player = config_entry.runtime_data.heos.players.get(player_id) + return { + "device": async_redact_data(device.dict_repr, TO_REDACT), + "entities": [ + { + "entity": entity.as_partial_dict, + "state": state.as_dict() + if (state := hass.states.get(entity.entity_id)) + else None, + } + for entity in entities + ], + "player": _as_dict(player, redact=True), + } diff --git a/homeassistant/components/heos/quality_scale.yaml b/homeassistant/components/heos/quality_scale.yaml index d48bcc492cd..cc110c627f0 100644 --- a/homeassistant/components/heos/quality_scale.yaml +++ b/homeassistant/components/heos/quality_scale.yaml @@ -37,7 +37,7 @@ rules: comment: 99% test coverage # Gold devices: done - diagnostics: todo + diagnostics: done discovery-update-info: status: todo comment: Explore if this is possible. diff --git a/tests/components/heos/conftest.py b/tests/components/heos/conftest.py index 122467c6b02..5312b8295ed 100644 --- a/tests/components/heos/conftest.py +++ b/tests/components/heos/conftest.py @@ -6,11 +6,13 @@ from collections.abc import AsyncIterator from unittest.mock import AsyncMock, Mock, patch from pyheos import ( - CONTROLS_ALL, Heos, HeosGroup, + HeosHost, + HeosNowPlayingMedia, HeosOptions, HeosPlayer, + HeosSystem, LineOutLevelType, MediaItem, MediaType, @@ -98,6 +100,33 @@ async def controller_fixture( yield mock_heos +@pytest.fixture(name="system") +def system_info_fixture() -> dict[str, str]: + """Create a system info fixture.""" + return HeosSystem( + "user@user.com", + "127.0.0.1", + hosts=[ + HeosHost( + "Test Player", + "HEOS Drive HS2", + "123456", + "1.0.0", + "127.0.0.1", + NetworkType.WIRED, + ), + HeosHost( + "Test Player 2", + "Speaker", + "123456", + "1.0.0", + "127.0.0.2", + NetworkType.WIFI, + ), + ], + ) + + @pytest.fixture(name="players") def players_fixture(quick_selects: dict[int, str]) -> dict[int, HeosPlayer]: """Create two mock HeosPlayers.""" @@ -121,20 +150,18 @@ def players_fixture(quick_selects: dict[int, str]) -> dict[int, HeosPlayer]: volume=25, heos=None, ) - player.now_playing_media = Mock() - player.now_playing_media.supported_controls = CONTROLS_ALL - player.now_playing_media.album_id = 1 - player.now_playing_media.queue_id = 1 - player.now_playing_media.source_id = 1 - player.now_playing_media.station = "Station Name" - player.now_playing_media.type = "Station" - player.now_playing_media.album = "Album" - player.now_playing_media.artist = "Artist" - player.now_playing_media.media_id = "1" - player.now_playing_media.duration = None - player.now_playing_media.current_position = None - player.now_playing_media.image_url = "http://" - player.now_playing_media.song = "Song" + player.now_playing_media = HeosNowPlayingMedia( + type=MediaType.STATION, + song="Song", + station="Station Name", + album="Album", + artist="Artist", + image_url="http://", + album_id="1", + media_id="1", + queue_id=1, + source_id=10, + ) player.add_to_queue = AsyncMock() player.clear_queue = AsyncMock() player.get_quick_selects = AsyncMock(return_value=quick_selects) diff --git a/tests/components/heos/snapshots/test_diagnostics.ambr b/tests/components/heos/snapshots/test_diagnostics.ambr new file mode 100644 index 00000000000..6de0a645f17 --- /dev/null +++ b/tests/components/heos/snapshots/test_diagnostics.ambr @@ -0,0 +1,371 @@ +# serializer version: 1 +# name: test_config_entry_diagnostics + dict({ + 'config_entry': dict({ + 'data': dict({ + 'host': '127.0.0.1', + }), + 'disabled_by': None, + 'discovery_keys': dict({ + }), + 'domain': 'heos', + 'minor_version': 1, + 'options': dict({ + }), + 'pref_disable_new_entities': False, + 'pref_disable_polling': False, + 'source': 'user', + 'title': 'HEOS System (via 127.0.0.1)', + 'unique_id': 'heos', + 'version': 1, + }), + 'favorites': dict({ + '1': dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': '123456789', + 'name': "Today's Hits Radio", + 'playable': True, + 'source_id': 1, + 'type': 'station', + }), + '2': dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 's1234', + 'name': 'Classical MPR (Classical Music)', + 'playable': True, + 'source_id': 3, + 'type': 'station', + }), + }), + 'groups': dict({ + '999': dict({ + 'group_id': 999, + 'is_muted': False, + 'lead_player_id': 1, + 'member_player_ids': list([ + 2, + ]), + 'name': 'Group', + 'volume': 0, + }), + }), + 'heos': dict({ + 'connection_state': 'disconnected', + 'current_credentials': None, + }), + 'inputs': list([ + dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 'inputs/aux_in_1', + 'name': 'HEOS Drive - Line In 1', + 'playable': True, + 'source_id': 1027, + 'type': 'station', + }), + dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 'inputs/aux_in_1', + 'name': 'Speaker - Line In 1', + 'playable': True, + 'source_id': 1027, + 'type': 'station', + }), + ]), + 'source_list': list([ + "Today's Hits Radio", + 'Classical MPR (Classical Music)', + 'HEOS Drive - Line In 1', + 'Speaker - Line In 1', + ]), + 'system': dict({ + 'connected_to_preferred_host': False, + 'host': '127.0.0.1', + 'hosts': list([ + dict({ + 'ip_address': '127.0.0.1', + 'model': 'HEOS Drive HS2', + 'name': 'Test Player', + 'network': 'wired', + 'serial': '**REDACTED**', + 'version': '1.0.0', + }), + dict({ + 'ip_address': '127.0.0.2', + 'model': 'Speaker', + 'name': 'Test Player 2', + 'network': 'wifi', + 'serial': '**REDACTED**', + 'version': '1.0.0', + }), + ]), + 'is_signed_in': True, + 'preferred_hosts': list([ + dict({ + 'ip_address': '127.0.0.1', + 'model': 'HEOS Drive HS2', + 'name': 'Test Player', + 'network': 'wired', + 'serial': '**REDACTED**', + 'version': '1.0.0', + }), + ]), + 'signed_in_username': '**REDACTED**', + }), + }) +# --- +# name: test_config_entry_diagnostics_error_getting_system + dict({ + 'config_entry': dict({ + 'data': dict({ + 'host': '127.0.0.1', + }), + 'disabled_by': None, + 'discovery_keys': dict({ + }), + 'domain': 'heos', + 'minor_version': 1, + 'options': dict({ + }), + 'pref_disable_new_entities': False, + 'pref_disable_polling': False, + 'source': 'user', + 'title': 'HEOS System (via 127.0.0.1)', + 'unique_id': 'heos', + 'version': 1, + }), + 'favorites': dict({ + '1': dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': '123456789', + 'name': "Today's Hits Radio", + 'playable': True, + 'source_id': 1, + 'type': 'station', + }), + '2': dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 's1234', + 'name': 'Classical MPR (Classical Music)', + 'playable': True, + 'source_id': 3, + 'type': 'station', + }), + }), + 'groups': dict({ + '999': dict({ + 'group_id': 999, + 'is_muted': False, + 'lead_player_id': 1, + 'member_player_ids': list([ + 2, + ]), + 'name': 'Group', + 'volume': 0, + }), + }), + 'heos': dict({ + 'connection_state': 'disconnected', + 'current_credentials': None, + }), + 'inputs': list([ + dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 'inputs/aux_in_1', + 'name': 'HEOS Drive - Line In 1', + 'playable': True, + 'source_id': 1027, + 'type': 'station', + }), + dict({ + 'album': None, + 'album_id': None, + 'artist': None, + 'browsable': False, + 'container_id': None, + 'image_url': '', + 'media_id': 'inputs/aux_in_1', + 'name': 'Speaker - Line In 1', + 'playable': True, + 'source_id': 1027, + 'type': 'station', + }), + ]), + 'source_list': list([ + "Today's Hits Radio", + 'Classical MPR (Classical Music)', + 'HEOS Drive - Line In 1', + 'Speaker - Line In 1', + ]), + 'system': dict({ + 'error': 'Not connected to device', + }), + }) +# --- +# name: test_device_diagnostics + dict({ + 'device': dict({ + 'area_id': None, + 'configuration_url': None, + 'connections': list([ + ]), + 'disabled_by': None, + 'entry_type': None, + 'hw_version': None, + 'identifiers': list([ + list([ + 'heos', + '1', + ]), + ]), + 'labels': list([ + ]), + 'manufacturer': 'HEOS', + 'model': 'Drive HS2', + 'model_id': None, + 'name': 'Test Player', + 'name_by_user': None, + 'serial_number': '**REDACTED**', + 'sw_version': '1.0.0', + 'via_device_id': None, + }), + 'entities': list([ + dict({ + 'entity': dict({ + 'area_id': None, + 'categories': dict({ + }), + 'disabled_by': None, + 'entity_category': None, + 'entity_id': 'media_player.test_player', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'labels': list([ + ]), + 'name': None, + 'options': dict({ + }), + 'original_name': None, + 'platform': 'heos', + 'translation_key': None, + 'unique_id': '1', + }), + 'state': dict({ + 'attributes': dict({ + 'entity_picture': 'http://', + 'friendly_name': 'Test Player', + 'group_members': list([ + 'media_player.test_player', + 'media_player.test_player_2', + ]), + 'is_volume_muted': False, + 'media_album_id': '1', + 'media_album_name': 'Album', + 'media_artist': 'Artist', + 'media_content_id': '1', + 'media_content_type': 'music', + 'media_queue_id': 1, + 'media_source_id': 10, + 'media_station': 'Station Name', + 'media_title': 'Song', + 'media_type': 'station', + 'repeat': 'off', + 'shuffle': False, + 'source_list': list([ + "Today's Hits Radio", + 'Classical MPR (Classical Music)', + 'HEOS Drive - Line In 1', + 'Speaker - Line In 1', + ]), + 'supported_features': 3079741, + 'volume_level': 0.25, + }), + 'context': dict({ + 'parent_id': None, + 'user_id': None, + }), + 'entity_id': 'media_player.test_player', + 'state': 'idle', + }), + }), + ]), + 'player': dict({ + 'available': True, + 'control': 0, + 'group_id': 999, + 'ip_address': '127.0.0.1', + 'is_muted': False, + 'line_out': 1, + 'model': 'HEOS Drive HS2', + 'name': 'Test Player', + 'network': 'wired', + 'now_playing_media': dict({ + 'album': 'Album', + 'album_id': '1', + 'artist': 'Artist', + 'current_position': None, + 'current_position_updated': None, + 'duration': None, + 'image_url': 'http://', + 'media_id': '1', + 'options': list([ + ]), + 'queue_id': 1, + 'song': 'Song', + 'source_id': 10, + 'station': 'Station Name', + 'supported_controls': list([ + 'play', + 'pause', + 'stop', + 'play_next', + 'play_previous', + ]), + 'type': 'station', + }), + 'playback_error': None, + 'player_id': 1, + 'repeat': 'off', + 'serial': '**REDACTED**', + 'shuffle': False, + 'state': 'stop', + 'version': '1.0.0', + 'volume': 25, + }), + }) +# --- diff --git a/tests/components/heos/snapshots/test_media_player.ambr b/tests/components/heos/snapshots/test_media_player.ambr index 7bfdac232cb..88d27f2073a 100644 --- a/tests/components/heos/snapshots/test_media_player.ambr +++ b/tests/components/heos/snapshots/test_media_player.ambr @@ -9,16 +9,16 @@ 'media_player.test_player_2', ]), 'is_volume_muted': False, - 'media_album_id': 1, + 'media_album_id': '1', 'media_album_name': 'Album', 'media_artist': 'Artist', 'media_content_id': '1', 'media_content_type': , 'media_queue_id': 1, - 'media_source_id': 1, + 'media_source_id': 10, 'media_station': 'Station Name', 'media_title': 'Song', - 'media_type': 'Station', + 'media_type': , 'repeat': , 'shuffle': False, 'source_list': list([ diff --git a/tests/components/heos/test_diagnostics.py b/tests/components/heos/test_diagnostics.py new file mode 100644 index 00000000000..d6fb8e1a8fe --- /dev/null +++ b/tests/components/heos/test_diagnostics.py @@ -0,0 +1,98 @@ +"""Tests for the HEOS diagnostics module.""" + +from unittest import mock + +from pyheos import Heos, HeosSystem +import pytest +from syrupy import SnapshotAssertion +from syrupy.filters import props + +from homeassistant.components.heos.const import DOMAIN +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr + +from tests.common import MockConfigEntry +from tests.components.diagnostics import ( + get_diagnostics_for_config_entry, + get_diagnostics_for_device, +) +from tests.typing import ClientSessionGenerator + + +async def test_config_entry_diagnostics( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + config_entry: MockConfigEntry, + controller: Heos, + system: HeosSystem, + snapshot: SnapshotAssertion, +) -> None: + """Test generating diagnostics for a config entry.""" + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + + with mock.patch.object( + controller, controller.get_system_info.__name__, return_value=system + ): + diagnostics = await get_diagnostics_for_config_entry( + hass, hass_client, config_entry + ) + + assert diagnostics == snapshot( + exclude=props("created_at", "modified_at", "entry_id") + ) + + +@pytest.mark.usefixtures("controller") +async def test_config_entry_diagnostics_error_getting_system( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + config_entry: MockConfigEntry, + snapshot: SnapshotAssertion, +) -> None: + """Test generating diagnostics with error during getting system info.""" + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + + # Not patching get_system_info to raise error 'Not connected to device' + + diagnostics = await get_diagnostics_for_config_entry( + hass, hass_client, config_entry + ) + + assert diagnostics == snapshot( + exclude=props("created_at", "modified_at", "entry_id") + ) + + +@pytest.mark.usefixtures("controller") +async def test_device_diagnostics( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + config_entry: MockConfigEntry, + snapshot: SnapshotAssertion, +) -> None: + """Test generating diagnostics for a config entry.""" + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + device_registry = dr.async_get(hass) + device = device_registry.async_get_device({(DOMAIN, "1")}) + + diagnostics = await get_diagnostics_for_device( + hass, hass_client, config_entry, device + ) + assert diagnostics == snapshot( + exclude=props( + "created_at", + "modified_at", + "config_entries", + "id", + "primary_config_entry", + "config_entry_id", + "device_id", + "entity_picture_local", + "last_changed", + "last_reported", + "last_updated", + ) + )