235 lines
10 KiB
Python
235 lines
10 KiB
Python
"""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)
|
|
|
|
# -- play / pause ------------------------------------------------------
|
|
def test_toggle_play_flips_what_the_speakers_report(self):
|
|
self.panel.scan()
|
|
self.assertEqual(self.panel.get_play_state("home400"), "play")
|
|
self.assertEqual(self.panel.toggle_play("home400"), "pause")
|
|
self.assertEqual(self.panel.get_play_state("home400"), "pause")
|
|
self.assertEqual(self.panel.toggle_play("home400"), "play")
|
|
|
|
def test_toggle_play_takes_an_explicit_state(self):
|
|
self.panel.scan()
|
|
self.assertEqual(self.panel.toggle_play("home400", "stop"), "stop")
|
|
self.assertEqual(self.heos.play_states[HOME400_PID], "stop")
|
|
|
|
def test_pair_is_controlled_through_one_of_its_speakers(self):
|
|
"""Playback is a player command -- a group has none of its own -- so
|
|
it goes to the pair's leader."""
|
|
self.panel.scan()
|
|
self.panel.toggle_play("living_room_group", "pause")
|
|
self.assertEqual(self.heos.play_states[3], "pause")
|
|
|
|
# -- 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"]))
|
|
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()
|