This commit is contained in:
+19
-2
@@ -1,10 +1,11 @@
|
||||
"""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
|
||||
"""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):
|
||||
@@ -21,6 +22,7 @@ class FakeSpotify(threading.Thread):
|
||||
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)
|
||||
|
||||
@@ -63,6 +65,21 @@ class FakeSpotify(threading.Thread):
|
||||
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):
|
||||
|
||||
@@ -37,6 +37,10 @@ class SpotifyTest(unittest.TestCase):
|
||||
self.client.resume("Kitchen")
|
||||
self.assertEqual(self.fake.transfers, [])
|
||||
|
||||
def test_seek_asks_for_the_position(self):
|
||||
self.client.seek(90000)
|
||||
self.assertEqual(self.fake.seeks, [90000])
|
||||
|
||||
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()
|
||||
|
||||
+17
-1
@@ -14,7 +14,7 @@ from urllib.parse import parse_qs, urlparse
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from zidoo import ZidooClient # noqa: E402
|
||||
from zidoo import ZidooClient, ZidooError # noqa: E402
|
||||
|
||||
PATH = "/storage/disk/Movies/A.Private.War.2018.1080p.mp4"
|
||||
PNG = b"\x89PNG\r\n\x1a\n" + b"\0" * 16
|
||||
@@ -25,6 +25,7 @@ class FakeZidoo(BaseHTTPRequestHandler):
|
||||
library = {} # file path -> getAggregationOfFile's answer
|
||||
posters = {} # poster id -> image bytes
|
||||
lookups = [] # paths getAggregationOfFile was asked about
|
||||
seeks = [] # every seekTo's position
|
||||
|
||||
def do_GET(self):
|
||||
url = urlparse(self.path)
|
||||
@@ -33,6 +34,11 @@ class FakeZidoo(BaseHTTPRequestHandler):
|
||||
if self.video is None:
|
||||
return self._send(404, b"")
|
||||
return self._json({"status": 200, "video": self.video})
|
||||
if url.path == "/ZidooVideoPlay/seekTo":
|
||||
if self.video is None:
|
||||
return self._send(404, b"")
|
||||
self.seeks.append(int(query["positon"])) # sic, as the real one spells it
|
||||
return self._json({"status": 200})
|
||||
if url.path == "/ZidooPoster/v2/getAggregationOfFile":
|
||||
self.lookups.append(query["path"])
|
||||
return self._json(self.library.get(query["path"], {"status": 804, "msg": "Error!!!"}))
|
||||
@@ -62,6 +68,7 @@ class ZidooTest(unittest.TestCase):
|
||||
"movie": {"id": 108, "name": "A Private War", "year": 2018}}}
|
||||
FakeZidoo.posters = {108: PNG}
|
||||
FakeZidoo.lookups = []
|
||||
FakeZidoo.seeks = []
|
||||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeZidoo)
|
||||
threading.Thread(target=self.server.serve_forever, args=(0.05,), daemon=True).start()
|
||||
self.zidoo = ZidooClient("127.0.0.1", self.server.server_address[1])
|
||||
@@ -95,6 +102,15 @@ class ZidooTest(unittest.TestCase):
|
||||
FakeZidoo.video = None
|
||||
self.assertIsNone(self.zidoo.now_playing())
|
||||
|
||||
def test_seek_moves_the_loaded_film(self):
|
||||
self.zidoo.seek(3600000)
|
||||
self.assertEqual(FakeZidoo.seeks, [3600000])
|
||||
|
||||
def test_seeking_with_nothing_loaded_is_an_error(self):
|
||||
FakeZidoo.video = None
|
||||
with self.assertRaises(ZidooError):
|
||||
self.zidoo.seek(3600000)
|
||||
|
||||
def test_poster_is_sniffed_from_its_bytes(self):
|
||||
self.assertEqual(self.zidoo.poster(108), (PNG, "image/png"))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user