mirror of
https://github.com/home-assistant/core.git
synced 2026-07-03 12:46:09 +01:00
Paul Bottein
a576aef9a4
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
410 lines
14 KiB
Python
410 lines
14 KiB
Python
"""Tests for the Yoto integration setup."""
|
|
|
|
from unittest.mock import MagicMock, Mock, patch
|
|
|
|
import aiohttp
|
|
from freezegun.api import FrozenDateTimeFactory
|
|
import pytest
|
|
from yoto_api import AuthenticationError, Device, YotoAPIError, YotoError, YotoPlayer
|
|
|
|
from homeassistant.components.yoto.const import (
|
|
DOMAIN,
|
|
SCAN_INTERVAL,
|
|
STATUS_PUSH_INTERVAL,
|
|
)
|
|
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import (
|
|
OAuth2TokenRequestError,
|
|
OAuth2TokenRequestReauthError,
|
|
)
|
|
from homeassistant.helpers import device_registry as dr
|
|
from homeassistant.helpers.config_entry_oauth2_flow import (
|
|
ImplementationUnavailableError,
|
|
)
|
|
|
|
from . import setup_integration
|
|
from .conftest import PLAYER_ID
|
|
|
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
|
|
|
pytestmark = pytest.mark.usefixtures("setup_credentials")
|
|
|
|
|
|
async def test_setup_unload(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""The integration loads and unloads cleanly."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
assert mock_config_entry.state is ConfigEntryState.LOADED
|
|
|
|
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
|
mock_yoto_client.disconnect_events.assert_called_once()
|
|
|
|
|
|
async def test_setup_retries_on_api_failure(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""A non-auth API failure surfaces as a setup retry."""
|
|
mock_yoto_client.refresh.side_effect = YotoAPIError("boom")
|
|
|
|
await setup_integration(hass, mock_config_entry)
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
|
|
|
|
|
async def test_mqtt_event_updates_entity(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""An MQTT event published by the broker refreshes the entity state."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
state_before = hass.states.get("media_player.nursery_yoto")
|
|
assert state_before is not None
|
|
|
|
# connect_events(device_ids, on_update) — invoke the registered on_update callback
|
|
on_update = mock_yoto_client.connect_events.call_args.args[1]
|
|
player = next(iter(mock_yoto_client.players.values()))
|
|
player.last_event.volume = 12
|
|
on_update(player)
|
|
await hass.async_block_till_done()
|
|
|
|
state_after = hass.states.get("media_player.nursery_yoto")
|
|
assert state_after is not None
|
|
assert state_after.attributes["volume_level"] == 12 / 16
|
|
assert state_after.last_updated > state_before.last_updated
|
|
|
|
|
|
async def test_status_push_tick(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""The status-push timer publishes a request every 60 s."""
|
|
mock_yoto_client.is_mqtt_connected = True
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.request_player_status.reset_mock()
|
|
|
|
freezer.tick(STATUS_PUSH_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
mock_yoto_client.request_player_status.assert_called_once_with("player-test")
|
|
|
|
|
|
async def test_status_push_skipped_when_mqtt_disconnected(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""The status-push timer is a no-op while MQTT is reconnecting."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.request_player_status.reset_mock()
|
|
mock_yoto_client.is_mqtt_connected = False
|
|
|
|
freezer.tick(STATUS_PUSH_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
mock_yoto_client.request_player_status.assert_not_called()
|
|
|
|
|
|
async def test_periodic_poll_refreshes_players(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""The coordinator refreshes the player list on every tick."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.refresh.reset_mock()
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
mock_yoto_client.refresh.assert_called_once()
|
|
|
|
|
|
async def test_setup_retries_when_implementation_missing(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""Missing OAuth2 implementation defers setup as not-ready."""
|
|
with patch(
|
|
"homeassistant.components.yoto.async_get_config_entry_implementation",
|
|
side_effect=ImplementationUnavailableError("gone"),
|
|
):
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"side_effect",
|
|
[
|
|
aiohttp.ClientError("boom"),
|
|
OAuth2TokenRequestError(request_info=Mock(), domain=DOMAIN),
|
|
],
|
|
)
|
|
async def test_setup_retries_on_token_validation_error(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
side_effect: Exception,
|
|
) -> None:
|
|
"""A failure refreshing the OAuth token defers setup."""
|
|
with patch(
|
|
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
|
|
side_effect=side_effect,
|
|
):
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
|
|
|
|
|
async def test_setup_reauth_on_invalid_refresh_token(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""An unrecoverable token refresh at setup starts a reauth flow."""
|
|
with patch(
|
|
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
|
|
side_effect=OAuth2TokenRequestReauthError(request_info=Mock(), domain=DOMAIN),
|
|
):
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
|
|
assert any(
|
|
flow["context"]["source"] == SOURCE_REAUTH
|
|
for flow in hass.config_entries.flow.async_progress()
|
|
)
|
|
|
|
|
|
async def test_setup_reauth_on_authentication_error(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""A rejected access token at setup starts a reauth flow."""
|
|
mock_yoto_client.refresh.side_effect = AuthenticationError("denied")
|
|
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
|
|
assert any(
|
|
flow["context"]["source"] == SOURCE_REAUTH
|
|
for flow in hass.config_entries.flow.async_progress()
|
|
)
|
|
|
|
|
|
async def test_poll_reauth_on_authentication_error(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""A rejected access token during the poll starts a reauth flow."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.refresh.side_effect = AuthenticationError("denied")
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
assert any(
|
|
flow["context"]["source"] == SOURCE_REAUTH
|
|
for flow in hass.config_entries.flow.async_progress()
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_yoto_client")
|
|
async def test_poll_reauth_on_invalid_refresh_token(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""An unrecoverable token refresh during the poll starts a reauth flow."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
with patch(
|
|
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
|
|
side_effect=OAuth2TokenRequestReauthError(request_info=Mock(), domain=DOMAIN),
|
|
):
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
assert any(
|
|
flow["context"]["source"] == SOURCE_REAUTH
|
|
for flow in hass.config_entries.flow.async_progress()
|
|
)
|
|
|
|
|
|
async def test_setup_retries_when_mqtt_unavailable(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""MQTT connect failure surfaces as a setup retry."""
|
|
mock_yoto_client.connect_events.side_effect = YotoError("mqtt down")
|
|
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
|
|
|
|
|
async def test_setup_succeeds_without_card_library(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
) -> None:
|
|
"""A library load failure doesn't block setup; titles and artwork stay empty."""
|
|
mock_yoto_client.update_library.side_effect = YotoError("library down")
|
|
|
|
await setup_integration(hass, mock_config_entry)
|
|
|
|
assert mock_config_entry.state is ConfigEntryState.LOADED
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"side_effect",
|
|
[
|
|
aiohttp.ClientError("boom"),
|
|
OAuth2TokenRequestError(request_info=Mock(), domain=DOMAIN),
|
|
],
|
|
)
|
|
@pytest.mark.usefixtures("mock_yoto_client")
|
|
async def test_periodic_poll_fails_on_token_validation_error(
|
|
hass: HomeAssistant,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
side_effect: Exception,
|
|
) -> None:
|
|
"""A failure refreshing the OAuth token marks the coordinator failed."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
with patch(
|
|
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
|
|
side_effect=side_effect,
|
|
):
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
coordinator = mock_config_entry.runtime_data
|
|
assert coordinator.last_update_success is False
|
|
|
|
|
|
async def test_periodic_poll_fails_on_api_error(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""A non-auth API error during periodic refresh marks the coordinator failed."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.refresh.side_effect = YotoAPIError("boom")
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
coordinator = mock_config_entry.runtime_data
|
|
assert coordinator.last_update_success is False
|
|
|
|
|
|
def _build_second_player() -> YotoPlayer:
|
|
"""Build a second Yoto player discovered after setup."""
|
|
return YotoPlayer(
|
|
device=Device(
|
|
device_id="player-2",
|
|
name="Playroom Yoto",
|
|
device_type="v3",
|
|
device_family="v3",
|
|
generation="gen3",
|
|
),
|
|
is_online=True,
|
|
)
|
|
|
|
|
|
async def test_dynamic_device_added(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""A player discovered after setup gets its entity without a reload."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
assert hass.states.get("media_player.nursery_yoto") is not None
|
|
assert hass.states.get("media_player.playroom_yoto") is None
|
|
|
|
mock_yoto_client.players["player-2"] = _build_second_player()
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
assert hass.states.get("media_player.playroom_yoto") is not None
|
|
assert hass.states.get("binary_sensor.playroom_yoto_charging") is not None
|
|
assert hass.states.get("sensor.playroom_yoto_battery") is not None
|
|
assert hass.states.get("time.playroom_yoto_day_mode_start") is not None
|
|
assert hass.states.get("number.playroom_yoto_day_mode_brightness") is not None
|
|
assert hass.states.get("select.playroom_yoto_day_mode_color") is not None
|
|
assert hass.states.get("switch.playroom_yoto_bluetooth_pairing") is not None
|
|
mock_yoto_client.subscribe_player_events.assert_called_once_with("player-2")
|
|
|
|
|
|
async def test_subscription_failure_does_not_fail_refresh(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""A subscription error keeps the refreshed data and retries next cycle."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
mock_yoto_client.players["player-2"] = _build_second_player()
|
|
mock_yoto_client.subscribe_player_events.side_effect = YotoAPIError("boom")
|
|
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
assert hass.states.get("media_player.playroom_yoto") is not None
|
|
assert mock_config_entry.runtime_data.last_update_success is True
|
|
|
|
mock_yoto_client.subscribe_player_events.side_effect = None
|
|
mock_yoto_client.subscribe_player_events.reset_mock()
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
mock_yoto_client.subscribe_player_events.assert_called_once_with("player-2")
|
|
|
|
|
|
async def test_stale_device_removed(
|
|
hass: HomeAssistant,
|
|
mock_yoto_client: MagicMock,
|
|
mock_config_entry: MockConfigEntry,
|
|
device_registry: dr.DeviceRegistry,
|
|
freezer: FrozenDateTimeFactory,
|
|
) -> None:
|
|
"""A player removed from the account has its device dropped."""
|
|
await setup_integration(hass, mock_config_entry)
|
|
assert (
|
|
device_registry.async_get_device(identifiers={(DOMAIN, PLAYER_ID)}) is not None
|
|
)
|
|
|
|
mock_yoto_client.players.clear()
|
|
freezer.tick(SCAN_INTERVAL)
|
|
async_fire_time_changed(hass)
|
|
await hass.async_block_till_done()
|
|
|
|
assert device_registry.async_get_device(identifiers={(DOMAIN, PLAYER_ID)}) is None
|
|
mock_yoto_client.unsubscribe_player_events.assert_called_once_with(PLAYER_ID)
|