"""Run against the fake hardware in fakes.py: python3 -m unittest discover -s tests -t . """ import sys import unittest from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from controller import Controller, stepped_level # noqa: E402 from tests.fakes import FakeHeos # noqa: E402 AVR_PID = 1 HOME400_PID = 2 LIVING_ROOM_PID = 3 # the Home 200 pair, one pid -- HEOS pairs it at the hardware level def build(): heos = FakeHeos() cfg = SimpleNamespace( HEOS_HOST="127.0.0.1", HEOS_PORT=heos.port, HOST_KEY="avr", ROOM_KEYS=["lego_room", "living_room"], TARGETS={ "avr": {"label": "Home Cinema", "heos_name": "Home Cinema"}, "lego_room": {"label": "Lego Room", "heos_name": "Lego Room"}, "living_room": {"label": "Living Room", "heos_name": "Denon Home 200 L"}, }, VOLUME_STEP=5, ) return Controller(cfg), heos class PanelTest(unittest.TestCase): def setUp(self): self.panel, self.heos = build() def group_pids(self): return {gid: set(pids) for gid, pids in self.heos.groups.items()} # -- resolving ------------------------------------------------------ def test_living_room_resolves_to_its_one_pid(self): self.panel.scan() self.assertEqual(self.panel.member_pids("living_room"), [LIVING_ROOM_PID]) self.assertEqual(self.panel.member_pids("lego_room"), [HOME400_PID]) def test_avr_resolves_to_a_player_even_while_it_leads_a_group(self): self.panel.join("lego_room") # HEOS now reports a *group* named "Home Cinema" as well as the player. self.assertEqual(self.panel.member_pids("avr"), [AVR_PID]) # -- grouping ------------------------------------------------------- def test_joining_adds_the_room(self): self.panel.join("living_room") self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, LIVING_ROOM_PID}}) self.assertEqual(self.panel.joined_keys(), ["living_room"]) def test_joining_keeps_whoever_is_already_grouped(self): self.panel.join("lego_room") self.panel.join("living_room") self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, HOME400_PID, LIVING_ROOM_PID}}) self.assertEqual(self.panel.joined_keys(), ["lego_room", "living_room"]) def test_joining_replays_the_avr_input_over_heos(self): """A room that has just joined sometimes stays silent until the AVR's input is reselected -- through HEOS's own browse/play_input, the way the HEOS app does it, not the AVR's Telnet port -- so join() pokes it with whatever is already playing.""" before = len(self.heos.commands) self.panel.join("lego_room") replays = [c for c in self.heos.commands[before:] if "browse/play_input" in c] self.assertEqual(len(replays), 1) self.assertIn(f"pid={AVR_PID}", replays[0]) self.assertIn("input=inputs/mediaplayer", replays[0]) def test_leaving_ungroups_the_room(self): self.panel.join("living_room") self.panel.leave("living_room") self.assertEqual(self.group_pids(), {}) # AVR alone, no group left self.assertEqual(self.panel.joined_keys(), []) def test_leaving_one_room_does_not_disturb_the_other(self): self.panel.join("lego_room") self.panel.join("living_room") before = [c for c in self.heos.commands if "set_group" in c] self.panel.leave("lego_room") after = [c for c in self.heos.commands if "set_group" in c] self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, LIVING_ROOM_PID}}) # Exactly one new set_group: the remaining room is never regrouped, # which is what stops its music restarting. self.assertEqual(len(after) - len(before), 1) def test_separate_everything(self): self.panel.set_membership(["lego_room", "living_room"]) self.panel.set_membership([]) self.assertEqual(self.group_pids(), {}) self.assertEqual(self.panel.joined_keys(), []) def test_resolving_survives_a_restart_while_merged(self): """member_pids reads the live player list, so a fresh process needs no memory of anything to find a room that is currently merged.""" self.panel.join("living_room") reborn = Controller(self.panel.cfg) reborn.scan() self.assertEqual(reborn.member_pids("living_room"), [LIVING_ROOM_PID]) reborn.leave("living_room") self.assertEqual(self.group_pids(), {}) # -- volume --------------------------------------------------------- def test_living_room_volume_is_its_own_player_volume(self): self.panel.scan() self.assertEqual(self.panel.set_volume("living_room", 42), 42) self.assertEqual(self.heos.volumes[LIVING_ROOM_PID], 42) self.assertEqual(self.panel.volume("living_room"), 42) def test_nudge_clamps_at_the_ends(self): self.panel.set_volume("lego_room", 98) self.assertEqual(self.panel.nudge_volume("lego_room", 5), 100) self.panel.set_volume("lego_room", 1) self.assertEqual(self.panel.nudge_volume("lego_room", -9), 0) # -- volume in whole steps ------------------------------------------- def test_a_tap_lands_on_the_next_multiple(self): self.panel.set_volume("lego_room", 23) self.assertEqual(self.panel.step_volume("lego_room", 1), 25) self.panel.set_volume("lego_room", 23) self.assertEqual(self.panel.step_volume("lego_room", -1), 20) def test_a_level_already_on_a_multiple_moves_a_whole_step(self): self.panel.set_volume("lego_room", 25) self.assertEqual(self.panel.step_volume("lego_room", 1), 30) self.panel.set_volume("lego_room", 25) self.assertEqual(self.panel.step_volume("lego_room", -1), 20) def test_a_burst_of_taps_snaps_once_then_moves_whole_steps(self): self.panel.set_volume("living_room", 23) self.assertEqual(self.panel.step_volume("living_room", 3), 35) # -- play / pause ------------------------------------------------------ def test_toggle_play_flips_what_the_speakers_report(self): self.panel.scan() self.assertEqual(self.panel.get_play_state("lego_room"), "play") self.assertEqual(self.panel.toggle_play("lego_room"), "pause") self.assertEqual(self.panel.get_play_state("lego_room"), "pause") self.assertEqual(self.panel.toggle_play("lego_room"), "play") def test_toggle_play_takes_an_explicit_state(self): self.panel.scan() self.assertEqual(self.panel.toggle_play("lego_room", "stop"), "stop") self.assertEqual(self.heos.play_states[HOME400_PID], "stop") def test_living_room_playback_goes_to_its_player(self): self.panel.scan() self.panel.toggle_play("living_room", "pause") self.assertEqual(self.heos.play_states[LIVING_ROOM_PID], "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_state_carries_the_song_a_room_is_playing(self): self.heos.now_playing_mid[HOME400_PID] = "spotify:track:4uLU6hMCjMI75M1A2tKUQC" self.heos.now_playing_track[HOME400_PID] = { "song": "Harvest Moon", "artist": "Neil Young", "image_url": "https://i.scdn.co/image/abc", } state = self.panel.state() self.assertEqual( [r["now_playing"] for r in state["rooms"]], [{"song": "Harvest Moon", "artist": "Neil Young", "image": "https://i.scdn.co/image/abc"}, None], ) def test_a_song_without_artist_or_cover_still_shows(self): self.heos.now_playing_mid[HOME400_PID] = "spotify:track:4uLU6hMCjMI75M1A2tKUQC" self.heos.now_playing_track[HOME400_PID] = {"song": "Harvest Moon", "artist": "", "image_url": ""} self.assertEqual( self.panel.state()["rooms"][0]["now_playing"], {"song": "Harvest Moon", "artist": None, "image": None}, ) def test_a_paused_room_keeps_its_song_and_a_stopped_one_drops_it(self): self.heos.now_playing_mid[HOME400_PID] = "spotify:track:4uLU6hMCjMI75M1A2tKUQC" self.heos.now_playing_track[HOME400_PID] = {"song": "Harvest Moon"} self.heos.play_states[HOME400_PID] = "pause" self.assertIsNotNone(self.panel.state()["rooms"][0]["now_playing"]) self.heos.play_states[HOME400_PID] = "stop" self.assertIsNone(self.panel.state()["rooms"][0]["now_playing"]) def test_an_avr_input_is_not_a_song(self): """HEOS puts an input's own name where the song goes.""" self.heos.now_playing_mid[HOME400_PID] = "inputs/mediaplayer" self.heos.now_playing_track[HOME400_PID] = {"song": "Z30 Pro"} self.assertIsNone(self.panel.state()["rooms"][0]["now_playing"]) 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 on its pid) -- there is no separate Telnet lookup needed for them.""" self.assertTrue(self.panel.avr_connected()) self.assertEqual( self.panel.avr_inputs(), [{"code": "inputs/mediaplayer", "name": "Z30 Pro"}, {"code": "inputs/game", "name": "Switch"}, {"code": "inputs/tvaudio", "name": "LG G5"}], ) self.assertEqual( self.panel.avr_current_input(), {"code": "inputs/mediaplayer", "name": "Z30 Pro"} ) def test_selecting_an_avr_input_goes_through_heos(self): """Selection has to be browse/play_input, not the AVR's own Telnet port -- that is what actually pushes the stream to a joined group, not just what the AVR itself is listening to.""" self.assertEqual( self.panel.avr_select_input("inputs/game"), {"code": "inputs/game", "name": "Switch"}, ) self.assertEqual( self.panel.avr_current_input(), {"code": "inputs/game", "name": "Switch"} ) self.assertTrue(any("browse/play_input" in c for c in self.heos.commands)) # -- the whole snapshot the UI renders ------------------------------- def test_state_snapshot(self): self.panel.join("lego_room") state = self.panel.state() self.assertTrue(state["heos_ok"]) self.assertEqual([r["key"] for r in state["rooms"]], ["lego_room", "living_room"]) self.assertEqual([r["grouped"] for r in state["rooms"]], [True, False]) self.assertTrue(all(isinstance(r["volume"], int) for r in state["rooms"])) self.assertEqual([r["play_state"] for r in state["rooms"]], ["play", "play"]) def test_state_reports_trouble_instead_of_blowing_up(self): self.panel.heos.host = "127.0.0.1" self.panel.heos.port = 1 # nothing is listening there self.panel.heos._close() state = self.panel.state() self.assertFalse(state["heos_ok"]) self.assertTrue(state["errors"]) self.assertTrue(all(r["available"] is False for r in state["rooms"])) class StepArithmeticTest(unittest.TestCase): """The rule on its own -- no speakers involved.""" def test_up_from_between_multiples(self): self.assertEqual([stepped_level(23, n, 5) for n in (1, 2, 3)], [25, 30, 35]) def test_down_from_between_multiples(self): self.assertEqual([stepped_level(23, -n, 5) for n in (1, 2, 3)], [20, 15, 10]) def test_from_a_multiple(self): self.assertEqual(stepped_level(25, 1, 5), 30) self.assertEqual(stepped_level(25, -1, 5), 20) def test_clamped_to_the_ends(self): self.assertEqual(stepped_level(98, 1, 5), 100) self.assertEqual(stepped_level(2, -1, 5), 0) self.assertEqual(stepped_level(0, -1, 5), 0) def test_no_taps_changes_nothing(self): self.assertEqual(stepped_level(23, 0, 5), 23) if __name__ == "__main__": unittest.main()