Add forked_daapd integration (#31953)

* New forked_daapd component

* Bunch of changes

Add config flow and zeroconf
Add zones on callback when added by server
Add password auth
Add async_play_media for TTS
Add media_image_url
Add support for pipe control/input from librespot-java
Improve update callbacks

* Refactor as per code review suggestions

Move config_flow connection testing to pypi library (v0.1.4)
Remove use of ForkedDaapdData class
Decouple Master and Zone data and functions
Add updater class to manage websocket and entity updates

* More changes as per code review

Avoid direct access to entities in tests
Bump pypi version
Mark entities unavailable when websocket disconnected
Move config tests to test_config_flow
Move full url creation from media_image_url to library
Move updater entity from master to hass.data
Remove default unmute volume option
Remove name from config_flow
Remove storage of entities in hass.data
Use async_write_ha_state
Use signal to trigger update_options
Use unittest.mock instead of asynctest.mock

* Yet more changes as per code review

Add more assertions in tests
Avoid patching asyncio
Make off state require player state stopped
Only send update to existing zones
Split up some tests
Use events instead of async_block_till_done
Use sets instead of lists where applicable
Wait for pause callback before continuing TTS

* Remove unnecessary use of Future()

* Add pipes and playlists as sources

* Add support for multiple servers

Change config options to add max_playlists+remove use_pipe_control
Create Machine ID in test_connection and also get from zeroconf
Modify hass.data storage
Update host for known configurations
Use Machine ID in unique_ids, entity names, config title, signals

* Use entry_id as basis for multiple entries

* Use f-strings and str.format, abort for same host

* Clean up check for same host

CODEOWNERS

@@ -130,6 +130,7 @@ homeassistant/components/flick_electric/* @ZephireNZ
 homeassistant/components/flock/* @fabaff
 homeassistant/components/flume/* @ChrisMandich @bdraco
 homeassistant/components/flunearyou/* @bachya
+homeassistant/components/forked_daapd/* @uvjustin
 homeassistant/components/fortigate/* @kifeo
 homeassistant/components/fortios/* @kimfrellsen
 homeassistant/components/foscam/* @skgsergio

homeassistant/components/forked_daapd/__init__.py

@@ -0,0 +1,34 @@
+"""The forked_daapd component."""
+from homeassistant.components.media_player import DOMAIN as MP_DOMAIN
+
+from .const import DOMAIN, HASS_DATA_REMOVE_LISTENERS_KEY, HASS_DATA_UPDATER_KEY
+
+
+async def async_setup(hass, config):
+    """Set up the forked-daapd component."""
+    return True
+
+
+async def async_setup_entry(hass, entry):
+    """Set up forked-daapd from a config entry by forwarding to platform."""
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, MP_DOMAIN)
+    )
+    return True
+
+
+async def async_unload_entry(hass, entry):
+    """Remove forked-daapd component."""
+    status = await hass.config_entries.async_forward_entry_unload(entry, MP_DOMAIN)
+    if status and hass.data.get(DOMAIN) and hass.data[DOMAIN].get(entry.entry_id):
+        hass.data[DOMAIN][entry.entry_id][
+            HASS_DATA_UPDATER_KEY
+        ].websocket_handler.cancel()
+        for remove_listener in hass.data[DOMAIN][entry.entry_id][
+            HASS_DATA_REMOVE_LISTENERS_KEY
+        ]:
+            remove_listener()
+        del hass.data[DOMAIN][entry.entry_id]
+        if not hass.data[DOMAIN]:
+            del hass.data[DOMAIN]
+    return status

homeassistant/components/forked_daapd/config_flow.py

@@ -0,0 +1,186 @@
+"""Config flow to configure forked-daapd devices."""
+import logging
+
+from pyforked_daapd import ForkedDaapdAPI
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT
+from homeassistant.core import callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import (  # pylint:disable=unused-import
+    CONF_LIBRESPOT_JAVA_PORT,
+    CONF_MAX_PLAYLISTS,
+    CONF_TTS_PAUSE_TIME,
+    CONF_TTS_VOLUME,
+    DEFAULT_PORT,
+    DEFAULT_TTS_PAUSE_TIME,
+    DEFAULT_TTS_VOLUME,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+# Can't use all vol types: https://github.com/home-assistant/core/issues/32819
+DATA_SCHEMA_DICT = {
+    vol.Required(CONF_HOST): str,
+    vol.Optional(CONF_PORT, default=DEFAULT_PORT): int,
+    vol.Optional(CONF_PASSWORD, default=""): str,
+}
+
+TEST_CONNECTION_ERROR_DICT = {
+    "ok": "ok",
+    "websocket_not_enabled": "websocket_not_enabled",
+    "wrong_host_or_port": "wrong_host_or_port",
+    "wrong_password": "wrong_password",
+    "wrong_server_type": "wrong_server_type",
+}
+
+
+class ForkedDaapdOptionsFlowHandler(config_entries.OptionsFlow):
+    """Handle a forked-daapd options flow."""
+
+    def __init__(self, config_entry):
+        """Initialize."""
+        self.config_entry = config_entry
+
+    async def async_step_init(self, user_input=None):
+        """Manage the options."""
+        if user_input is not None:
+            return self.async_create_entry(title="options", data=user_input)
+
+        return self.async_show_form(
+            step_id="init",
+            data_schema=vol.Schema(
+                {
+                    vol.Optional(
+                        CONF_TTS_PAUSE_TIME,
+                        default=self.config_entry.options.get(
+                            CONF_TTS_PAUSE_TIME, DEFAULT_TTS_PAUSE_TIME
+                        ),
+                    ): float,
+                    vol.Optional(
+                        CONF_TTS_VOLUME,
+                        default=self.config_entry.options.get(
+                            CONF_TTS_VOLUME, DEFAULT_TTS_VOLUME
+                        ),
+                    ): float,
+                    vol.Optional(
+                        CONF_LIBRESPOT_JAVA_PORT,
+                        default=self.config_entry.options.get(
+                            CONF_LIBRESPOT_JAVA_PORT, 24879
+                        ),
+                    ): int,
+                    vol.Optional(
+                        CONF_MAX_PLAYLISTS,
+                        default=self.config_entry.options.get(CONF_MAX_PLAYLISTS, 10),
+                    ): int,
+                }
+            ),
+        )
+
+
+def fill_in_schema_dict(some_input):
+    """Fill in schema dict defaults from user_input."""
+    schema_dict = {}
+    for field, _type in DATA_SCHEMA_DICT.items():
+        if some_input.get(str(field)):
+            schema_dict[
+                vol.Optional(str(field), default=some_input[str(field)])
+            ] = _type
+        else:
+            schema_dict[field] = _type
+    return schema_dict
+
+
+class ForkedDaapdFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a forked-daapd config flow."""
+
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
+
+    def __init__(self):
+        """Initialize."""
+        self.discovery_schema = None
+
+    @staticmethod
+    @callback
+    def async_get_options_flow(config_entry):
+        """Return options flow handler."""
+        return ForkedDaapdOptionsFlowHandler(config_entry)
+
+    async def validate_input(self, user_input):
+        """Validate the user input."""
+        websession = async_get_clientsession(self.hass)
+        validate_result = await ForkedDaapdAPI.test_connection(
+            websession=websession,
+            host=user_input[CONF_HOST],
+            port=user_input[CONF_PORT],
+            password=user_input[CONF_PASSWORD],
+        )
+        validate_result[0] = TEST_CONNECTION_ERROR_DICT.get(
+            validate_result[0], "unknown_error"
+        )
+        return validate_result
+
+    async def async_step_user(self, user_input=None):
+        """Handle a forked-daapd config flow start.
+
+        Manage device specific parameters.
+        """
+        if user_input is not None:
+            # check for any entries with same host, abort if found
+            for entry in self._async_current_entries():
+                if entry.data[CONF_HOST] == user_input[CONF_HOST]:
+                    return self.async_abort(reason="already_configured")
+            validate_result = await self.validate_input(user_input)
+            if validate_result[0] == "ok":  # success
+                _LOGGER.debug("Connected successfully. Creating entry")
+                return self.async_create_entry(
+                    title=validate_result[1], data=user_input
+                )
+            return self.async_show_form(
+                step_id="user",
+                data_schema=vol.Schema(fill_in_schema_dict(user_input)),
+                errors={"base": validate_result[0]},
+            )
+        if self.discovery_schema:  # stop at form to allow user to set up manually
+            return self.async_show_form(
+                step_id="user", data_schema=self.discovery_schema, errors={}
+            )
+        return self.async_show_form(
+            step_id="user", data_schema=vol.Schema(DATA_SCHEMA_DICT), errors={}
+        )
+
+    async def async_step_zeroconf(self, discovery_info):
+        """Prepare configuration for a discovered forked-daapd device."""
+        if not (
+            discovery_info.get("properties")
+            and discovery_info["properties"].get("mtd-version")
+            and discovery_info["properties"].get("Machine Name")
+        ):
+            return self.async_abort(reason="not_forked_daapd")
+
+        # Update title and abort if we already have an entry for this host
+        for entry in self._async_current_entries():
+            if entry.data[CONF_HOST] != discovery_info["host"]:
+                continue
+            self.hass.config_entries.async_update_entry(
+                entry, title=discovery_info["properties"]["Machine Name"],
+            )
+            return self.async_abort(reason="already_configured")
+
+        await self.async_set_unique_id(discovery_info["properties"]["Machine Name"])
+        self._abort_if_unique_id_configured()
+
+        zeroconf_data = {
+            CONF_HOST: discovery_info["host"],
+            CONF_PORT: int(discovery_info["port"]),
+            CONF_NAME: discovery_info["properties"]["Machine Name"],
+        }
+        self.discovery_schema = vol.Schema(fill_in_schema_dict(zeroconf_data))
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        self.context.update({"title_placeholders": zeroconf_data})
+        return await self.async_step_user()

homeassistant/components/forked_daapd/const.py

@@ -0,0 +1,85 @@
+"""Const for forked-daapd."""
+from homeassistant.components.media_player.const import (
+    SUPPORT_CLEAR_PLAYLIST,
+    SUPPORT_NEXT_TRACK,
+    SUPPORT_PAUSE,
+    SUPPORT_PLAY,
+    SUPPORT_PLAY_MEDIA,
+    SUPPORT_PREVIOUS_TRACK,
+    SUPPORT_SEEK,
+    SUPPORT_SELECT_SOURCE,
+    SUPPORT_SHUFFLE_SET,
+    SUPPORT_STOP,
+    SUPPORT_TURN_OFF,
+    SUPPORT_TURN_ON,
+    SUPPORT_VOLUME_MUTE,
+    SUPPORT_VOLUME_SET,
+)
+
+CALLBACK_TIMEOUT = 8  # max time between command and callback from forked-daapd server
+CONF_LIBRESPOT_JAVA_PORT = "librespot_java_port"
+CONF_MAX_PLAYLISTS = "max_playlists"
+CONF_TTS_PAUSE_TIME = "tts_pause_time"
+CONF_TTS_VOLUME = "tts_volume"
+DEFAULT_PORT = 3689
+DEFAULT_SERVER_NAME = "My Server"
+DEFAULT_TTS_PAUSE_TIME = 1.2
+DEFAULT_TTS_VOLUME = 0.8
+DEFAULT_UNMUTE_VOLUME = 0.6
+DOMAIN = "forked_daapd"  # key for hass.data
+FD_NAME = "forked-daapd"
+HASS_DATA_REMOVE_LISTENERS_KEY = "REMOVE_LISTENERS"
+HASS_DATA_UPDATER_KEY = "UPDATER"
+KNOWN_PIPES = {"librespot-java"}
+PIPE_FUNCTION_MAP = {
+    "librespot-java": {
+        "async_media_play": "player_resume",
+        "async_media_pause": "player_pause",
+        "async_media_stop": "player_pause",
+        "async_media_previous_track": "player_prev",
+        "async_media_next_track": "player_next",
+    }
+}
+SIGNAL_ADD_ZONES = "forked-daapd_add_zones {}"
+SIGNAL_CONFIG_OPTIONS_UPDATE = "forked-daapd_config_options_update {}"
+SIGNAL_UPDATE_DATABASE = "forked-daapd_update_database {}"
+SIGNAL_UPDATE_MASTER = "forked-daapd_update_master {}"
+SIGNAL_UPDATE_OUTPUTS = "forked-daapd_update_outputs {}"
+SIGNAL_UPDATE_PLAYER = "forked-daapd_update_player {}"
+SIGNAL_UPDATE_QUEUE = "forked-daapd_update_queue {}"
+SOURCE_NAME_CLEAR = "Clear queue"
+SOURCE_NAME_DEFAULT = "Default (no pipe)"
+STARTUP_DATA = {
+    "player": {
+        "state": "stop",
+        "repeat": "off",
+        "consume": False,
+        "shuffle": False,
+        "volume": 0,
+        "item_id": 0,
+        "item_length_ms": 0,
+        "item_progress_ms": 0,
+    },
+    "queue": {"version": 0, "count": 0, "items": []},
+    "outputs": [],
+}
+SUPPORTED_FEATURES = (
+    SUPPORT_PLAY
+    | SUPPORT_PAUSE
+    | SUPPORT_STOP
+    | SUPPORT_SEEK
+    | SUPPORT_VOLUME_SET
+    | SUPPORT_VOLUME_MUTE
+    | SUPPORT_PREVIOUS_TRACK
+    | SUPPORT_NEXT_TRACK
+    | SUPPORT_CLEAR_PLAYLIST
+    | SUPPORT_SELECT_SOURCE
+    | SUPPORT_SHUFFLE_SET
+    | SUPPORT_TURN_ON
+    | SUPPORT_TURN_OFF
+    | SUPPORT_PLAY_MEDIA
+)
+SUPPORTED_FEATURES_ZONE = (
+    SUPPORT_VOLUME_SET | SUPPORT_VOLUME_MUTE | SUPPORT_TURN_ON | SUPPORT_TURN_OFF
+)
+TTS_TIMEOUT = 20  # max time to wait between TTS getting sent and starting to play

homeassistant/components/forked_daapd/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "forked_daapd",
+  "name": "forked-daapd",
+  "documentation": "https://www.home-assistant.io/integrations/forked-daapd",
+  "codeowners": ["@uvjustin"],
+  "requirements": ["pyforked-daapd==0.1.8", "pylibrespot-java==0.1.0"],
+  "config_flow": true,
+  "zeroconf": ["_daap._tcp.local."]
+}

homeassistant/components/forked_daapd/media_player.py

@@ -0,0 +1,858 @@
+"""This library brings support for forked_daapd to Home Assistant."""
+import asyncio
+from collections import defaultdict
+import logging
+
+from pyforked_daapd import ForkedDaapdAPI
+from pylibrespot_java import LibrespotJavaAPI
+
+from homeassistant.components.media_player import MediaPlayerDevice
+from homeassistant.components.media_player.const import MEDIA_TYPE_MUSIC
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_PORT,
+    STATE_IDLE,
+    STATE_OFF,
+    STATE_ON,
+    STATE_PAUSED,
+    STATE_PLAYING,
+)
+from homeassistant.core import callback
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.dispatcher import (
+    async_dispatcher_connect,
+    async_dispatcher_send,
+)
+from homeassistant.util.dt import utcnow
+
+from .const import (
+    CALLBACK_TIMEOUT,
+    CONF_LIBRESPOT_JAVA_PORT,
+    CONF_MAX_PLAYLISTS,
+    CONF_TTS_PAUSE_TIME,
+    CONF_TTS_VOLUME,
+    DEFAULT_TTS_PAUSE_TIME,
+    DEFAULT_TTS_VOLUME,
+    DEFAULT_UNMUTE_VOLUME,
+    DOMAIN,
+    FD_NAME,
+    HASS_DATA_REMOVE_LISTENERS_KEY,
+    HASS_DATA_UPDATER_KEY,
+    KNOWN_PIPES,
+    PIPE_FUNCTION_MAP,
+    SIGNAL_ADD_ZONES,
+    SIGNAL_CONFIG_OPTIONS_UPDATE,
+    SIGNAL_UPDATE_DATABASE,
+    SIGNAL_UPDATE_MASTER,
+    SIGNAL_UPDATE_OUTPUTS,
+    SIGNAL_UPDATE_PLAYER,
+    SIGNAL_UPDATE_QUEUE,
+    SOURCE_NAME_CLEAR,
+    SOURCE_NAME_DEFAULT,
+    STARTUP_DATA,
+    SUPPORTED_FEATURES,
+    SUPPORTED_FEATURES_ZONE,
+    TTS_TIMEOUT,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+WS_NOTIFY_EVENT_TYPES = ["player", "outputs", "volume", "options", "queue", "database"]
+WEBSOCKET_RECONNECT_TIME = 30  # seconds
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up forked-daapd from a config entry."""
+    host = config_entry.data[CONF_HOST]
+    port = config_entry.data[CONF_PORT]
+    password = config_entry.data[CONF_PASSWORD]
+    forked_daapd_api = ForkedDaapdAPI(
+        async_get_clientsession(hass), host, port, password
+    )
+    forked_daapd_master = ForkedDaapdMaster(
+        clientsession=async_get_clientsession(hass),
+        api=forked_daapd_api,
+        ip_address=host,
+        api_port=port,
+        api_password=password,
+        config_entry=config_entry,
+    )
+
+    @callback
+    def async_add_zones(api, outputs):
+        zone_entities = []
+        for output in outputs:
+            zone_entities.append(ForkedDaapdZone(api, output, config_entry.entry_id))
+        async_add_entities(zone_entities, False)
+
+    remove_add_zones_listener = async_dispatcher_connect(
+        hass, SIGNAL_ADD_ZONES.format(config_entry.entry_id), async_add_zones
+    )
+    remove_entry_listener = config_entry.add_update_listener(update_listener)
+
+    if not hass.data.get(DOMAIN):
+        hass.data[DOMAIN] = {config_entry.entry_id: {}}
+    hass.data[DOMAIN][config_entry.entry_id] = {
+        HASS_DATA_REMOVE_LISTENERS_KEY: [
+            remove_add_zones_listener,
+            remove_entry_listener,
+        ]
+    }
+    async_add_entities([forked_daapd_master], False)
+    forked_daapd_updater = ForkedDaapdUpdater(
+        hass, forked_daapd_api, config_entry.entry_id
+    )
+    await forked_daapd_updater.async_init()
+    hass.data[DOMAIN][config_entry.entry_id][
+        HASS_DATA_UPDATER_KEY
+    ] = forked_daapd_updater
+
+
+async def update_listener(hass, entry):
+    """Handle options update."""
+    async_dispatcher_send(
+        hass, SIGNAL_CONFIG_OPTIONS_UPDATE.format(entry.entry_id), entry.options
+    )
+
+
+class ForkedDaapdZone(MediaPlayerDevice):
+    """Representation of a forked-daapd output."""
+
+    def __init__(self, api, output, entry_id):
+        """Initialize the ForkedDaapd Zone."""
+        self._api = api
+        self._output = output
+        self._output_id = output["id"]
+        self._last_volume = DEFAULT_UNMUTE_VOLUME  # used for mute/unmute
+        self._available = True
+        self._entry_id = entry_id
+
+    async def async_added_to_hass(self):
+        """Use lifecycle hooks."""
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_OUTPUTS.format(self._entry_id),
+                self._async_update_output_callback,
+            )
+        )
+
+    @callback
+    def _async_update_output_callback(self, outputs, _event=None):
+        new_output = next(
+            (output for output in outputs if output["id"] == self._output_id), None
+        )
+        self._available = bool(new_output)
+        if self._available:
+            self._output = new_output
+        self.async_write_ha_state()
+
+    @property
+    def unique_id(self):
+        """Return unique ID."""
+        return f"{self._entry_id}-{self._output_id}"
+
+    @property
+    def should_poll(self) -> bool:
+        """Entity pushes its state to HA."""
+        return False
+
+    async def async_toggle(self):
+        """Toggle the power on the zone."""
+        if self.state == STATE_OFF:
+            await self.async_turn_on()
+        else:
+            await self.async_turn_off()
+
+    @property
+    def available(self) -> bool:
+        """Return whether the zone is available."""
+        return self._available
+
+    async def async_turn_on(self):
+        """Enable the output."""
+        await self._api.change_output(self._output_id, selected=True)
+
+    async def async_turn_off(self):
+        """Disable the output."""
+        await self._api.change_output(self._output_id, selected=False)
+
+    @property
+    def name(self):
+        """Return the name of the zone."""
+        return f"{FD_NAME} output ({self._output['name']})"
+
+    @property
+    def state(self):
+        """State of the zone."""
+        if self._output["selected"]:
+            return STATE_ON
+        return STATE_OFF
+
+    @property
+    def volume_level(self):
+        """Volume level of the media player (0..1)."""
+        return self._output["volume"] / 100
+
+    @property
+    def is_volume_muted(self):
+        """Boolean if volume is currently muted."""
+        return self._output["volume"] == 0
+
+    async def async_mute_volume(self, mute):
+        """Mute the volume."""
+        if mute:
+            if self.volume_level == 0:
+                return
+            self._last_volume = self.volume_level  # store volume level to restore later
+            target_volume = 0
+        else:
+            target_volume = self._last_volume  # restore volume level
+        await self.async_set_volume_level(volume=target_volume)
+
+    async def async_set_volume_level(self, volume):
+        """Set volume - input range [0,1]."""
+        await self._api.set_volume(volume=volume * 100, output_id=self._output_id)
+
+    @property
+    def supported_features(self):
+        """Flag media player features that are supported."""
+        return SUPPORTED_FEATURES_ZONE
+
+
+class ForkedDaapdMaster(MediaPlayerDevice):
+    """Representation of the main forked-daapd device."""
+
+    def __init__(
+        self, clientsession, api, ip_address, api_port, api_password, config_entry
+    ):
+        """Initialize the ForkedDaapd Master Device."""
+        self._api = api
+        self._player = STARTUP_DATA[
+            "player"
+        ]  # _player, _outputs, and _queue are loaded straight from api
+        self._outputs = STARTUP_DATA["outputs"]
+        self._queue = STARTUP_DATA["queue"]
+        self._track_info = defaultdict(
+            str
+        )  # _track info is found by matching _player data with _queue data
+        self._last_outputs = None  # used for device on/off
+        self._last_volume = DEFAULT_UNMUTE_VOLUME
+        self._player_last_updated = None
+        self._pipe_control_api = {}
+        self._ip_address = (
+            ip_address  # need to save this because pipe control is on same ip
+        )
+        self._tts_pause_time = DEFAULT_TTS_PAUSE_TIME
+        self._tts_volume = DEFAULT_TTS_VOLUME
+        self._tts_requested = False
+        self._tts_queued = False
+        self._tts_playing_event = asyncio.Event()
+        self._on_remove = None
+        self._available = False
+        self._clientsession = clientsession
+        self._config_entry = config_entry
+        self.update_options(config_entry.options)
+        self._paused_event = asyncio.Event()
+        self._pause_requested = False
+        self._sources_uris = {}
+        self._source = SOURCE_NAME_DEFAULT
+        self._max_playlists = None
+
+    async def async_added_to_hass(self):
+        """Use lifecycle hooks."""
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_PLAYER.format(self._config_entry.entry_id),
+                self._update_player,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_QUEUE.format(self._config_entry.entry_id),
+                self._update_queue,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_OUTPUTS.format(self._config_entry.entry_id),
+                self._update_outputs,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_MASTER.format(self._config_entry.entry_id),
+                self._update_callback,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_CONFIG_OPTIONS_UPDATE.format(self._config_entry.entry_id),
+                self.update_options,
+            )
+        )
+        self.async_on_remove(
+            async_dispatcher_connect(
+                self.hass,
+                SIGNAL_UPDATE_DATABASE.format(self._config_entry.entry_id),
+                self._update_database,
+            )
+        )
+
+    @callback
+    def _update_callback(self, available):
+        """Call update method."""
+        self._available = available
+        self.async_write_ha_state()
+
+    @callback
+    def update_options(self, options):
+        """Update forked-daapd server options."""
+        if CONF_LIBRESPOT_JAVA_PORT in options:
+            self._pipe_control_api["librespot-java"] = LibrespotJavaAPI(
+                self._clientsession, self._ip_address, options[CONF_LIBRESPOT_JAVA_PORT]
+            )
+        if CONF_TTS_PAUSE_TIME in options:
+            self._tts_pause_time = options[CONF_TTS_PAUSE_TIME]
+        if CONF_TTS_VOLUME in options:
+            self._tts_volume = options[CONF_TTS_VOLUME]
+        if CONF_MAX_PLAYLISTS in options:
+            # sources not updated until next _update_database call
+            self._max_playlists = options[CONF_MAX_PLAYLISTS]
+
+    @callback
+    def _update_player(self, player, event):
+        self._player = player
+        self._player_last_updated = utcnow()
+        self._update_track_info()
+        if self._tts_queued:
+            self._tts_playing_event.set()
+            self._tts_queued = False
+        if self._pause_requested:
+            self._paused_event.set()
+            self._pause_requested = False
+        event.set()
+
+    @callback
+    def _update_queue(self, queue, event):
+        self._queue = queue
+        if (
+            self._tts_requested
+            and self._queue["count"] == 1
+            and self._queue["items"][0]["uri"].find("tts_proxy") != -1
+        ):
+            self._tts_requested = False
+            self._tts_queued = True
+        self._update_track_info()
+        event.set()
+
+    @callback
+    def _update_outputs(self, outputs, event=None):
+        if event:  # Calling without event is meant for zone, so ignore
+            self._outputs = outputs
+            event.set()
+
+    @callback
+    def _update_database(self, pipes, playlists, event):
+        self._sources_uris = {SOURCE_NAME_CLEAR: None, SOURCE_NAME_DEFAULT: None}
+        if pipes:
+            self._sources_uris.update(
+                {
+                    f"{pipe['title']} (pipe)": pipe["uri"]
+                    for pipe in pipes
+                    if pipe["title"] in KNOWN_PIPES
+                }
+            )
+        if playlists:
+            self._sources_uris.update(
+                {
+                    f"{playlist['name']} (playlist)": playlist["uri"]
+                    for playlist in playlists[: self._max_playlists]
+                }
+            )
+        event.set()
+
+    def _update_track_info(self):  # run during every player or queue update
+        try:
+            self._track_info = next(
+                track
+                for track in self._queue["items"]
+                if track["id"] == self._player["item_id"]
+            )
+        except (StopIteration, TypeError, KeyError):
+            _LOGGER.debug("Could not get track info")
+            self._track_info = defaultdict(str)
+
+    @property
+    def unique_id(self):
+        """Return unique ID."""
+        return self._config_entry.entry_id
+
+    @property
+    def should_poll(self) -> bool:
+        """Entity pushes its state to HA."""
+        return False
+
+    @property
+    def available(self) -> bool:
+        """Return whether the master is available."""
+        return self._available
+
+    async def async_turn_on(self):
+        """Restore the last on outputs state."""
+        # restore state
+        if self._last_outputs:
+            futures = []
+            for output in self._last_outputs:
+                futures.append(
+                    self._api.change_output(
+                        output["id"],
+                        selected=output["selected"],
+                        volume=output["volume"],
+                    )
+                )
+            await asyncio.wait(futures)
+        else:
+            selected = []
+            for output in self._outputs:
+                selected.append(output["id"])
+            await self._api.set_enabled_outputs(selected)
+
+    async def async_turn_off(self):
+        """Pause player and store outputs state."""
+        await self.async_media_pause()
+        if any(
+            [output["selected"] for output in self._outputs]
+        ):  # only store output state if some output is selected
+            self._last_outputs = self._outputs
+            await self._api.set_enabled_outputs([])
+
+    async def async_toggle(self):
+        """Toggle the power on the device.
+
+        Default media player component method counts idle as off.
+        We consider idle to be on but just not playing.
+        """
+        if self.state == STATE_OFF:
+            await self.async_turn_on()
+        else:
+            await self.async_turn_off()
+
+    @property
+    def name(self):
+        """Return the name of the device."""
+        return f"{FD_NAME} server"
+
+    @property
+    def state(self):
+        """State of the player."""
+        if self._player["state"] == "play":
+            return STATE_PLAYING
+        if self._player["state"] == "pause":
+            return STATE_PAUSED
+        if not any([output["selected"] for output in self._outputs]):
+            return STATE_OFF
+        if self._player["state"] == "stop":  # this should catch all remaining cases
+            return STATE_IDLE
+
+    @property
+    def volume_level(self):
+        """Volume level of the media player (0..1)."""
+        return self._player["volume"] / 100
+
+    @property
+    def is_volume_muted(self):
+        """Boolean if volume is currently muted."""
+        return self._player["volume"] == 0
+
+    @property
+    def media_content_id(self):
+        """Content ID of current playing media."""
+        return self._player["item_id"]
+
+    @property
+    def media_content_type(self):
+        """Content type of current playing media."""
+        return self._track_info["media_kind"]
+
+    @property
+    def media_duration(self):
+        """Duration of current playing media in seconds."""
+        return self._player["item_length_ms"] / 1000
+
+    @property
+    def media_position(self):
+        """Position of current playing media in seconds."""
+        return self._player["item_progress_ms"] / 1000
+
+    @property
+    def media_position_updated_at(self):
+        """When was the position of the current playing media valid."""
+        return self._player_last_updated
+
+    @property
+    def media_title(self):
+        """Title of current playing media."""
+        return self._track_info["title"]
+
+    @property
+    def media_artist(self):
+        """Artist of current playing media, music track only."""
+        return self._track_info["artist"]
+
+    @property
+    def media_album_name(self):
+        """Album name of current playing media, music track only."""
+        return self._track_info["album"]
+
+    @property
+    def media_album_artist(self):
+        """Album artist of current playing media, music track only."""
+        return self._track_info["album_artist"]
+
+    @property
+    def media_track(self):
+        """Track number of current playing media, music track only."""
+        return self._track_info["track_number"]
+
+    @property
+    def shuffle(self):
+        """Boolean if shuffle is enabled."""
+        return self._player["shuffle"]
+
+    @property
+    def supported_features(self):
+        """Flag media player features that are supported."""
+        return SUPPORTED_FEATURES
+
+    @property
+    def source(self):
+        """Name of the current input source."""
+        return self._source
+
+    @property
+    def source_list(self):
+        """List of available input sources."""
+        return [*self._sources_uris]
+
+    async def async_mute_volume(self, mute):
+        """Mute the volume."""
+        if mute:
+            if self.volume_level == 0:
+                return
+            self._last_volume = self.volume_level  # store volume level to restore later
+            target_volume = 0
+        else:
+            target_volume = self._last_volume  # restore volume level
+        await self._api.set_volume(volume=target_volume * 100)
+
+    async def async_set_volume_level(self, volume):
+        """Set volume - input range [0,1]."""
+        await self._api.set_volume(volume=volume * 100)
+
+    async def async_media_play(self):
+        """Start playback."""
+        if self._use_pipe_control():
+            await self._pipe_call(self._use_pipe_control(), "async_media_play")
+        else:
+            await self._api.start_playback()
+
+    async def async_media_pause(self):
+        """Pause playback."""
+        if self._use_pipe_control():
+            await self._pipe_call(self._use_pipe_control(), "async_media_pause")
+        else:
+            await self._api.pause_playback()
+
+    async def async_media_stop(self):
+        """Stop playback."""
+        if self._use_pipe_control():
+            await self._pipe_call(self._use_pipe_control(), "async_media_stop")
+        else:
+            await self._api.stop_playback()
+
+    async def async_media_previous_track(self):
+        """Skip to previous track."""
+        if self._use_pipe_control():
+            await self._pipe_call(
+                self._use_pipe_control(), "async_media_previous_track"
+            )
+        else:
+            await self._api.previous_track()
+
+    async def async_media_next_track(self):
+        """Skip to next track."""
+        if self._use_pipe_control():
+            await self._pipe_call(self._use_pipe_control(), "async_media_next_track")
+        else:
+            await self._api.next_track()
+
+    async def async_media_seek(self, position):
+        """Seek to position."""
+        await self._api.seek(position_ms=position * 1000)
+
+    async def async_clear_playlist(self):
+        """Clear playlist."""
+        await self._api.clear_queue()
+
+    async def async_set_shuffle(self, shuffle):
+        """Enable/disable shuffle mode."""
+        await self._api.shuffle(shuffle)
+
+    @property
+    def media_image_url(self):
+        """Image url of current playing media."""
+        url = self._track_info.get("artwork_url")
+        if url:
+            url = self._api.full_url(url)
+        return url
+
+    async def _set_tts_volumes(self):
+        if self._outputs:
+            futures = []
+            for output in self._outputs:
+                futures.append(
+                    self._api.change_output(
+                        output["id"], selected=True, volume=self._tts_volume * 100
+                    )
+                )
+            await asyncio.wait(futures)
+            await self._api.set_volume(volume=self._tts_volume * 100)
+
+    async def _pause_and_wait_for_callback(self):
+        """Send pause and wait for the pause callback to be received."""
+        self._pause_requested = True
+        await self.async_media_pause()
+        try:
+            await asyncio.wait_for(
+                self._paused_event.wait(), timeout=CALLBACK_TIMEOUT
+            )  # wait for paused
+        except asyncio.TimeoutError:
+            self._pause_requested = False
+        self._paused_event.clear()
+
+    async def async_play_media(self, media_type, media_id, **kwargs):
+        """Play a URI."""
+        if media_type == MEDIA_TYPE_MUSIC:
+            saved_state = self.state  # save play state
+            if any([output["selected"] for output in self._outputs]):  # save outputs
+                self._last_outputs = self._outputs
+            await self._api.set_enabled_outputs([])  # turn off outputs
+            sleep_future = asyncio.create_task(
+                asyncio.sleep(self._tts_pause_time)
+            )  # start timing now, but not exact because of fd buffer + tts latency
+            await self._pause_and_wait_for_callback()
+            await self._set_tts_volumes()
+            # save position
+            saved_song_position = self._player["item_progress_ms"]
+            saved_queue = (
+                self._queue if self._queue["count"] > 0 else None
+            )  # stash queue
+            if saved_queue:
+                saved_queue_position = next(
+                    i
+                    for i, item in enumerate(saved_queue["items"])
+                    if item["id"] == self._player["item_id"]
+                )
+            self._tts_requested = True
+            await sleep_future
+            await self._api.add_to_queue(uris=media_id, playback="start", clear=True)
+            try:
+                await asyncio.wait_for(
+                    self._tts_playing_event.wait(), timeout=TTS_TIMEOUT
+                )
+                # we have started TTS, now wait for completion
+                await asyncio.sleep(
+                    self._queue["items"][0]["length_ms"]
+                    / 1000  # player may not have updated yet so grab length from queue
+                    + self._tts_pause_time
+                )
+            except asyncio.TimeoutError:
+                self._tts_requested = False
+                _LOGGER.warning("TTS request timed out")
+            self._tts_playing_event.clear()
+            # TTS done, return to normal
+            await self.async_turn_on()  # restores outputs
+            if self._use_pipe_control():  # resume pipe
+                await self._api.add_to_queue(
+                    uris=self._sources_uris[self._source], clear=True
+                )
+                if saved_state == STATE_PLAYING:
+                    await self.async_media_play()
+            else:  # restore stashed queue
+                if saved_queue:
+                    uris = ""
+                    for item in saved_queue["items"]:
+                        uris += item["uri"] + ","
+                    await self._api.add_to_queue(
+                        uris=uris,
+                        playback="start",
+                        playback_from_position=saved_queue_position,
+                        clear=True,
+                    )
+                    await self._api.seek(position_ms=saved_song_position)
+                    if saved_state == STATE_PAUSED:
+                        await self.async_media_pause()
+                    elif saved_state != STATE_PLAYING:
+                        await self.async_media_stop()
+        else:
+            _LOGGER.debug("Media type '%s' not supported", media_type)
+
+    async def select_source(self, source):
+        """Change source.
+
+        Source name reflects whether in default mode or pipe mode.
+        Selecting playlists/clear sets the playlists/clears but ends up in default mode.
+        """
+        if source != self._source:
+            if (
+                self._use_pipe_control()
+            ):  # if pipe was playing, we need to stop it first
+                await self._pause_and_wait_for_callback()
+            self._source = source
+            if not self._use_pipe_control():  # playlist or clear ends up at default
+                self._source = SOURCE_NAME_DEFAULT
+            if self._sources_uris.get(source):  # load uris for pipes or playlists
+                await self._api.add_to_queue(
+                    uris=self._sources_uris[source], clear=True
+                )
+            elif source == SOURCE_NAME_CLEAR:  # clear playlist
+                await self._api.clear_queue()
+            self.async_write_ha_state()
+
+    def _use_pipe_control(self):
+        """Return which pipe control from KNOWN_PIPES to use."""
+        if self._source[-7:] == " (pipe)":
+            return self._source[:-7]
+        return ""
+
+    async def _pipe_call(self, pipe_name, base_function_name):
+        if self._pipe_control_api.get(pipe_name):
+            return await getattr(
+                self._pipe_control_api[pipe_name],
+                PIPE_FUNCTION_MAP[pipe_name][base_function_name],
+            )()
+        _LOGGER.warning("No pipe control available for %s", pipe_name)
+
+
+class ForkedDaapdUpdater:
+    """Manage updates for the forked-daapd device."""
+
+    def __init__(self, hass, api, entry_id):
+        """Initialize."""
+        self.hass = hass
+        self._api = api
+        self.websocket_handler = None
+        self._all_output_ids = set()
+        self._entry_id = entry_id
+
+    async def async_init(self):
+        """Perform async portion of class initialization."""
+        server_config = await self._api.get_request("config")
+        websocket_port = server_config.get("websocket_port")
+        if websocket_port:
+            self.websocket_handler = asyncio.create_task(
+                self._api.start_websocket_handler(
+                    server_config["websocket_port"],
+                    WS_NOTIFY_EVENT_TYPES,
+                    self._update,
+                    WEBSOCKET_RECONNECT_TIME,
+                    self._disconnected_callback,
+                )
+            )
+        else:
+            _LOGGER.error("Invalid websocket port")
+
+    def _disconnected_callback(self):
+        async_dispatcher_send(
+            self.hass, SIGNAL_UPDATE_MASTER.format(self._entry_id), False
+        )
+        async_dispatcher_send(
+            self.hass, SIGNAL_UPDATE_OUTPUTS.format(self._entry_id), []
+        )
+
+    async def _update(self, update_types):
+        """Private update method."""
+        update_types = set(update_types)
+        update_events = {}
+        _LOGGER.debug("Updating %s", update_types)
+        if (
+            "queue" in update_types
+        ):  # update queue, queue before player for async_play_media
+            queue = await self._api.get_request("queue")
+            update_events["queue"] = asyncio.Event()
+            async_dispatcher_send(
+                self.hass,
+                SIGNAL_UPDATE_QUEUE.format(self._entry_id),
+                queue,
+                update_events["queue"],
+            )
+        # order of below don't matter
+        if not {"outputs", "volume"}.isdisjoint(update_types):  # update outputs
+            outputs = (await self._api.get_request("outputs"))["outputs"]
+            update_events[
+                "outputs"
+            ] = asyncio.Event()  # only for master, zones should ignore
+            async_dispatcher_send(
+                self.hass,
+                SIGNAL_UPDATE_OUTPUTS.format(self._entry_id),
+                outputs,
+                update_events["outputs"],
+            )
+            self._add_zones(outputs)
+        if not {"database"}.isdisjoint(update_types):
+            pipes, playlists = await asyncio.gather(
+                self._api.get_pipes(), self._api.get_playlists()
+            )
+            update_events["database"] = asyncio.Event()
+            async_dispatcher_send(
+                self.hass,
+                SIGNAL_UPDATE_DATABASE.format(self._entry_id),
+                pipes,
+                playlists,
+                update_events["database"],
+            )
+        if not {"update", "config"}.isdisjoint(update_types):  # not supported
+            _LOGGER.debug("update/config notifications neither requested nor supported")
+        if not {"player", "options", "volume"}.isdisjoint(
+            update_types
+        ):  # update player
+            player = await self._api.get_request("player")
+            update_events["player"] = asyncio.Event()
+            if update_events.get("queue"):
+                await update_events[
+                    "queue"
+                ].wait()  # make sure queue done before player for async_play_media
+            async_dispatcher_send(
+                self.hass,
+                SIGNAL_UPDATE_PLAYER.format(self._entry_id),
+                player,
+                update_events["player"],
+            )
+        if update_events:
+            await asyncio.wait(
+                [event.wait() for event in update_events.values()]
+            )  # make sure callbacks done before update
+            async_dispatcher_send(
+                self.hass, SIGNAL_UPDATE_MASTER.format(self._entry_id), True
+            )
+
+    def _add_zones(self, outputs):
+        outputs_to_add = []
+        for output in outputs:
+            if output["id"] not in self._all_output_ids:
+                self._all_output_ids.add(output["id"])
+                outputs_to_add.append(output)
+        if outputs_to_add:
+            async_dispatcher_send(
+                self.hass,
+                SIGNAL_ADD_ZONES.format(self._entry_id),
+                self._api,
+                outputs_to_add,
+            )

homeassistant/components/forked_daapd/strings.json

@@ -0,0 +1,41 @@
+{
+  "config": {
+    "flow_title": "forked-daapd server: {name} ({host})",
+    "step": {
+      "user": {
+        "title": "Set up forked-daapd device",
+        "data": {
+          "name": "Friendly name",
+          "host": "Host",
+          "port": "API port",
+          "password": "API password (leave blank if no password)"
+        }
+      }
+    },
+    "error": {
+      "websocket_not_enabled": "forked-daapd server websocket not enabled.",
+      "wrong_host_or_port": "Unable to connect. Please check host and port.",
+      "wrong_password": "Incorrect password.",
+      "wrong_server_type": "Not a forked-daapd server.",
+      "unknown_error": "Unknown error."
+    },
+    "abort": {
+      "already_configured": "Device is already configured.",
+      "not_forked_daapd": "Device is not a forked-daapd server."
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "title": "Configure forked-daapd options",
+        "description": "Set various options for the forked-daapd integration.",
+        "data": {
+          "librespot_java_port": "Port for librespot-java pipe control (if used)",
+          "max_playlists": "Max number of playlists used as sources",
+          "tts_volume": "TTS volume (float in range [0,1])",
+          "tts_pause_time": "Seconds to pause before and after TTS"
+        }
+      }
+    }
+  }
+}

homeassistant/components/forked_daapd/translations/en.json

@@ -0,0 +1,41 @@
+{
+  "config": {
+    "flow_title": "forked-daapd server: {name} ({host})",
+    "step": {
+      "user": {
+        "title": "Set up forked-daapd device",
+        "data": {
+          "name": "Friendly name",
+          "host": "Host",
+          "port": "API port",
+          "password": "API password (leave blank if no password)"
+        }
+      }
+    },
+    "error": {
+      "websocket_not_enabled": "forked-daapd server websocket not enabled.",
+      "wrong_host_or_port": "Unable to connect. Please check host and port.",
+      "wrong_password": "Incorrect password.",
+      "wrong_server_type": "Not a forked-daapd server.",
+      "unknown_error": "Unknown error."
+    },
+    "abort": {
+      "already_configured": "Device is already configured.",
+      "not_forked_daapd": "Device is not a forked-daapd server."
+    }
+  },
+  "options": {
+    "step": {
+      "init": {
+        "title": "Configure forked-daapd options",
+        "description": "Set various options for the forked-daapd integration.",
+        "data": {
+          "librespot_java_port": "Port for librespot-java pipe control (if used)",
+          "max_playlists": "Max number of playlists used as sources",
+          "tts_volume": "TTS volume (float in range [0,1])",
+          "tts_pause_time": "Seconds to pause before and after TTS"
+        }
+      }
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -40,6 +40,7 @@ FLOWS = [
     "flick_electric",
     "flume",
     "flunearyou",
+    "forked_daapd",
     "freebox",
     "fritzbox",
     "garmin_connect",

homeassistant/generated/zeroconf.py

@@ -10,6 +10,9 @@ ZEROCONF = {
         "axis",
         "doorbird"
     ],
+    "_daap._tcp.local.": [
+        "forked_daapd"
+    ],
     "_elg._tcp.local.": [
         "elgato"
     ],

requirements_all.txt

@@ -1331,6 +1331,9 @@ pyflunearyou==1.0.7
 # homeassistant.components.futurenow
 pyfnip==0.2
 
+# homeassistant.components.forked_daapd
+pyforked-daapd==0.1.8
+
 # homeassistant.components.fritzbox
 pyfritzhome==0.4.2
 
@@ -1416,6 +1419,9 @@ pylaunches==0.2.0
 # homeassistant.components.lg_netcast
 pylgnetcast-homeassistant==0.2.0.dev0
 
+# homeassistant.components.forked_daapd
+pylibrespot-java==0.1.0
+
 # homeassistant.components.linky
 pylinky==0.4.0
 

requirements_test_all.txt

@@ -553,6 +553,9 @@ pyflume==0.4.0
 # homeassistant.components.flunearyou
 pyflunearyou==1.0.7
 
+# homeassistant.components.forked_daapd
+pyforked-daapd==0.1.8
+
 # homeassistant.components.fritzbox
 pyfritzhome==0.4.2
 
@@ -593,6 +596,9 @@ pykira==0.1.1
 # homeassistant.components.lastfm
 pylast==3.2.1
 
+# homeassistant.components.forked_daapd
+pylibrespot-java==0.1.0
+
 # homeassistant.components.linky
 pylinky==0.4.0
 

tests/components/forked_daapd/__init__.py

@@ -0,0 +1 @@
+"""Tests for the forked_daapd component."""

tests/components/forked_daapd/test_config_flow.py

@@ -0,0 +1,181 @@
+"""The config flow tests for the forked_daapd media player platform."""
+import pytest
+
+from homeassistant import data_entry_flow
+from homeassistant.components.forked_daapd.const import (
+    CONF_LIBRESPOT_JAVA_PORT,
+    CONF_MAX_PLAYLISTS,
+    CONF_TTS_PAUSE_TIME,
+    CONF_TTS_VOLUME,
+    DOMAIN,
+)
+from homeassistant.config_entries import (
+    CONN_CLASS_LOCAL_PUSH,
+    SOURCE_USER,
+    SOURCE_ZEROCONF,
+)
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT
+
+from tests.async_mock import patch
+from tests.common import MockConfigEntry
+
+SAMPLE_CONFIG = {
+    "websocket_port": 3688,
+    "version": "25.0",
+    "buildoptions": [
+        "ffmpeg",
+        "iTunes XML",
+        "Spotify",
+        "LastFM",
+        "MPD",
+        "Device verification",
+        "Websockets",
+        "ALSA",
+    ],
+}
+
+
+@pytest.fixture(name="config_entry")
+def config_entry_fixture():
+    """Create hass config_entry fixture."""
+    data = {
+        CONF_HOST: "192.168.1.1",
+        CONF_PORT: "2345",
+        CONF_PASSWORD: "",
+    }
+    return MockConfigEntry(
+        version=1,
+        domain=DOMAIN,
+        title="",
+        data=data,
+        options={},
+        system_options={},
+        source=SOURCE_USER,
+        connection_class=CONN_CLASS_LOCAL_PUSH,
+        entry_id=1,
+    )
+
+
+async def test_show_form(hass):
+    """Test that the form is served with no input."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == SOURCE_USER
+
+
+async def test_config_flow(hass, config_entry):
+    """Test that the user step works."""
+    with patch(
+        "homeassistant.components.forked_daapd.config_flow.ForkedDaapdAPI.test_connection"
+    ) as mock_test_connection, patch(
+        "homeassistant.components.forked_daapd.media_player.ForkedDaapdAPI.get_request",
+        autospec=True,
+    ) as mock_get_request:
+        mock_get_request.return_value = SAMPLE_CONFIG
+        mock_test_connection.return_value = ["ok", "My Music on myhost"]
+        config_data = config_entry.data
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}, data=config_data
+        )
+        await hass.async_block_till_done()
+        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+        assert result["title"] == "My Music on myhost"
+        assert result["data"][CONF_HOST] == config_data[CONF_HOST]
+        assert result["data"][CONF_PORT] == config_data[CONF_PORT]
+        assert result["data"][CONF_PASSWORD] == config_data[CONF_PASSWORD]
+
+        # Also test that creating a new entry with the same host aborts
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}, data=config_entry.data,
+        )
+        await hass.async_block_till_done()
+        assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+
+
+async def test_zeroconf_updates_title(hass, config_entry):
+    """Test that zeroconf updates title and aborts with same host."""
+    MockConfigEntry(domain=DOMAIN, data={CONF_HOST: "different host"}).add_to_hass(hass)
+    config_entry.add_to_hass(hass)
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 2
+    discovery_info = {
+        "host": "192.168.1.1",
+        "port": 23,
+        "properties": {"mtd-version": 1, "Machine Name": "zeroconf_test"},
+    }
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info
+    )
+    await hass.async_block_till_done()
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert config_entry.title == "zeroconf_test"
+    assert len(hass.config_entries.async_entries(DOMAIN)) == 2
+
+
+async def test_config_flow_no_websocket(hass, config_entry):
+    """Test config flow setup without websocket enabled on server."""
+    with patch(
+        "homeassistant.components.forked_daapd.config_flow.ForkedDaapdAPI.test_connection"
+    ) as mock_test_connection:
+        # test invalid config data
+        mock_test_connection.return_value = ["websocket_not_enabled"]
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}, data=config_entry.data
+        )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+async def test_config_flow_zeroconf_invalid(hass):
+    """Test that an invalid zeroconf entry doesn't work."""
+    discovery_info = {"host": "127.0.0.1", "port": 23}
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info
+    )  # doesn't create the entry, tries to show form but gets abort
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "not_forked_daapd"
+
+
+async def test_config_flow_zeroconf_valid(hass):
+    """Test that a valid zeroconf entry works."""
+    discovery_info = {
+        "host": "192.168.1.1",
+        "port": 23,
+        "properties": {
+            "mtd-version": 1,
+            "Machine Name": "zeroconf_test",
+            "Machine ID": "5E55EEFF",
+        },
+    }
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+async def test_options_flow(hass, config_entry):
+    """Test config flow options."""
+
+    with patch(
+        "homeassistant.components.forked_daapd.media_player.ForkedDaapdAPI.get_request",
+        autospec=True,
+    ) as mock_get_request:
+        mock_get_request.return_value = SAMPLE_CONFIG
+        config_entry.add_to_hass(hass)
+        await config_entry.async_setup(hass)
+        await hass.async_block_till_done()
+
+        result = await hass.config_entries.options.async_init(config_entry.entry_id)
+        assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+        result = await hass.config_entries.options.async_configure(
+            result["flow_id"],
+            user_input={
+                CONF_TTS_PAUSE_TIME: 0.05,
+                CONF_TTS_VOLUME: 0.8,
+                CONF_LIBRESPOT_JAVA_PORT: 0,
+                CONF_MAX_PLAYLISTS: 8,
+            },
+        )
+        assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY

tests/components/forked_daapd/test_media_player.py

@@ -0,0 +1,726 @@
+"""The media player tests for the forked_daapd media player platform."""
+
+import pytest
+
+from homeassistant.components.forked_daapd.const import (
+    CONF_LIBRESPOT_JAVA_PORT,
+    CONF_MAX_PLAYLISTS,
+    CONF_TTS_PAUSE_TIME,
+    CONF_TTS_VOLUME,
+    DOMAIN,
+    SOURCE_NAME_CLEAR,
+    SOURCE_NAME_DEFAULT,
+    SUPPORTED_FEATURES,
+    SUPPORTED_FEATURES_ZONE,
+)
+from homeassistant.components.media_player import (
+    SERVICE_CLEAR_PLAYLIST,
+    SERVICE_MEDIA_NEXT_TRACK,
+    SERVICE_MEDIA_PAUSE,
+    SERVICE_MEDIA_PLAY,
+    SERVICE_MEDIA_PREVIOUS_TRACK,
+    SERVICE_MEDIA_SEEK,
+    SERVICE_MEDIA_STOP,
+    SERVICE_PLAY_MEDIA,
+    SERVICE_SELECT_SOURCE,
+    SERVICE_SHUFFLE_SET,
+    SERVICE_TOGGLE,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+    SERVICE_VOLUME_MUTE,
+    SERVICE_VOLUME_SET,
+)
+from homeassistant.components.media_player.const import (
+    ATTR_INPUT_SOURCE,
+    ATTR_MEDIA_ALBUM_ARTIST,
+    ATTR_MEDIA_ALBUM_NAME,
+    ATTR_MEDIA_ARTIST,
+    ATTR_MEDIA_CONTENT_ID,
+    ATTR_MEDIA_CONTENT_TYPE,
+    ATTR_MEDIA_DURATION,
+    ATTR_MEDIA_POSITION,
+    ATTR_MEDIA_SEEK_POSITION,
+    ATTR_MEDIA_SHUFFLE,
+    ATTR_MEDIA_TITLE,
+    ATTR_MEDIA_TRACK,
+    ATTR_MEDIA_VOLUME_LEVEL,
+    ATTR_MEDIA_VOLUME_MUTED,
+    DOMAIN as MP_DOMAIN,
+    MEDIA_TYPE_MUSIC,
+    MEDIA_TYPE_TVSHOW,
+)
+from homeassistant.config_entries import CONN_CLASS_LOCAL_PUSH, SOURCE_USER
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    ATTR_FRIENDLY_NAME,
+    ATTR_SUPPORTED_FEATURES,
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_PORT,
+    STATE_ON,
+    STATE_PAUSED,
+    STATE_UNAVAILABLE,
+)
+
+from tests.async_mock import patch
+from tests.common import MockConfigEntry
+
+TEST_MASTER_ENTITY_NAME = "media_player.forked_daapd_server"
+TEST_ZONE_ENTITY_NAMES = [
+    "media_player.forked_daapd_output_" + x
+    for x in ["kitchen", "computer", "daapd_fifo"]
+]
+
+OPTIONS_DATA = {
+    CONF_LIBRESPOT_JAVA_PORT: "123",
+    CONF_MAX_PLAYLISTS: 8,
+    CONF_TTS_PAUSE_TIME: 0,
+    CONF_TTS_VOLUME: 0.25,
+}
+
+SAMPLE_PLAYER_PAUSED = {
+    "state": "pause",
+    "repeat": "off",
+    "consume": False,
+    "shuffle": False,
+    "volume": 20,
+    "item_id": 12322,
+    "item_length_ms": 50,
+    "item_progress_ms": 5,
+}
+
+SAMPLE_PLAYER_PLAYING = {
+    "state": "play",
+    "repeat": "off",
+    "consume": False,
+    "shuffle": False,
+    "volume": 50,
+    "item_id": 12322,
+    "item_length_ms": 50,
+    "item_progress_ms": 5,
+}
+
+SAMPLE_PLAYER_STOPPED = {
+    "state": "stop",
+    "repeat": "off",
+    "consume": False,
+    "shuffle": False,
+    "volume": 0,
+    "item_id": 12322,
+    "item_length_ms": 50,
+    "item_progress_ms": 5,
+}
+
+SAMPLE_TTS_QUEUE = {
+    "version": 833,
+    "count": 1,
+    "items": [
+        {
+            "id": 12322,
+            "position": 0,
+            "track_id": 1234,
+            "title": "Short TTS file",
+            "artist": "Google",
+            "album": "No album",
+            "album_artist": "The xx",
+            "artwork_url": "http://art",
+            "length_ms": 0,
+            "track_number": 1,
+            "media_kind": "music",
+            "uri": "tts_proxy_somefile.mp3",
+        }
+    ],
+}
+
+SAMPLE_CONFIG = {
+    "websocket_port": 3688,
+    "version": "25.0",
+    "buildoptions": [
+        "ffmpeg",
+        "iTunes XML",
+        "Spotify",
+        "LastFM",
+        "MPD",
+        "Device verification",
+        "Websockets",
+        "ALSA",
+    ],
+}
+
+SAMPLE_CONFIG_NO_WEBSOCKET = {
+    "websocket_port": 0,
+    "version": "25.0",
+    "buildoptions": [
+        "ffmpeg",
+        "iTunes XML",
+        "Spotify",
+        "LastFM",
+        "MPD",
+        "Device verification",
+        "Websockets",
+        "ALSA",
+    ],
+}
+
+
+SAMPLE_OUTPUTS_ON = (
+    {
+        "id": "123456789012345",
+        "name": "kitchen",
+        "type": "AirPlay",
+        "selected": True,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 50,
+    },
+    {
+        "id": "0",
+        "name": "Computer",
+        "type": "ALSA",
+        "selected": True,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 19,
+    },
+    {
+        "id": "100",
+        "name": "daapd-fifo",
+        "type": "fifo",
+        "selected": False,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 0,
+    },
+)
+
+
+SAMPLE_OUTPUTS_UNSELECTED = [
+    {
+        "id": "123456789012345",
+        "name": "kitchen",
+        "type": "AirPlay",
+        "selected": False,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 0,
+    },
+    {
+        "id": "0",
+        "name": "Computer",
+        "type": "ALSA",
+        "selected": False,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 19,
+    },
+    {
+        "id": "100",
+        "name": "daapd-fifo",
+        "type": "fifo",
+        "selected": False,
+        "has_password": False,
+        "requires_auth": False,
+        "needs_auth_key": False,
+        "volume": 0,
+    },
+]
+
+SAMPLE_PIPES = [
+    {
+        "id": 1,
+        "title": "librespot-java",
+        "media_kind": "music",
+        "data_kind": "pipe",
+        "path": "/music/srv/input.pipe",
+        "uri": "library:track:1",
+    }
+]
+
+SAMPLE_PLAYLISTS = [{"id": 7, "name": "test_playlist", "uri": "library:playlist:2"}]
+
+
+@pytest.fixture(name="config_entry")
+def config_entry_fixture():
+    """Create hass config_entry fixture."""
+    data = {
+        CONF_HOST: "192.168.1.1",
+        CONF_PORT: "2345",
+        CONF_PASSWORD: "",
+    }
+    return MockConfigEntry(
+        version=1,
+        domain=DOMAIN,
+        title="",
+        data=data,
+        options={CONF_TTS_PAUSE_TIME: 0},
+        system_options={},
+        source=SOURCE_USER,
+        connection_class=CONN_CLASS_LOCAL_PUSH,
+        entry_id=1,
+    )
+
+
+@pytest.fixture(name="get_request_return_values")
+async def get_request_return_values_fixture():
+    """Get request return values we can change later."""
+    return {
+        "config": SAMPLE_CONFIG,
+        "outputs": SAMPLE_OUTPUTS_ON,
+        "player": SAMPLE_PLAYER_PAUSED,
+        "queue": SAMPLE_TTS_QUEUE,
+    }
+
+
+@pytest.fixture(name="mock_api_object")
+async def mock_api_object_fixture(hass, config_entry, get_request_return_values):
+    """Create mock api fixture."""
+
+    async def get_request_side_effect(update_type):
+        if update_type == "outputs":
+            return {"outputs": get_request_return_values["outputs"]}
+        return get_request_return_values[update_type]
+
+    with patch(
+        "homeassistant.components.forked_daapd.media_player.ForkedDaapdAPI",
+        autospec=True,
+    ) as mock_api:
+        mock_api.return_value.get_request.side_effect = get_request_side_effect
+        mock_api.return_value.full_url.return_value = ""
+        mock_api.return_value.get_pipes.return_value = SAMPLE_PIPES
+        mock_api.return_value.get_playlists.return_value = SAMPLE_PLAYLISTS
+        config_entry.add_to_hass(hass)
+        await config_entry.async_setup(hass)
+        await hass.async_block_till_done()
+
+    mock_api.return_value.start_websocket_handler.assert_called_once()
+    mock_api.return_value.get_request.assert_called_once()
+    updater_update = mock_api.return_value.start_websocket_handler.call_args[0][2]
+    await updater_update(["player", "outputs", "queue"])
+    await hass.async_block_till_done()
+
+    async def add_to_queue_side_effect(
+        uris, playback=None, playback_from_position=None, clear=None
+    ):
+        await updater_update(["queue", "player"])
+
+    mock_api.return_value.add_to_queue.side_effect = (
+        add_to_queue_side_effect  # for play_media testing
+    )
+
+    async def pause_side_effect():
+        await updater_update(["player"])
+
+    mock_api.return_value.pause_playback.side_effect = pause_side_effect
+
+    return mock_api.return_value
+
+
+async def test_unload_config_entry(hass, config_entry, mock_api_object):
+    """Test the player is removed when the config entry is unloaded."""
+    assert hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert hass.states.get(TEST_ZONE_ENTITY_NAMES[0])
+    await config_entry.async_unload(hass)
+    assert not hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert not hass.states.get(TEST_ZONE_ENTITY_NAMES[0])
+
+
+def test_master_state(hass, mock_api_object):
+    """Test master state attributes."""
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.state == STATE_PAUSED
+    assert state.attributes[ATTR_FRIENDLY_NAME] == "forked-daapd server"
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORTED_FEATURES
+    assert not state.attributes[ATTR_MEDIA_VOLUME_MUTED]
+    assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.2
+    assert state.attributes[ATTR_MEDIA_CONTENT_ID] == 12322
+    assert state.attributes[ATTR_MEDIA_CONTENT_TYPE] == MEDIA_TYPE_MUSIC
+    assert state.attributes[ATTR_MEDIA_DURATION] == 0.05
+    assert state.attributes[ATTR_MEDIA_POSITION] == 0.005
+    assert state.attributes[ATTR_MEDIA_TITLE] == "Short TTS file"
+    assert state.attributes[ATTR_MEDIA_ARTIST] == "Google"
+    assert state.attributes[ATTR_MEDIA_ALBUM_NAME] == "No album"
+    assert state.attributes[ATTR_MEDIA_ALBUM_ARTIST] == "The xx"
+    assert state.attributes[ATTR_MEDIA_TRACK] == 1
+    assert not state.attributes[ATTR_MEDIA_SHUFFLE]
+
+
+async def _service_call(
+    hass, entity_name, service, additional_service_data=None, blocking=True
+):
+    if additional_service_data is None:
+        additional_service_data = {}
+    return await hass.services.async_call(
+        MP_DOMAIN,
+        service,
+        service_data={ATTR_ENTITY_ID: entity_name, **additional_service_data},
+        blocking=blocking,
+    )
+
+
+async def test_zone(hass, mock_api_object):
+    """Test zone attributes and methods."""
+    zone_entity_name = TEST_ZONE_ENTITY_NAMES[0]
+    state = hass.states.get(zone_entity_name)
+    assert state.attributes[ATTR_FRIENDLY_NAME] == "forked-daapd output (kitchen)"
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == SUPPORTED_FEATURES_ZONE
+    assert state.state == STATE_ON
+    assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.5
+    assert not state.attributes[ATTR_MEDIA_VOLUME_MUTED]
+    await _service_call(hass, zone_entity_name, SERVICE_TURN_ON)
+    await _service_call(hass, zone_entity_name, SERVICE_TURN_OFF)
+    await _service_call(hass, zone_entity_name, SERVICE_TOGGLE)
+    await _service_call(
+        hass, zone_entity_name, SERVICE_VOLUME_SET, {ATTR_MEDIA_VOLUME_LEVEL: 0.3}
+    )
+    await _service_call(
+        hass, zone_entity_name, SERVICE_VOLUME_MUTE, {ATTR_MEDIA_VOLUME_MUTED: True}
+    )
+    await _service_call(
+        hass, zone_entity_name, SERVICE_VOLUME_MUTE, {ATTR_MEDIA_VOLUME_MUTED: False}
+    )
+    zone_entity_name = TEST_ZONE_ENTITY_NAMES[2]
+    await _service_call(hass, zone_entity_name, SERVICE_TOGGLE)
+    await _service_call(
+        hass, zone_entity_name, SERVICE_VOLUME_MUTE, {ATTR_MEDIA_VOLUME_MUTED: True}
+    )
+    output_id = SAMPLE_OUTPUTS_ON[0]["id"]
+    initial_volume = SAMPLE_OUTPUTS_ON[0]["volume"]
+    mock_api_object.change_output.assert_any_call(output_id, selected=True)
+    mock_api_object.change_output.assert_any_call(output_id, selected=False)
+    mock_api_object.set_volume.assert_any_call(output_id=output_id, volume=30)
+    mock_api_object.set_volume.assert_any_call(output_id=output_id, volume=0)
+    mock_api_object.set_volume.assert_any_call(
+        output_id=output_id, volume=initial_volume
+    )
+    output_id = SAMPLE_OUTPUTS_ON[2]["id"]
+    mock_api_object.change_output.assert_any_call(output_id, selected=True)
+
+
+async def test_last_outputs_master(hass, mock_api_object):
+    """Test restoration of _last_outputs."""
+    # Test turning on sends API call
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TURN_ON)
+    assert mock_api_object.change_output.call_count == 0
+    assert mock_api_object.set_enabled_outputs.call_count == 1
+    await _service_call(
+        hass, TEST_MASTER_ENTITY_NAME, SERVICE_TURN_OFF
+    )  # should have stored last outputs
+    assert mock_api_object.change_output.call_count == 0
+    assert mock_api_object.set_enabled_outputs.call_count == 2
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TURN_ON)
+    assert mock_api_object.change_output.call_count == 3
+    assert mock_api_object.set_enabled_outputs.call_count == 2
+
+
+async def test_bunch_of_stuff_master(hass, mock_api_object, get_request_return_values):
+    """Run bunch of stuff."""
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TURN_ON)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TURN_OFF)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TOGGLE)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_MEDIA_VOLUME_MUTED: True},
+    )
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_MEDIA_VOLUME_MUTED: False},
+    )
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_VOLUME_SET,
+        {ATTR_MEDIA_VOLUME_LEVEL: 0.5},
+    )
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PAUSE)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PLAY)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_STOP)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PREVIOUS_TRACK)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_NEXT_TRACK)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_MEDIA_SEEK,
+        {ATTR_MEDIA_SEEK_POSITION: 35},
+    )
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_CLEAR_PLAYLIST)
+    await _service_call(
+        hass, TEST_MASTER_ENTITY_NAME, SERVICE_SHUFFLE_SET, {ATTR_MEDIA_SHUFFLE: False}
+    )
+    # stop player and run more stuff
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.2
+    get_request_return_values["player"] = SAMPLE_PLAYER_STOPPED
+    updater_update = mock_api_object.start_websocket_handler.call_args[0][2]
+    await updater_update(["player"])
+    await hass.async_block_till_done()
+    # mute from volume==0
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_MEDIA_VOLUME_MUTED: True},
+    )
+    # now turn off (stopped and all outputs unselected)
+    get_request_return_values["outputs"] = SAMPLE_OUTPUTS_UNSELECTED
+    await updater_update(["outputs"])
+    await hass.async_block_till_done()
+    # toggle from off
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_TOGGLE)
+    for output in SAMPLE_OUTPUTS_ON:
+        mock_api_object.change_output.assert_any_call(
+            output["id"], selected=output["selected"], volume=output["volume"],
+        )
+    mock_api_object.set_volume.assert_any_call(volume=0)
+    mock_api_object.set_volume.assert_any_call(volume=SAMPLE_PLAYER_PAUSED["volume"])
+    mock_api_object.set_volume.assert_any_call(volume=50)
+    mock_api_object.set_enabled_outputs.assert_any_call(
+        [output["id"] for output in SAMPLE_OUTPUTS_ON]
+    )
+    mock_api_object.set_enabled_outputs.assert_any_call([])
+    mock_api_object.start_playback.assert_called_once()
+    assert mock_api_object.pause_playback.call_count == 3
+    mock_api_object.stop_playback.assert_called_once()
+    mock_api_object.previous_track.assert_called_once()
+    mock_api_object.next_track.assert_called_once()
+    mock_api_object.seek.assert_called_once()
+    mock_api_object.shuffle.assert_called_once()
+    mock_api_object.clear_queue.assert_called_once()
+
+
+async def test_async_play_media_from_paused(hass, mock_api_object):
+    """Test async play media from paused."""
+    initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_PLAY_MEDIA,
+        {
+            ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
+            ATTR_MEDIA_CONTENT_ID: "somefile.mp3",
+        },
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.state == initial_state.state
+    assert state.last_updated > initial_state.last_updated
+
+
+async def test_async_play_media_from_stopped(
+    hass, get_request_return_values, mock_api_object
+):
+    """Test async play media from stopped."""
+    updater_update = mock_api_object.start_websocket_handler.call_args[0][2]
+
+    get_request_return_values["player"] = SAMPLE_PLAYER_STOPPED
+    await updater_update(["player"])
+    await hass.async_block_till_done()
+    initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_PLAY_MEDIA,
+        {
+            ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
+            ATTR_MEDIA_CONTENT_ID: "somefile.mp3",
+        },
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.state == initial_state.state
+    assert state.last_updated > initial_state.last_updated
+
+
+async def test_async_play_media_unsupported(hass, mock_api_object):
+    """Test async play media on unsupported media type."""
+    initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_PLAY_MEDIA,
+        {
+            ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_TVSHOW,
+            ATTR_MEDIA_CONTENT_ID: "wontwork.mp4",
+        },
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.last_updated == initial_state.last_updated
+
+
+async def test_async_play_media_tts_timeout(hass, mock_api_object):
+    """Test async play media with TTS timeout."""
+    mock_api_object.add_to_queue.side_effect = None
+    with patch("homeassistant.components.forked_daapd.media_player.TTS_TIMEOUT", 0):
+        initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+        await _service_call(
+            hass,
+            TEST_MASTER_ENTITY_NAME,
+            SERVICE_PLAY_MEDIA,
+            {
+                ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
+                ATTR_MEDIA_CONTENT_ID: "somefile.mp3",
+            },
+        )
+        state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+        assert state.state == initial_state.state
+        assert state.last_updated > initial_state.last_updated
+
+
+async def test_use_pipe_control_with_no_api(hass, mock_api_object):
+    """Test using pipe control with no api set."""
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_INPUT_SOURCE: "librespot-java (pipe)"},
+    )
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PLAY)
+    assert mock_api_object.start_playback.call_count == 0
+
+
+async def test_clear_source(hass, mock_api_object):
+    """Test changing source to clear."""
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_INPUT_SOURCE: SOURCE_NAME_CLEAR},
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.attributes[ATTR_INPUT_SOURCE] == SOURCE_NAME_DEFAULT
+
+
+@pytest.fixture(name="pipe_control_api_object")
+async def pipe_control_api_object_fixture(
+    hass, config_entry, get_request_return_values, mock_api_object
+):
+    """Fixture for mock librespot_java api."""
+    with patch(
+        "homeassistant.components.forked_daapd.media_player.LibrespotJavaAPI",
+        autospec=True,
+    ) as pipe_control_api:
+        hass.config_entries.async_update_entry(config_entry, options=OPTIONS_DATA)
+        await hass.async_block_till_done()
+    get_request_return_values["player"] = SAMPLE_PLAYER_PLAYING
+    updater_update = mock_api_object.start_websocket_handler.call_args[0][2]
+    await updater_update(["player"])
+    await hass.async_block_till_done()
+
+    async def pause_side_effect():
+        await updater_update(["player"])
+
+    pipe_control_api.return_value.player_pause.side_effect = pause_side_effect
+
+    await updater_update(["database"])  # load in sources
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_INPUT_SOURCE: "librespot-java (pipe)"},
+    )
+
+    return pipe_control_api.return_value
+
+
+async def test_librespot_java_stuff(hass, pipe_control_api_object):
+    """Test options update and librespot-java stuff."""
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.attributes[ATTR_INPUT_SOURCE] == "librespot-java (pipe)"
+    # call some basic services
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_STOP)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PREVIOUS_TRACK)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_NEXT_TRACK)
+    await _service_call(hass, TEST_MASTER_ENTITY_NAME, SERVICE_MEDIA_PLAY)
+    pipe_control_api_object.player_pause.assert_called_once()
+    pipe_control_api_object.player_prev.assert_called_once()
+    pipe_control_api_object.player_next.assert_called_once()
+    pipe_control_api_object.player_resume.assert_called_once()
+    # switch away
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_INPUT_SOURCE: SOURCE_NAME_DEFAULT},
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.attributes[ATTR_INPUT_SOURCE] == SOURCE_NAME_DEFAULT
+
+
+async def test_librespot_java_play_media(hass, pipe_control_api_object):
+    """Test play media with librespot-java pipe."""
+    initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    await _service_call(
+        hass,
+        TEST_MASTER_ENTITY_NAME,
+        SERVICE_PLAY_MEDIA,
+        {
+            ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
+            ATTR_MEDIA_CONTENT_ID: "somefile.mp3",
+        },
+    )
+    state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+    assert state.state == initial_state.state
+    assert state.last_updated > initial_state.last_updated
+
+
+async def test_librespot_java_play_media_pause_timeout(hass, pipe_control_api_object):
+    """Test play media with librespot-java pipe."""
+    # test media play with pause timeout
+    pipe_control_api_object.player_pause.side_effect = None
+    with patch(
+        "homeassistant.components.forked_daapd.media_player.CALLBACK_TIMEOUT", 0
+    ):
+        initial_state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+        await _service_call(
+            hass,
+            TEST_MASTER_ENTITY_NAME,
+            SERVICE_PLAY_MEDIA,
+            {
+                ATTR_MEDIA_CONTENT_TYPE: MEDIA_TYPE_MUSIC,
+                ATTR_MEDIA_CONTENT_ID: "somefile.mp3",
+            },
+        )
+        state = hass.states.get(TEST_MASTER_ENTITY_NAME)
+        assert state.state == initial_state.state
+        assert state.last_updated > initial_state.last_updated
+
+
+async def test_unsupported_update(hass, mock_api_object):
+    """Test unsupported update type."""
+    last_updated = hass.states.get(TEST_MASTER_ENTITY_NAME).last_updated
+    updater_update = mock_api_object.start_websocket_handler.call_args[0][2]
+    await updater_update(["config"])
+    await hass.async_block_till_done()
+    assert hass.states.get(TEST_MASTER_ENTITY_NAME).last_updated == last_updated
+
+
+async def test_invalid_websocket_port(hass, config_entry):
+    """Test invalid websocket port on async_init."""
+    with patch(
+        "homeassistant.components.forked_daapd.media_player.ForkedDaapdAPI",
+        autospec=True,
+    ) as mock_api:
+        mock_api.return_value.get_request.return_value = SAMPLE_CONFIG_NO_WEBSOCKET
+        config_entry.add_to_hass(hass)
+        await config_entry.async_setup(hass)
+        await hass.async_block_till_done()
+        assert hass.states.get(TEST_MASTER_ENTITY_NAME).state == STATE_UNAVAILABLE
+
+
+async def test_websocket_disconnect(hass, mock_api_object):
+    """Test websocket disconnection."""
+    assert hass.states.get(TEST_MASTER_ENTITY_NAME).state != STATE_UNAVAILABLE
+    assert hass.states.get(TEST_ZONE_ENTITY_NAMES[0]).state != STATE_UNAVAILABLE
+    updater_disconnected = mock_api_object.start_websocket_handler.call_args[0][4]
+    updater_disconnected()
+    await hass.async_block_till_done()
+    assert hass.states.get(TEST_MASTER_ENTITY_NAME).state == STATE_UNAVAILABLE
+    assert hass.states.get(TEST_ZONE_ENTITY_NAMES[0]).state == STATE_UNAVAILABLE