118 lines
5.0 KiB
Python
118 lines
5.0 KiB
Python
"""Stand-in Spotify Web API: just enough of /api/token, .../player/devices,
|
|
.../player and .../player/seek 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
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
|
|
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.seeks = [] # every PUT /v1/me/player/seek's position_ms
|
|
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
|
|
if self.path.startswith("/v1/me/player/seek?"):
|
|
if not self._authorized():
|
|
self._send(401, {"error": {"message": "The access token expired"}})
|
|
return
|
|
query = parse_qs(urlparse(self.path).query)
|
|
fake.seeks.append(int(query["position_ms"][0]))
|
|
# The real one answers some player commands with a
|
|
# non-JSON body rather than an empty 204.
|
|
body = b"a1b2c3d4"
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/plain")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
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}"
|