"""Tests for the Sonos Media Player platform."""

from collections.abc import Generator
from datetime import UTC, datetime
from typing import Any
from unittest.mock import MagicMock, patch

from freezegun import freeze_time
import pytest
from soco.data_structures import (
    DidlAudioBroadcast,
    DidlAudioLineIn,
    DidlPlaylistContainer,
    SearchResult,
)
from sonos_websocket.exception import SonosWebsocketError
from syrupy.assertion import SnapshotAssertion

from homeassistant.components.media_player import (
    ATTR_INPUT_SOURCE,
    ATTR_INPUT_SOURCE_LIST,
    ATTR_MEDIA_ALBUM_NAME,
    ATTR_MEDIA_ANNOUNCE,
    ATTR_MEDIA_ARTIST,
    ATTR_MEDIA_CHANNEL,
    ATTR_MEDIA_CONTENT_ID,
    ATTR_MEDIA_CONTENT_TYPE,
    ATTR_MEDIA_DURATION,
    ATTR_MEDIA_ENQUEUE,
    ATTR_MEDIA_EXTRA,
    ATTR_MEDIA_PLAYLIST,
    ATTR_MEDIA_POSITION,
    ATTR_MEDIA_POSITION_UPDATED_AT,
    ATTR_MEDIA_REPEAT,
    ATTR_MEDIA_SHUFFLE,
    ATTR_MEDIA_TITLE,
    ATTR_MEDIA_VOLUME_LEVEL,
    DOMAIN as MP_DOMAIN,
    SERVICE_CLEAR_PLAYLIST,
    SERVICE_PLAY_MEDIA,
    SERVICE_SELECT_SOURCE,
    MediaPlayerEnqueue,
    RepeatMode,
)
from homeassistant.components.sonos.const import (
    DOMAIN,
    MEDIA_TYPE_DIRECTORY,
    SOURCE_LINEIN,
    SOURCE_TV,
)
from homeassistant.components.sonos.media_player import (
    LONG_SERVICE_TIMEOUT,
    VOLUME_INCREMENT,
)
from homeassistant.components.sonos.services import (
    ATTR_ALARM_ID,
    ATTR_ENABLED,
    ATTR_INCLUDE_LINKED_ZONES,
    ATTR_QUEUE_POSITION,
    ATTR_VOLUME,
    SERVICE_GET_QUEUE,
    SERVICE_RESTORE,
    SERVICE_SNAPSHOT,
    SERVICE_UPDATE_ALARM,
)
from homeassistant.const import (
    ATTR_ENTITY_ID,
    ATTR_ENTITY_PICTURE,
    ATTR_TIME,
    SERVICE_MEDIA_NEXT_TRACK,
    SERVICE_MEDIA_PAUSE,
    SERVICE_MEDIA_PLAY,
    SERVICE_MEDIA_PREVIOUS_TRACK,
    SERVICE_MEDIA_STOP,
    SERVICE_REPEAT_SET,
    SERVICE_SHUFFLE_SET,
    SERVICE_VOLUME_DOWN,
    SERVICE_VOLUME_SET,
    SERVICE_VOLUME_UP,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import area_registry as ar, entity_registry as er
from homeassistant.helpers.device_registry import (
    CONNECTION_NETWORK_MAC,
    CONNECTION_UPNP,
    DeviceRegistry,
)
from homeassistant.setup import async_setup_component

from .conftest import MockMusicServiceItem, MockSoCo, SoCoMockFactory, SonosMockEvent


@pytest.fixture(autouse=True)
def mock_token() -> Generator[MagicMock]:
    """Mock token generator."""
    with patch("secrets.token_hex", return_value="123456789") as token:
        yield token


async def test_device_registry(
    hass: HomeAssistant, device_registry: DeviceRegistry, async_autosetup_sonos, soco
) -> None:
    """Test sonos device registered in the device registry."""
    reg_device = device_registry.async_get_device(
        identifiers={("sonos", "RINCON_test")}
    )
    assert reg_device is not None
    assert reg_device.model == "Model Name"
    assert reg_device.model_id == "S12"
    assert reg_device.sw_version == "13.1"
    assert reg_device.connections == {
        (CONNECTION_NETWORK_MAC, "00:11:22:33:44:55"),
        (CONNECTION_UPNP, "uuid:RINCON_test"),
    }
    assert reg_device.manufacturer == "Sonos"
    assert reg_device.name == "Zone A"
    # Default device provides battery info, area should not be suggested
    assert reg_device.area_id is None


async def test_device_registry_not_portable(
    hass: HomeAssistant,
    area_registry: ar.AreaRegistry,
    device_registry: DeviceRegistry,
    async_setup_sonos,
    soco,
) -> None:
    """Test non-portable sonos device has area suggested."""
    soco.get_battery_info.return_value = {}
    await async_setup_sonos()

    reg_device = device_registry.async_get_device(
        identifiers={("sonos", "RINCON_test")}
    )
    assert reg_device is not None
    assert reg_device.area_id == area_registry.async_get_area_by_name("Zone A").id


async def test_entity_basic(
    hass: HomeAssistant,
    async_autosetup_sonos,
    discover,
    entity_registry: er.EntityRegistry,
    snapshot: SnapshotAssertion,
) -> None:
    """Test basic state and attributes."""
    entity_id = "media_player.zone_a"
    entity_entry = entity_registry.async_get(entity_id)
    assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
    state = hass.states.get(entity_entry.entity_id)
    assert state == snapshot(name=f"{entity_entry.entity_id}-state")


@pytest.mark.parametrize(
    ("media_content_type", "media_content_id", "enqueue", "test_result"),
    [
        (
            "artist",
            "A:ALBUMARTIST/Beatles",
            MediaPlayerEnqueue.REPLACE,
            {
                "title": "All",
                "item_id": "A:ALBUMARTIST/Beatles/",
                "clear_queue": 1,
                "position": None,
                "play": 1,
                "expected_play_args": [0],
            },
        ),
        (
            "genre",
            "A:GENRE/Classic%20Rock",
            MediaPlayerEnqueue.ADD,
            {
                "title": "All",
                "item_id": "A:GENRE/Classic%20Rock/",
                "clear_queue": 0,
                "position": None,
                "play": 0,
            },
        ),
        (
            "album",
            "A:ALBUM/Abbey%20Road",
            MediaPlayerEnqueue.NEXT,
            {
                "title": "Abbey Road",
                "item_id": "A:ALBUM/Abbey%20Road",
                "clear_queue": 0,
                "position": 1,
                "play": 0,
            },
        ),
        (
            "composer",
            "A:COMPOSER/Carlos%20Santana",
            MediaPlayerEnqueue.PLAY,
            {
                "title": "All",
                "item_id": "A:COMPOSER/Carlos%20Santana/",
                "clear_queue": 0,
                "position": 1,
                "play": 1,
                "expected_play_args": [9],
            },
        ),
        (
            "artist",
            "A:ALBUMARTIST/Beatles/Abbey%20Road",
            MediaPlayerEnqueue.REPLACE,
            {
                "title": "Abbey Road",
                "item_id": "A:ALBUMARTIST/Beatles/Abbey%20Road",
                "clear_queue": 1,
                "position": None,
                "play": 1,
                "expected_play_args": [0],
            },
        ),
        (
            MEDIA_TYPE_DIRECTORY,
            "S://192.168.1.1/music/elton%20john",
            MediaPlayerEnqueue.REPLACE,
            {
                "title": None,
                "item_id": "S://192.168.1.1/music/elton%20john",
                "clear_queue": 1,
                "position": None,
                "play": 1,
                "expected_play_args": [0],
            },
        ),
    ],
)
async def test_play_media_library(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    media_content_type,
    media_content_id,
    enqueue,
    test_result,
) -> None:
    """Test playing local library with a variety of options."""
    sock_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: media_content_type,
            ATTR_MEDIA_CONTENT_ID: media_content_id,
            ATTR_MEDIA_ENQUEUE: enqueue,
        },
        blocking=True,
    )
    assert sock_mock.clear_queue.call_count == test_result["clear_queue"]
    assert sock_mock.add_to_queue.call_count == 1
    assert (
        sock_mock.add_to_queue.call_args_list[0].args[0].title == test_result["title"]
    )
    assert (
        sock_mock.add_to_queue.call_args_list[0].args[0].item_id
        == test_result["item_id"]
    )
    assert (
        sock_mock.add_to_queue.call_args_list[0].kwargs.get("position")
        == test_result["position"]
    )

    assert (
        sock_mock.add_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert sock_mock.play_from_queue.call_count == test_result["play"]
    actual_play_args = [
        call.args[0] for call in sock_mock.play_from_queue.call_args_list
    ]
    assert actual_play_args == test_result.get("expected_play_args", [])


@pytest.mark.parametrize(
    ("media_content_type", "media_content_id", "message"),
    [
        (
            "artist",
            "A:ALBUM/UnknowAlbum",
            "Could not find media in library: A:ALBUM/UnknowAlbum",
        ),
        (
            "UnknownContent",
            "A:ALBUM/UnknowAlbum",
            "Sonos does not support media content type: UnknownContent",
        ),
        (
            MEDIA_TYPE_DIRECTORY,
            "S://192.168.1.1/music/error",
            "Could not find media in library: S://192.168.1.1/music/error",
        ),
    ],
)
async def test_play_media_library_content_error(
    hass: HomeAssistant,
    async_autosetup_sonos,
    media_content_type,
    media_content_id,
    message,
) -> None:
    """Test playing local library errors on content and content type."""
    with pytest.raises(
        ServiceValidationError,
        match=message,
    ):
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: media_content_type,
                ATTR_MEDIA_CONTENT_ID: media_content_id,
            },
            blocking=True,
        )


