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

Add tests for join and unjoin service calls in Sonos (#145602)

Browse Source 
* fix: add tests for join and unjoin

* fix: update comments

* fix: update comments

* fix: refactor to common functions

* fix: refactor to common functions

* fix: add type def

* fix: add return types

* fix: add return types

* fix: correct type annontation for uui_ds

* fix: update comments

* fix: merge issues

* fix: merge issue

* fix: raise homeassistanterror on timeout

* fix: add comments

* fix: simplify test

* fix: simplify test

* fix: simplify test
This commit is contained in:
Pete Sage 2025-06-25 05:38:51 -04:00 committed by GitHub
parent f897a728f1
commit c9e9575a3d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 240 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/sonos/speaker.py
View File

@ -1172,8 +1172,15 @@ class SonosSpeaker:
while not _test_groups(groups):
await config_entry.runtime_data.topology_condition.wait()
except TimeoutError:
_LOGGER.warning("Timeout waiting for target groups %s", groups)
group_description = [
f"{group[0].zone_name}: {', '.join(speaker.zone_name for speaker in group)}"
for group in groups
]
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="timeout_join",
translation_placeholders={"group_description": str(group_description)},
) from TimeoutError
any_speaker = next(iter(config_entry.runtime_data.discovered.values()))
any_speaker.soco.zone_group_state.clear_cache()

3
homeassistant/components/sonos/strings.json
View File

@ -194,6 +194,9 @@
},
"announce_media_error": {
"message": "Announcing clip {media_id} failed {response}"
},
"timeout_join": {
"message": "Timeout while waiting for Sonos player to join the group {group_description}"
}
}
}

63
tests/components/sonos/conftest.py
View File

