Add spotify integration

tests/fake_spotify.py

@@ -0,0 +1,100 @@
+"""Stand-in Spotify Web API: just enough of /api/token, .../player/devices
+and .../player to test spotify.py against, the same way fakes.py stands in
+for real HEOS hardware."""
+
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+
+class FakeSpotify(threading.Thread):
+    """Issues a fresh access token per refresh, tracks which one is
+    currently valid, and remembers every playback transfer it was asked
+    to make."""
+
+    def __init__(self, devices=None):
+        super().__init__(daemon=True)
+        self.devices_list = devices if devices is not None else [
+            {"id": "dev-1", "name": "Lego Room", "type": "Speaker"},
+            {"id": "dev-2", "name": "Home Cinema", "type": "AVR"},
+        ]
+        self.valid_token = None
+        self.tokens_issued = 0
+        self.transfers = []            # every PUT /v1/me/player body
+        self.reject_refresh = False    # simulate a revoked refresh token
+        self.player = None             # GET /v1/me/player's body; None is no session (204)
+
+        fake = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def _send(self, status, payload=None):
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                if payload is not None:
+                    self.wfile.write(json.dumps(payload).encode())
+
+            def _authorized(self):
+                header = self.headers.get("Authorization", "")
+                return header == f"Bearer {fake.valid_token}" and fake.valid_token is not None
+
+            def do_POST(self):
+                if self.path == "/api/token":
+                    if fake.reject_refresh:
+                        self._send(400, {"error": "invalid_grant",
+                                          "error_description": "refresh token revoked"})
+                        return
+                    fake.tokens_issued += 1
+                    fake.valid_token = f"token-{fake.tokens_issued}"
+                    self._send(200, {"access_token": fake.valid_token, "expires_in": 3600})
+                    return
+                self._send(404, {"error": {"message": "not found"}})
+
+            def do_PUT(self):
+                if self.path == "/v1/me/player":
+                    if not self._authorized():
+                        self._send(401, {"error": {"message": "The access token expired"}})
+                        return
+                    length = int(self.headers.get("Content-Length", 0))
+                    fake.transfers.append(json.loads(self.rfile.read(length) or b"{}"))
+                    self.send_response(204)
+                    self.end_headers()
+                    return
+                self._send(404, {"error": {"message": "not found"}})
+
+            def do_GET(self):
+                if self.path == "/v1/me/player/devices":
+                    if not self._authorized():
+                        self._send(401, {"error": {"message": "The access token expired"}})
+                        return
+                    self._send(200, {"devices": fake.devices_list})
+                    return
+                if self.path == "/v1/me/player":
+                    if not self._authorized():
+                        self._send(401, {"error": {"message": "The access token expired"}})
+                        return
+                    if fake.player is None:
+                        self.send_response(204)
+                        self.end_headers()
+                    else:
+                        self._send(200, fake.player)
+                    return
+                self._send(404, {"error": {"message": "not found"}})
+
+        self.server = HTTPServer(("127.0.0.1", 0), Handler)
+        self.port = self.server.server_port
+        self.start()
+
+    def run(self):
+        self.server.serve_forever(poll_interval=0.05)
+
+    def stop(self):
+        self.server.shutdown()
+        self.server.server_close()
+
+    @property
+    def base_url(self):
+        return f"http://127.0.0.1:{self.port}"

tests/fakes.py

@@ -31,6 +31,7 @@ class FakeHeos(threading.Thread):
         self.play_states = {pid: "play" for pid in self.NAMES}
         self.group_volumes = {3: 25}
         self.now_playing_mid = {1: "inputs/mediaplayer"}  # pid -> what get_now_playing_media reports
+        self.now_playing_sid = {}                     # pid -> its source id; 4 is Spotify
         self.commands = []                            # everything we were asked to do
         self.server = socket.socket()
         self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -120,7 +121,10 @@ class FakeHeos(threading.Thread):
             pid = int(args["pid"])
             mid = self.now_playing_mid.get(pid, "")
             name = next((s["name"] for s in self.AVR_INPUTS if s["mid"] == mid), mid)