_track_url = "S://192.168.42.100/music/iTunes/The%20Beatles/A%20Hard%20Day%2fs%I%20Should%20Have%20Known%20Better.mp3"


async def test_play_media_lib_track_play(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Tests playing media track with enqueue mode play."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "track",
            ATTR_MEDIA_CONTENT_ID: _track_url,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.PLAY,
        },
        blocking=True,
    )
    assert soco_mock.add_uri_to_queue.call_count == 1
    assert soco_mock.add_uri_to_queue.call_args_list[0].args[0] == _track_url
    assert soco_mock.add_uri_to_queue.call_args_list[0].kwargs["position"] == 1
    assert (
        soco_mock.add_uri_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_from_queue.call_count == 1
    assert soco_mock.play_from_queue.call_args_list[0].args[0] == 9


async def test_play_media_lib_track_next(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Tests playing media track with enqueue mode next."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "track",
            ATTR_MEDIA_CONTENT_ID: _track_url,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.NEXT,
        },
        blocking=True,
    )
    assert soco_mock.add_uri_to_queue.call_count == 1
    assert soco_mock.add_uri_to_queue.call_args_list[0].args[0] == _track_url
    assert soco_mock.add_uri_to_queue.call_args_list[0].kwargs["position"] == 1
    assert (
        soco_mock.add_uri_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_from_queue.call_count == 0


async def test_play_media_lib_track_replace(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Tests playing media track with enqueue mode replace."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "track",
            ATTR_MEDIA_CONTENT_ID: _track_url,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.REPLACE,
        },
        blocking=True,
    )
    assert soco_mock.play_uri.call_count == 1
    assert soco_mock.play_uri.call_args_list[0].args[0] == _track_url
    assert soco_mock.play_uri.call_args_list[0].kwargs["force_radio"] is False


