franzz/heos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove group management
Deploy HEOS panel / deploy (push) Successful in 25s
Details

Browse Source
This commit is contained in:
franzz
2026-09-16 13:43:31 +02:00
parent 5b82a24ac4
commit 58a2dedc03
15 changed files with 358 additions and 588 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_panel.py
+64 -81
View File
@@ -4,7 +4,6 @@
"""
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
@@ -14,62 +13,56 @@ 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
PAIR = {3, 4} # the two Home 200s
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(tmpdir):
def build():
heos = FakeHeos()
cfg = SimpleNamespace(
HEOS_HOST="127.0.0.1", HEOS_PORT=heos.port,
HOST_KEY="avr",
ROOM_KEYS=["home400", "living_room_group"],
ROOM_KEYS=["lego_room", "living_room"],
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"},
"lego_room": {"label": "Lego Room", "heos_name": "Lego Room"},
"living_room": {"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
class PanelTest(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.panel, self.heos = build(self.tmp.name)
self.panel, self.heos = build()
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."""
def test_living_room_resolves_to_its_one_pid(self):
self.panel.scan()
self.assertEqual(set(self.panel.member_pids("living_room_group")), PAIR)
self.assertEqual(self.panel.member_pids("home400"), [HOME400_PID])
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("home400")
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_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_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("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"])
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
@@ -77,102 +70,92 @@ class PanelTest(unittest.TestCase):
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("home400")
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_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
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("home400")
self.panel.join("living_room_group")
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("home400")
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} | PAIR})
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(["home400", "living_room_group"])
self.panel.set_membership(["lego_room", "living_room"])
self.panel.set_membership([])
self.assertEqual(self.group_pids(), {3: PAIR})
self.assertEqual(self.group_pids(), {})
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")
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(set(reborn.member_pids("living_room_group")), PAIR)
reborn.leave("living_room_group")
self.assertEqual(self.group_pids(), {3: PAIR})
self.assertEqual(reborn.member_pids("living_room"), [LIVING_ROOM_PID])
reborn.leave("living_room")
self.assertEqual(self.group_pids(), {})
# -- volume ---------------------------------------------------------
def test_pair_uses_group_volume_when_it_stands_alone(self):
def test_living_room_volume_is_its_own_player_volume(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)
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("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)
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("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)
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("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)
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_group", 23)
self.assertEqual(self.panel.step_volume("living_room_group", 3), 35)
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("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")
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("home400", "stop"), "stop")
self.assertEqual(self.panel.toggle_play("lego_room", "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."""
def test_living_room_playback_goes_to_its_player(self):
self.panel.scan()
self.panel.toggle_play("living_room_group", "pause")
self.assertEqual(self.heos.play_states[3], "pause")
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
@@ -249,10 +232,10 @@ class PanelTest(unittest.TestCase):
# -- the whole snapshot the UI renders -------------------------------
def test_state_snapshot(self):
self.panel.join("home400")
self.panel.join("lego_room")
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["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"])