fix AVR input swap

tests/fakes.py

@@ -1,4 +1,4 @@
-"""Stand-in Denon hardware: just enough HEOS and Telnet to test against.
+"""Stand-in HEOS hardware: just enough of the CLI protocol to test against.
 
 The grouping rules are the part worth pinning down -- what a set_group
 call does to a stereo pair is the kind of thing you do not want to find
@@ -15,12 +15,22 @@ class FakeHeos(threading.Thread):
 
     NAMES = {1: "Home Cinema", 2: "Lego Room", 3: "Denon Home 200 L", 4: "Denon Home 200 R"}
 
+    # What browse/browse?sid=<AVR pid> reports: HEOS's own list of the
+    # AVR's local inputs, already under whatever names you gave them in
+    # its setup menu -- HEOS carries the renamed labels itself.
+    AVR_INPUTS = [
+        {"name": "Z30 Pro", "mid": "inputs/mediaplayer"},
+        {"name": "Switch", "mid": "inputs/game"},
+        {"name": "LG G5", "mid": "inputs/tvaudio"},
+    ]
+
     def __init__(self):
         super().__init__(daemon=True)
         self.groups = {3: [3, 4]}                     # gid -> pids, leader first
         self.volumes = {pid: 20 for pid in self.NAMES}
         self.play_states = {pid: "play" for pid in self.NAMES}
         self.group_volumes = {3: 25}
+        self.now_playing_mid = {1: "inputs/mediaplayer"}  # pid -> what get_now_playing_media reports
         self.commands = []                            # everything we were asked to do
         self.server = socket.socket()
         self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -106,6 +116,22 @@ class FakeHeos(threading.Thread):
             self.play_states[pid] = args["state"]
             return self._ok(path, message=f"pid={pid}&state={args['state']}")
 
+        if path == "player/get_now_playing_media":
+            pid = int(args["pid"])
+            mid = self.now_playing_mid.get(pid, "")
+            name = next((s["name"] for s in self.AVR_INPUTS if s["mid"] == mid), mid)
+            return self._ok(path, payload={"mid": mid, "station": name} if mid else {})
+
+        if path == "browse/browse":
+            sid = int(args["sid"])
+            payload = list(self.AVR_INPUTS) if sid == 1 else []
+            return self._ok(path, payload=payload)
+
+        if path == "browse/play_input":
+            pid = int(args["pid"])
+            self.now_playing_mid[pid] = args["input"]
+            return self._ok(path)
+
         if path.endswith("/toggle_mute") or path == "system/heart_beat":
             return self._ok(path)
 
@@ -125,61 +151,3 @@ class FakeHeos(threading.Thread):
         if payload is not None:
             reply["payload"] = payload
         return reply
-
-
-class FakeAvr(threading.Thread):
-    """A Denon Telnet server that knows SI and SSFUN."""
-
-    SOURCES = [("MPLAY", "Apple TV"), ("GAME", "PlayStation"),
-               ("SAT/CBL", "TV Box"), ("DVD", "Old DVD")]
-    DELETED = {"DVD"}          # switched off in the AVR's setup menu
-
-    def __init__(self):
-        super().__init__(daemon=True)
-        self.input = "MPLAY"
-        self.server = socket.socket()
-        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        self.server.bind(("127.0.0.1", 0))
-        self.server.listen(4)
-        self.port = self.server.getsockname()[1]
-        self.start()
-
-    def run(self):
-        while True:
-            try:
-                conn, _ = self.server.accept()
-            except OSError:
-                return
-            threading.Thread(target=self._serve, args=(conn,), daemon=True).start()
-
-    def _serve(self, conn):
-        buffer = b""
-        with conn:
-            while True:
-                try:
-                    chunk = conn.recv(1024)
-                except OSError:
-                    return
-                if not chunk:
-                    return
-                buffer += chunk
-                while b"\r" in buffer:
-                    line, buffer = buffer.split(b"\r", 1)
-                    for reply in self.handle(line.decode().strip()):
-                        conn.sendall(reply.encode() + b"\r")
-
-    def handle(self, command: str) -> list:
-        if command == "SSFUN ?":
-            # The real AVR pads the names out with spaces.
-            return [f"SSFUN{code} {name}      " for code, name in self.SOURCES] + ["SSFUN END"]
-        if command == "SSSOD ?":
-            return [f"SSSOD{code} {'DEL' if code in self.DELETED else 'USE'}"
-                    for code, _ in self.SOURCES] + ["SSSOD END"]
-        if command == "SI?":
-            return [f"SI{self.input}"]
-        if command.startswith("SI"):
-            self.input = command[2:]
-            return [f"SI{self.input}"]
-        if command == "PW?":
-            return ["PWON"]
-        return []

tests/test_panel.py

@@ -5,7 +5,6 @@
 
 import sys
 import tempfile
-import time
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
@@ -13,7 +12,7 @@ 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
+from tests.fakes import FakeHeos    # noqa: E402
 
 PAIR = {3, 4}          # the two Home 200s
 AVR_PID = 1
@@ -21,10 +20,9 @@ HOME400_PID = 2
 
 
 def build(tmpdir):
-    heos, avr = FakeHeos(), FakeAvr()
+    heos = FakeHeos()
     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={
@@ -36,14 +34,14 @@ def build(tmpdir):
         VOLUME_STEP=5,
         MEMBERS_FILE=str(Path(tmpdir) / "members.json"),
     )
-    return Controller(cfg), heos, avr
+    return Controller(cfg), heos
 
 
 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)
+        self.panel, self.heos = build(self.tmp.name)
 
     def group_pids(self):
         return {gid: set(pids) for gid, pids in self.heos.groups.items()}
@@ -73,6 +71,18 @@ class PanelTest(unittest.TestCase):
         self.assertEqual(self.group_pids(), {AVR_PID: {AVR_PID, HOME400_PID} | PAIR})
         self.assertEqual(self.panel.joined_keys(), ["home400", "living_room_group"])
 
+    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("home400")
+        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")
@@ -164,29 +174,33 @@ class PanelTest(unittest.TestCase):
         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)
-
+    # -- 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": "MPLAY", "name": "Apple TV"},
-             {"code": "GAME", "name": "PlayStation"},
-             {"code": "SAT/CBL", "name": "TV Box"}],
+            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"}
         )
-        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")
+    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):