async def test_play_media_lib_track_add(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Tests playing media track with enqueue mode add."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "track",
            ATTR_MEDIA_CONTENT_ID: _track_url,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.ADD,
        },
        blocking=True,
    )
    assert soco_mock.add_uri_to_queue.call_count == 1
    assert soco_mock.add_uri_to_queue.call_args_list[0].args[0] == _track_url
    assert (
        soco_mock.add_uri_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_from_queue.call_count == 0


_share_link: str = "spotify:playlist:abcdefghij0123456789XY"
_share_link_title: str = "playlist title"


async def test_play_media_share_link_add(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    soco_sharelink,
) -> None:
    """Tests playing a share link with enqueue option add."""
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: _share_link,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.ADD,
            ATTR_MEDIA_EXTRA: {"title": _share_link_title},
        },
        blocking=True,
    )
    assert soco_sharelink.add_share_link_to_queue.call_count == 1
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].args[0] == _share_link
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["dc_title"]
        == _share_link_title
    )


async def test_play_media_share_link_next(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    soco_sharelink,
) -> None:
    """Tests playing a share link with enqueue option next."""
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: _share_link,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.NEXT,
        },
        blocking=True,
    )
    assert soco_sharelink.add_share_link_to_queue.call_count == 1
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].args[0] == _share_link
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["position"] == 1
    )
    assert (
        "dc_title"
        not in soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs
    )


async def test_play_media_share_link_play(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    soco_sharelink,
) -> None:
    """Tests playing a share link with enqueue option play."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: _share_link,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.PLAY,
        },
        blocking=True,
    )
    assert soco_sharelink.add_share_link_to_queue.call_count == 1
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].args[0] == _share_link
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["position"] == 1
    )
    assert soco_mock.play_from_queue.call_count == 1
    soco_mock.play_from_queue.assert_called_with(9)


async def test_play_media_share_link_replace(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    soco_sharelink,
) -> None:
    """Tests playing a share link with enqueue option replace."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: _share_link,
            ATTR_MEDIA_ENQUEUE: MediaPlayerEnqueue.REPLACE,
        },
        blocking=True,
    )
    assert soco_mock.clear_queue.call_count == 1
    assert soco_sharelink.add_share_link_to_queue.call_count == 1
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].args[0] == _share_link
    )
    assert (
        soco_sharelink.add_share_link_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_from_queue.call_count == 1
    soco_mock.play_from_queue.assert_called_with(0)


_mock_playlists = [
    MockMusicServiceItem(
        "playlist1",
        "S://192.168.1.68/music/iTunes/iTunes%20Music%20Library.xml#GUID_1",
        "A:PLAYLISTS",
        "object.container.playlistContainer",
    ),
    MockMusicServiceItem(
        "playlist2",
        "S://192.168.1.68/music/iTunes/iTunes%20Music%20Library.xml#GUID_2",
        "A:PLAYLISTS",
        "object.container.playlistContainer",
    ),
]


@pytest.mark.parametrize(
    ("media_content_id", "expected_item_id"),
    [
        (
            _mock_playlists[0].item_id,
            _mock_playlists[0].item_id,
        ),
        (
            f"S:{_mock_playlists[1].title}",
            _mock_playlists[1].item_id,
        ),
    ],
)
async def test_play_media_music_library_playlist(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    discover,
    media_content_id,
    expected_item_id,
) -> None:
    """Test that playlists can be found by id or title."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    soco_mock.music_library.get_playlists.return_value = _mock_playlists

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: media_content_id,
        },
        blocking=True,
    )

    assert soco_mock.clear_queue.call_count == 1
    assert soco_mock.add_to_queue.call_count == 1
    assert soco_mock.add_to_queue.call_args_list[0].args[0].item_id == expected_item_id
    assert soco_mock.play_from_queue.call_count == 1


async def test_play_media_music_library_playlist_dne(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test error handling when attempting to play a non-existent playlist ."""
    media_content_id = "S:nonexistent"
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    soco_mock.music_library.get_playlists.return_value = _mock_playlists

    with pytest.raises(
        ServiceValidationError,
        match=f"Could not find Sonos playlist: {media_content_id}",
    ):
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: "playlist",
                ATTR_MEDIA_CONTENT_ID: media_content_id,
            },
            blocking=True,
        )
    assert soco_mock.play_uri.call_count == 0


async def test_play_sonos_playlist(
    hass: HomeAssistant,
    async_autosetup_sonos,
    soco: MockSoCo,
    sonos_playlists: SearchResult,
) -> None:
    """Test that sonos playlists can be played."""

    # Test a successful call
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "playlist",
            ATTR_MEDIA_CONTENT_ID: "sample playlist",
        },
        blocking=True,
    )
    assert soco.clear_queue.call_count == 1
    assert soco.add_to_queue.call_count == 1
    soco.add_to_queue.asset_called_with(
        sonos_playlists[0], timeout=LONG_SERVICE_TIMEOUT
    )

    # Test playing a non-existent playlist
    soco.clear_queue.reset_mock()
    soco.add_to_queue.reset_mock()
    media_content_id: str = "bad playlist"
    with pytest.raises(
        ServiceValidationError,
        match=f"Could not find Sonos playlist: {media_content_id}",
    ):
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: "playlist",
                ATTR_MEDIA_CONTENT_ID: media_content_id,
            },
            blocking=True,
        )
    assert soco.clear_queue.call_count == 0
    assert soco.add_to_queue.call_count == 0


