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

Add streaming media platform to Teslemetry (#140482)

Browse Source 
* Update media player

* Add media player platform with tests and bump firmware
This commit is contained in:
Brett Adams 2025-03-15 00:27:18 +10:00 committed by GitHub
parent de0efd61d1
commit 251bb30dc7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 523 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

312
homeassistant/components/teslemetry/media_player.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from tesla_fleet_api import VehicleSpecific
from tesla_fleet_api.const import Scope
from homeassistant.components.media_player import (
@ -12,9 +13,14 @@ from homeassistant.components.media_player import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from . import TeslemetryConfigEntry
from .entity import TeslemetryVehicleEntity
from .entity import (
TeslemetryRootEntity,
TeslemetryVehicleEntity,
TeslemetryVehicleStreamEntity,
)
from .helpers import handle_vehicle_command
from .models import TeslemetryVehicleData
@ -24,8 +30,16 @@ STATES = {
"Stopped": MediaPlayerState.IDLE,
"Off": MediaPlayerState.OFF,
}
VOLUME_MAX = 11.0
VOLUME_STEP = 1.0 / 3
DISPLAY_STATES = {
"On": MediaPlayerState.IDLE,
"Accessory": MediaPlayerState.IDLE,
"Charging": MediaPlayerState.OFF,
"Sentry": MediaPlayerState.OFF,
"Off": MediaPlayerState.OFF,
}
# Tesla uses 31 steps, in 0.333 increments up to 10.333
VOLUME_STEP = 1 / 31
VOLUME_FACTOR = 31 / 3 # 10.333
PARALLEL_UPDATES = 0
@ -38,68 +52,99 @@ async def async_setup_entry(
"""Set up the Teslemetry Media platform from a config entry."""
async_add_entities(
TeslemetryMediaEntity(vehicle, Scope.VEHICLE_CMDS in entry.runtime_data.scopes)
TeslemetryPollingMediaEntity(vehicle, entry.runtime_data.scopes)
if vehicle.api.pre2021 or vehicle.firmware < "2025.2.6"
else TeslemetryStreamingMediaEntity(vehicle, entry.runtime_data.scopes)
for vehicle in entry.runtime_data.vehicles
)
class TeslemetryMediaEntity(TeslemetryVehicleEntity, MediaPlayerEntity):
"""Vehicle media player class."""
class TeslemetryMediaEntity(TeslemetryRootEntity, MediaPlayerEntity):
"""Base vehicle media player class."""
api: VehicleSpecific
_attr_device_class = MediaPlayerDeviceClass.SPEAKER
_attr_supported_features = (
MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.VOLUME_SET
)
_volume_max: float = VOLUME_MAX
_attr_volume_step = VOLUME_STEP
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(self.api.adjust_volume(volume * VOLUME_FACTOR))
self._attr_volume_level = volume
self.async_write_ha_state()
async def async_media_play(self) -> None:
"""Send play command."""
if self.state != MediaPlayerState.PLAYING:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(self.api.media_toggle_playback())
self._attr_state = MediaPlayerState.PLAYING
self.async_write_ha_state()
async def async_media_pause(self) -> None:
"""Send pause command."""
if self.state == MediaPlayerState.PLAYING:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(self.api.media_toggle_playback())
self._attr_state = MediaPlayerState.PAUSED
self.async_write_ha_state()
async def async_media_next_track(self) -> None:
"""Send next track command."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(self.api.media_next_track())
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(self.api.media_prev_track())
class TeslemetryPollingMediaEntity(TeslemetryVehicleEntity, TeslemetryMediaEntity):
"""Polling vehicle media player class."""
def __init__(
self,
data: TeslemetryVehicleData,
scoped: bool,
scopes: list[Scope],
) -> None:
"""Initialize the media player entity."""
super().__init__(data, "media")
self.scoped = scoped
if not scoped:
self._attr_supported_features = (
MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.VOLUME_SET
)
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = MediaPlayerEntityFeature(0)
def _async_update_attrs(self) -> None:
"""Update entity attributes."""
self._volume_max = (
self.get("vehicle_state_media_info_audio_volume_max") or VOLUME_MAX
)
self._attr_state = STATES.get(
self.get("vehicle_state_media_info_media_playback_status") or "Off",
)
self._attr_volume_step = (
1.0
/ self._volume_max
/ (
self.get("vehicle_state_media_info_audio_volume_increment")
or VOLUME_STEP
)
)
state = self.get("vehicle_state_media_info_media_playback_status")
self._attr_state = STATES.get(state) if state else None
self._attr_volume_level = (
self.get("vehicle_state_media_info_audio_volume") or 0
) / VOLUME_FACTOR
if volume := self.get("vehicle_state_media_info_audio_volume"):
self._attr_volume_level = volume / self._volume_max
else:
self._attr_volume_level = None
duration = self.get("vehicle_state_media_info_now_playing_duration")
self._attr_media_duration = duration / 1000 if duration is not None else None
if duration := self.get("vehicle_state_media_info_now_playing_duration"):
self._attr_media_duration = duration / 1000
else:
self._attr_media_duration = None
if duration and (
position := self.get("vehicle_state_media_info_now_playing_elapsed")
):
self._attr_media_position = position / 1000
else:
self._attr_media_position = None
# Return media position only when a media duration is > 0.
elapsed = self.get("vehicle_state_media_info_now_playing_elapsed")
self._attr_media_position = (
elapsed / 1000 if duration and elapsed is not None else None
)
self._attr_media_title = self.get("vehicle_state_media_info_now_playing_title")
self._attr_media_artist = self.get(
@ -113,42 +158,151 @@ class TeslemetryMediaEntity(TeslemetryVehicleEntity, MediaPlayerEntity):
)
self._attr_source = self.get("vehicle_state_media_info_now_playing_source")
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(
self.api.adjust_volume(int(volume * self._volume_max))
class TeslemetryStreamingMediaEntity(
TeslemetryVehicleStreamEntity, TeslemetryMediaEntity, RestoreEntity
):
"""Streaming vehicle media player class."""
def __init__(
self,
data: TeslemetryVehicleData,
scopes: list[Scope],
) -> None:
"""Initialize the media player entity."""
super().__init__(data, "media")
self._attr_supported_features = (
MediaPlayerEntityFeature.NEXT_TRACK
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PREVIOUS_TRACK
| MediaPlayerEntityFeature.VOLUME_SET
)
self._attr_volume_level = volume
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = MediaPlayerEntityFeature(0)
async def async_added_to_hass(self) -> None:
"""Call when entity is added to hass."""
await super().async_added_to_hass()
if (state := await self.async_get_last_state()) is not None:
try:
self._attr_state = MediaPlayerState(state.state)
except ValueError:
self._attr_state = None
self._attr_volume_level = state.attributes.get("volume_level")
self._attr_media_title = state.attributes.get("media_title")
self._attr_media_artist = state.attributes.get("media_artist")
self._attr_media_album_name = state.attributes.get("media_album_name")
self._attr_media_playlist = state.attributes.get("media_playlist")
self._attr_media_duration = state.attributes.get("media_duration")
self._attr_media_position = state.attributes.get("media_position")
self._attr_source = state.attributes.get("source")
self.async_write_ha_state()
self.async_on_remove(
self.vehicle.stream_vehicle.listen_CenterDisplay(
self._async_handle_center_display
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaPlaybackStatus(
self._async_handle_media_playback_status
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaPlaybackSource(
self._async_handle_media_playback_source
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaAudioVolume(
self._async_handle_media_audio_volume
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingDuration(
self._async_handle_media_now_playing_duration
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingElapsed(
self._async_handle_media_now_playing_elapsed
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingArtist(
self._async_handle_media_now_playing_artist
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingAlbum(
self._async_handle_media_now_playing_album
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingTitle(
self._async_handle_media_now_playing_title
)
)
self.async_on_remove(
self.vehicle.stream_vehicle.listen_MediaNowPlayingStation(
self._async_handle_media_now_playing_station
)
)
def _async_handle_center_display(self, value: str | None) -> None:
"""Update entity attributes."""
if value is not None:
self._attr_state = DISPLAY_STATES.get(value)
self.async_write_ha_state()
def _async_handle_media_playback_status(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_state = MediaPlayerState.OFF if value is None else STATES.get(value)
self.async_write_ha_state()
async def async_media_play(self) -> None:
"""Send play command."""
if self.state != MediaPlayerState.PLAYING:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.media_toggle_playback())
self._attr_state = MediaPlayerState.PLAYING
self.async_write_ha_state()
def _async_handle_media_playback_source(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_source = value
self.async_write_ha_state()
async def async_media_pause(self) -> None:
"""Send pause command."""
if self.state == MediaPlayerState.PLAYING:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.media_toggle_playback())
self._attr_state = MediaPlayerState.PAUSED
self.async_write_ha_state()
def _async_handle_media_audio_volume(self, value: float | None) -> None:
"""Update entity attributes."""
self._attr_volume_level = None if value is None else value / VOLUME_FACTOR
self.async_write_ha_state()
async def async_media_next_track(self) -> None:
"""Send next track command."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.media_next_track())
def _async_handle_media_now_playing_duration(self, value: int | None) -> None:
"""Update entity attributes."""
self._attr_media_duration = None if value is None else int(value / 1000)
self.async_write_ha_state()
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.media_prev_track())
def _async_handle_media_now_playing_elapsed(self, value: int | None) -> None:
"""Update entity attributes."""
self._attr_media_position = None if value is None else int(value / 1000)
self.async_write_ha_state()
def _async_handle_media_now_playing_artist(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_media_artist = value # Check if this is album artist or not
self.async_write_ha_state()
def _async_handle_media_now_playing_album(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_media_album_name = value
self.async_write_ha_state()
def _async_handle_media_now_playing_title(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_media_title = value
self.async_write_ha_state()
def _async_handle_media_now_playing_station(self, value: str | None) -> None:
"""Update entity attributes."""
self._attr_media_channel = (
value # could also be _attr_media_playlist when Spotify
)
self.async_write_ha_state()

3
tests/components/teslemetry/const.py
View File

@ -18,7 +18,6 @@ VEHICLE_DATA_ALT = load_json_object_fixture("vehicle_data_alt.json", DOMAIN)
LIVE_STATUS = load_json_object_fixture("live_status.json", DOMAIN)
SITE_INFO = load_json_object_fixture("site_info.json", DOMAIN)
ENERGY_HISTORY = load_json_object_fixture("energy_history.json", DOMAIN)
METADATA = load_json_object_fixture("metadata.json", DOMAIN)
COMMAND_OK = {"response": {"result": True, "reason": ""}}
COMMAND_REASON = {"response": {"result": False, "reason": "already closed"}}
@ -52,7 +51,7 @@ METADATA = {
"proxy": False,
"access": True,
"polling": True,
"firmware": "2024.44.25",
"firmware": "2026.0.0",
}
},
}

22
tests/components/teslemetry/fixtures/metadata.json
View File

@ -1,22 +0,0 @@
{
"uid": "abc-123",
"region": "NA",
"scopes": [
"openid",
"offline_access",
"user_data",
"vehicle_device_data",
"vehicle_cmds",
"vehicle_charging_cmds",
"energy_device_data",
"energy_cmds"
],
"vehicles": {
"LRW3F7EK4NC700000": {
"access": true,
"polling": true,
"proxy": true,
"firmware": "2024.44.25"
}
}
}

2
tests/components/teslemetry/fixtures/vehicle_data.json
View File

@ -192,7 +192,7 @@
"api_version": 71,
"autopark_state_v2": "unavailable",
"calendar_supported": true,
"car_version": "2024.44.25 06f534d46010",
"car_version": "2026.0.0 06f534d46010",
"center_display_state": 0,
"dashcam_clip_save_available": true,
"dashcam_state": "Recording",

180
tests/components/teslemetry/snapshots/test_binary_sensor.ambr
View File

@ -1371,6 +1371,147 @@
'state': 'unknown',
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_favorite-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.test_located_at_favorite',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Located at favorite',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'located_at_favorite',
'unique_id': 'LRW3F7EK4NC700000-located_at_favorite',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_favorite-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at favorite',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_favorite',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_home-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.test_located_at_home',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Located at home',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'located_at_home',
'unique_id': 'LRW3F7EK4NC700000-located_at_home',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_home-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at home',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_home',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_work-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'binary_sensor',
'entity_category': None,
'entity_id': 'binary_sensor.test_located_at_work',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Located at work',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'located_at_work',
'unique_id': 'LRW3F7EK4NC700000-located_at_work',
'unit_of_measurement': None,
})
# ---
# name: test_binary_sensor[binary_sensor.test_located_at_work-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at work',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_work',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor[binary_sensor.test_offroad_lightbar-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -2801,6 +2942,45 @@
'state': 'unknown',
})
# ---
# name: test_binary_sensor_refresh[binary_sensor.test_located_at_favorite-statealt]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at favorite',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_favorite',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor_refresh[binary_sensor.test_located_at_home-statealt]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at home',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_home',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor_refresh[binary_sensor.test_located_at_work-statealt]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Located at work',
}),
'context': <ANY>,
'entity_id': 'binary_sensor.test_located_at_work',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_binary_sensor_refresh[binary_sensor.test_offroad_lightbar-statealt]
StateSnapshot({
'attributes': ReadOnlyDict({

2
tests/components/teslemetry/snapshots/test_diagnostics.ambr
View File

@ -375,7 +375,7 @@
'vehicle_state_api_version': 71,
'vehicle_state_autopark_state_v2': 'unavailable',
'vehicle_state_calendar_supported': True,
'vehicle_state_car_version': '2024.44.25 06f534d46010',
'vehicle_state_car_version': '2026.0.0 06f534d46010',
'vehicle_state_center_display_state': 0,
'vehicle_state_dashcam_clip_save_available': True,
'vehicle_state_dashcam_state': 'Recording',

42
tests/components/teslemetry/snapshots/test_media_player.ambr
View File

@ -47,7 +47,7 @@
'media_title': 'Chapter 51: Cybertruck: Tesla, 2018–2019',
'source': 'Audible',
'supported_features': <MediaPlayerEntityFeature: 16437>,
'volume_level': 0.16129355359011466,
'volume_level': 0.16129354838709678,
}),
'context': <ANY>,
'entity_id': 'media_player.test_media_player',
@ -64,10 +64,12 @@
'friendly_name': 'Test Media player',
'media_album_name': '',
'media_artist': '',
'media_duration': 0.0,
'media_playlist': '',
'media_title': '',
'source': 'Spotify',
'supported_features': <MediaPlayerEntityFeature: 16437>,
'volume_level': 0.0,
}),
'context': <ANY>,
'entity_id': 'media_player.test_media_player',
@ -125,7 +127,43 @@
'media_title': 'Chapter 51: Cybertruck: Tesla, 2018–2019',
'source': 'Audible',
'supported_features': <MediaPlayerEntityFeature: 0>,
'volume_level': 0.16129355359011466,
'volume_level': 0.16129354838709678,
}),
'context': <ANY>,
'entity_id': 'media_player.test_media_player',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'playing',
})
# ---
# name: test_update_streaming[off]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'speaker',
'friendly_name': 'Test Media player',
'supported_features': <MediaPlayerEntityFeature: 16437>,
}),
'context': <ANY>,
'entity_id': 'media_player.test_media_player',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_update_streaming[on]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'speaker',
'friendly_name': 'Test Media player',
'media_album_name': 'Test Album',
'media_artist': 'Test Artist',
'media_duration': 60,
'media_position': 5,
'source': 'Spotify',
'supported_features': <MediaPlayerEntityFeature: 16437>,
'volume_level': 0.1935483870967742,
}),
'context': <ANY>,
'entity_id': 'media_player.test_media_player',

