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

Add Bang & Olufsen websocket testing (#123075)

Browse Source 
* Add websocket.py testing
Convert media_player testing dispatch events to the proper websocket events

* Fix WebSocket testing by using callbacks

* Add typing

* Add caplog checking
Check dispatch events directly
Check event bus directly
Avoid using internals

* Fix event and / dispatch callbacks not necessarily being checked

* Remove unnecessary caplog log level handling
This commit is contained in:
Markus Jacobsen 2024-08-22 16:31:15 +02:00 committed by GitHub
parent 404a7bab18
commit 890c6e97fd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 257 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
tests/components/bang_olufsen/conftest.py
View File

@ -245,8 +245,11 @@ def mock_mozart_client() -> Generator[AsyncMock]:
# Non-REST API client methods
client.check_device_connection = AsyncMock()
client.close_api_client = AsyncMock()
# WebSocket listener
client.connect_notifications = AsyncMock()
client.disconnect_notifications = Mock()
client.websocket_connected = False
yield client

166
tests/components/bang_olufsen/test_media_player.py
View File

@ -5,14 +5,18 @@ from contextlib import nullcontext as does_not_raise
import logging
from unittest.mock import AsyncMock, patch
from mozart_api.models import PlaybackContentMetadata, RenderingState, Source
from mozart_api.models import (
PlaybackContentMetadata,
RenderingState,
Source,
WebsocketNotificationTag,
)
import pytest
from homeassistant.components.bang_olufsen.const import (
BANG_OLUFSEN_STATES,
DOMAIN,
BangOlufsenSource,
WebsocketNotification,
)
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
@ -38,7 +42,6 @@ from homeassistant.components.media_player import (
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
from .const import (
@ -59,7 +62,6 @@ from .const import (
TEST_PLAYBACK_STATE_TURN_OFF,
TEST_RADIO_STATION,
TEST_SEEK_POSITION_HOME_ASSISTANT_FORMAT,
TEST_SERIAL_NUMBER,
TEST_SOURCES,
TEST_VIDEO_SOURCES,
TEST_VOLUME,
@ -131,6 +133,29 @@ async def test_async_update_sources_outdated_api(
)
async def test_async_update_sources_remote(
hass: HomeAssistant, mock_mozart_client, mock_config_entry: MockConfigEntry
) -> None:
"""Test _async_update_sources is called when there are new video sources."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
notification_callback = mock_mozart_client.get_notification_notifications.call_args[
0
][0]
# This is not an ideal check, but I couldn't get anything else to work
assert mock_mozart_client.get_available_sources.call_count == 1
assert mock_mozart_client.get_remote_menu.call_count == 1
# Send the remote menu Websocket event
notification_callback(WebsocketNotificationTag(value="remoteMenuChanged"))
assert mock_mozart_client.get_available_sources.call_count == 2
assert mock_mozart_client.get_remote_menu.call_count == 2
async def test_async_update_playback_metadata(
hass: HomeAssistant,
mock_mozart_client: AsyncMock,
@ -141,6 +166,10 @@ async def test_async_update_playback_metadata(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
playback_metadata_callback = (
mock_mozart_client.get_playback_metadata_notifications.call_args[0][0]
)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert ATTR_MEDIA_DURATION not in states.attributes
assert ATTR_MEDIA_TITLE not in states.attributes
@ -150,11 +179,7 @@ async def test_async_update_playback_metadata(
assert ATTR_MEDIA_CHANNEL not in states.attributes
# Send the WebSocket event dispatch
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_METADATA}",
TEST_PLAYBACK_METADATA,
)
playback_metadata_callback(TEST_PLAYBACK_METADATA)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert (
@ -181,13 +206,13 @@ async def test_async_update_playback_error(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
# The async_dispatcher_send function seems to swallow exceptions, making pytest.raises unusable
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_ERROR}",
TEST_PLAYBACK_ERROR,
playback_error_callback = (
mock_mozart_client.get_playback_error_notifications.call_args[0][0]
)
# The async_dispatcher_send function seems to swallow exceptions, making pytest.raises unusable
playback_error_callback(TEST_PLAYBACK_ERROR)
assert (
"Exception in _async_update_playback_error when dispatching '11111111_playback_error': (PlaybackError(error='Test error', item=None),)"
in caplog.text
@ -204,16 +229,16 @@ async def test_async_update_playback_progress(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
playback_progress_callback = (
mock_mozart_client.get_playback_progress_notifications.call_args[0][0]
)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert ATTR_MEDIA_POSITION not in states.attributes
old_updated_at = states.attributes[ATTR_MEDIA_POSITION_UPDATED_AT]
assert old_updated_at
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_PROGRESS}",
TEST_PLAYBACK_PROGRESS,
)
playback_progress_callback(TEST_PLAYBACK_PROGRESS)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.attributes[ATTR_MEDIA_POSITION] == TEST_PLAYBACK_PROGRESS.progress
@ -232,14 +257,14 @@ async def test_async_update_playback_state(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
playback_state_callback = (
mock_mozart_client.get_playback_state_notifications.call_args[0][0]
)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.state == MediaPlayerState.PLAYING
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_STATE}",
TEST_PLAYBACK_STATE_PAUSED,
)
playback_state_callback(TEST_PLAYBACK_STATE_PAUSED)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.state == TEST_PLAYBACK_STATE_PAUSED.value
@ -314,29 +339,26 @@ async def test_async_update_source_change(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
playback_progress_callback = (
mock_mozart_client.get_playback_progress_notifications.call_args[0][0]
)
playback_metadata_callback = (
mock_mozart_client.get_playback_metadata_notifications.call_args[0][0]
)
source_change_callback = (
mock_mozart_client.get_source_change_notifications.call_args[0][0]
)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert ATTR_INPUT_SOURCE not in states.attributes
assert states.attributes[ATTR_MEDIA_CONTENT_TYPE] == MediaType.MUSIC
# Simulate progress attribute being available
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_PROGRESS}",
TEST_PLAYBACK_PROGRESS,
)
playback_progress_callback(TEST_PLAYBACK_PROGRESS)
# Simulate metadata
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_METADATA}",
metadata,
)
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.SOURCE_CHANGE}",
reported_source,
)
playback_metadata_callback(metadata)
source_change_callback(reported_source)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.attributes[ATTR_INPUT_SOURCE] == real_source.name
@ -354,6 +376,10 @@ async def test_async_turn_off(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
playback_state_callback = (
mock_mozart_client.get_playback_state_notifications.call_args[0][0]
)
await hass.services.async_call(
"media_player",
"turn_off",
@ -361,11 +387,7 @@ async def test_async_turn_off(
blocking=True,
)
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_STATE}",
TEST_PLAYBACK_STATE_TURN_OFF,
)
playback_state_callback(TEST_PLAYBACK_STATE_TURN_OFF)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.state == BANG_OLUFSEN_STATES[TEST_PLAYBACK_STATE_TURN_OFF.value]
@ -384,6 +406,8 @@ async def test_async_set_volume_level(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
volume_callback = mock_mozart_client.get_volume_notifications.call_args[0][0]
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert ATTR_MEDIA_VOLUME_LEVEL not in states.attributes
@ -398,11 +422,7 @@ async def test_async_set_volume_level(
)
# The service call will trigger a WebSocket notification
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.VOLUME}",
TEST_VOLUME,
)
volume_callback(TEST_VOLUME)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert (
@ -424,6 +444,8 @@ async def test_async_mute_volume(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
volume_callback = mock_mozart_client.get_volume_notifications.call_args[0][0]
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert ATTR_MEDIA_VOLUME_MUTED not in states.attributes
@ -438,11 +460,7 @@ async def test_async_mute_volume(
)
# The service call will trigger a WebSocket notification
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.VOLUME}",
TEST_VOLUME_MUTED,
)
volume_callback(TEST_VOLUME_MUTED)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert (
@ -476,13 +494,13 @@ async def test_async_media_play_pause(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Set the initial state
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_STATE}",
initial_state,
playback_state_callback = (
mock_mozart_client.get_playback_state_notifications.call_args[0][0]
)
# Set the initial state
playback_state_callback(initial_state)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.state == BANG_OLUFSEN_STATES[initial_state.value]
@ -506,13 +524,13 @@ async def test_async_media_stop(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Set the state to playing
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.PLAYBACK_STATE}",
TEST_PLAYBACK_STATE_PLAYING,
playback_state_callback = (
mock_mozart_client.get_playback_state_notifications.call_args[0][0]
)
# Set the state to playing
playback_state_callback(TEST_PLAYBACK_STATE_PLAYING)
states = hass.states.get(TEST_MEDIA_PLAYER_ENTITY_ID)
assert states.state == BANG_OLUFSEN_STATES[TEST_PLAYBACK_STATE_PLAYING.value]
@ -569,13 +587,13 @@ async def test_async_media_seek(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
# Set the source
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.SOURCE_CHANGE}",
source,
source_change_callback = (
mock_mozart_client.get_source_change_notifications.call_args[0][0]
)
# Set the source
source_change_callback(source)
# Check results
with expected_result:
await hass.services.async_call(
@ -801,12 +819,10 @@ async def test_async_play_media_overlay_offset_volume_tts(
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
volume_callback = mock_mozart_client.get_volume_notifications.call_args[0][0]
# Set the volume to enable offset
async_dispatcher_send(
hass,
f"{TEST_SERIAL_NUMBER}_{WebsocketNotification.VOLUME}",
TEST_VOLUME,
)
volume_callback(TEST_VOLUME)
await hass.services.async_call(
"media_player",

163
tests/components/bang_olufsen/test_websocket.py Normal file
View File

@ -0,0 +1,163 @@
"""Test the Bang & Olufsen WebSocket listener."""
import logging
from unittest.mock import AsyncMock, Mock
from mozart_api.models import SoftwareUpdateState
import pytest
from homeassistant.components.bang_olufsen.const import (
BANG_OLUFSEN_WEBSOCKET_EVENT,
CONNECTION_STATUS,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceRegistry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import TEST_NAME
from tests.common import MockConfigEntry
async def test_connection(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_config_entry: MockConfigEntry,
mock_mozart_client: AsyncMock,
) -> None:
"""Test on_connection and on_connection_lost logs and calls correctly."""
mock_mozart_client.websocket_connected = True
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
connection_callback = mock_mozart_client.get_on_connection.call_args[0][0]
caplog.set_level(logging.DEBUG)
mock_connection_callback = Mock()
async_dispatcher_connect(
hass,
f"{mock_config_entry.unique_id}_{CONNECTION_STATUS}",
mock_connection_callback,
)
# Call the WebSocket connection status method
connection_callback()
await hass.async_block_till_done()
mock_connection_callback.assert_called_once_with(True)
assert f"Connected to the {TEST_NAME} notification channel" in caplog.text
async def test_connection_lost(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mock_config_entry: MockConfigEntry,
mock_mozart_client: AsyncMock,
) -> None:
"""Test on_connection_lost logs and calls correctly."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
connection_lost_callback = mock_mozart_client.get_on_connection_lost.call_args[0][0]
mock_connection_lost_callback = Mock()
async_dispatcher_connect(
hass,
f"{mock_config_entry.unique_id}_{CONNECTION_STATUS}",
mock_connection_lost_callback,
)
connection_lost_callback()
await hass.async_block_till_done()
mock_connection_lost_callback.assert_called_once_with(False)
assert f"Lost connection to the {TEST_NAME}" in caplog.text
async def test_on_software_update_state(
hass: HomeAssistant,
device_registry: DeviceRegistry,
mock_config_entry: MockConfigEntry,
mock_mozart_client: AsyncMock,
) -> None:
"""Test software version is updated through on_software_update_state."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
software_update_state_callback = (
mock_mozart_client.get_software_update_state_notifications.call_args[0][0]
)
# Trigger the notification
await software_update_state_callback(SoftwareUpdateState())
await hass.async_block_till_done()
device = device_registry.async_get_device(
identifiers={(DOMAIN, mock_config_entry.unique_id)}
)
assert device.sw_version == "1.0.0"
async def test_on_all_notifications_raw(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
device_registry: DeviceRegistry,
mock_config_entry: MockConfigEntry,
mock_mozart_client: AsyncMock,
) -> None:
"""Test on_all_notifications_raw logs and fires as expected."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
all_notifications_raw_callback = (
mock_mozart_client.get_all_notifications_raw.call_args[0][0]
)
raw_notification = {
"eventData": {
"default": {"level": 40},
"level": {"level": 40},
"maximum": {"level": 100},
"muted": {"muted": False},
},
"eventType": "WebSocketEventVolume",
}
raw_notification_full = raw_notification
# Get device ID for the modified notification that is sent as an event and in the log
device = device_registry.async_get_device(
identifiers={(DOMAIN, mock_config_entry.unique_id)}
)
raw_notification_full.update(
{
"device_id": device.id,
"serial_number": mock_config_entry.unique_id,
}
)
caplog.set_level(logging.DEBUG)
mock_event_callback = Mock()
# Listen to BANG_OLUFSEN_WEBSOCKET_EVENT events
hass.bus.async_listen(BANG_OLUFSEN_WEBSOCKET_EVENT, mock_event_callback)
# Trigger the notification
all_notifications_raw_callback(raw_notification)
await hass.async_block_till_done()
assert str(raw_notification_full) in caplog.text
mocked_call = mock_event_callback.call_args[0][0].as_dict()
assert mocked_call["event_type"] == BANG_OLUFSEN_WEBSOCKET_EVENT
assert mocked_call["data"] == raw_notification_full