@pytest.mark.parametrize(
    ("source", "result"),
    [
        (
            SOURCE_LINEIN,
            {
                "switch_to_line_in": 1,
            },
        ),
        (
            SOURCE_TV,
            {
                "switch_to_tv": 1,
            },
        ),
    ],
)
async def test_select_source_line_in_tv(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    source: str,
    result: dict[str, Any],
) -> None:
    """Test the select_source method with a variety of inputs."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_SELECT_SOURCE,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_INPUT_SOURCE: source,
        },
        blocking=True,
    )
    assert soco_mock.switch_to_line_in.call_count == result.get("switch_to_line_in", 0)
    assert soco_mock.switch_to_tv.call_count == result.get("switch_to_tv", 0)


@pytest.mark.parametrize(
    ("source", "result"),
    [
        (
            "James Taylor Radio",
            {
                "play_uri": 1,
                "play_uri_uri": "x-sonosapi-radio:ST%3aetc",
                "play_uri_title": "James Taylor Radio",
                "play_uri_meta": (
                    '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/"'
                    ' xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"'
                    ' xmlns:r="urn:schemas-rinconnetworks-com:metadata-1-0/"'
                    ' xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/">'
                    '<item id="100c2068ST%3a1683194971234567890"'
                    ' parentID="10fe2064myStations" restricted="true">'
                    "<dc:title>James Taylor Radio</dc:title>"
                    "<upnp:class>object.item.audioItem.audioBroadcast"
                    ".#station</upnp:class>"
                    '<desc id="cdudn"'
                    ' nameSpace="urn:schemas-rinconnetworks-com:metadata-1-0/">'
                    "SA_RINCON60423_X_#Svc60423-99999999-Token"
                    "</desc></item></DIDL-Lite>"
                ),
            },
        ),
        (
            "66 - Watercolors",
            {
                "play_uri": 1,
                "play_uri_uri": (
                    "x-sonosapi-hls:Api%3atune%3aliveAudio%3ajazzcafe%3aetc"
                ),
                "play_uri_title": "66 - Watercolors",
                "play_uri_meta": (
                    '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/"'
                    ' xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"'
                    ' xmlns:r="urn:schemas-rinconnetworks-com:metadata-1-0/"'
                    ' xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/">'
                    '<item id="10090120Api%3atune%3aliveAudio%3ajazzcafe'
                    '%3ae4b5402c-9999-9999-9999-4bc8e2cdccce"'
                    ' parentID="10086064live%3f93b0b9cb-9999-9999-9999'
                    '-bcf75971fcfe" restricted="false">'
                    "<dc:title>66 - Watercolors</dc:title>"
                    "<upnp:class>object.item.audioItem.audioBroadcast"
                    "</upnp:class>"
                    '<desc id="cdudn"'
                    ' nameSpace="urn:schemas-rinconnetworks-com:metadata-1-0/">'
                    "SA_RINCON9479_X_#Svc9479-99999999-Token"
                    "</desc></item></DIDL-Lite>"
                ),
            },
        ),
        (
            "American Tall Tales",
            {
                "play_uri": 1,
                "play_uri_uri": (
                    "x-rincon-cpcontainer:101340c8reftitle"
                    "%C9F27_com?sid=239&flags=16584&sn=5"
                ),
                "play_uri_title": "American Tall Tales",
                "play_uri_meta": (
                    '<DIDL-Lite xmlns:dc="http://purl.org/dc/elements/1.1/"'
                    ' xmlns:upnp="urn:schemas-upnp-org:metadata-1-0/upnp/"'
                    ' xmlns:r="urn:schemas-rinconnetworks-com:metadata-1-0/"'
                    ' xmlns="urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/">'
                    '<item id="101340c8reftitleC9F27_com"'
                    ' parentID="101340c8reftitleC9F27_com" restricted="true">'
                    "<dc:title>American Tall Tales</dc:title>"
                    "<upnp:class>object.item.audioItem.audioBook</upnp:class>"
                    '<desc id="cdudn"'
                    ' nameSpace="urn:schemas-rinconnetworks-com:metadata-1-0/">'
                    "SA_RINCON61191_X_#Svc6-0-Token"
                    "</desc></item></DIDL-Lite>"
                ),
            },
        ),
    ],
)
async def test_select_source_play_uri(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    source: str,
    result: dict[str, Any],
) -> None:
    """Test the select_source method with a variety of inputs."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_SELECT_SOURCE,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_INPUT_SOURCE: source,
        },
        blocking=True,
    )
    assert soco_mock.play_uri.call_count == result.get("play_uri")
    soco_mock.play_uri.assert_called_with(
        result.get("play_uri_uri"),
        title=result.get("play_uri_title"),
        meta=result.get("play_uri_meta"),
        timeout=LONG_SERVICE_TIMEOUT,
    )


