"""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_KEY="avr", ROOM_KEYS=[], TARGETS={"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()