diff --git a/.strict-typing b/.strict-typing
index 6a6918543ad544f65030f491203a9585017b6a5a..4bfacaa64f44ac135d8c865e5d8d84817abb42c3 100644
--- a/.strict-typing
+++ b/.strict-typing
@@ -324,7 +324,6 @@ homeassistant.components.moon.*
homeassistant.components.mopeka.*
homeassistant.components.motionmount.*
homeassistant.components.mqtt.*
-homeassistant.components.music_assistant.*
homeassistant.components.my.*
homeassistant.components.mysensors.*
homeassistant.components.myuplink.*
diff --git a/CODEOWNERS b/CODEOWNERS
index 99cfefa81c6048b3df171852984e0c4b12ec0be1..5cda5610f6c7f6a00935dbca14c6163a4641288b 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -954,8 +954,6 @@ build.json @home-assistant/supervisor
/homeassistant/components/msteams/ @peroyvind
/homeassistant/components/mullvad/ @meichthys
/tests/components/mullvad/ @meichthys
-/homeassistant/components/music_assistant/ @music-assistant
-/tests/components/music_assistant/ @music-assistant
/homeassistant/components/mutesync/ @currentoor
/tests/components/mutesync/ @currentoor
/homeassistant/components/my/ @home-assistant/core
diff --git a/homeassistant/components/music_assistant/__init__.py b/homeassistant/components/music_assistant/__init__.py
deleted file mode 100644
index 9f0fc1aad276012a52d6e57d75adec0e244143a3..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/__init__.py
+++ /dev/null
@@ -1,164 +0,0 @@
-"""Music Assistant (music-assistant.io) integration."""
-
-from __future__ import annotations
-
-import asyncio
-from dataclasses import dataclass
-from typing import TYPE_CHECKING
-
-from music_assistant_client import MusicAssistantClient
-from music_assistant_client.exceptions import CannotConnect, InvalidServerVersion
-from music_assistant_models.enums import EventType
-from music_assistant_models.errors import MusicAssistantError
-
-from homeassistant.config_entries import ConfigEntry, ConfigEntryState
-from homeassistant.const import CONF_URL, EVENT_HOMEASSISTANT_STOP, Platform
-from homeassistant.core import Event, HomeAssistant
-from homeassistant.exceptions import ConfigEntryNotReady
-from homeassistant.helpers import device_registry as dr
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.issue_registry import (
- IssueSeverity,
- async_create_issue,
- async_delete_issue,
-)
-
-from .const import DOMAIN, LOGGER
-
-if TYPE_CHECKING:
- from music_assistant_models.event import MassEvent
-
-type MusicAssistantConfigEntry = ConfigEntry[MusicAssistantEntryData]
-
-PLATFORMS = [Platform.MEDIA_PLAYER]
-
-CONNECT_TIMEOUT = 10
-LISTEN_READY_TIMEOUT = 30
-
-
-@dataclass
-class MusicAssistantEntryData:
- """Hold Mass data for the config entry."""
-
- mass: MusicAssistantClient
- listen_task: asyncio.Task
-
-
-async def async_setup_entry(
- hass: HomeAssistant, entry: MusicAssistantConfigEntry
-) -> bool:
- """Set up from a config entry."""
- http_session = async_get_clientsession(hass, verify_ssl=False)
- mass_url = entry.data[CONF_URL]
- mass = MusicAssistantClient(mass_url, http_session)
-
- try:
- async with asyncio.timeout(CONNECT_TIMEOUT):
- await mass.connect()
- except (TimeoutError, CannotConnect) as err:
- raise ConfigEntryNotReady(
- f"Failed to connect to music assistant server {mass_url}"
- ) from err
- except InvalidServerVersion as err:
- async_create_issue(
- hass,
- DOMAIN,
- "invalid_server_version",
- is_fixable=False,
- severity=IssueSeverity.ERROR,
- translation_key="invalid_server_version",
- )
- raise ConfigEntryNotReady(f"Invalid server version: {err}") from err
- except MusicAssistantError as err:
- LOGGER.exception("Failed to connect to music assistant server", exc_info=err)
- raise ConfigEntryNotReady(
- f"Unknown error connecting to the Music Assistant server {mass_url}"
- ) from err
-
- async_delete_issue(hass, DOMAIN, "invalid_server_version")
-
- async def on_hass_stop(event: Event) -> None:
- """Handle incoming stop event from Home Assistant."""
- await mass.disconnect()
-
- entry.async_on_unload(
- hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, on_hass_stop)
- )
-
- # launch the music assistant client listen task in the background
- # use the init_ready event to wait until initialization is done
- init_ready = asyncio.Event()
- listen_task = asyncio.create_task(_client_listen(hass, entry, mass, init_ready))
-
- try:
- async with asyncio.timeout(LISTEN_READY_TIMEOUT):
- await init_ready.wait()
- except TimeoutError as err:
- listen_task.cancel()
- raise ConfigEntryNotReady("Music Assistant client not ready") from err
-
- entry.runtime_data = MusicAssistantEntryData(mass, listen_task)
-
- # If the listen task is already failed, we need to raise ConfigEntryNotReady
- if listen_task.done() and (listen_error := listen_task.exception()) is not None:
- await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
- try:
- await mass.disconnect()
- finally:
- raise ConfigEntryNotReady(listen_error) from listen_error
-
- # initialize platforms
- await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
-
- # register listener for removed players
- async def handle_player_removed(event: MassEvent) -> None:
- """Handle Mass Player Removed event."""
- if event.object_id is None:
- return
- dev_reg = dr.async_get(hass)
- if hass_device := dev_reg.async_get_device({(DOMAIN, event.object_id)}):
- dev_reg.async_update_device(
- hass_device.id, remove_config_entry_id=entry.entry_id
- )
-
- entry.async_on_unload(
- mass.subscribe(handle_player_removed, EventType.PLAYER_REMOVED)
- )
-
- return True
-
-
-async def _client_listen(
- hass: HomeAssistant,
- entry: ConfigEntry,
- mass: MusicAssistantClient,
- init_ready: asyncio.Event,
-) -> None:
- """Listen with the client."""
- try:
- await mass.start_listening(init_ready)
- except MusicAssistantError as err:
- if entry.state != ConfigEntryState.LOADED:
- raise
- LOGGER.error("Failed to listen: %s", err)
- except Exception as err: # pylint: disable=broad-except
- # We need to guard against unknown exceptions to not crash this task.
- if entry.state != ConfigEntryState.LOADED:
- raise
- LOGGER.exception("Unexpected exception: %s", err)
-
- if not hass.is_stopping:
- LOGGER.debug("Disconnected from server. Reloading integration")
- hass.async_create_task(hass.config_entries.async_reload(entry.entry_id))
-
-
-async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
- """Unload a config entry."""
- unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
-
- if unload_ok:
- mass_entry_data: MusicAssistantEntryData = entry.runtime_data
- mass_entry_data.listen_task.cancel()
- await mass_entry_data.mass.disconnect()
-
- return unload_ok
diff --git a/homeassistant/components/music_assistant/config_flow.py b/homeassistant/components/music_assistant/config_flow.py
deleted file mode 100644
index fc50a2d654bf7b3811db70c751dcdb2ae75fc5f2..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/config_flow.py
+++ /dev/null
@@ -1,137 +0,0 @@
-"""Config flow for MusicAssistant integration."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Any
-
-from music_assistant_client import MusicAssistantClient
-from music_assistant_client.exceptions import (
- CannotConnect,
- InvalidServerVersion,
- MusicAssistantClientException,
-)
-from music_assistant_models.api import ServerInfoMessage
-import voluptuous as vol
-
-from homeassistant.components import zeroconf
-from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
-from homeassistant.const import CONF_URL
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers import aiohttp_client
-
-from .const import DOMAIN, LOGGER
-
-DEFAULT_URL = "http://mass.local:8095"
-DEFAULT_TITLE = "Music Assistant"
-
-
-def get_manual_schema(user_input: dict[str, Any]) -> vol.Schema:
- """Return a schema for the manual step."""
- default_url = user_input.get(CONF_URL, DEFAULT_URL)
- return vol.Schema(
- {
- vol.Required(CONF_URL, default=default_url): str,
- }
- )
-
-
-async def get_server_info(hass: HomeAssistant, url: str) -> ServerInfoMessage:
- """Validate the user input allows us to connect."""
- async with MusicAssistantClient(
- url, aiohttp_client.async_get_clientsession(hass)
- ) as client:
- if TYPE_CHECKING:
- assert client.server_info is not None
- return client.server_info
-
-
-class MusicAssistantConfigFlow(ConfigFlow, domain=DOMAIN):
- """Handle a config flow for MusicAssistant."""
-
- VERSION = 1
-
- def __init__(self) -> None:
- """Set up flow instance."""
- self.server_info: ServerInfoMessage | None = None
-
- async def async_step_user(
- self, user_input: dict[str, Any] | None = None
- ) -> ConfigFlowResult:
- """Handle a manual configuration."""
- errors: dict[str, str] = {}
- if user_input is not None:
- try:
- self.server_info = await get_server_info(
- self.hass, user_input[CONF_URL]
- )
- await self.async_set_unique_id(
- self.server_info.server_id, raise_on_progress=False
- )
- self._abort_if_unique_id_configured(
- updates={CONF_URL: self.server_info.base_url},
- reload_on_update=True,
- )
- except CannotConnect:
- errors["base"] = "cannot_connect"
- except InvalidServerVersion:
- errors["base"] = "invalid_server_version"
- except MusicAssistantClientException:
- LOGGER.exception("Unexpected exception")
- errors["base"] = "unknown"
- else:
- return self.async_create_entry(
- title=DEFAULT_TITLE,
- data={
- CONF_URL: self.server_info.base_url,
- },
- )
-
- return self.async_show_form(
- step_id="user", data_schema=get_manual_schema(user_input), errors=errors
- )
-
- return self.async_show_form(step_id="user", data_schema=get_manual_schema({}))
-
- async def async_step_zeroconf(
- self, discovery_info: zeroconf.ZeroconfServiceInfo
- ) -> ConfigFlowResult:
- """Handle a discovered Mass server.
-
- This flow is triggered by the Zeroconf component. It will check if the
- host is already configured and delegate to the import step if not.
- """
- # abort if discovery info is not what we expect
- if "server_id" not in discovery_info.properties:
- return self.async_abort(reason="missing_server_id")
- # abort if we already have exactly this server_id
- # reload the integration if the host got updated
- self.server_info = ServerInfoMessage.from_dict(discovery_info.properties)
- await self.async_set_unique_id(self.server_info.server_id)
- self._abort_if_unique_id_configured(
- updates={CONF_URL: self.server_info.base_url},
- reload_on_update=True,
- )
- try:
- await get_server_info(self.hass, self.server_info.base_url)
- except CannotConnect:
- return self.async_abort(reason="cannot_connect")
- return await self.async_step_discovery_confirm()
-
- async def async_step_discovery_confirm(
- self, user_input: dict[str, Any] | None = None
- ) -> ConfigFlowResult:
- """Handle user-confirmation of discovered server."""
- if TYPE_CHECKING:
- assert self.server_info is not None
- if user_input is not None:
- return self.async_create_entry(
- title=DEFAULT_TITLE,
- data={
- CONF_URL: self.server_info.base_url,
- },
- )
- self._set_confirm_only()
- return self.async_show_form(
- step_id="discovery_confirm",
- description_placeholders={"url": self.server_info.base_url},
- )
diff --git a/homeassistant/components/music_assistant/const.py b/homeassistant/components/music_assistant/const.py
deleted file mode 100644
index 6512f58b96c74b04ab7ad98a330dae12cffc509c..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/const.py
+++ /dev/null
@@ -1,18 +0,0 @@
-"""Constants for Music Assistant Component."""
-
-import logging
-
-DOMAIN = "music_assistant"
-DOMAIN_EVENT = f"{DOMAIN}_event"
-
-DEFAULT_NAME = "Music Assistant"
-
-ATTR_IS_GROUP = "is_group"
-ATTR_GROUP_MEMBERS = "group_members"
-ATTR_GROUP_PARENTS = "group_parents"
-
-ATTR_MASS_PLAYER_TYPE = "mass_player_type"
-ATTR_ACTIVE_QUEUE = "active_queue"
-ATTR_STREAM_TITLE = "stream_title"
-
-LOGGER = logging.getLogger(__package__)
diff --git a/homeassistant/components/music_assistant/entity.py b/homeassistant/components/music_assistant/entity.py
deleted file mode 100644
index f5b6d92b0cfa9474c666e426e91cb918070347d8..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/entity.py
+++ /dev/null
@@ -1,86 +0,0 @@
-"""Base entity model."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from music_assistant_models.enums import EventType
-from music_assistant_models.event import MassEvent
-from music_assistant_models.player import Player
-
-from homeassistant.helpers.device_registry import DeviceInfo
-from homeassistant.helpers.entity import Entity
-
-from .const import DOMAIN
-
-if TYPE_CHECKING:
- from music_assistant_client import MusicAssistantClient
-
-
-class MusicAssistantEntity(Entity):
- """Base Entity from Music Assistant Player."""
-
- _attr_has_entity_name = True
- _attr_should_poll = False
-
- def __init__(self, mass: MusicAssistantClient, player_id: str) -> None:
- """Initialize MediaPlayer entity."""
- self.mass = mass
- self.player_id = player_id
- provider = self.mass.get_provider(self.player.provider)
- if TYPE_CHECKING:
- assert provider is not None
- self._attr_device_info = DeviceInfo(
- identifiers={(DOMAIN, player_id)},
- manufacturer=self.player.device_info.manufacturer or provider.name,
- model=self.player.device_info.model or self.player.name,
- name=self.player.display_name,
- configuration_url=f"{mass.server_url}/#/settings/editplayer/{player_id}",
- )
-
- async def async_added_to_hass(self) -> None:
- """Register callbacks."""
- await self.async_on_update()
- self.async_on_remove(
- self.mass.subscribe(
- self.__on_mass_update, EventType.PLAYER_UPDATED, self.player_id
- )
- )
- self.async_on_remove(
- self.mass.subscribe(
- self.__on_mass_update,
- EventType.QUEUE_UPDATED,
- )
- )
-
- @property
- def player(self) -> Player:
- """Return the Mass Player attached to this HA entity."""
- return self.mass.players[self.player_id]
-
- @property
- def unique_id(self) -> str | None:
- """Return unique id for entity."""
- _base = self.player_id
- if hasattr(self, "entity_description"):
- return f"{_base}_{self.entity_description.key}"
- return _base
-
- @property
- def available(self) -> bool:
- """Return availability of entity."""
- return self.player.available and bool(self.mass.connection.connected)
-
- async def __on_mass_update(self, event: MassEvent) -> None:
- """Call when we receive an event from MusicAssistant."""
- if event.event == EventType.QUEUE_UPDATED and event.object_id not in (
- self.player.active_source,
- self.player.active_group,
- self.player.player_id,
- ):
- return
- await self.async_on_update()
- self.async_write_ha_state()
-
- async def async_on_update(self) -> None:
- """Handle player updates."""
diff --git a/homeassistant/components/music_assistant/manifest.json b/homeassistant/components/music_assistant/manifest.json
deleted file mode 100644
index c3e05d7a55f3d7ce8e07eb34441a9be556d1f71f..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/manifest.json
+++ /dev/null
@@ -1,13 +0,0 @@
-{
- "domain": "music_assistant",
- "name": "Music Assistant",
- "after_dependencies": ["media_source", "media_player"],
- "codeowners": ["@music-assistant"],
- "config_flow": true,
- "documentation": "https://music-assistant.io",
- "iot_class": "local_push",
- "issue_tracker": "https://github.com/music-assistant/hass-music-assistant/issues",
- "loggers": ["music_assistant"],
- "requirements": ["music-assistant-client==1.0.3"],
- "zeroconf": ["_mass._tcp.local."]
-}
diff --git a/homeassistant/components/music_assistant/media_player.py b/homeassistant/components/music_assistant/media_player.py
deleted file mode 100644
index f0f3675ee32cefd44506b87eb886e63e014e73b1..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/media_player.py
+++ /dev/null
@@ -1,557 +0,0 @@
-"""MediaPlayer platform for Music Assistant integration."""
-
-from __future__ import annotations
-
-import asyncio
-from collections.abc import Awaitable, Callable, Coroutine, Mapping
-from contextlib import suppress
-import functools
-import os
-from typing import TYPE_CHECKING, Any
-
-from music_assistant_models.enums import (
- EventType,
- MediaType,
- PlayerFeature,
- QueueOption,
- RepeatMode as MassRepeatMode,
-)
-from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
-from music_assistant_models.event import MassEvent
-from music_assistant_models.media_items import ItemMapping, MediaItemType, Track
-
-from homeassistant.components import media_source
-from homeassistant.components.media_player import (
- ATTR_MEDIA_EXTRA,
- BrowseMedia,
- MediaPlayerDeviceClass,
- MediaPlayerEnqueue,
- MediaPlayerEntity,
- MediaPlayerEntityFeature,
- MediaPlayerState,
- MediaType as HAMediaType,
- RepeatMode,
- async_process_play_media_url,
-)
-from homeassistant.const import STATE_OFF
-from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
-from homeassistant.helpers import entity_registry as er
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.util.dt import utc_from_timestamp
-
-from . import MusicAssistantConfigEntry
-from .const import ATTR_ACTIVE_QUEUE, ATTR_MASS_PLAYER_TYPE, DOMAIN
-from .entity import MusicAssistantEntity
-
-if TYPE_CHECKING:
- from music_assistant_client import MusicAssistantClient
- from music_assistant_models.player import Player
- from music_assistant_models.player_queue import PlayerQueue
-
-SUPPORTED_FEATURES = (
- MediaPlayerEntityFeature.PAUSE
- | MediaPlayerEntityFeature.VOLUME_SET
- | MediaPlayerEntityFeature.STOP
- | MediaPlayerEntityFeature.PREVIOUS_TRACK
- | MediaPlayerEntityFeature.NEXT_TRACK
- | MediaPlayerEntityFeature.SHUFFLE_SET
- | MediaPlayerEntityFeature.REPEAT_SET
- | MediaPlayerEntityFeature.TURN_ON
- | MediaPlayerEntityFeature.TURN_OFF
- | MediaPlayerEntityFeature.PLAY
- | MediaPlayerEntityFeature.PLAY_MEDIA
- | MediaPlayerEntityFeature.VOLUME_STEP
- | MediaPlayerEntityFeature.CLEAR_PLAYLIST
- | MediaPlayerEntityFeature.BROWSE_MEDIA
- | MediaPlayerEntityFeature.MEDIA_ENQUEUE
- | MediaPlayerEntityFeature.MEDIA_ANNOUNCE
- | MediaPlayerEntityFeature.SEEK
-)
-
-QUEUE_OPTION_MAP = {
- # map from HA enqueue options to MA enqueue options
- # which are the same but just in case
- MediaPlayerEnqueue.ADD: QueueOption.ADD,
- MediaPlayerEnqueue.NEXT: QueueOption.NEXT,
- MediaPlayerEnqueue.PLAY: QueueOption.PLAY,
- MediaPlayerEnqueue.REPLACE: QueueOption.REPLACE,
-}
-
-ATTR_RADIO_MODE = "radio_mode"
-ATTR_MEDIA_ID = "media_id"
-ATTR_MEDIA_TYPE = "media_type"
-ATTR_ARTIST = "artist"
-ATTR_ALBUM = "album"
-ATTR_URL = "url"
-ATTR_USE_PRE_ANNOUNCE = "use_pre_announce"
-ATTR_ANNOUNCE_VOLUME = "announce_volume"
-ATTR_SOURCE_PLAYER = "source_player"
-ATTR_AUTO_PLAY = "auto_play"
-
-
-def catch_musicassistant_error[_R, **P](
- func: Callable[..., Awaitable[_R]],
-) -> Callable[..., Coroutine[Any, Any, _R | None]]:
- """Check and log commands to players."""
-
- @functools.wraps(func)
- async def wrapper(
- self: MusicAssistantPlayer, *args: P.args, **kwargs: P.kwargs
- ) -> _R | None:
- """Catch Music Assistant errors and convert to Home Assistant error."""
- try:
- return await func(self, *args, **kwargs)
- except MusicAssistantError as err:
- error_msg = str(err) or err.__class__.__name__
- raise HomeAssistantError(error_msg) from err
-
- return wrapper
-
-
-async def async_setup_entry(
- hass: HomeAssistant,
- entry: MusicAssistantConfigEntry,
- async_add_entities: AddEntitiesCallback,
-) -> None:
- """Set up Music Assistant MediaPlayer(s) from Config Entry."""
- mass = entry.runtime_data.mass
- added_ids = set()
-
- async def handle_player_added(event: MassEvent) -> None:
- """Handle Mass Player Added event."""
- if TYPE_CHECKING:
- assert event.object_id is not None
- if event.object_id in added_ids:
- return
- added_ids.add(event.object_id)
- async_add_entities([MusicAssistantPlayer(mass, event.object_id)])
-
- # register listener for new players
- entry.async_on_unload(mass.subscribe(handle_player_added, EventType.PLAYER_ADDED))
- mass_players = []
- # add all current players
- for player in mass.players:
- added_ids.add(player.player_id)
- mass_players.append(MusicAssistantPlayer(mass, player.player_id))
-
- async_add_entities(mass_players)
-
-
-class MusicAssistantPlayer(MusicAssistantEntity, MediaPlayerEntity):
- """Representation of MediaPlayerEntity from Music Assistant Player."""
-
- _attr_name = None
- _attr_media_image_remotely_accessible = True
- _attr_media_content_type = HAMediaType.MUSIC
-
- def __init__(self, mass: MusicAssistantClient, player_id: str) -> None:
- """Initialize MediaPlayer entity."""
- super().__init__(mass, player_id)
- self._attr_icon = self.player.icon.replace("mdi-", "mdi:")
- self._attr_supported_features = SUPPORTED_FEATURES
- if PlayerFeature.SYNC in self.player.supported_features:
- self._attr_supported_features |= MediaPlayerEntityFeature.GROUPING
- self._attr_device_class = MediaPlayerDeviceClass.SPEAKER
- self._prev_time: float = 0
-
- async def async_added_to_hass(self) -> None:
- """Register callbacks."""
- await super().async_added_to_hass()
-
- # we subscribe to player queue time update but we only
- # accept a state change on big time jumps (e.g. seeking)
- async def queue_time_updated(event: MassEvent) -> None:
- if event.object_id != self.player.active_source:
- return
- if abs((self._prev_time or 0) - event.data) > 5:
- await self.async_on_update()
- self.async_write_ha_state()
- self._prev_time = event.data
-
- self.async_on_remove(
- self.mass.subscribe(
- queue_time_updated,
- EventType.QUEUE_TIME_UPDATED,
- )
- )
-
- @property
- def active_queue(self) -> PlayerQueue | None:
- """Return the active queue for this player (if any)."""
- if not self.player.active_source:
- return None
- return self.mass.player_queues.get(self.player.active_source)
-
- @property
- def extra_state_attributes(self) -> Mapping[str, Any]:
- """Return additional state attributes."""
- return {
- ATTR_MASS_PLAYER_TYPE: self.player.type.value,
- ATTR_ACTIVE_QUEUE: (
- self.active_queue.queue_id if self.active_queue else None
- ),
- }
-
- async def async_on_update(self) -> None:
- """Handle player updates."""
- if not self.available:
- return
- player = self.player
- active_queue = self.active_queue
- # update generic attributes
- if player.powered and active_queue is not None:
- self._attr_state = MediaPlayerState(active_queue.state.value)
- if player.powered and player.state is not None:
- self._attr_state = MediaPlayerState(player.state.value)
- else:
- self._attr_state = MediaPlayerState(STATE_OFF)
- group_members_entity_ids: list[str] = []
- if player.group_childs:
- # translate MA group_childs to HA group_members as entity id's
- entity_registry = er.async_get(self.hass)
- group_members_entity_ids = [
- entity_id
- for child_id in player.group_childs
- if (
- entity_id := entity_registry.async_get_entity_id(
- self.platform.domain, DOMAIN, child_id
- )
- )
- ]
- self._attr_group_members = group_members_entity_ids
- self._attr_volume_level = (
- player.volume_level / 100 if player.volume_level is not None else None
- )
- self._attr_is_volume_muted = player.volume_muted
- self._update_media_attributes(player, active_queue)
- self._update_media_image_url(player, active_queue)
-
- @catch_musicassistant_error
- async def async_media_play(self) -> None:
- """Send play command to device."""
- await self.mass.players.player_command_play(self.player_id)
-
- @catch_musicassistant_error
- async def async_media_pause(self) -> None:
- """Send pause command to device."""
- await self.mass.players.player_command_pause(self.player_id)
-
- @catch_musicassistant_error
- async def async_media_stop(self) -> None:
- """Send stop command to device."""
- await self.mass.players.player_command_stop(self.player_id)
-
- @catch_musicassistant_error
- async def async_media_next_track(self) -> None:
- """Send next track command to device."""
- await self.mass.players.player_command_next_track(self.player_id)
-
- @catch_musicassistant_error
- async def async_media_previous_track(self) -> None:
- """Send previous track command to device."""
- await self.mass.players.player_command_previous_track(self.player_id)
-
- @catch_musicassistant_error
- async def async_media_seek(self, position: float) -> None:
- """Send seek command."""
- position = int(position)
- await self.mass.players.player_command_seek(self.player_id, position)
-
- @catch_musicassistant_error
- async def async_mute_volume(self, mute: bool) -> None:
- """Mute the volume."""
- await self.mass.players.player_command_volume_mute(self.player_id, mute)
-
- @catch_musicassistant_error
- async def async_set_volume_level(self, volume: float) -> None:
- """Send new volume_level to device."""
- volume = int(volume * 100)
- await self.mass.players.player_command_volume_set(self.player_id, volume)
-
- @catch_musicassistant_error
- async def async_volume_up(self) -> None:
- """Send new volume_level to device."""
- await self.mass.players.player_command_volume_up(self.player_id)
-
- @catch_musicassistant_error
- async def async_volume_down(self) -> None:
- """Send new volume_level to device."""
- await self.mass.players.player_command_volume_down(self.player_id)
-
- @catch_musicassistant_error
- async def async_turn_on(self) -> None:
- """Turn on device."""
- await self.mass.players.player_command_power(self.player_id, True)
-
- @catch_musicassistant_error
- async def async_turn_off(self) -> None:
- """Turn off device."""
- await self.mass.players.player_command_power(self.player_id, False)
-
- @catch_musicassistant_error
- async def async_set_shuffle(self, shuffle: bool) -> None:
- """Set shuffle state."""
- if not self.active_queue:
- return
- await self.mass.player_queues.queue_command_shuffle(
- self.active_queue.queue_id, shuffle
- )
-
- @catch_musicassistant_error
- async def async_set_repeat(self, repeat: RepeatMode) -> None:
- """Set repeat state."""
- if not self.active_queue:
- return
- await self.mass.player_queues.queue_command_repeat(
- self.active_queue.queue_id, MassRepeatMode(repeat)
- )
-
- @catch_musicassistant_error
- async def async_clear_playlist(self) -> None:
- """Clear players playlist."""
- if TYPE_CHECKING:
- assert self.player.active_source is not None
- if queue := self.mass.player_queues.get(self.player.active_source):
- await self.mass.player_queues.queue_command_clear(queue.queue_id)
-
- @catch_musicassistant_error
- async def async_play_media(
- self,
- media_type: MediaType | str,
- media_id: str,
- enqueue: MediaPlayerEnqueue | None = None,
- announce: bool | None = None,
- **kwargs: Any,
- ) -> None:
- """Send the play_media command to the media player."""
- if media_source.is_media_source_id(media_id):
- # Handle media_source
- sourced_media = await media_source.async_resolve_media(
- self.hass, media_id, self.entity_id
- )
- media_id = sourced_media.url
- media_id = async_process_play_media_url(self.hass, media_id)
-
- if announce:
- await self._async_handle_play_announcement(
- media_id,
- use_pre_announce=kwargs[ATTR_MEDIA_EXTRA].get("use_pre_announce"),
- announce_volume=kwargs[ATTR_MEDIA_EXTRA].get("announce_volume"),
- )
- return
-
- # forward to our advanced play_media handler
- await self._async_handle_play_media(
- media_id=[media_id],
- enqueue=enqueue,
- media_type=media_type,
- radio_mode=kwargs[ATTR_MEDIA_EXTRA].get(ATTR_RADIO_MODE),
- )
-
- @catch_musicassistant_error
- async def async_join_players(self, group_members: list[str]) -> None:
- """Join `group_members` as a player group with the current player."""
- player_ids: list[str] = []
- for child_entity_id in group_members:
- # resolve HA entity_id to MA player_id
- if (hass_state := self.hass.states.get(child_entity_id)) is None:
- continue
- if (mass_player_id := hass_state.attributes.get("mass_player_id")) is None:
- continue
- player_ids.append(mass_player_id)
- await self.mass.players.player_command_sync_many(self.player_id, player_ids)
-
- @catch_musicassistant_error
- async def async_unjoin_player(self) -> None:
- """Remove this player from any group."""
- await self.mass.players.player_command_unsync(self.player_id)
-
- @catch_musicassistant_error
- async def _async_handle_play_media(
- self,
- media_id: list[str],
- enqueue: MediaPlayerEnqueue | QueueOption | None = None,
- radio_mode: bool | None = None,
- media_type: str | None = None,
- ) -> None:
- """Send the play_media command to the media player."""
- media_uris: list[str] = []
- item: MediaItemType | ItemMapping | None = None
- # work out (all) uri(s) to play
- for media_id_str in media_id:
- # URL or URI string
- if "://" in media_id_str:
- media_uris.append(media_id_str)
- continue
- # try content id as library id
- if media_type and media_id_str.isnumeric():
- with suppress(MediaNotFoundError):
- item = await self.mass.music.get_item(
- MediaType(media_type), media_id_str, "library"
- )
- if isinstance(item, MediaItemType | ItemMapping) and item.uri:
- media_uris.append(item.uri)
- continue
- # try local accessible filename
- elif await asyncio.to_thread(os.path.isfile, media_id_str):
- media_uris.append(media_id_str)
- continue
-
- if not media_uris:
- raise HomeAssistantError(
- f"Could not resolve {media_id} to playable media item"
- )
-
- # determine active queue to send the play request to
- if TYPE_CHECKING:
- assert self.player.active_source is not None
- if queue := self.mass.player_queues.get(self.player.active_source):
- queue_id = queue.queue_id
- else:
- queue_id = self.player_id
-
- await self.mass.player_queues.play_media(
- queue_id,
- media=media_uris,
- option=self._convert_queueoption_to_media_player_enqueue(enqueue),
- radio_mode=radio_mode if radio_mode else False,
- )
-
- @catch_musicassistant_error
- async def _async_handle_play_announcement(
- self,
- url: str,
- use_pre_announce: bool | None = None,
- announce_volume: int | None = None,
- ) -> None:
- """Send the play_announcement command to the media player."""
- await self.mass.players.play_announcement(
- self.player_id, url, use_pre_announce, announce_volume
- )
-
- async def async_browse_media(
- self,
- media_content_type: MediaType | str | None = None,
- media_content_id: str | None = None,
- ) -> BrowseMedia:
- """Implement the websocket media browsing helper."""
- return await media_source.async_browse_media(
- self.hass,
- media_content_id,
- content_filter=lambda item: item.media_content_type.startswith("audio/"),
- )
-
- def _update_media_image_url(
- self, player: Player, queue: PlayerQueue | None
- ) -> None:
- """Update image URL for the active queue item."""
- if queue is None or queue.current_item is None:
- self._attr_media_image_url = None
- return
- if image_url := self.mass.get_media_item_image_url(queue.current_item):
- self._attr_media_image_remotely_accessible = (
- self.mass.server_url not in image_url
- )
- self._attr_media_image_url = image_url
- return
- self._attr_media_image_url = None
-
- def _update_media_attributes(
- self, player: Player, queue: PlayerQueue | None
- ) -> None:
- """Update media attributes for the active queue item."""
- # pylint: disable=too-many-statements
- self._attr_media_artist = None
- self._attr_media_album_artist = None
- self._attr_media_album_name = None
- self._attr_media_title = None
- self._attr_media_content_id = None
- self._attr_media_duration = None
- self._attr_media_position = None
- self._attr_media_position_updated_at = None
-
- if queue is None and player.current_media:
- # player has some external source active
- self._attr_media_content_id = player.current_media.uri
- self._attr_app_id = player.active_source
- self._attr_media_title = player.current_media.title
- self._attr_media_artist = player.current_media.artist
- self._attr_media_album_name = player.current_media.album
- self._attr_media_duration = player.current_media.duration
- # shuffle and repeat are not (yet) supported for external sources
- self._attr_shuffle = None
- self._attr_repeat = None
- if TYPE_CHECKING:
- assert player.elapsed_time is not None
- self._attr_media_position = int(player.elapsed_time)
- self._attr_media_position_updated_at = (
- utc_from_timestamp(player.elapsed_time_last_updated)
- if player.elapsed_time_last_updated
- else None
- )
- if TYPE_CHECKING:
- assert player.elapsed_time is not None
- self._prev_time = player.elapsed_time
- return
-
- if queue is None:
- # player has no MA queue active
- self._attr_source = player.active_source
- self._attr_app_id = player.active_source
- return
-
- # player has an MA queue active (either its own queue or some group queue)
- self._attr_app_id = DOMAIN
- self._attr_shuffle = queue.shuffle_enabled
- self._attr_repeat = queue.repeat_mode.value
- if not (cur_item := queue.current_item):
- # queue is empty
- return
-
- self._attr_media_content_id = queue.current_item.uri
- self._attr_media_duration = queue.current_item.duration
- self._attr_media_position = int(queue.elapsed_time)
- self._attr_media_position_updated_at = utc_from_timestamp(
- queue.elapsed_time_last_updated
- )
- self._prev_time = queue.elapsed_time
-
- # handle stream title (radio station icy metadata)
- if (stream_details := cur_item.streamdetails) and stream_details.stream_title:
- self._attr_media_album_name = cur_item.name
- if " - " in stream_details.stream_title:
- stream_title_parts = stream_details.stream_title.split(" - ", 1)
- self._attr_media_title = stream_title_parts[1]
- self._attr_media_artist = stream_title_parts[0]
- else:
- self._attr_media_title = stream_details.stream_title
- return
-
- if not (media_item := cur_item.media_item):
- # queue is not playing a regular media item (edge case?!)
- self._attr_media_title = cur_item.name
- return
-
- # queue is playing regular media item
- self._attr_media_title = media_item.name
- # for tracks we can extract more info
- if media_item.media_type == MediaType.TRACK:
- if TYPE_CHECKING:
- assert isinstance(media_item, Track)
- self._attr_media_artist = media_item.artist_str
- if media_item.version:
- self._attr_media_title += f" ({media_item.version})"
- if media_item.album:
- self._attr_media_album_name = media_item.album.name
- self._attr_media_album_artist = getattr(
- media_item.album, "artist_str", None
- )
-
- def _convert_queueoption_to_media_player_enqueue(
- self, queue_option: MediaPlayerEnqueue | QueueOption | None
- ) -> QueueOption | None:
- """Convert a QueueOption to a MediaPlayerEnqueue."""
- if isinstance(queue_option, MediaPlayerEnqueue):
- queue_option = QUEUE_OPTION_MAP.get(queue_option)
- return queue_option
diff --git a/homeassistant/components/music_assistant/strings.json b/homeassistant/components/music_assistant/strings.json
deleted file mode 100644
index f15b0b1b3065f17655de0d526b94fac0dcd548a3..0000000000000000000000000000000000000000
--- a/homeassistant/components/music_assistant/strings.json
+++ /dev/null
@@ -1,51 +0,0 @@
-{
- "config": {
- "step": {
- "init": {
- "data": {
- "url": "URL of the Music Assistant server"
- }
- },
- "manual": {
- "title": "Manually add Music Assistant Server",
- "description": "Enter the URL to your already running Music Assistant Server. If you do not have the Music Assistant Server running, you should install it first.",
- "data": {
- "url": "URL of the Music Assistant server"
- }
- },
- "discovery_confirm": {
- "description": "Do you want to add the Music Assistant Server `{url}` to Home Assistant?",
- "title": "Discovered Music Assistant Server"
- }
- },
- "error": {
- "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
- "invalid_server_version": "The Music Assistant server is not the correct version",
- "unknown": "[%key:common::config_flow::error::unknown%]"
- },
- "abort": {
- "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
- "already_in_progress": "Configuration flow is already in progress",
- "reconfiguration_successful": "Successfully reconfigured the Music Assistant integration.",
- "cannot_connect": "Failed to connect",
- "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
- }
- },
- "issues": {
- "invalid_server_version": {
- "title": "The Music Assistant server is not the correct version",
- "description": "Check if there are updates available for the Music Assistant Server and/or integration."
- }
- },
- "selector": {
- "enqueue": {
- "options": {
- "play": "Play",
- "next": "Play next",
- "add": "Add to queue",
- "replace": "Play now and clear queue",
- "replace_next": "Play next and clear queue"
- }
- }
- }
-}
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index 981409555527059f8b86b29ddb5a420015415cae..e80238c47a479dfc252c1f7684309d049d1666be 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -383,7 +383,6 @@ FLOWS = {
"mpd",
"mqtt",
"mullvad",
- "music_assistant",
"mutesync",
"mysensors",
"mystrom",
diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json
index 7d8383c90cd2771b2af011a09b605607a46718dd..6e0ab856b5791727cab10b9c7750431542f7c103 100644
--- a/homeassistant/generated/integrations.json
+++ b/homeassistant/generated/integrations.json
@@ -3944,12 +3944,6 @@
"iot_class": "cloud_polling",
"single_config_entry": true
},
- "music_assistant": {
- "name": "Music Assistant",
- "integration_type": "hub",
- "config_flow": true,
- "iot_class": "local_push"
- },
"mutesync": {
"name": "mutesync",
"integration_type": "hub",
diff --git a/homeassistant/generated/zeroconf.py b/homeassistant/generated/zeroconf.py
index 1fbd6337fdb0fb8045c5d2e930bfdeef80ad6a2d..eb3c1b3a105a08f9188ece78cff0371a653e7fec 100644
--- a/homeassistant/generated/zeroconf.py
+++ b/homeassistant/generated/zeroconf.py
@@ -639,11 +639,6 @@ ZEROCONF = {
},
},
],
- "_mass._tcp.local.": [
- {
- "domain": "music_assistant",
- },
- ],
"_matter._tcp.local.": [
{
"domain": "matter",
diff --git a/mypy.ini b/mypy.ini
index 1b98877759425867cac299d7185d8b795dfc2ee0..794579eb48f0acee77b4035aaaa790f0b388f603 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -2995,16 +2995,6 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
-[mypy-homeassistant.components.music_assistant.*]
-check_untyped_defs = true
-disallow_incomplete_defs = true
-disallow_subclassing_any = true
-disallow_untyped_calls = true
-disallow_untyped_decorators = true
-disallow_untyped_defs = true
-warn_return_any = true
-warn_unreachable = true
-
[mypy-homeassistant.components.my.*]
check_untyped_defs = true
disallow_incomplete_defs = true
diff --git a/requirements_all.txt b/requirements_all.txt
index 4be98eea7358e9f0735c3d3055d8bf2d3e95ffde..329b227d01a78218bd4f819641d052e33feda90e 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -1405,9 +1405,6 @@ mozart-api==4.1.1.116.0
# homeassistant.components.mullvad
mullvad-api==1.0.0
-# homeassistant.components.music_assistant
-music-assistant-client==1.0.3
-
# homeassistant.components.tts
mutagen==1.47.0
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 7596dd5e23b126aad547f26ddbfebfbf87a50fe7..052b5307bcfdbffa9f4acaf8ccb90e6689c0c33a 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -1174,9 +1174,6 @@ mozart-api==4.1.1.116.0
# homeassistant.components.mullvad
mullvad-api==1.0.0
-# homeassistant.components.music_assistant
-music-assistant-client==1.0.3
-
# homeassistant.components.tts
mutagen==1.47.0
diff --git a/tests/components/music_assistant/__init__.py b/tests/components/music_assistant/__init__.py
deleted file mode 100644
index 6893b862e2d08d82afb2d9b377e597f6e1693033..0000000000000000000000000000000000000000
--- a/tests/components/music_assistant/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-"""The tests for the Music Assistant component."""
diff --git a/tests/components/music_assistant/conftest.py b/tests/components/music_assistant/conftest.py
deleted file mode 100644
index b03a56ab4a6ea73c3f650204d8fef100af5359fa..0000000000000000000000000000000000000000
--- a/tests/components/music_assistant/conftest.py
+++ /dev/null
@@ -1,35 +0,0 @@
-"""Music Assistant test fixtures."""
-
-from collections.abc import Generator
-from unittest.mock import patch
-
-from music_assistant_models.api import ServerInfoMessage
-import pytest
-
-from homeassistant.components.music_assistant.config_flow import CONF_URL
-from homeassistant.components.music_assistant.const import DOMAIN
-
-from tests.common import AsyncMock, MockConfigEntry, load_fixture
-
-
-@pytest.fixture
-def mock_get_server_info() -> Generator[AsyncMock]:
- """Mock the function to get server info."""
- with patch(
- "homeassistant.components.music_assistant.config_flow.get_server_info"
- ) as mock_get_server_info:
- mock_get_server_info.return_value = ServerInfoMessage.from_json(
- load_fixture("server_info_message.json", DOMAIN)
- )
- yield mock_get_server_info
-
-
-@pytest.fixture
-def mock_config_entry() -> MockConfigEntry:
- """Mock a config entry."""
- return MockConfigEntry(
- domain=DOMAIN,
- title="Music Assistant",
- data={CONF_URL: "http://localhost:8095"},
- unique_id="1234",
- )
diff --git a/tests/components/music_assistant/fixtures/server_info_message.json b/tests/components/music_assistant/fixtures/server_info_message.json
deleted file mode 100644
index 907ec8af820ae2c1548131a5a67bee38ed16cfa3..0000000000000000000000000000000000000000
--- a/tests/components/music_assistant/fixtures/server_info_message.json
+++ /dev/null
@@ -1,9 +0,0 @@
-{
- "server_id": "1234",
- "server_version": "0.0.0",
- "schema_version": 23,
- "min_supported_schema_version": 23,
- "base_url": "http://localhost:8095",
- "homeassistant_addon": false,
- "onboard_done": false
-}
diff --git a/tests/components/music_assistant/test_config_flow.py b/tests/components/music_assistant/test_config_flow.py
deleted file mode 100644
index c700060889c2175d5ce7e0ce5e9b0da94217b933..0000000000000000000000000000000000000000
--- a/tests/components/music_assistant/test_config_flow.py
+++ /dev/null
@@ -1,217 +0,0 @@
-"""Define tests for the Music Assistant Integration config flow."""
-
-from copy import deepcopy
-from ipaddress import ip_address
-from unittest import mock
-from unittest.mock import AsyncMock
-
-from music_assistant_client.exceptions import (
- CannotConnect,
- InvalidServerVersion,
- MusicAssistantClientException,
-)
-from music_assistant_models.api import ServerInfoMessage
-import pytest
-
-from homeassistant.components.music_assistant.config_flow import CONF_URL
-from homeassistant.components.music_assistant.const import DEFAULT_NAME, DOMAIN
-from homeassistant.components.zeroconf import ZeroconfServiceInfo
-from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
-from homeassistant.core import HomeAssistant
-from homeassistant.data_entry_flow import FlowResultType
-
-from tests.common import MockConfigEntry, load_fixture
-
-SERVER_INFO = {
- "server_id": "1234",
- "base_url": "http://localhost:8095",
- "server_version": "0.0.0",
- "schema_version": 23,
- "min_supported_schema_version": 23,
- "homeassistant_addon": True,
-}
-
-ZEROCONF_DATA = ZeroconfServiceInfo(
- ip_address=ip_address("127.0.0.1"),
- ip_addresses=[ip_address("127.0.0.1")],
- hostname="mock_hostname",
- port=None,
- type=mock.ANY,
- name=mock.ANY,
- properties=SERVER_INFO,
-)
-
-
-async def test_full_flow(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
-) -> None:
- """Test full flow."""
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_USER},
- )
- assert result["type"] is FlowResultType.FORM
- assert result["step_id"] == "user"
-
- result = await hass.config_entries.flow.async_configure(
- result["flow_id"],
- {CONF_URL: "http://localhost:8095"},
- )
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert result["title"] == DEFAULT_NAME
- assert result["data"] == {
- CONF_URL: "http://localhost:8095",
- }
- assert result["result"].unique_id == "1234"
-
-
-async def test_zero_conf_flow(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
-) -> None:
- """Test zeroconf flow."""
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_ZEROCONF},
- data=ZEROCONF_DATA,
- )
- assert result["type"] is FlowResultType.FORM
- assert result["step_id"] == "discovery_confirm"
- result = await hass.config_entries.flow.async_configure(
- result["flow_id"],
- {},
- )
- assert result["type"] is FlowResultType.CREATE_ENTRY
- assert result["title"] == DEFAULT_NAME
- assert result["data"] == {
- CONF_URL: "http://localhost:8095",
- }
- assert result["result"].unique_id == "1234"
-
-
-async def test_zero_conf_missing_server_id(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
-) -> None:
- """Test zeroconf flow with missing server id."""
- bad_zero_conf_data = deepcopy(ZEROCONF_DATA)
- bad_zero_conf_data.properties.pop("server_id")
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_ZEROCONF},
- data=bad_zero_conf_data,
- )
- await hass.async_block_till_done()
- assert result["type"] is FlowResultType.ABORT
- assert result["reason"] == "missing_server_id"
-
-
-async def test_duplicate_user(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
- mock_config_entry: MockConfigEntry,
-) -> None:
- """Test duplicate user flow."""
- mock_config_entry.add_to_hass(hass)
-
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_USER},
- )
- await hass.async_block_till_done()
- assert result["type"] is FlowResultType.FORM
- assert result["step_id"] == "user"
-
- result = await hass.config_entries.flow.async_configure(
- result["flow_id"],
- {CONF_URL: "http://localhost:8095"},
- )
- await hass.async_block_till_done()
-
- assert result["type"] is FlowResultType.ABORT
- assert result["reason"] == "already_configured"
-
-
-async def test_duplicate_zeroconf(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
- mock_config_entry: MockConfigEntry,
-) -> None:
- """Test duplicate zeroconf flow."""
- mock_config_entry.add_to_hass(hass)
-
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_ZEROCONF},
- data=ZEROCONF_DATA,
- )
- await hass.async_block_till_done()
-
- assert result["type"] is FlowResultType.ABORT
- assert result["reason"] == "already_configured"
-
-
-@pytest.mark.parametrize(
- ("exception", "error_message"),
- [
- (InvalidServerVersion("invalid_server_version"), "invalid_server_version"),
- (CannotConnect("cannot_connect"), "cannot_connect"),
- (MusicAssistantClientException("unknown"), "unknown"),
- ],
-)
-async def test_flow_user_server_version_invalid(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
- exception: MusicAssistantClientException,
- error_message: str,
-) -> None:
- """Test user flow when server url is invalid."""
- mock_get_server_info.side_effect = exception
-
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_USER},
- )
- await hass.async_block_till_done()
- assert result["type"] is FlowResultType.FORM
- assert result["step_id"] == "user"
-
- result = await hass.config_entries.flow.async_configure(
- result["flow_id"],
- {CONF_URL: "http://localhost:8095"},
- )
- await hass.async_block_till_done()
- assert result["errors"] == {"base": error_message}
-
- mock_get_server_info.side_effect = None
- mock_get_server_info.return_value = ServerInfoMessage.from_json(
- load_fixture("server_info_message.json", DOMAIN)
- )
-
- assert result["type"] is FlowResultType.FORM
- result = await hass.config_entries.flow.async_configure(
- result["flow_id"],
- {CONF_URL: "http://localhost:8095"},
- )
- await hass.async_block_till_done()
-
- assert result["type"] is FlowResultType.CREATE_ENTRY
-
-
-async def test_flow_zeroconf_connect_issue(
- hass: HomeAssistant,
- mock_get_server_info: AsyncMock,
-) -> None:
- """Test zeroconf flow when server connect be reached."""
- mock_get_server_info.side_effect = CannotConnect("cannot_connect")
-
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": SOURCE_ZEROCONF},
- data=ZEROCONF_DATA,
- )
- await hass.async_block_till_done()
-
- assert result["type"] is FlowResultType.ABORT
- assert result["reason"] == "cannot_connect"