franzzandClaude Opus 5
1a1fcc84d6
A tap now moves to the next multiple of VOLUME_STEP rather than adding it, so 23 goes to 25 and 25 goes to 30 and the levels stay round. The panel counts taps and lets the speakers do the rounding from whatever level they are actually at, since the phone's copy can be seconds old; the same rule is mirrored in JS so the optimistic number never has to correct itself when the reply lands. The input picker also asks the AVR which sources are still switched on (SSSOD ?) and leaves out the ones deleted in its setup menu. Sources it does not mention are kept, so a model that ignores the command shows its whole list rather than nothing; deleted sources also keep their names, in case the AVR is sitting on one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
214 lines
9.0 KiB
Python
214 lines
9.0 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)
|
|
|
|
# -- 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()
|