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

Update HEOS tests to not interact directly with integration internals (#136177)

Browse Source
This commit is contained in:
Andrew Sayre 2025-01-21 09:00:34 -06:00 committed by GitHub
parent b11b36b523
commit 9bf2996ea0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 71 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,12 +2,11 @@
from __future__ import annotations
from collections.abc import Sequence
from collections.abc import AsyncIterator
from unittest.mock import AsyncMock, Mock, patch
from pyheos import (
CONTROLS_ALL,
Dispatcher,
Heos,
HeosGroup,
HeosOptions,
@ -24,15 +23,8 @@ from pyheos import (
import pytest
import pytest_asyncio
from homeassistant.components.heos import (
CONF_PASSWORD,
DOMAIN,
ControllerManager,
GroupManager,
HeosRuntimeData,
SourceManager,
)
from homeassistant.const import CONF_HOST, CONF_USERNAME
from homeassistant.components.heos import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_DEVICE_TYPE,
ATTR_UPNP_FRIENDLY_NAME,
@ -48,54 +40,39 @@ from tests.common import MockConfigEntry
@pytest.fixture(name="config_entry")
def config_entry_fixture(heos_runtime_data):
def config_entry_fixture() -> MockConfigEntry:
"""Create a mock HEOS config entry."""
entry = MockConfigEntry(
return MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1"},
title="HEOS System (via 127.0.0.1)",
unique_id=DOMAIN,
)
entry.runtime_data = heos_runtime_data
return entry
@pytest.fixture(name="config_entry_options")
def config_entry_options_fixture(heos_runtime_data):
def config_entry_options_fixture() -> MockConfigEntry:
"""Create a mock HEOS config entry with options."""
entry = MockConfigEntry(
return MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "127.0.0.1"},
title="HEOS System (via 127.0.0.1)",
options={CONF_USERNAME: "user", CONF_PASSWORD: "pass"},
unique_id=DOMAIN,
)
entry.runtime_data = heos_runtime_data
return entry
@pytest.fixture(name="heos_runtime_data")
def heos_runtime_data_fixture(controller_manager, players):
"""Create a mock HeosRuntimeData fixture."""
return HeosRuntimeData(
controller_manager, Mock(GroupManager), Mock(SourceManager), players
)
@pytest.fixture(name="controller_manager")
def controller_manager_fixture(controller):
"""Create a mock controller manager fixture."""
mock_controller_manager = Mock(ControllerManager)
mock_controller_manager.controller = controller
return mock_controller_manager
@pytest.fixture(name="controller")
def controller_fixture(
players, favorites, input_sources, playlists, change_data, dispatcher, group
):
@pytest_asyncio.fixture(name="controller", autouse=True)
async def controller_fixture(
players: dict[int, HeosPlayer],
favorites: dict[int, MediaItem],
input_sources: list[MediaItem],
playlists: list[MediaItem],
change_data: PlayerUpdateResult,
group: dict[int, HeosGroup],
) -> AsyncIterator[Heos]:
"""Create a mock Heos controller fixture."""
mock_heos = Heos(HeosOptions(host="127.0.0.1", dispatcher=dispatcher))
mock_heos = Heos(HeosOptions(host="127.0.0.1"))
for player in players.values():
player.heos = mock_heos
mock_heos.connect = AsyncMock()
@ -121,7 +98,7 @@ def controller_fixture(
@pytest.fixture(name="players")
def player_fixture(quick_selects):
def players_fixture(quick_selects: dict[int, str]) -> dict[int, HeosPlayer]:
"""Create two mock HeosPlayers."""
players = {}
for i in (1, 2):
@ -179,12 +156,11 @@ def player_fixture(quick_selects):
@pytest.fixture(name="group")
def group_fixture():
def group_fixture() -> dict[int, HeosGroup]:
"""Create a HEOS group consisting of two players."""
group = HeosGroup(
name="Group", group_id=999, lead_player_id=1, member_player_ids=[2]
)
return {group.group_id: group}
@ -215,7 +191,7 @@ def favorites_fixture() -> dict[int, MediaItem]:
@pytest.fixture(name="input_sources")
def input_sources_fixture() -> Sequence[MediaItem]:
def input_sources_fixture() -> list[MediaItem]:
"""Create a set of input sources for testing."""
source = MediaItem(
source_id=1,
@ -230,14 +206,8 @@ def input_sources_fixture() -> Sequence[MediaItem]:
return [source]
@pytest_asyncio.fixture(name="dispatcher")
async def dispatcher_fixture() -> Dispatcher:
"""Create a dispatcher for testing."""
return Dispatcher()
@pytest.fixture(name="discovery_data")
def discovery_data_fixture() -> dict:
def discovery_data_fixture() -> SsdpServiceInfo:
"""Return mock discovery data for testing."""
return SsdpServiceInfo(
ssdp_usn="mock_usn",
@ -256,7 +226,7 @@ def discovery_data_fixture() -> dict:
@pytest.fixture(name="discovery_data_bedroom")
def discovery_data_fixture_bedroom() -> dict:
def discovery_data_fixture_bedroom() -> SsdpServiceInfo:
"""Return mock discovery data for testing."""
return SsdpServiceInfo(
ssdp_usn="mock_usn",
@ -288,7 +258,7 @@ def quick_selects_fixture() -> dict[int, str]:
@pytest.fixture(name="playlists")
def playlists_fixture() -> Sequence[MediaItem]:
def playlists_fixture() -> list[MediaItem]:
"""Create favorites fixture."""
playlist = MediaItem(
source_id=const.MUSIC_SOURCE_PLAYLISTS,

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

@ -220,6 +220,7 @@ async def test_options_flow_signs_in(
) -> None:
"""Test options flow signs-in with entered credentials."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Start the options flow. Entry has not current options.
assert CONF_USERNAME not in config_entry.options
@ -258,6 +259,7 @@ async def test_options_flow_signs_out(
) -> None:
"""Test options flow signs-out when credentials cleared."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Start the options flow. Entry has not current options.
result = await hass.config_entries.options.async_init(config_entry.entry_id)
@ -305,6 +307,7 @@ async def test_options_flow_missing_one_param_recovers(
) -> None:
"""Test options flow signs-in after recovering from only username or password being entered."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Start the options flow. Entry has not current options.
assert CONF_USERNAME not in config_entry.options
@ -353,6 +356,7 @@ async def test_reauth_signs_in_aborts(
) -> None:
"""Test reauth flow signs-in with entered credentials and aborts."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
result = await config_entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
@ -390,6 +394,7 @@ async def test_reauth_signs_out(
) -> None:
"""Test reauth flow signs-out when credentials cleared and aborts."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
result = await config_entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
@ -438,6 +443,7 @@ async def test_reauth_flow_missing_one_param_recovers(
) -> None:
"""Test reauth flow signs-in after recovering from only username or password being entered."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Start the options flow. Entry has not current options.
result = await config_entry.start_reauth_flow(hass)

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

@ -1,6 +1,5 @@
"""Tests for the init module."""
import asyncio
from typing import cast
from pyheos import (
@ -71,7 +70,7 @@ async def test_async_setup_entry_auth_failure_starts_reauth(
# Simulates what happens when the controller can't sign-in during connection
async def connect_send_auth_failure() -> None:
controller._signed_in_username = None
controller.dispatcher.send(
await controller.dispatcher.wait_send(
SignalType.HEOS_EVENT, SignalHeosEvent.USER_CREDENTIALS_INVALID
)
@ -151,7 +150,6 @@ async def test_update_sources_retry(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test update sources retries on failures to max attempts."""
config_entry.add_to_hass(hass)
@ -162,12 +160,10 @@ async def test_update_sources_retry(
source_manager.retry_delay = 0
source_manager.max_retry_attempts = 1
controller.get_favorites.side_effect = CommandFailedError("Test", "test", 0)
controller.dispatcher.send(
await controller.dispatcher.wait_send(
SignalType.CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {}
)
# Wait until it's finished
while "Unable to update sources" not in caplog.text:
await asyncio.sleep(0.1)
await hass.async_block_till_done()
assert controller.get_favorites.call_count == 2

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

@ -1,7 +1,5 @@
"""Tests for the Heos Media Player platform."""
import asyncio
from collections.abc import Sequence
import re
from typing import Any
@ -21,7 +19,7 @@ import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
from homeassistant.components.heos.const import DOMAIN, SIGNAL_HEOS_UPDATED
from homeassistant.components.heos.const import DOMAIN
from homeassistant.components.media_player import (
ATTR_GROUP_MEMBERS,
ATTR_INPUT_SOURCE,
@ -60,12 +58,10 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("controller")
async def test_state_attributes(
hass: HomeAssistant, config_entry: MockConfigEntry, snapshot: SnapshotAssertion
) -> None:
@ -94,7 +90,7 @@ async def test_updates_from_signals(
# Test player does not update for other players
player.state = PlayState.PLAY
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT, 2, const.EVENT_PLAYER_STATE_CHANGED
)
await hass.async_block_till_done()
@ -103,7 +99,7 @@ async def test_updates_from_signals(
# Test player_update standard events
player.state = PlayState.PLAY
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
)
await hass.async_block_till_done()
@ -114,7 +110,7 @@ async def test_updates_from_signals(
# Test player_update progress events
player.now_playing_media.duration = 360000
player.now_playing_media.current_position = 1000
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT,
player.player_id,
const.EVENT_PLAYER_NOW_PLAYING_PROGRESS,
@ -136,38 +132,36 @@ async def test_updates_from_connection_event(
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
event = asyncio.Event()
async def set_signal():
event.set()
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
# Connected
player.available = True
player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED)
await event.wait()
await player.heos.dispatcher.wait_send(
SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED
)
await hass.async_block_till_done()
state = hass.states.get("media_player.test_player")
assert state.state == STATE_IDLE
assert controller.load_players.call_count == 1
# Disconnected
event.clear()
controller.load_players.reset_mock()
player.available = False
player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.DISCONNECTED)
await event.wait()
await player.heos.dispatcher.wait_send(
SignalType.HEOS_EVENT, SignalHeosEvent.DISCONNECTED
)
await hass.async_block_till_done()
state = hass.states.get("media_player.test_player")
assert state.state == STATE_UNAVAILABLE
assert controller.load_players.call_count == 0
# Connected handles refresh failure
event.clear()
controller.load_players.reset_mock()
controller.load_players.side_effect = CommandFailedError(None, "Failure", 1)
player.available = True
player.heos.dispatcher.send(SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED)
await event.wait()
await player.heos.dispatcher.wait_send(
SignalType.HEOS_EVENT, SignalHeosEvent.CONNECTED
)
await hass.async_block_till_done()
state = hass.states.get("media_player.test_player")
assert state.state == STATE_IDLE
assert controller.load_players.call_count == 1
@ -178,28 +172,23 @@ async def test_updates_from_sources_updated(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
input_sources: Sequence[MediaItem],
input_sources: list[MediaItem],
) -> None:
"""Tests player updates from changes in sources list."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
event = asyncio.Event()
async def set_signal():
event.set()
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
input_sources.clear()
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.CONTROLLER_EVENT, const.EVENT_SOURCES_CHANGED, {}
)
await event.wait()
source_list = config_entry.runtime_data.source_manager.source_list
assert len(source_list) == 2
await hass.async_block_till_done()
state = hass.states.get("media_player.test_player")
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == source_list
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == [
"Today's Hits Radio",
"Classical MPR (Classical Music)",
]
async def test_updates_from_players_changed(
@ -212,19 +201,12 @@ async def test_updates_from_players_changed(
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
event = asyncio.Event()
async def set_signal():
event.set()
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
assert hass.states.get("media_player.test_player").state == STATE_IDLE
player.state = PlayState.PLAY
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.CONTROLLER_EVENT, const.EVENT_PLAYERS_CHANGED, change_data
)
await event.wait()
await hass.async_block_till_done()
assert hass.states.get("media_player.test_player").state == STATE_PLAYING
@ -241,7 +223,6 @@ async def test_updates_from_players_changed_new_ids(
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
event = asyncio.Event()
# Assert device registry matches current id
assert device_registry.async_get_device(identifiers={(DOMAIN, "1")})
@ -251,17 +232,12 @@ async def test_updates_from_players_changed_new_ids(
== "media_player.test_player"
)
# Trigger update
async def set_signal():
event.set()
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.CONTROLLER_EVENT,
const.EVENT_PLAYERS_CHANGED,
change_data_mapped_ids,
)
await event.wait()
await hass.async_block_till_done()
# Assert device registry identifiers were updated
assert len(device_registry.devices) == 2
@ -275,28 +251,23 @@ async def test_updates_from_players_changed_new_ids(
async def test_updates_from_user_changed(
hass: HomeAssistant, config_entry: MockConfigEntry, controller: Heos
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
) -> None:
"""Tests player updates from changes in user."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
player = controller.players[1]
event = asyncio.Event()
async def set_signal():
event.set()
async_dispatcher_connect(hass, SIGNAL_HEOS_UPDATED, set_signal)
controller._signed_in_username = None
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.CONTROLLER_EVENT, const.EVENT_USER_CHANGED, None
)
await event.wait()
source_list = config_entry.runtime_data.source_manager.source_list
assert len(source_list) == 1
await hass.async_block_till_done()
state = hass.states.get("media_player.test_player")
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == source_list
assert state.attributes[ATTR_INPUT_SOURCE_LIST] == ["HEOS Drive - Line In 1"]
async def test_clear_playlist(
@ -650,7 +621,7 @@ async def test_select_favorite(
player.play_preset_station.assert_called_once_with(1)
# Test state is matched by station name
player.now_playing_media.station = favorite.name
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
)
await hass.async_block_till_done()
@ -680,7 +651,7 @@ async def test_select_radio_favorite(
# Test state is matched by album id
player.now_playing_media.station = "Classical"
player.now_playing_media.album_id = favorite.media_id
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
)
await hass.async_block_till_done()
@ -721,7 +692,7 @@ async def test_select_input_source(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
input_sources: Sequence[MediaItem],
input_sources: list[MediaItem],
) -> None:
"""Tests selecting input source and state."""
config_entry.add_to_hass(hass)
@ -742,7 +713,7 @@ async def test_select_input_source(
# Test state is matched by media id
player.now_playing_media.source_id = const.MUSIC_SOURCE_AUX_INPUT
player.now_playing_media.media_id = const.INPUT_AUX_IN_1
player.heos.dispatcher.send(
await player.heos.dispatcher.wait_send(
SignalType.PLAYER_EVENT, player.player_id, const.EVENT_PLAYER_STATE_CHANGED
)
await hass.async_block_till_done()
@ -750,7 +721,6 @@ async def test_select_input_source(
assert state.attributes[ATTR_INPUT_SOURCE] == input_source.name
@pytest.mark.usefixtures("controller")
async def test_select_input_unknown_raises(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
@ -773,7 +743,7 @@ async def test_select_input_command_error(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
input_sources: Sequence[MediaItem],
input_sources: list[MediaItem],
) -> None:
"""Tests selecting an unknown input."""
config_entry.add_to_hass(hass)
@ -797,7 +767,6 @@ async def test_select_input_command_error(
player.play_input_source.assert_called_once_with(input_source.media_id)
@pytest.mark.usefixtures("controller")
async def test_unload_config_entry(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:
@ -926,7 +895,7 @@ async def test_play_media_playlist(
hass: HomeAssistant,
config_entry: MockConfigEntry,
controller: Heos,
playlists: Sequence[MediaItem],
playlists: list[MediaItem],
enqueue: Any,
criteria: AddCriteriaType,
) -> None:
@ -1026,7 +995,6 @@ async def test_play_media_favorite_error(
assert player.play_preset_station.call_count == 0
@pytest.mark.usefixtures("controller")
async def test_play_media_invalid_type(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None:

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

@ -81,7 +81,6 @@ async def test_sign_in_unknown_error(
assert "Unable to sign in" in caplog.text
@pytest.mark.usefixtures("controller")
async def test_sign_in_not_loaded_raises(
hass: HomeAssistant, config_entry: MockConfigEntry
) -> None: