heos/tests/fakes.py

"""Stand-in Denon hardware: just enough HEOS and Telnet 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
out by experimenting on the speakers at eleven at night.
"""

import json
import socket
import threading


class FakeHeos(threading.Thread):
    """A HEOS CLI server on localhost, with four players and a pair."""

    NAMES = {1: "Home Cinema", 2: "Lego Room", 3: "Denon Home 200 L", 4: "Denon Home 200 R"}

    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.group_volumes = {3: 25}
        self.commands = []                            # everything we were asked to do
        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(8)
        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(4096)
                except OSError:
                    return
                if not chunk:
                    return
                buffer += chunk
                while b"\r\n" in buffer:
                    line, buffer = buffer.split(b"\r\n", 1)
                    reply = self.handle(line.decode().strip())
                    conn.sendall(json.dumps(reply).encode() + b"\r\n")

    # -- command handling ------------------------------------------------
    def handle(self, command: str) -> dict:
        self.commands.append(command)
        path, _, query = command[len("heos://"):].partition("?")
        args = dict(part.split("=", 1) for part in query.split("&") if "=" in part)

        if path == "player/get_players":
            return self._ok(path, payload=[
                {"name": name, "pid": pid, "model": "Fake"} for pid, name in self.NAMES.items()
            ])

        if path == "group/get_groups":
            payload = []
            for gid, pids in self.groups.items():
                payload.append({
                    "name": self.NAMES[gid], "gid": gid,
                    "players": [
                        {"name": self.NAMES[pid], "pid": pid,
                         "role": "leader" if pid == gid else "member"}
                        for pid in pids
                    ],
                })
            return self._ok(path, payload=payload)

        if path == "group/set_group":
            pids = [int(p) for p in args["pid"].split(",")]
            for gid in list(self.groups):
                self.groups[gid] = [p for p in self.groups[gid] if p not in pids]
                if len(self.groups[gid]) < 2:
                    del self.groups[gid]          # HEOS dissolves a group of one
            if len(pids) > 1:
                self.groups[pids[0]] = pids
                self.group_volumes.setdefault(pids[0], 25)
            return self._ok(path, message=args["pid"])

        if path.endswith("/get_volume"):
            store, key = self._store(path, args)
            return self._ok(path, message=f"{key[0]}={key[1]}&level={store[key[1]]}")

        if path.endswith("/set_volume"):
            store, key = self._store(path, args)
            store[key[1]] = int(args["level"])
            return self._ok(path, message=f"{key[0]}={key[1]}&level={args['level']}")

        if path.endswith("/toggle_mute") or path == "system/heart_beat":
            return self._ok(path)

        return {"heos": {"command": path, "result": "fail", "message": "eid=2&text=Not+supported"}}

    def _store(self, path, args):
        if path.startswith("group/"):
            gid = int(args["gid"])
            if gid not in self.groups:
                raise KeyError(gid)
            return self.group_volumes, ("gid", gid)
        return self.volumes, ("pid", int(args["pid"]))

    @staticmethod
    def _ok(path, message="", payload=None):
        reply = {"heos": {"command": path, "result": "success", "message": message}}
        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 []