@ -1,5 +1,7 @@
"""Configuration for Sonos tests."""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine, Generator
from copy import copy
@ -107,13 +109,31 @@ class SonosMockAlarmClock(SonosMockService):
class SonosMockEvent:
"""Mock a sonos Event used in callbacks."""
def __init__(self, soco, service, variables) -> None:
"""Initialize the instance."""
def __init__(
self,
soco: MockSoCo,
service: SonosMockService,
variables: dict[str, str],
zone_player_uui_ds_in_group: str | None = None,
) -> None:
"""Initialize the instance.
Args:
soco: The mock SoCo device associated with this event.
service: The Sonos mock service that generated the event.
variables: A dictionary of event variables and their values.
zone_player_uui_ds_in_group: Optional comma-separated string of unique zone IDs in the group.
"""
self.sid = f"{soco.uid}_sub0000000001"
self.seq = "0"
self.timestamp = 1621000000.0
self.service = service
self.variables = variables
# In Soco events of the same type may or may not have this attribute present.
# Only create the attribute if it should be present.
if zone_player_uui_ds_in_group:
self.zone_player_uui_ds_in_group = zone_player_uui_ds_in_group
def increment_variable(self, var_name):
"""Increment the value of the var_name key in variables dict attribute.
@ -823,3 +843,42 @@ async def sonos_setup_two_speakers(
)
await hass.async_block_till_done()
return [soco_lr, soco_br]
def create_zgs_sonos_event(
fixture_file: str,
soco_1: MockSoCo,
soco_2: MockSoCo,
create_uui_ds_in_group: bool = True,
) -> SonosMockEvent:
"""Create a Sonos Event for zone group state, with the option of creating the uui_ds_in_group."""
zgs = load_fixture(fixture_file, DOMAIN)
variables = {}
variables["ZoneGroupState"] = zgs
# Sonos does not always send this variable with zgs events
if create_uui_ds_in_group:
variables["zone_player_uui_ds_in_group"] = f"{soco_1.uid},{soco_2.uid}"
zone_player_uui_ds_in_group = (
f"{soco_1.uid},{soco_2.uid}" if create_uui_ds_in_group else None
)
return SonosMockEvent(
soco_1, soco_1.zoneGroupTopology, variables, zone_player_uui_ds_in_group
)
def group_speakers(coordinator: MockSoCo, group_member: MockSoCo) -> None:
"""Generate events to group two speakers together."""
event = create_zgs_sonos_event(
"zgs_group.xml", coordinator, group_member, create_uui_ds_in_group=True
)
coordinator.zoneGroupTopology.subscribe.return_value._callback(event)
group_member.zoneGroupTopology.subscribe.return_value._callback(event)
def ungroup_speakers(coordinator: MockSoCo, group_member: MockSoCo) -> None:
"""Generate events to ungroup two speakers."""
event = create_zgs_sonos_event(
"zgs_two_single.xml", coordinator, group_member, create_uui_ds_in_group=False
)
coordinator.zoneGroupTopology.subscribe.return_value._callback(event)
group_member.zoneGroupTopology.subscribe.return_value._callback(event)

184
tests/components/sonos/test_services.py
View File

@ -1,53 +1,191 @@
"""Tests for Sonos services."""
import asyncio
from contextlib import asynccontextmanager
import logging
import re
from unittest.mock import Mock, patch
import pytest
from homeassistant.components.media_player import DOMAIN as MP_DOMAIN, SERVICE_JOIN
from homeassistant.components.media_player import (
DOMAIN as MP_DOMAIN,
SERVICE_JOIN,
SERVICE_UNJOIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from tests.common import MockConfigEntry
from .conftest import MockSoCo, group_speakers, ungroup_speakers
async def test_media_player_join(
hass: HomeAssistant, async_autosetup_sonos, config_entry: MockConfigEntry
hass: HomeAssistant,
sonos_setup_two_speakers: list[MockSoCo],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test join service."""
valid_entity_id = "media_player.zone_a"
mocked_entity_id = "media_player.mocked"
"""Test joining two speakers together."""
soco_living_room = sonos_setup_two_speakers[0]
soco_bedroom = sonos_setup_two_speakers[1]
# Ensure an error is raised if the entity is unknown
with pytest.raises(HomeAssistantError):
# After dispatching the join to the speakers, the integration waits for the
# group to be updated before returning. To simulate this we will dispatch
# a ZGS event to group the speaker. This event is
# triggered by the firing of the join_complete_event in the join mock.
join_complete_event = asyncio.Event()
def mock_join(*args, **kwargs) -> None:
hass.loop.call_soon_threadsafe(join_complete_event.set)
soco_bedroom.join = Mock(side_effect=mock_join)
with caplog.at_level(logging.WARNING):
caplog.clear()
await hass.services.async_call(
MP_DOMAIN,
SERVICE_JOIN,
{"entity_id": valid_entity_id, "group_members": mocked_entity_id},
{
"entity_id": "media_player.living_room",
"group_members": ["media_player.bedroom"],
},
blocking=False,
)
await join_complete_event.wait()
# Fire the ZGS event to update the speaker grouping as the join method is waiting
# for the speakers to be regrouped.
group_speakers(soco_living_room, soco_bedroom)
await hass.async_block_till_done(wait_background_tasks=True)
# Code logs warning messages if the join is not successful, so we check
# that no warning messages were logged.
assert len(caplog.records) == 0
# The API joins the group members to the entity_id speaker.
assert soco_bedroom.join.call_count == 1
assert soco_bedroom.join.call_args[0][0] == soco_living_room
assert soco_living_room.join.call_count == 0
async def test_media_player_join_bad_entity(
hass: HomeAssistant,
sonos_setup_two_speakers: list[MockSoCo],
) -> None:
"""Test error handling of joining with a bad entity."""
# Ensure an error is raised if the entity is unknown
with pytest.raises(HomeAssistantError) as excinfo:
await hass.services.async_call(
MP_DOMAIN,
SERVICE_JOIN,
{
"entity_id": "media_player.living_room",
"group_members": "media_player.bad_entity",
},
blocking=True,
)
assert "media_player.bad_entity" in str(excinfo.value)
# Ensure SonosSpeaker.join_multi is called if entity is found
mocked_speaker = Mock()
mock_entity_id_mappings = {mocked_entity_id: mocked_speaker}
@asynccontextmanager
async def instant_timeout(*args, **kwargs) -> None:
"""Mock a timeout error."""
raise TimeoutError
# This is never reached, but is needed to satisfy the asynccontextmanager
yield # pylint: disable=unreachable
async def test_media_player_join_timeout(
hass: HomeAssistant,
sonos_setup_two_speakers: list[MockSoCo],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test joining of two speakers with timeout error."""
soco_living_room = sonos_setup_two_speakers[0]
soco_bedroom = sonos_setup_two_speakers[1]
expected = (
"Timeout while waiting for Sonos player to join the "
"group ['Living Room: Living Room, Bedroom']"
)
with (
patch.dict(
config_entry.runtime_data.entity_id_mappings,
mock_entity_id_mappings,
),
patch(
"homeassistant.components.sonos.speaker.SonosSpeaker.join_multi"
) as mock_join_multi,
"homeassistant.components.sonos.speaker.asyncio.timeout", instant_timeout
),
pytest.raises(HomeAssistantError, match=re.escape(expected)),
):
await hass.services.async_call(
MP_DOMAIN,
SERVICE_JOIN,
{"entity_id": valid_entity_id, "group_members": mocked_entity_id},
{
"entity_id": "media_player.living_room",
"group_members": ["media_player.bedroom"],
},
blocking=True,
)
assert soco_bedroom.join.call_count == 1
assert soco_bedroom.join.call_args[0][0] == soco_living_room
assert soco_living_room.join.call_count == 0
async def test_media_player_unjoin(
hass: HomeAssistant,
sonos_setup_two_speakers: list[MockSoCo],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test unjoing two speaker."""
soco_living_room = sonos_setup_two_speakers[0]
soco_bedroom = sonos_setup_two_speakers[1]
# First group the speakers together
group_speakers(soco_living_room, soco_bedroom)
await hass.async_block_till_done(wait_background_tasks=True)
# Now that the speaker are joined, test unjoining
unjoin_complete_event = asyncio.Event()
def mock_unjoin(*args, **kwargs):
hass.loop.call_soon_threadsafe(unjoin_complete_event.set)
soco_bedroom.unjoin = Mock(side_effect=mock_unjoin)
with caplog.at_level(logging.WARNING):
caplog.clear()
await hass.services.async_call(
MP_DOMAIN,
SERVICE_UNJOIN,
{"entity_id": "media_player.bedroom"},
blocking=False,
)
await unjoin_complete_event.wait()
# Fire the ZGS event to ungroup the speakers as the unjoin method is waiting
# for the speakers to be ungrouped.
ungroup_speakers(soco_living_room, soco_bedroom)
await hass.async_block_till_done(wait_background_tasks=True)
assert len(caplog.records) == 0
assert soco_bedroom.unjoin.call_count == 1
assert soco_living_room.unjoin.call_count == 0
async def test_media_player_unjoin_already_unjoined(
hass: HomeAssistant,
sonos_setup_two_speakers: list[MockSoCo],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test unjoining when already unjoined."""
soco_living_room = sonos_setup_two_speakers[0]
soco_bedroom = sonos_setup_two_speakers[1]
with caplog.at_level(logging.WARNING):
caplog.clear()
await hass.services.async_call(
MP_DOMAIN,
SERVICE_UNJOIN,
{"entity_id": "media_player.bedroom"},
blocking=True,
)
found_speaker = config_entry.runtime_data.entity_id_mappings[valid_entity_id]
mock_join_multi.assert_called_with(
hass, config_entry, found_speaker, [mocked_speaker]
)
assert len(caplog.records) == 0
# Should not have called unjoin, since the speakers are already unjoined.
assert soco_bedroom.unjoin.call_count == 0
assert soco_living_room.unjoin.call_count == 0

39
tests/components/sonos/test_speaker.py
View File

@ -13,12 +13,11 @@ from homeassistant.components.sonos.const import SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from .conftest import MockSoCo, SonosMockEvent
from .conftest import MockSoCo, SonosMockEvent, group_speakers, ungroup_speakers
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
load_fixture,
load_json_value_fixture,
)
@ -81,22 +80,6 @@ async def test_subscription_creation_fails(
assert speaker._subscriptions
def _create_zgs_sonos_event(
fixture_file: str, soco_1: MockSoCo, soco_2: MockSoCo, create_uui_ds: bool = True
) -> SonosMockEvent:
"""Create a Sonos Event for zone group state, with the option of creating the uui_ds_in_group."""
zgs = load_fixture(fixture_file, DOMAIN)
variables = {}
variables["ZoneGroupState"] = zgs
# Sonos does not always send this variable with zgs events
if create_uui_ds:
variables["zone_player_uui_ds_in_group"] = f"{soco_1.uid},{soco_2.uid}"
event = SonosMockEvent(soco_1, soco_1.zoneGroupTopology, variables)
if create_uui_ds:
event.zone_player_uui_ds_in_group = f"{soco_1.uid},{soco_2.uid}"
return event
def _create_avtransport_sonos_event(
fixture_file: str, soco: MockSoCo
) -> SonosMockEvent:
@ -142,11 +125,8 @@ async def test_zgs_event_group_speakers(
soco_br.play.reset_mock()
# Test 2 - Group the speakers, living room is the coordinator
event = _create_zgs_sonos_event(
"zgs_group.xml", soco_lr, soco_br, create_uui_ds=True
)
soco_lr.zoneGroupTopology.subscribe.return_value._callback(event)
soco_br.zoneGroupTopology.subscribe.return_value._callback(event)
group_speakers(soco_lr, soco_br)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get("media_player.living_room")
assert state.attributes["group_members"] == [
@ -168,11 +148,8 @@ async def test_zgs_event_group_speakers(
soco_br.play.reset_mock()
# Test 3 - Ungroup the speakers
event = _create_zgs_sonos_event(
"zgs_two_single.xml", soco_lr, soco_br, create_uui_ds=False
)
soco_lr.zoneGroupTopology.subscribe.return_value._callback(event)
soco_br.zoneGroupTopology.subscribe.return_value._callback(event)
ungroup_speakers(soco_lr, soco_br)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get("media_player.living_room")
assert state.attributes["group_members"] == ["media_player.living_room"]
@ -206,11 +183,7 @@ async def test_zgs_avtransport_group_speakers(
soco_br.play.reset_mock()
# Test 2- Send a zgs event to return living room to its own coordinator
event = _create_zgs_sonos_event(
"zgs_two_single.xml", soco_lr, soco_br, create_uui_ds=False
)
soco_lr.zoneGroupTopology.subscribe.return_value._callback(event)
soco_br.zoneGroupTopology.subscribe.return_value._callback(event)
ungroup_speakers(soco_lr, soco_br)
await hass.async_block_till_done(wait_background_tasks=True)
# Call should route to the living room
await _media_play(hass, "media_player.living_room")