2
tests/components/teslemetry/snapshots/test_update.ambr
View File

@ -41,7 +41,7 @@
'entity_picture': 'https://brands.home-assistant.io/_/teslemetry/icon.png',
'friendly_name': 'Test Update',
'in_progress': False,
'installed_version': '2024.44.25',
'installed_version': '2026.0.0',
'latest_version': '2024.12.0.0',
'release_summary': None,
'release_url': None,

67
tests/components/teslemetry/test_media_player.py
View File

@ -2,7 +2,9 @@
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from teslemetry_stream import Signal
from homeassistant.components.media_player import (
ATTR_MEDIA_VOLUME_LEVEL,
@ -18,7 +20,7 @@ from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import assert_entities, assert_entities_alt, setup_platform
from . import assert_entities, assert_entities_alt, reload_platform, setup_platform
from .const import COMMAND_OK, METADATA_NOSCOPE, VEHICLE_DATA_ALT
@ -26,6 +28,7 @@ async def test_media_player(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the media player entities are correct."""
@ -38,6 +41,7 @@ async def test_media_player_alt(
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data: AsyncMock,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the media player entities are correct."""
@ -51,6 +55,7 @@ async def test_media_player_noscope(
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_metadata: AsyncMock,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the media player entities are correct without required scope."""
@ -62,6 +67,7 @@ async def test_media_player_noscope(
async def test_media_player_services(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the media player services work."""
@ -137,3 +143,62 @@ async def test_media_player_services(
)
state = hass.states.get(entity_id)
call.assert_called_once()
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_update_streaming(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_vehicle_data: AsyncMock,
mock_add_listener: AsyncMock,
) -> None:
"""Tests that the media player entities with streaming are correct."""
entry = await setup_platform(hass, [Platform.MEDIA_PLAYER])
# Stream update
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.CENTER_DISPLAY: "Off",
Signal.MEDIA_PLAYBACK_STATUS: None,
Signal.MEDIA_PLAYBACK_SOURCE: None,
Signal.MEDIA_AUDIO_VOLUME: None,
Signal.MEDIA_NOW_PLAYING_DURATION: None,
Signal.MEDIA_NOW_PLAYING_ELAPSED: None,
Signal.MEDIA_NOW_PLAYING_ARTIST: None,
Signal.MEDIA_NOW_PLAYING_ALBUM: None,
},
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
state = hass.states.get("media_player.test_media_player")
assert state == snapshot(name="off")
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.CENTER_DISPLAY: "Driving",
Signal.MEDIA_PLAYBACK_STATUS: "Playing",
Signal.MEDIA_PLAYBACK_SOURCE: "Spotify",
Signal.MEDIA_AUDIO_VOLUME: 2,
Signal.MEDIA_NOW_PLAYING_DURATION: 60000,
Signal.MEDIA_NOW_PLAYING_ELAPSED: 5000,
Signal.MEDIA_NOW_PLAYING_ARTIST: "Test Artist",
Signal.MEDIA_NOW_PLAYING_ALBUM: "Test Album",
},
"createdAt": "2024-10-04T10:55:17.000Z",
}
)
await hass.async_block_till_done()
state = hass.states.get("media_player.test_media_player")
assert state == snapshot(name="on")
await reload_platform(hass, entry, [Platform.MEDIA_PLAYER])
# Ensure the restored state is the same as the previous state
state = hass.states.get("media_player.test_media_player")
assert state == snapshot(name="on")