"""Run against the fake hardware in fakes.py: python3 -m unittest discover -s tests -t . """ import sys import tempfile import time 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 FakeAvr, FakeHeos # noqa: E402 PAIR = {3, 4} # the two Home 200s AVR_PID = 1 HOME400_PID = 2 def build(tmpdir): heos, avr = FakeHeos(), FakeAvr() cfg = SimpleNamespace( HEOS_HOST="127.0.0.1", HEOS_PORT=heos.port, AVR_HOST="127.0.0.1", AVR_PORT=avr.port, HOST_KEY="avr", ROOM_KEYS=["home400", "living_room_group"], TARGETS={ "avr": {"label": "Home Cinema", "heos_name": "Home Cinema"}, "home400": {"label": "Lego Room", "heos_name": "Lego Room"}, "living_room_group": {"label": "Living Room", "heos_name": "Denon Home 200 L"}, }, AVR_INPUT_CODES=[], VOLUME_STEP=5, MEMBERS_FILE=str(Path(tmpdir) / "members.json"), ) return Controller(cfg), heos, avr class PanelTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.panel, self.heos, self.avr = build(self.tmp.name) def group_pids(self): return {gid: set(pids) for gid, pids in self.heos.groups.items()} # -- resolving ------------------------------------------------------ def test_pair_resolves_to_both_speakers(self): """The bug this replaces: grouping used only the pair's leader, which left the second Home 200 behind.""" self.panel.scan() self.assertEqual(set(self.panel.member_pids("living_room_group")), PAIR) self.assertEqual(self.panel.member_pids("home400"), [HOME400_PID]) def test_avr_resolves_to_a_player_even_while_it_leads_a_group(self): self.panel.join("home400") # 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_takes_the_whole_pair(self): self.panel.join("living_room_group") self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR}) self.assertEqual(self.panel.joined_keys(), ["living_room_group"]) def test_joining_keeps_whoever_is_already_grouped(self): self.panel.join("home400") self.panel.join("living_room_group") self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, HOME400_PID} | PAIR}) self.assertEqual(self.panel.joined_keys(), ["home400", "living_room_group"]) def test_leaving_rebuilds_the_stereo_pair(self): self.panel.join("living_room_group") self.panel.leave("living_room_group") self.assertEqual(self.group_pids(), {3: PAIR}) # pair back, AVR alone self.assertEqual(self.panel.joined_keys(), []) def test_leaving_one_room_does_not_disturb_the_other(self): self.panel.join("home400") self.panel.join("living_room_group") before = [c for c in self.heos.commands if "set_group" in c] self.panel.leave("home400") after = [c for c in self.heos.commands if "set_group" in c] self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID} | PAIR}) # 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(["home400", "living_room_group"]) self.panel.set_membership([]) self.assertEqual(self.group_pids(), {3: PAIR}) self.assertEqual(self.panel.joined_keys(), []) def test_membership_survives_a_restart_while_merged(self): """Once merged, the pair's own group is gone from HEOS, so a fresh process has to fall back on what it learned earlier.""" self.panel.join("living_room_group") reborn = Controller(self.panel.cfg) reborn.scan() self.assertEqual(set(reborn.member_pids("living_room_group")), PAIR) reborn.leave("living_room_group") self.assertEqual(self.group_pids(), {3: PAIR}) # -- volume --------------------------------------------------------- def test_pair_uses_group_volume_when_it_stands_alone(self): self.panel.scan() self.assertEqual(self.panel.set_volume("living_room_group", 42), 42) self.assertEqual(self.heos.group_volumes[3], 42) def test_pair_uses_player_volume_once_merged(self): """Its gid stops existing the moment it joins the AVR, so the old bridge's get_volume?gid= call would simply fail here.""" self.panel.join("living_room_group") self.assertEqual(self.panel.set_volume("living_room_group", 31), 31) self.assertEqual(self.heos.volumes[3], 31) self.assertEqual(self.heos.volumes[4], 31) self.assertEqual(self.panel.volume("living_room_group"), 31) def test_nudge_clamps_at_the_ends(self): self.panel.set_volume("home400", 98) self.assertEqual(self.panel.nudge_volume("home400", 5), 100) self.panel.set_volume("home400", 1) self.assertEqual(self.panel.nudge_volume("home400", -9), 0) # -- volume in whole steps ------------------------------------------- def test_a_tap_lands_on_the_next_multiple(self): self.panel.set_volume("home400", 23) self.assertEqual(self.panel.step_volume("home400", 1), 25) self.panel.set_volume("home400", 23) self.assertEqual(self.panel.step_volume("home400", -1), 20) def test_a_level_already_on_a_multiple_moves_a_whole_step(self): self.panel.set_volume("home400", 25) self.assertEqual(self.panel.step_volume("home400", 1), 30) self.panel.set_volume("home400", 25) self.assertEqual(self.panel.step_volume("home400", -1), 20) def test_a_burst_of_taps_snaps_once_then_moves_whole_steps(self): self.panel.set_volume("living_room_group", 23) self.assertEqual(self.panel.step_volume("living_room_group", 3), 35) # -- the AVR --------------------------------------------------------- def test_renamed_inputs_and_selection(self): deadline = time.time() + 5 while not self.panel.avr.connected and time.time() < deadline: time.sleep(0.05) self.assertTrue(self.panel.avr.connected) self.assertEqual( self.panel.avr.inputs(), [{"code": "MPLAY", "name": "Apple TV"}, {"code": "GAME", "name": "PlayStation"}, {"code": "SAT/CBL", "name": "TV Box"}], ) self.assertEqual(self.panel.avr.current_input(), {"code": "MPLAY", "name": "Apple TV"}) # "Old DVD" is deleted in the AVR's setup menu, so the picker skips # it -- but it keeps its name, in case the AVR is sitting on it. self.assertNotIn("DVD", [s["code"] for s in self.panel.avr.inputs()]) self.assertIn({"code": "DVD", "name": "Old DVD"}, self.panel.avr.all_inputs()) self.assertEqual(self.panel.avr.name_for("DVD"), "Old DVD") self.assertEqual(self.panel.avr.select_input("GAME"), {"code": "GAME", "name": "PlayStation"}) self.assertEqual(self.avr.input, "GAME") # -- the whole snapshot the UI renders ------------------------------- def test_state_snapshot(self): self.panel.join("home400") state = self.panel.state() self.assertTrue(state["heos_ok"]) self.assertEqual([r["key"] for r in state["rooms"]], ["home400", "living_room_group"]) self.assertEqual([r["grouped"] for r in state["rooms"]], [True, False]) self.assertTrue(all(isinstance(r["volume"], int) for r in state["rooms"])) 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()