@pytest.mark.parametrize(
    ("source", "result"),
    [
        (
            "1984",
            {
                "add_to_queue": 1,
                "add_to_queue_item_id": "A:ALBUMARTIST/Aerosmith/1984",
                "clear_queue": 1,
                "play_from_queue": 1,
            },
        ),
    ],
)
async def test_select_source_play_queue(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
    source: str,
    result: dict[str, Any],
) -> None:
    """Test the select_source method with a variety of inputs."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_SELECT_SOURCE,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_INPUT_SOURCE: source,
        },
        blocking=True,
    )
    assert soco_mock.clear_queue.call_count == result.get("clear_queue")
    assert soco_mock.add_to_queue.call_count == result.get("add_to_queue")
    assert soco_mock.add_to_queue.call_args_list[0].args[0].item_id == result.get(
        "add_to_queue_item_id"
    )
    assert (
        soco_mock.add_to_queue.call_args_list[0].kwargs["timeout"]
        == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_from_queue.call_count == result.get("play_from_queue")
    soco_mock.play_from_queue.assert_called_with(0)


async def test_select_source_error(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Test the select_source method with a variety of inputs."""
    with pytest.raises(ServiceValidationError) as sve:
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_SELECT_SOURCE,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_INPUT_SOURCE: "invalid_source",
            },
            blocking=True,
        )
    assert "invalid_source" in str(sve.value)
    assert "Could not find a Sonos favorite" in str(sve.value)


async def test_shuffle_set(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
) -> None:
    """Test the set shuffle method."""
    assert soco.play_mode == "NORMAL"

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_SHUFFLE_SET,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_SHUFFLE: True,
        },
        blocking=True,
    )
    assert soco.play_mode == "SHUFFLE_NOREPEAT"

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_SHUFFLE_SET,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_SHUFFLE: False,
        },
        blocking=True,
    )
    assert soco.play_mode == "NORMAL"


async def test_shuffle_get(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    no_media_event: SonosMockEvent,
) -> None:
    """Test the get shuffle attribute by simulating a Sonos Event."""
    subscription = soco.avTransport.subscribe.return_value
    sub_callback = subscription.callback

    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_SHUFFLE] is False

    no_media_event.variables["current_play_mode"] = "SHUFFLE_NOREPEAT"
    sub_callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_SHUFFLE] is True

    # The integration keeps a copy of the last event to check for
    # changes, so we create a new event.
    no_media_event = SonosMockEvent(
        soco, soco.avTransport, no_media_event.variables.copy()
    )
    no_media_event.variables["current_play_mode"] = "NORMAL"
    sub_callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_SHUFFLE] is False


async def test_repeat_set(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
) -> None:
    """Test the set repeat method."""
    assert soco.play_mode == "NORMAL"
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_REPEAT_SET,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_REPEAT: RepeatMode.ALL,
        },
        blocking=True,
    )
    assert soco.play_mode == "REPEAT_ALL"

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_REPEAT_SET,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_REPEAT: RepeatMode.ONE,
        },
        blocking=True,
    )
    assert soco.play_mode == "REPEAT_ONE"

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_REPEAT_SET,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_REPEAT: RepeatMode.OFF,
        },
        blocking=True,
    )
    assert soco.play_mode == "NORMAL"


async def test_repeat_get(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    no_media_event: SonosMockEvent,
) -> None:
    """Test the get repeat attribute by simulating a Sonos Event."""
    subscription = soco.avTransport.subscribe.return_value
    sub_callback = subscription.callback

    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.OFF

    no_media_event.variables["current_play_mode"] = "REPEAT_ALL"
    sub_callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.ALL

    no_media_event = SonosMockEvent(
        soco, soco.avTransport, no_media_event.variables.copy()
    )
    no_media_event.variables["current_play_mode"] = "REPEAT_ONE"
    sub_callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.ONE

    no_media_event = SonosMockEvent(
        soco, soco.avTransport, no_media_event.variables.copy()
    )
    no_media_event.variables["current_play_mode"] = "NORMAL"
    sub_callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get("media_player.zone_a")
    assert state.attributes[ATTR_MEDIA_REPEAT] == RepeatMode.OFF


