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

Add HEOS diagnostics (#136663)

Browse Source
This commit is contained in:
Andrew Sayre 2025-01-28 07:02:15 -06:00 committed by GitHub
parent 6278d36981
commit c2da844f76
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 610 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/heos/coordinator.py
View File

@ -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

90
homeassistant/components/heos/diagnostics.py Normal file
View File

@ -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),
}

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

@ -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.

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

@ -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)

371
tests/components/heos/snapshots/test_diagnostics.ambr Normal file
View File

@ -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,
}),
})
# ---

6
tests/components/heos/snapshots/test_media_player.ambr
View File

@ -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': <MediaType.MUSIC: 'music'>,
'media_queue_id': 1,
'media_source_id': 1,
'media_source_id': 10,
'media_station': 'Station Name',
'media_title': 'Song',
'media_type': 'Station',
'media_type': <MediaType.STATION: 'station'>,
'repeat': <RepeatMode.OFF: 'off'>,
'shuffle': False,
'source_list': list([

98
tests/components/heos/test_diagnostics.py Normal file
View File

@ -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",
)
)