-            return self._ok(path, payload={"mid": mid, "station": name} if mid else {})
+            payload = {"mid": mid, "station": name} if mid else {}
+            if pid in self.now_playing_sid:
+                payload["sid"] = self.now_playing_sid[pid]
+            return self._ok(path, payload=payload)
 
         if path == "browse/browse":
             sid = int(args["sid"])

tests/test_panel.py

@@ -174,6 +174,18 @@ class PanelTest(unittest.TestCase):
         self.panel.toggle_play("living_room_group", "pause")
         self.assertEqual(self.heos.play_states[3], "pause")
 
+    def test_state_says_which_rooms_are_playing_spotify(self):
+        """The cards only offer play/pause for a Spotify stream, so the
+        snapshot has to say which rooms have one."""
+        self.heos.now_playing_mid[HOME400_PID] = "spotify:track:4uLU6hMCjMI75M1A2tKUQC"
+        self.heos.now_playing_sid[HOME400_PID] = 4
+        state = self.panel.state()
+        self.assertEqual([r["spotify"] for r in state["rooms"]], [True, False])
+
+    def test_an_avr_input_is_not_spotify(self):
+        self.heos.now_playing_sid[AVR_PID] = 1027
+        self.assertFalse(self.panel.on_spotify("avr"))
+
     # -- the AVR, entirely over HEOS --------------------------------------
     def test_avr_inputs_carry_your_renamed_labels(self):
         """HEOS reports the AVR's own renamed sources itself (browse/browse

tests/test_spotify.py

@@ -0,0 +1,66 @@
+"""Run against the fake Spotify Web API in fake_spotify.py:
+
+    python3 -m unittest discover -s tests -t .
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from spotify import SpotifyClient, SpotifyError   # noqa: E402
+from tests.fake_spotify import FakeSpotify         # noqa: E402
+
+
+class SpotifyTest(unittest.TestCase):
+    def setUp(self):
+        self.fake = FakeSpotify()
+        self.addCleanup(self.fake.stop)
+        self.client = SpotifyClient(
+            "client-id", "client-secret", "refresh-token",
+            accounts_url=self.fake.base_url, api_url=self.fake.base_url,
+        )
+
+    def test_devices_lists_what_spotify_reports(self):
+        names = [d["name"] for d in self.client.devices()]
+        self.assertEqual(names, ["Lego Room", "Home Cinema"])
+        self.assertEqual(self.fake.tokens_issued, 1)   # one refresh for the whole call
+
+    def test_resume_transfers_playback_to_the_matched_device(self):
+        device = self.client.resume("Lego Room")
+        self.assertEqual(device["id"], "dev-1")
+        self.assertEqual(self.fake.transfers, [{"device_ids": ["dev-1"], "play": True}])
+
+    def test_resume_raises_when_no_device_has_that_name(self):
+        with self.assertRaises(SpotifyError):
+            self.client.resume("Kitchen")
+        self.assertEqual(self.fake.transfers, [])
+
+    def test_playback_reports_the_device_and_whether_it_plays(self):
+        self.fake.player = {"device": {"id": "dev-1", "name": "Lego Room"}, "is_playing": True}
+        player = self.client.playback()
+        self.assertEqual(player["device"]["name"], "Lego Room")
+        self.assertTrue(player["is_playing"])
+
+    def test_playback_is_empty_without_a_session(self):
+        self.assertEqual(self.client.playback(), {})
+
+    def test_a_rejected_access_token_is_refreshed_and_retried_once(self):
+        self.client.devices()                       # get a real token first
+        self.assertEqual(self.fake.tokens_issued, 1)
+        self.client._access_token = "stale-but-not-yet-expired"
+        # _expires_at is untouched, so only the 401 -- not the pre-call
+        # expiry check -- can be what forces this to work.
+        names = [d["name"] for d in self.client.devices()]
+        self.assertEqual(names, ["Lego Room", "Home Cinema"])
+        self.assertEqual(self.fake.tokens_issued, 2)
+
+    def test_a_revoked_refresh_token_raises_a_clear_error(self):
+        self.fake.reject_refresh = True
+        with self.assertRaises(SpotifyError):
+            self.client.devices()
+
+
+if __name__ == "__main__":
+    unittest.main()