hermes-agent/tests/tools/test_spotify_client.py

from __future__ import annotations

import json

import pytest

from plugins.spotify import client as spotify_mod
from plugins.spotify import tools as spotify_tool


class _FakeResponse:
    def __init__(self, status_code: int, payload: dict | None = None, *, text: str = "", headers: dict | None = None):
        self.status_code = status_code
        self._payload = payload
        self.text = text or (json.dumps(payload) if payload is not None else "")
        self.headers = headers or {"content-type": "application/json"}
        self.content = self.text.encode("utf-8") if self.text else b""

    def json(self):
        if self._payload is None:
            raise ValueError("no json")
        return self._payload


class _StubSpotifyClient:
    def __init__(self, payload):
        self.payload = payload

    def get_currently_playing(self, *, market=None):
        return self.payload


def test_spotify_client_retries_once_after_401(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[str] = []
    tokens = iter([
        {
            "access_token": "token-1",
            "base_url": "https://api.spotify.com/v1",
        },
        {
            "access_token": "token-2",
            "base_url": "https://api.spotify.com/v1",
        },
    ])

    monkeypatch.setattr(
        spotify_mod,
        "resolve_spotify_runtime_credentials",
        lambda **kwargs: next(tokens),
    )

    def fake_request(method, url, headers=None, params=None, json=None, timeout=None):
        calls.append(headers["Authorization"])
        if len(calls) == 1:
            return _FakeResponse(401, {"error": {"message": "expired token"}})
        return _FakeResponse(200, {"devices": [{"id": "dev-1"}]})

    monkeypatch.setattr(spotify_mod.httpx, "request", fake_request)

    client = spotify_mod.SpotifyClient()
    payload = client.get_devices()

    assert payload["devices"][0]["id"] == "dev-1"
    assert calls == ["Bearer token-1", "Bearer token-2"]


def test_normalize_spotify_uri_accepts_urls() -> None:
    uri = spotify_mod.normalize_spotify_uri(
        "https://open.spotify.com/track/7ouMYWpwJ422jRcDASZB7P",
        "track",
    )
    assert uri == "spotify:track:7ouMYWpwJ422jRcDASZB7P"


@pytest.mark.parametrize(
    ("status_code", "path", "payload", "expected"),
    [
        (
            403,
            "/me/player/play",
            {"error": {"message": "Premium required"}},
            "Spotify rejected this playback request. Playback control usually requires a Spotify Premium account and an active Spotify Connect device.",
        ),
        (
            404,
            "/me/player",
            {"error": {"message": "Device not found"}},
            "Spotify could not find an active playback device or player session for this request.",
        ),
        (
            429,
            "/search",
            {"error": {"message": "rate limit"}},
            "Spotify rate limit exceeded. Retry after 7 seconds.",
        ),
    ],
)
def test_spotify_client_formats_friendly_api_errors(
    monkeypatch: pytest.MonkeyPatch,
    status_code: int,
    path: str,
    payload: dict,
    expected: str,
) -> None:
    monkeypatch.setattr(
        spotify_mod,
        "resolve_spotify_runtime_credentials",
        lambda **kwargs: {
            "access_token": "token-1",
            "base_url": "https://api.spotify.com/v1",
        },
    )

    def fake_request(method, url, headers=None, params=None, json=None, timeout=None):
        return _FakeResponse(status_code, payload, headers={"content-type": "application/json", "Retry-After": "7"})

    monkeypatch.setattr(spotify_mod.httpx, "request", fake_request)

    client = spotify_mod.SpotifyClient()
    with pytest.raises(spotify_mod.SpotifyAPIError) as exc:
        client.request("GET", path)

    assert str(exc.value) == expected


def test_get_currently_playing_returns_explanatory_empty_payload(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        spotify_mod,
        "resolve_spotify_runtime_credentials",
        lambda **kwargs: {
            "access_token": "token-1",
            "base_url": "https://api.spotify.com/v1",
        },
    )

    def fake_request(method, url, headers=None, params=None, json=None, timeout=None):
        return _FakeResponse(204, None, text="", headers={"content-type": "application/json"})

    monkeypatch.setattr(spotify_mod.httpx, "request", fake_request)

    client = spotify_mod.SpotifyClient()
    payload = client.get_currently_playing()

    assert payload == {
        "status_code": 204,
        "empty": True,
        "message": "Spotify is not currently playing anything. Start playback in Spotify and try again.",
    }


def test_spotify_playback_get_currently_playing_returns_explanatory_empty_result(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        spotify_tool,
        "_spotify_client",
        lambda: _StubSpotifyClient({
            "status_code": 204,
            "empty": True,
            "message": "Spotify is not currently playing anything. Start playback in Spotify and try again.",
        }),
    )

    payload = json.loads(spotify_tool._handle_spotify_playback({"action": "get_currently_playing"}))

    assert payload == {
        "success": True,
        "action": "get_currently_playing",
        "is_playing": False,
        "status_code": 204,
        "message": "Spotify is not currently playing anything. Start playback in Spotify and try again.",
    }


def test_library_contains_uses_generic_library_endpoint(monkeypatch: pytest.MonkeyPatch) -> None:
    seen: list[tuple[str, str, dict | None]] = []

    monkeypatch.setattr(
        spotify_mod,
        "resolve_spotify_runtime_credentials",
        lambda **kwargs: {
            "access_token": "token-1",
            "base_url": "https://api.spotify.com/v1",
        },
    )

    def fake_request(method, url, headers=None, params=None, json=None, timeout=None):
        seen.append((method, url, params))
        return _FakeResponse(200, [True])

    monkeypatch.setattr(spotify_mod.httpx, "request", fake_request)

    client = spotify_mod.SpotifyClient()
    payload = client.library_contains(uris=["spotify:album:abc", "spotify:track:def"])

    assert payload == [True]
    assert seen == [
        (
            "GET",
            "https://api.spotify.com/v1/me/library/contains",
            {"uris": "spotify:album:abc,spotify:track:def"},
        )
    ]


@pytest.mark.parametrize(
    ("method_name", "item_key", "item_value", "expected_uris"),
    [
        ("remove_saved_tracks", "track_ids", ["track-a", "track-b"], ["spotify:track:track-a", "spotify:track:track-b"]),
        ("remove_saved_albums", "album_ids", ["album-a"], ["spotify:album:album-a"]),
    ],
)
def test_library_remove_uses_generic_library_endpoint(
    monkeypatch: pytest.MonkeyPatch,
    method_name: str,
    item_key: str,
    item_value: list[str],
    expected_uris: list[str],
) -> None:
    seen: list[tuple[str, str, dict | None]] = []

    monkeypatch.setattr(
        spotify_mod,
        "resolve_spotify_runtime_credentials",
        lambda **kwargs: {
            "access_token": "token-1",
            "base_url": "https://api.spotify.com/v1",
        },
    )

    def fake_request(method, url, headers=None, params=None, json=None, timeout=None):
        seen.append((method, url, params))
        return _FakeResponse(200, {})

    monkeypatch.setattr(spotify_mod.httpx, "request", fake_request)

    client = spotify_mod.SpotifyClient()
    getattr(client, method_name)(**{item_key: item_value})

    assert seen == [
        (
            "DELETE",
            "https://api.spotify.com/v1/me/library",
            {"uris": ",".join(expected_uris)},
        )
    ]



def test_spotify_library_tracks_list_routes_to_saved_tracks(monkeypatch: pytest.MonkeyPatch) -> None:
    seen: list[str] = []

    class _LibStub:
        def get_saved_tracks(self, **kw):
            seen.append("tracks")
            return {"items": [], "total": 0}

        def get_saved_albums(self, **kw):
            seen.append("albums")
            return {"items": [], "total": 0}

    monkeypatch.setattr(spotify_tool, "_spotify_client", lambda: _LibStub())
    json.loads(spotify_tool._handle_spotify_library({"kind": "tracks", "action": "list"}))
    assert seen == ["tracks"]


def test_spotify_library_albums_list_routes_to_saved_albums(monkeypatch: pytest.MonkeyPatch) -> None:
    seen: list[str] = []

    class _LibStub:
        def get_saved_tracks(self, **kw):
            seen.append("tracks")
            return {"items": [], "total": 0}

        def get_saved_albums(self, **kw):
            seen.append("albums")
            return {"items": [], "total": 0}

    monkeypatch.setattr(spotify_tool, "_spotify_client", lambda: _LibStub())
    json.loads(spotify_tool._handle_spotify_library({"kind": "albums", "action": "list"}))
    assert seen == ["albums"]


def test_spotify_library_rejects_missing_kind() -> None:
    payload = json.loads(spotify_tool._handle_spotify_library({"action": "list"}))
    assert "kind" in (payload.get("error") or "").lower()


def test_spotify_playback_recently_played_action(monkeypatch: pytest.MonkeyPatch) -> None:
    """recently_played is now an action on spotify_playback (folded from spotify_activity)."""
    seen: list[dict] = []

    class _RecentStub:
        def get_recently_played(self, **kw):
            seen.append(kw)
            return {"items": [{"track": {"name": "x"}}]}

    monkeypatch.setattr(spotify_tool, "_spotify_client", lambda: _RecentStub())
    payload = json.loads(spotify_tool._handle_spotify_playback({"action": "recently_played", "limit": 5}))
    assert seen and seen[0]["limit"] == 5
    assert isinstance(payload, dict)