async def test_play_media_favorite_item_id(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
    async_autosetup_sonos,
) -> None:
    """Test playing media with a favorite item id."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "favorite_item_id",
            ATTR_MEDIA_CONTENT_ID: "FV:2/4",
        },
        blocking=True,
    )
    assert soco_mock.play_uri.call_count == 1
    assert (
        soco_mock.play_uri.call_args_list[0].args[0]
        == "x-sonosapi-hls:Api%3atune%3aliveAudio%3ajazzcafe%3aetc"
    )
    assert (
        soco_mock.play_uri.call_args_list[0].kwargs["timeout"] == LONG_SERVICE_TIMEOUT
    )
    assert soco_mock.play_uri.call_args_list[0].kwargs["title"] == "66 - Watercolors"

    # Test exception handling with an invalid id.
    with pytest.raises(ValueError) as sve:
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: "favorite_item_id",
                ATTR_MEDIA_CONTENT_ID: "UNKNOWN_ID",
            },
            blocking=True,
        )
    assert "UNKNOWN_ID" in str(sve.value)


async def _setup_hass(hass: HomeAssistant):
    await async_setup_component(
        hass,
        DOMAIN,
        {
            "sonos": {
                "media_player": {
                    "interface_addr": "127.0.0.1",
                    "hosts": ["10.10.10.1", "10.10.10.2"],
                }
            }
        },
    )
    await hass.async_block_till_done()


async def test_service_snapshot_restore(
    hass: HomeAssistant,
    soco_factory: SoCoMockFactory,
) -> None:
    """Test the snapshot and restore services."""
    soco_factory.cache_mock(MockSoCo(), "10.10.10.1", "Living Room")
    soco_factory.cache_mock(MockSoCo(), "10.10.10.2", "Bedroom")
    await _setup_hass(hass)
    with patch(
        "homeassistant.components.sonos.speaker.Snapshot.snapshot"
    ) as mock_snapshot:
        await hass.services.async_call(
            DOMAIN,
            SERVICE_SNAPSHOT,
            {
                ATTR_ENTITY_ID: ["media_player.living_room", "media_player.bedroom"],
            },
            blocking=True,
        )
    assert mock_snapshot.call_count == 2

    with patch(
        "homeassistant.components.sonos.speaker.Snapshot.restore"
    ) as mock_restore:
        await hass.services.async_call(
            DOMAIN,
            SERVICE_RESTORE,
            {
                ATTR_ENTITY_ID: ["media_player.living_room", "media_player.bedroom"],
            },
            blocking=True,
        )
    assert mock_restore.call_count == 2


async def test_volume(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
) -> None:
    """Test the media player volume services."""
    initial_volume = soco.volume

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_VOLUME_UP,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
        },
        blocking=True,
    )
    assert soco.volume == initial_volume + VOLUME_INCREMENT

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_VOLUME_DOWN,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
        },
        blocking=True,
    )
    assert soco.volume == initial_volume

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_VOLUME_SET,
        {ATTR_ENTITY_ID: "media_player.zone_a", ATTR_MEDIA_VOLUME_LEVEL: 0.57},
        blocking=True,
    )
    # SoCo uses 0..100 for its range.
    assert soco.volume == 57


@pytest.mark.parametrize(
    ("service", "client_call"),
    [
        (SERVICE_MEDIA_PLAY, "play"),
        (SERVICE_MEDIA_PAUSE, "pause"),
        (SERVICE_MEDIA_STOP, "stop"),
        (SERVICE_MEDIA_NEXT_TRACK, "next"),
        (SERVICE_MEDIA_PREVIOUS_TRACK, "previous"),
        (SERVICE_CLEAR_PLAYLIST, "clear_queue"),
    ],
)
async def test_media_transport(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    service: str,
    client_call: str,
) -> None:
    """Test the media player transport services."""
    await hass.services.async_call(
        MP_DOMAIN,
        service,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
        },
        blocking=True,
    )
    assert getattr(soco, client_call).call_count == 1


@pytest.mark.parametrize(
    ("transport_state", "title", "expected_state"),
    [
        ("PAUSED_PLAYBACK", "Something", "paused"),
        ("PAUSED_PLAYBACK", None, "idle"),
        ("PAUSED_PLAYBACK", " ", "idle"),
        ("STOPPED", "Something", "paused"),
        ("STOPPED", None, "idle"),
        ("STOPPED", " ", "idle"),
    ],
)
async def test_state_paused_idle(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    no_media_event: SonosMockEvent,
    transport_state: str,
    title: str | None,
    expected_state: str,
) -> None:
    """Test that idle is returned when title is None or whitespace, paused otherwise."""
    soco.get_current_track_info.return_value = {
        "title": title,
        "artist": "",
        "album": "",
        "album_art": "",
        "position": "NOT_IMPLEMENTED",
        "playlist_position": "1",
        "duration": "NOT_IMPLEMENTED",
        "uri": "x-file-cifs://192.168.42.10/music/track.mp3",
        "metadata": "NOT_IMPLEMENTED",
    }
    no_media_event.variables["transport_state"] = transport_state
    soco.avTransport.subscribe.return_value.callback(no_media_event)
    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("media_player.zone_a")
    assert state.state == expected_state


async def test_play_media_announce(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    sonos_websocket,
) -> None:
    """Test playing media with the announce."""
    content_id: str = "http://10.0.0.1:8123/local/sounds/doorbell.mp3"
    volume: float = 0.30

    # Test the success path
    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "music",
            ATTR_MEDIA_CONTENT_ID: content_id,
            ATTR_MEDIA_ANNOUNCE: True,
            ATTR_MEDIA_EXTRA: {"volume": volume},
        },
        blocking=True,
    )
    assert sonos_websocket.play_clip.call_count == 1
    sonos_websocket.play_clip.assert_called_with(content_id, volume=volume)

    # Test receiving a websocket exception
    sonos_websocket.play_clip.reset_mock()
    sonos_websocket.play_clip.side_effect = SonosWebsocketError("Error Message")
    with pytest.raises(
        HomeAssistantError, match="Error when calling Sonos websocket: Error Message"
    ):
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: "music",
                ATTR_MEDIA_CONTENT_ID: content_id,
                ATTR_MEDIA_ANNOUNCE: True,
            },
            blocking=True,
        )
    assert sonos_websocket.play_clip.call_count == 1
    sonos_websocket.play_clip.assert_called_with(content_id, volume=None)

    # Test receiving a non success result
    sonos_websocket.play_clip.reset_mock()
    sonos_websocket.play_clip.side_effect = None
    retval = {"success": 0}
    sonos_websocket.play_clip.return_value = [retval, {}]
    with pytest.raises(
        HomeAssistantError, match=f"Announcing clip {content_id} failed {retval}"
    ):
        await hass.services.async_call(
            MP_DOMAIN,
            SERVICE_PLAY_MEDIA,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_MEDIA_CONTENT_TYPE: "music",
                ATTR_MEDIA_CONTENT_ID: content_id,
                ATTR_MEDIA_ANNOUNCE: True,
            },
            blocking=True,
        )
    assert sonos_websocket.play_clip.call_count == 1

    # Test speakers that do not support announce. This
    # will result in playing the clip directly via play_uri
    sonos_websocket.play_clip.reset_mock()
    sonos_websocket.play_clip.side_effect = None
    retval = {"success": 0, "type": "globalError"}
    sonos_websocket.play_clip.return_value = [retval, {}]

    await hass.services.async_call(
        MP_DOMAIN,
        SERVICE_PLAY_MEDIA,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_MEDIA_CONTENT_TYPE: "music",
            ATTR_MEDIA_CONTENT_ID: content_id,
            ATTR_MEDIA_ANNOUNCE: True,
        },
        blocking=True,
    )
    assert sonos_websocket.play_clip.call_count == 1
    soco.play_uri.assert_called_with(content_id, force_radio=False)


async def test_media_get_queue(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    soco_factory,
    snapshot: SnapshotAssertion,
) -> None:
    """Test getting the media queue."""
    soco_mock = soco_factory.mock_list.get("192.168.42.2")
    result = await hass.services.async_call(
        DOMAIN,
        SERVICE_GET_QUEUE,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
        },
        blocking=True,
        return_response=True,
    )
    soco_mock.get_queue.assert_called_with(max_items=0)
    assert result == snapshot


@pytest.mark.parametrize(
    ("speaker_model", "source_list"),
    [
        ("Sonos Arc Ultra", [SOURCE_TV]),
        ("Sonos Arc", [SOURCE_TV]),
        ("Sonos Playbar", [SOURCE_TV]),
        ("Sonos Connect", [SOURCE_LINEIN]),
        ("Sonos Play:5", [SOURCE_LINEIN]),
        ("Sonos Amp", [SOURCE_LINEIN, SOURCE_TV]),
        ("Sonos Era", None),
    ],
    indirect=["speaker_model"],
)
async def test_media_source_list(
    hass: HomeAssistant,
    async_autosetup_sonos,
    speaker_model: str,
    source_list: list[str] | None,
) -> None:
    """Test the mapping between the speaker model name and source_list."""
    state = hass.states.get("media_player.zone_a")
    assert state.attributes.get(ATTR_INPUT_SOURCE_LIST) == source_list


async def test_service_update_alarm(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
) -> None:
    """Test updating an alarm."""

    await hass.services.async_call(
        DOMAIN,
        SERVICE_UPDATE_ALARM,
        {
            ATTR_ENTITY_ID: "media_player.zone_a",
            ATTR_ALARM_ID: 14,
            ATTR_TIME: "07:15:00",
            ATTR_VOLUME: 0.25,
            ATTR_INCLUDE_LINKED_ZONES: True,
            ATTR_ENABLED: True,
        },
        blocking=True,
    )

    assert soco.alarmClock.UpdateAlarm.call_count == 1
    assert soco.alarmClock.UpdateAlarm.call_args.args[0] == [
        ("ID", "14"),
        ("StartLocalTime", "07:15:00"),
        ("Duration", "02:00:00"),
        ("Recurrence", "DAILY"),
        ("Enabled", "1"),
        ("RoomUUID", "RINCON_test"),
        ("ProgramURI", "x-rincon-buzzer:0"),
        ("ProgramMetaData", ""),
        ("PlayMode", "SHUFFLE_NOREPEAT"),
        ("Volume", 25),
        ("IncludeLinkedZones", "1"),
    ]


async def test_service_update_alarm_dne(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
) -> None:
    """Test updating an alarm that does not exist."""

    with pytest.raises(
        ServiceValidationError,
        match="Alarm 99 does not exist and cannot be updated",
    ):
        await hass.services.async_call(
            DOMAIN,
            SERVICE_UPDATE_ALARM,
            {
                ATTR_ENTITY_ID: "media_player.zone_a",
                ATTR_ALARM_ID: 99,
                ATTR_TIME: "07:15:00",
                ATTR_VOLUME: 0.25,
                ATTR_INCLUDE_LINKED_ZONES: True,
                ATTR_ENABLED: True,
            },
            blocking=True,
        )
    assert soco.alarmClock.UpdateAlarm.call_count == 0


@pytest.mark.freeze_time("2024-01-01T12:00:00Z")
async def test_position_updates(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    media_event: SonosMockEvent,
    current_track_info: dict[str, Any],
) -> None:
    """Test the media player position updates."""

    soco.get_current_track_info.return_value = current_track_info
    soco.avTransport.subscribe.return_value.callback(media_event)
    await hass.async_block_till_done(wait_background_tasks=True)

    entity_id = "media_player.zone_a"
    state = hass.states.get(entity_id)

    assert state.attributes[ATTR_MEDIA_POSITION] == 42
    # updated_at should be recent
    updated_at = state.attributes[ATTR_MEDIA_POSITION_UPDATED_AT]
    assert updated_at == datetime.now(UTC)

    # Position only updated by 1 second; should not update attributes
    new_track_info = current_track_info.copy()
    new_track_info["position"] = "00:00:43"
    soco.get_current_track_info.return_value = new_track_info
    new_media_event = SonosMockEvent(
        soco, soco.avTransport, media_event.variables.copy()
    )
    new_media_event.variables["position"] = "00:00:43"
    with freeze_time("2024-01-01T12:00:01Z"):
        soco.avTransport.subscribe.return_value.callback(new_media_event)
        await hass.async_block_till_done(wait_background_tasks=True)
    state = hass.states.get(entity_id)
    assert state.attributes[ATTR_MEDIA_POSITION] == 42
    assert state.attributes[ATTR_MEDIA_POSITION_UPDATED_AT] == updated_at

    # Position jumped by more than 1.5 seconds; should update position
    new_track_info = current_track_info.copy()
    new_track_info["position"] = "00:01:10"
    soco.get_current_track_info.return_value = new_track_info
    new_media_event = SonosMockEvent(
        soco, soco.avTransport, media_event.variables.copy()
    )
    new_media_event.variables["position"] = "00:01:10"
    with freeze_time("2024-01-01T12:00:11Z"):
        soco.avTransport.subscribe.return_value.callback(new_media_event)
        await hass.async_block_till_done(wait_background_tasks=True)
        state = hass.states.get(entity_id)
        assert state.attributes[ATTR_MEDIA_POSITION] == 70
        assert state.attributes[ATTR_MEDIA_POSITION_UPDATED_AT] == datetime.now(UTC)


@pytest.mark.parametrize(
    ("track_info", "event_variables"),
    [
        (
            {
                "title": "Something",
                "artist": "The Beatles",
                "album": "Abbey Road",
                "album_art": "http://example.com/albumart.jpg",
                "position": "00:00:42",
                "playlist_position": "5",
                "duration": "00:02:36",
                "uri": "x-file-cifs://192.168.42.10/music/The%20Beatles/Abbey%20Road/03%20Something.mp3",
                "metadata": "NOT_IMPLEMENTED",
            },
            {},
        ),
        (
            {
                "title": "Something",
                "artist": "The Beatles",
                "album": "Abbey Road",
                "position": "00:00:42",
                "playlist_position": "5",
                "duration": "00:02:36",
                "uri": "x-file-cifs://192.168.42.10/music/The%20Beatles/Abbey%20Road/03%20Something.mp3",
                "metadata": "NOT_IMPLEMENTED",
            },
            {},
        ),
        (
            {
                "title": "Something",
                "artist": "The Beatles",
                "album": "Abbey Road",
                "album_art": "http://example.com/albumart.jpg",
                "playlist_position": "5",
                "uri": "x-file-cifs://192.168.42.10/music/The%20Beatles/Abbey%20Road/03%20Something.mp3",
                "metadata": "NOT_IMPLEMENTED",
            },
            {},
        ),
        (
            {
                "uri": "x-rincon-stream:0",
                "metadata": "NOT_IMPLEMENTED",
            },
            {
                "current_track_uri": "x-rincon-stream:0",
                "current_track_meta_data": DidlAudioLineIn("Line-in", "-1", "-1"),
            },
        ),
        (
            {
                "title": "Something",
                "artist": "The Beatles",
                "album": "Abbey Road",
                "album_art": "http://example.com/albumart.jpg",
                "playlist_position": "5",
                "uri": "x-file-cifs://192.168.42.10/music/The%20Beatles/Abbey%20Road/03%20Something.mp3",
                "metadata": "NOT_IMPLEMENTED",
            },
            {
                "enqueued_transport_uri_meta_data": DidlPlaylistContainer(
                    "My Playlist", "-1", "-1"
                )
            },
        ),
        (
            {
                "album_art": "http://example.com/albumart.jpg",
                "position": "00:00:42",
                "duration": "00:02:36",
                "uri": "x-sonosapi-stream:1234",
                "metadata": "NOT_IMPLEMENTED",
            },
            {
                "enqueued_transport_uri_meta_data": DidlAudioBroadcast(
                    "World News", "-1", "-1"
                ),
                "current_track_uri": "x-sonosapi-stream:1234",
            },
        ),
        (
            {
                "album_art": "http://example.com/albumart.jpg",
                "position": "00:00:42",
                "duration": "00:02:36",
                "uri": "x-sonosapi-stream:1234",
                "metadata": "NOT_IMPLEMENTED",
            },
            {
                "enqueued_transport_uri_meta_data": DidlAudioBroadcast(
                    "World News", "-1", "-1"
                ),
                "current_track_uri": "x-sonosapi-stream:1234",
                "current_track_meta_data": DidlAudioBroadcast(
                    "World News", "-1", "-1", radio_show="Live at 6"
                ),
            },
        ),
    ],
    ids=[
        "basic_track",
        "basic_track_no_art",
        "basic_track_no_position",
        "line_in",
        "playlist_container",
        "radio_station",
        "radio_station_with_show",
    ],
)
async def test_media_info_attributes(
    hass: HomeAssistant,
    soco: MockSoCo,
    async_autosetup_sonos,
    media_event: SonosMockEvent,
    track_info: dict[str, Any],
    event_variables: dict[str, Any],
    snapshot: SnapshotAssertion,
) -> None:
    """Test the media player info attributes using a variety of inputs."""
    media_event.variables.update(event_variables)
    soco.get_current_track_info.return_value = track_info
    soco.avTransport.subscribe.return_value.callback(media_event)

    await hass.async_block_till_done(wait_background_tasks=True)

    state = hass.states.get("media_player.zone_a")

    snapshot_keys = [
        ATTR_MEDIA_ALBUM_NAME,
        ATTR_MEDIA_ARTIST,
        ATTR_MEDIA_CONTENT_ID,
        ATTR_MEDIA_CONTENT_TYPE,
        ATTR_MEDIA_DURATION,
        ATTR_MEDIA_POSITION,
        ATTR_MEDIA_TITLE,
        ATTR_QUEUE_POSITION,
        ATTR_ENTITY_PICTURE,
        ATTR_INPUT_SOURCE,
        ATTR_MEDIA_PLAYLIST,
        ATTR_MEDIA_CHANNEL,
    ]

    # Create a filtered dict of only those attributes
    filtered_attrs = {k: state.attributes.get(k) for k in snapshot_keys}

    # Use the snapshot assertion
    assert filtered_attrs == snapshot
