204 lines
8.4 KiB
Python
204 lines
8.4 KiB
Python
"""The Zidoo client against a pretend Zidoo, answering the way a real one
|
|
does (see zidoo.py).
|
|
|
|
python3 -m unittest discover -s tests -t .
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from controller import Controller, TargetError # 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
|
|
|
|
|
|
class FakeZidoo(BaseHTTPRequestHandler):
|
|
video = None # getPlayStatus's "video", or None for nothing loaded
|
|
library = {} # file path -> getAggregationOfFile's answer
|
|
posters = {} # poster id -> image bytes
|
|
lookups = [] # paths getAggregationOfFile was asked about
|
|
seeks = [] # every seekTo's position
|
|
keys = [] # every sendkey the remote endpoint was handed
|
|
accepts = {"Key.MediaPlay", "Key.MediaPause"} # the rest come back 805
|
|
|
|
def do_GET(self):
|
|
url = urlparse(self.path)
|
|
query = {key: values[0] for key, values in parse_qs(url.query).items()}
|
|
if url.path == "/ZidooVideoPlay/getPlayStatus":
|
|
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 == "/ZidooControlCenter/RemoteControl/sendkey":
|
|
key = query.get("key")
|
|
if key not in self.accepts:
|
|
return self._json({"status": 805, "msg": "The parameter is error or no parameter is submitted"})
|
|
# The real box answers before the player has acted on the press,
|
|
# so getPlayStatus keeps reporting the old state for a moment.
|
|
self.keys.append(key)
|
|
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!!!"}))
|
|
if url.path == "/ZidooPoster/getFile/getPoster":
|
|
image = self.posters.get(int(query["id"]))
|
|
return self._send(200, image) if image else self._json({"status": 804, "msg": "Error!!!"})
|
|
self._send(404, b"")
|
|
|
|
def _json(self, payload):
|
|
self._send(200, json.dumps(payload).encode())
|
|
|
|
def _send(self, code, body):
|
|
self.send_response(code) # no Content-Type, like the real thing
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
class ZidooTest(unittest.TestCase):
|
|
def setUp(self):
|
|
FakeZidoo.video = {"status": 1, "title": "A Private War", "path": PATH,
|
|
"currentPosition": 1589745, "duration": 6635092}
|
|
FakeZidoo.library = {PATH: {"path": PATH, "type": "movie", "video": {"id": 2248, "parentId": 108},
|
|
"movie": {"id": 108, "name": "A Private War", "year": 2018}}}
|
|
FakeZidoo.posters = {108: PNG}
|
|
FakeZidoo.lookups = []
|
|
FakeZidoo.seeks = []
|
|
FakeZidoo.keys = []
|
|
FakeZidoo.accepts = {"Key.MediaPlay", "Key.MediaPause"}
|
|
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])
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
def test_a_film_carries_its_year_progress_and_poster(self):
|
|
self.assertEqual(self.zidoo.now_playing(), {
|
|
"song": "A Private War", "artist": "2018", "image": None, "poster_id": 108,
|
|
"play_state": "play", "position_ms": 1589745, "duration_ms": 6635092,
|
|
})
|
|
|
|
def test_a_paused_film_is_still_there_but_paused(self):
|
|
FakeZidoo.video["status"] = 0
|
|
self.assertEqual(self.zidoo.now_playing()["play_state"], "pause")
|
|
|
|
def test_the_poster_is_looked_up_once_per_file(self):
|
|
self.zidoo.now_playing()
|
|
self.zidoo.now_playing()
|
|
self.assertEqual(FakeZidoo.lookups, [PATH])
|
|
|
|
def test_a_file_outside_the_library_has_no_poster_or_year(self):
|
|
FakeZidoo.library = {}
|
|
track = self.zidoo.now_playing()
|
|
self.assertIsNone(track["poster_id"])
|
|
self.assertIsNone(track["artist"])
|
|
|
|
def test_nothing_loaded_is_nothing_playing(self):
|
|
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_play_and_pause_send_the_matching_media_key(self):
|
|
self.zidoo.play_state("pause")
|
|
self.zidoo.play_state("play")
|
|
self.assertEqual(FakeZidoo.keys, ["Key.MediaPause", "Key.MediaPlay"])
|
|
|
|
def test_a_state_the_box_has_no_key_for_is_an_error(self):
|
|
"""HEOS rooms take "stop" as well; the Zidoo's remote has no such key,
|
|
so /api/playback's third state has to be refused rather than guessed at."""
|
|
with self.assertRaises(ZidooError):
|
|
self.zidoo.play_state("stop")
|
|
self.assertEqual(FakeZidoo.keys, [])
|
|
|
|
def test_a_key_the_box_rejects_is_an_error(self):
|
|
FakeZidoo.accepts = set()
|
|
with self.assertRaises(ZidooError):
|
|
self.zidoo.play_state("pause")
|
|
|
|
def test_an_unreachable_box_is_an_error_not_silence(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
with self.assertRaises(ZidooError):
|
|
self.zidoo.play_state("pause")
|
|
|
|
def test_poster_is_sniffed_from_its_bytes(self):
|
|
self.assertEqual(self.zidoo.poster(108), (PNG, "image/png"))
|
|
|
|
def test_an_id_without_a_poster_is_none_not_its_json_error(self):
|
|
self.assertIsNone(self.zidoo.poster(2248))
|
|
|
|
|
|
class ZidooThroughTheControllerTest(unittest.TestCase):
|
|
"""zidoo_play_state() only ever talks to the Zidoo, so this needs no HEOS
|
|
behind it -- HeosClient does not dial until something asks it to."""
|
|
|
|
def setUp(self):
|
|
FakeZidoo.video = {"status": 1, "title": "A Private War", "path": PATH,
|
|
"currentPosition": 1589745, "duration": 6635092}
|
|
FakeZidoo.library = {}
|
|
FakeZidoo.keys = []
|
|
FakeZidoo.accepts = {"Key.MediaPlay", "Key.MediaPause"}
|
|
self.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeZidoo)
|
|
threading.Thread(target=self.server.serve_forever, args=(0.05,), daemon=True).start()
|
|
self.cfg = SimpleNamespace(
|
|
HEOS_HOST="127.0.0.1", HEOS_PORT=1255,
|
|
HOST_ROOM="avr", GUEST_ROOMS=[],
|
|
ROOMS={"avr": {"label": "Home Cinema", "heos_name": "Home Cinema"}},
|
|
VOLUME_STEP=5,
|
|
ZIDOO_HOST="127.0.0.1", ZIDOO_PORT=self.server.server_address[1],
|
|
)
|
|
self.panel = Controller(self.cfg)
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
|
|
def test_an_explicit_state_goes_straight_through(self):
|
|
self.assertEqual(self.panel.zidoo_play_state("pause"), "pause")
|
|
self.assertEqual(FakeZidoo.keys, ["Key.MediaPause"])
|
|
|
|
def test_no_state_flips_what_the_zidoo_itself_reports(self):
|
|
self.assertEqual(self.panel.zidoo_play_state(), "pause")
|
|
self.assertEqual(FakeZidoo.keys, ["Key.MediaPause"])
|
|
|
|
def test_no_state_starts_a_film_the_zidoo_has_paused(self):
|
|
FakeZidoo.video["status"] = 0
|
|
self.assertEqual(self.panel.zidoo_play_state(), "play")
|
|
self.assertEqual(FakeZidoo.keys, ["Key.MediaPlay"])
|
|
|
|
def test_no_zidoo_configured_is_a_target_error(self):
|
|
panel = Controller(SimpleNamespace(**{**vars(self.cfg), "ZIDOO_HOST": None}))
|
|
with self.assertRaises(TargetError):
|
|
panel.zidoo_play_state("pause")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|