diff --git a/README.md b/README.md index 268bc42..7848440 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,7 @@ All three install the same commands: |---|---| | `ragnarosd` | the daemon (run by the `ragnarosd` user unit) | | `ragnarosctl` | drive the running daemon (see [Control CLI](#control-cli)) | -| `ragnaros` | `install-system`, `install-service`, `paths` | +| `ragnaros` | `install-system`, `install-service`, `paths`, `diagnose` (see [Reporting a new device](#reporting-a-new-device)) | | `ragnaros-fetch-gifs` | opt-in anime GIF themes for the strip (see [GIF themes](#gif-themes)) | | `ragnaros-claude-hook` | Claude Code hook (see [Notices that wait for you](#notices-that-wait-for-you)) | | `ragnaros-strip-gif` | convert any GIF to the 704x124 full-strip format | @@ -312,6 +312,47 @@ connected, so the deck becomes a blank input device; the grab is released the moment the client disconnects. Profile switches, layer changes and alerts are published on the same stream. +## Reporting a new device + +Ragnaros drives one deck, the SS552 (`0200:3001`). Other Redragon decks share its USB +vendor ID `0200` (which `lsusb` labels "TP-Link") and may or may not speak the same +protocol. If you have one, two things show whether it can be added. + +**1. A device report.** `ragnaros diagnose` prints a Markdown report on every vendor-`0200` +device plugged in: IDs and strings, interfaces and endpoints, the HID report descriptors, +the hidraw and USB device nodes with their permissions, which udev rules and usbhid quirks +cover it, and the ragnaros, kernel and distribution versions. It only reads sysfs and the +permissions in `/dev`: the deck is never opened, so it needs no root and is safe while the +daemon runs. It leaves out the serial number, the hostname and user names. + +```bash +ragnaros diagnose > report.md # --device VID:PID reports any other device +``` + +Paste `report.md` into an issue, with how many keys, knobs and screens the deck has. What +to try next goes to the terminal, not to the file. + +**2. An experimental test.** `ragnarosd --experimental-pid 0x2000` (or +`RAGNAROS_EXPERIMENTAL_PID=0x2000`) also opens that product ID and drives it with the SS552 +protocol. It is off unless you ask for it, it is marked experimental in the daemon's log and +in `ragnarosctl status`, and it logs every raw input report the deck sends. Nothing about +it is known to work. + +The udev rules only give your user the SS552, so hand yourself the deck's USB node (the +`device node` line of the report) until you unplug it, instead of widening the rules: + +```bash +sudo chown "$USER" /dev/bus/usb/005/002 # the device node from the report +systemctl --user stop ragnarosd # if it runs +ragnarosd --experimental-pid 0x2000 # stays in the foreground; Ctrl+C stops it +ragnarosctl watch --grab # in a second terminal: press every key, turn every knob +``` + +`--grab` keeps the profile's actions from running while you press things. Report whether +icons show up and on which keys (a photo helps), and paste the output of both terminals. +Unplugging the deck undoes the `chown`. From a git checkout the commands are +`python3 -m ragnaros.cli diagnose`, `python3 daemon/ragnarosd.py` and `tools/ragnarosctl`. + ## Tools The user-facing tools are installed as commands (table above). The checkout also keeps: diff --git a/ragnaros/cli.py b/ragnaros/cli.py index fe2d17f..1e8ce92 100644 --- a/ragnaros/cli.py +++ b/ragnaros/cli.py @@ -3,6 +3,7 @@ ragnaros install-system [--print] udev rules + usbhid quirk (sudo, one reboot) ragnaros install-service [--print] systemd user unit for ragnarosd ragnaros paths where profiles, assets and GIF themes live + ragnaros diagnose [--device VID:PID] read-only device report to paste in an issue The .deb and AUR packages already install the udev rules, the quirk and the user units; there you only run `systemctl --user enable --now ragnarosd`. @@ -15,7 +16,7 @@ import subprocess import sys -from . import __version__, control, paths, theme +from . import __version__, control, diagnose, paths, theme UDEV_RULES = "70-ragnaros.rules" QUIRK_CONF = "ragnaros-usbhid.conf" @@ -188,6 +189,13 @@ def show_paths(args): return 0 +def device_arg(text): + try: + return diagnose.parse_device(text) + except ValueError as error: + raise argparse.ArgumentTypeError(str(error)) from None + + def main(argv=None): parser = argparse.ArgumentParser( prog="ragnaros", description="Setup helpers for the Ragnaros deck daemon.", @@ -213,6 +221,11 @@ def main(argv=None): p = sub.add_parser("paths", help="show where profiles, assets and GIF themes live") p.set_defaults(func=show_paths) + p = sub.add_parser("diagnose", help="read-only device report to paste in an issue") + p.add_argument("--device", type=device_arg, metavar="VID:PID", + help="report this USB device instead of every vendor-0200 one") + p.set_defaults(func=diagnose.run) + args = parser.parse_args(argv) if not args.command: parser.print_help() diff --git a/ragnaros/daemon.py b/ragnaros/daemon.py index 7cc03bf..c7f33e4 100755 --- a/ragnaros/daemon.py +++ b/ragnaros/daemon.py @@ -141,8 +141,12 @@ def run_action(action): class Deck: - def __init__(self): + pids = (protocol.PID,) # product IDs to open, in order + experimental_id = None # "0200:2000" while driving an untested product ID + + def __init__(self, pids=(protocol.PID,)): self.dev = None + self.pids = tuple(pids) self._wait_for_device() self.profile = load_profile(saved_profile_name()) self.last_activity = time.monotonic() @@ -199,9 +203,13 @@ def _wait_for_device(self): announced = False while True: try: - self.dev = protocol.Ragnaros() + self.dev = protocol.Ragnaros(self.pids) if announced: print("ragnaros: device attached, resuming", file=sys.stderr) + if getattr(self.dev, "experimental", False): + self.experimental_id = protocol.device_id(self.dev.pid) + print(f"ragnaros: EXPERIMENTAL: opened untested device " + f"{self.experimental_id} with the SS552 protocol", file=sys.stderr) return except RuntimeError as error: if "libusb-1.0 is required" in str(error): @@ -1017,17 +1025,20 @@ def handle_control(self, argv): return {"ok": False, "error": "empty command"} command, args = argv[0], argv[1:] if command in ("status", "state"): - return {"ok": True, "profile": self.profile.get("_name"), - "profiles": available_profiles(), - "layer": self.layer or "base", "layers": sorted(self.layers()), - "strip": self.strip_state(), "brightness": self.brightness, - "locked": self.locked, "dashboard": self.pinned == "dashboard", - "player": self.player and {k: self.player[k] - for k in ("title", "artist", "length")}, - "alert": self.alert and self.alert["summary"], - "watchers": len(self.control.subscribers) if self.control else 0, - "pomodoro": self.pomo and self.pomo["phase"], - "obs": self.obs_state} + reply = {"ok": True, "profile": self.profile.get("_name"), + "profiles": available_profiles(), + "layer": self.layer or "base", "layers": sorted(self.layers()), + "strip": self.strip_state(), "brightness": self.brightness, + "locked": self.locked, "dashboard": self.pinned == "dashboard", + "player": self.player and {k: self.player[k] + for k in ("title", "artist", "length")}, + "alert": self.alert and self.alert["summary"], + "watchers": len(self.control.subscribers) if self.control else 0, + "pomodoro": self.pomo and self.pomo["phase"], + "obs": self.obs_state} + if self.experimental_id: + reply["experimental"] = f"{self.experimental_id} (untested, SS552 protocol)" + return reply if command in ("profile", "profiles"): if not args or args[0] == "list": return {"ok": True, "profiles": available_profiles(), @@ -1164,23 +1175,49 @@ def run(self): self.dev.close() +def experimental_arg(argv): + """The value of --experimental-pid X / --experimental-pid=X, or None.""" + for i, arg in enumerate(argv): + if arg == "--experimental-pid": + if i + 1 >= len(argv): + raise ValueError("--experimental-pid needs a product ID, e.g. 0x2000") + return argv[i + 1] + if arg.startswith("--experimental-pid="): + return arg.partition("=")[2] + return None + + def main(argv=None): argv = sys.argv[1:] if argv is None else argv if argv and argv[0] in ("-h", "--help"): - print("usage: ragnarosd\n\n" + print("usage: ragnarosd [--experimental-pid PID]\n\n" "Run the Ragnaros deck daemon in the foreground (normally started by\n" "the ragnarosd.service systemd user unit).\n\n" + " --experimental-pid PID also open the untested vendor-0200 deck PID\n" + " (hex, e.g. 0x2000) with the SS552 protocol;\n" + " for testing only, see `ragnaros diagnose`\n\n" f" profiles {', '.join(paths.profile_dirs())}\n" f" assets {', '.join(paths.asset_roots())}\n" f" control {control.socket_path()} (drive it with ragnarosctl)\n\n" "Environment: RAGNAROS_CONFIG, RAGNAROS_DATA, RAGNAROS_IDLE_SECONDS,\n" - "RAGNAROS_IDLE_FACE_SECONDS, RAGNAROS_USAGE_LOG, RAGNAROS_NO_KEEPALIVE") + "RAGNAROS_IDLE_FACE_SECONDS, RAGNAROS_USAGE_LOG, RAGNAROS_NO_KEEPALIVE,\n" + f"{protocol.EXPERIMENTAL_ENV} (same as --experimental-pid)") return 0 if argv and argv[0] == "--version": from . import __version__ print(f"ragnarosd {__version__}") return 0 - Deck().run() + try: + pids = protocol.product_ids(experimental_arg(argv)) + except ValueError as error: + print(f"ragnarosd: {error}", file=sys.stderr) + return 2 + for pid in pids[1:]: + print(f"ragnaros: EXPERIMENTAL: also accepting {protocol.device_id(pid)}, which is " + "untested, and driving it with the SS552 protocol. Keys, knobs and the strip " + "may not line up. Please report what works in " + "https://github.com/ValentinTorassa/VT-Ragnaros/issues", file=sys.stderr) + Deck(pids).run() return 0 diff --git a/ragnaros/diagnose.py b/ragnaros/diagnose.py new file mode 100644 index 0000000..b42e6c1 --- /dev/null +++ b/ragnaros/diagnose.py @@ -0,0 +1,458 @@ +"""`ragnaros diagnose`: a device report to paste into an issue. + +Read-only: everything comes from sysfs, the /dev nodes' permissions and the +udev rules directories. The device is never opened, so it needs no root and +is safe while ragnarosd owns the deck. The serial number, the hostname and +user names are left out. + +By default it reports every USB device under vendor 0200 (the SS552 and its +siblings); `--device VID:PID` picks any other one. +""" +import ctypes.util +import fnmatch +import grp +import os +import platform +import pwd +import re +import stat +import sys + +from . import __version__, protocol + +ROOT = "/" +SUPPORTED = {(protocol.VID, protocol.PID): "Redragon SS552"} +UDEV_DIRS = ("etc/udev/rules.d", "run/udev/rules.d", "usr/local/lib/udev/rules.d", + "usr/lib/udev/rules.d", "lib/udev/rules.d") +HID_DEVICE = re.compile(r"^[0-9A-Fa-f]{4}:[0-9A-Fa-f]{4}:[0-9A-Fa-f]{4}\.[0-9A-Fa-f]+$") +QUIRK = re.compile(r"^0x([0-9a-f]+):0x([0-9a-f]+):0x([0-9a-f]+)$", re.I) +QUIRK_NAMES = {0x4: "IGNORE"} +CLASSES = {"03": "HID", "ff": "vendor specific"} +BOOT_PROTOCOLS = {("01", "01"): "boot keyboard", ("01", "02"): "boot mouse"} +RULE_MATCH = re.compile( + r'\b(ATTRS?\{idVendor\}|ATTRS?\{idProduct\}|ENV\{ID_VENDOR_ID\}|ENV\{ID_MODEL_ID\}' + r'|SUBSYSTEMS?)\s*==\s*"([^"]*)"') + + +def parse_device(text): + """'0200:2000' -> (0x0200, 0x2000).""" + vendor, sep, product = text.strip().lower().partition(":") + try: + ids = int(vendor, 16), int(product, 16) + except ValueError: + ids = None + if not sep or ids is None or not all(0 <= i <= 0xFFFF for i in ids): + raise ValueError(f"expected VID:PID in hex, e.g. 0200:2000, not {text!r}") + return ids + + +def _path(root, *parts): + return os.path.join(root, *parts) + + +def read_attr(directory, name): + try: + with open(os.path.join(directory, name), errors="replace") as f: + return f.read().strip() + except OSError: + return None + + +def read_bytes(path): + try: + with open(path, "rb") as f: + return f.read() + except OSError: + return None + + +def hexdump(data, width=16): + return "\n".join(data[i:i + width].hex(" ") for i in range(0, len(data), width)) + + +# -- finding devices -------------------------------------------------------------- +def find_devices(root, vid, pid=None): + """sysfs directories of the USB devices with this vendor (and product).""" + base = _path(root, "sys", "bus", "usb", "devices") + try: + names = sorted(os.listdir(base)) + except OSError: + return [] + found = [] + for name in names: + if ":" in name: # an interface, not a device + continue + directory = os.path.realpath(os.path.join(base, name)) + try: + ids = int(read_attr(directory, "idVendor") or "", 16), \ + int(read_attr(directory, "idProduct") or "", 16) + except ValueError: + continue + if ids[0] == vid and (pid is None or ids[1] == pid): + found.append(directory) + return found + + +def interfaces(directory): + """Each interface of the device: class, driver, endpoints and HID children.""" + name = os.path.basename(directory) + result = [] + for entry in sorted(os.listdir(directory)): + if not entry.startswith(name + ":"): + continue + path = os.path.join(directory, entry) + driver = os.path.realpath(os.path.join(path, "driver")) + endpoints = [] + for ep in sorted(os.listdir(path)): + if ep.startswith("ep_"): + endpoints.append({k: read_attr(os.path.join(path, ep), k) for k in ( + "bEndpointAddress", "direction", "type", "wMaxPacketSize", + "bInterval", "interval")}) + hid = [] + for child in sorted(os.listdir(path)): + if HID_DEVICE.match(child): + hid_dir = os.path.join(path, child) + try: + nodes = sorted(os.listdir(os.path.join(hid_dir, "hidraw"))) + except OSError: + nodes = [] + hid.append({"report_descriptor": + read_bytes(os.path.join(hid_dir, "report_descriptor")), + "hidraw": nodes}) + result.append({ + "number": read_attr(path, "bInterfaceNumber"), + "class": read_attr(path, "bInterfaceClass"), + "subclass": read_attr(path, "bInterfaceSubClass"), + "protocol": read_attr(path, "bInterfaceProtocol"), + "driver": os.path.basename(driver) if os.path.lexists( + os.path.join(path, "driver")) else None, + "endpoints": endpoints, + "hid": hid, + }) + return result + + +def hid_descriptors(raw): + """{interface number: (bcdHID, report descriptor length)} from the raw + configuration descriptors, which sysfs keeps even with no driver bound.""" + found = {} + interface = None + i = 0 + while raw and i + 2 <= len(raw): + length, kind = raw[i], raw[i + 1] + if length < 2: + break + if kind == 0x04 and length >= 3: + interface = raw[i + 2] + elif kind == 0x21 and length >= 9 and interface is not None: + bcd = raw[i + 3] << 8 | raw[i + 2] + for j in range(i + 6, i + length - 2, 3): + if raw[j] == 0x22: + found[interface] = (f"{bcd >> 8:x}.{bcd & 0xFF:02x}", + raw[j + 1] | raw[j + 2] << 8) + i += length + return found + + +# -- permissions, udev, quirks ---------------------------------------------------------- +def _account(ident, lookup, mine, users): + """A system account by name; a person's account only as "you" or anonymous.""" + if ident == mine and ident != 0: + return "you" if users else "your group" + if ident < 1000: + try: + return lookup(ident)[0] + except KeyError: + return str(ident) + return "another user" if users else "a user group" + + +def node_access(path): + """'crw-rw---- root:plugdev, ACL, you: read+write', or None if absent.""" + try: + st = os.stat(path) + except OSError: + return None + owner = _account(st.st_uid, pwd.getpwuid, os.getuid(), True) + group = _account(st.st_gid, grp.getgrgid, os.getgid(), False) + try: + acl = ", ACL" if "system.posix_acl_access" in os.listxattr(path) else "" + except OSError: + acl = "" + if os.access(path, os.R_OK | os.W_OK): + you = "read+write" + elif os.access(path, os.R_OK): + you = "read only" + else: + you = "no access" + return f"{stat.filemode(st.st_mode)} {owner}:{group}{acl}, you: {you}" + + +def usbhid_quirks(root): + """[(vid, pid, flags)] the running usbhid was loaded with, or None if unknown.""" + text = read_attr(_path(root, "sys", "module", "usbhid", "parameters"), "quirks") + if text is None: + return None + quirks = [] + for item in text.split(","): + match = QUIRK.match(item.strip()) + if match: + quirks.append(tuple(int(g, 16) for g in match.groups())) + return quirks + + +def udev_rule_files(root): + """The rules files udev uses: the first of each name, /etc before /usr/lib.""" + chosen = {} + for directory in UDEV_DIRS: + try: + names = os.listdir(_path(root, directory)) + except OSError: + continue + for name in names: + if name.endswith(".rules") and name not in chosen: + chosen[name] = _path(root, directory, name) + return [chosen[name] for name in sorted(chosen)] + + +def _matches(pattern, value): + return any(fnmatch.fnmatchcase(value, alt) for alt in pattern.split("|")) + + +def rule_matches(text, vid, pid): + """Subsystems of the rules in this file that name this vendor and product + (a rule naming only the vendor covers every product).""" + found = [] + for line in text.replace("\\\n", " ").splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + keys = {"vendor": [], "product": [], "subsystem": []} + for key, value in RULE_MATCH.findall(line): + if "Vendor" in key or "VENDOR" in key: + keys["vendor"].append(value) + elif "SUBSYSTEM" in key: + keys["subsystem"].append(value) + else: + keys["product"].append(value) + if not keys["vendor"]: + continue + if all(_matches(p, f"{vid:04x}") for p in keys["vendor"]) and \ + all(_matches(p, f"{pid:04x}") for p in keys["product"]): + subsystem = "/".join(keys["subsystem"]) or "any" + if subsystem not in found: + found.append(subsystem) + return found + + +def udev_coverage(root, vid, pid): + covered = [] + for path in udev_rule_files(root): + try: + with open(path, errors="replace") as f: + subsystems = rule_matches(f.read(), vid, pid) + except OSError: + continue + if subsystems: + shown = "/" + os.path.relpath(path, root) + covered.append(f"`{shown}` ({', '.join(subsystems)})") + return covered + + +# -- the report --------------------------------------------------------------------------- +def distribution(root): + for rel in (("etc", "os-release"), ("usr", "lib", "os-release")): + try: + with open(_path(root, *rel)) as f: + for line in f: + if line.startswith("PRETTY_NAME="): + return line.partition("=")[2].strip().strip('"') + except OSError: + continue + return "unknown" + + +def table(rows): + lines = ["| | |", "|---|---|"] + lines += [f"| {key} | {value} |" for key, value in rows] + return "\n".join(lines) + + +def environment_rows(root, quirks): + experimental = os.environ.get(protocol.EXPERIMENTAL_ENV, "").strip() + return [ + ("ragnaros", __version__), + ("Python", platform.python_version()), + ("kernel", platform.release()), + ("distribution", distribution(root)), + ("libusb-1.0", ctypes.util.find_library("usb-1.0") or "not found"), + ("usbhid quirks", "unknown (usbhid not loaded?)" if quirks is None else + ", ".join(f"`0x{v:04x}:0x{p:04x}:0x{f:08x}`" for v, p, f in quirks) or "none"), + (f"`{protocol.EXPERIMENTAL_ENV}`", f"`{experimental}`" if experimental else "not set"), + ] + + +def _endpoint_row(ep): + address = ep["bEndpointAddress"] or "?" + size = ep["wMaxPacketSize"] + try: + size = str(int(size, 16) & 0x7FF) + except (TypeError, ValueError): + size = "?" + return (f"0x{address} {(ep['direction'] or '?').upper()}", ep["type"] or "?", size, + f"{ep['interval'] or '?'} (bInterval {ep['bInterval'] or '?'})") + + +def device_section(root, directory, quirks): + """(markdown, facts) for one device; facts feed the hints.""" + vid = int(read_attr(directory, "idVendor"), 16) + pid = int(read_attr(directory, "idProduct"), 16) + ident = f"{vid:04x}:{pid:04x}" + product = read_attr(directory, "product") + busnum, devnum = read_attr(directory, "busnum"), read_attr(directory, "devnum") + try: + node = f"/dev/bus/usb/{int(busnum):03d}/{int(devnum):03d}" + except (TypeError, ValueError): + node = None + access = node_access(_path(root, node.lstrip("/"))) if node else None + raw = read_bytes(os.path.join(directory, "descriptors")) or b"" + lengths = hid_descriptors(raw) + intfs = interfaces(directory) + experimental = os.environ.get(protocol.EXPERIMENTAL_ENV, "").strip() + try: + tried = vid == protocol.VID and experimental and protocol.parse_pid(experimental) == pid + except ValueError: + tried = False + + if (vid, pid) in SUPPORTED: + status = f"Supported ({SUPPORTED[(vid, pid)]})." + elif tried: + status = (f"Not supported yet; `{protocol.EXPERIMENTAL_ENV}` opts it into " + "the experimental test.") + else: + status = "Not supported yet." + quirk = [f for v, p, f in (quirks or []) if (v, p) == (vid, pid)] + rules = udev_coverage(root, vid, pid) + speed = read_attr(directory, "speed") + rows = [ + ("manufacturer", read_attr(directory, "manufacturer") or "none"), + ("product", product or "none"), + # never read: only whether the device has one + ("serial number", "present, not shown" + if os.path.exists(os.path.join(directory, "serial")) else "none"), + ("USB", f"{read_attr(directory, 'version') or '?'}, {speed or '?'} Mbit/s, " + f"bcdDevice {read_attr(directory, 'bcdDevice') or '?'}, " + f"class {read_attr(directory, 'bDeviceClass') or '?'}, " + f"{read_attr(directory, 'bNumConfigurations') or '?'} configuration(s), " + f"{(read_attr(directory, 'bNumInterfaces') or '?').strip()} interface(s), " + f"{read_attr(directory, 'bMaxPower') or '?'}"), + ("sysfs", f"`{os.path.basename(directory)}`"), + ("device node", f"`{node}` {access}" if access else f"`{node}` (missing)"), + ("usbhid quirk", ", ".join( + f"`0x{f:08x}` ({QUIRK_NAMES.get(f, 'flags')})" for f in quirk) or "none"), + ("udev rules", ", ".join(rules) or "none match"), + ] + out = [f"#### {ident} {product or ''}".rstrip(), "", status, "", table(rows), ""] + + out += ["| interface | class | driver | endpoint | type | max packet | interval |", + "|---|---|---|---|---|---|---|"] + for intf in intfs: + klass = intf["class"] or "?" + label = CLASSES.get(klass, "") + boot = BOOT_PROTOCOLS.get((intf["subclass"], intf["protocol"])) + label = f"{label}, {boot}" if boot and klass == "03" else label + head = [str(int(intf["number"], 16)) if intf["number"] else "?", + f"{klass}/{intf['subclass'] or '?'}/{intf['protocol'] or '?'}" + + (f" {label}" if label else ""), + intf["driver"] or "none"] + eps = intf["endpoints"] or [None] + for n, ep in enumerate(eps): + cells = (head if n == 0 else ["", "", ""]) + \ + (list(_endpoint_row(ep)) if ep else ["none", "", "", ""]) + out.append("| " + " | ".join(cells) + " |") + out.append("") + + for intf in intfs: + number = int(intf["number"], 16) if intf["number"] else None + declared = lengths.get(number) + size = f" ({declared[1]} bytes declared, HID {declared[0]})" if declared else "" + descriptors = [h for h in intf["hid"] if h["report_descriptor"]] + if not descriptors: + if intf["class"] == "03": + why = (f"no HID driver bound (driver: {intf['driver'] or 'none'})" + if not intf["hid"] else "not readable") + out += [f"HID report descriptor, interface {number}{size}: {why}.", ""] + continue + for hid in descriptors: + nodes = [] + for name in hid["hidraw"]: + hid_access = node_access(_path(root, "dev", name)) + nodes.append(f"`/dev/{name}` {hid_access}" if hid_access + else f"`/dev/{name}` (missing)") + out += [f"HID report descriptor, interface {number}{size}" + + (f", {', '.join(nodes)}" if nodes else ""), "", + "```", hexdump(hid["report_descriptor"]), "```", ""] + + if raw: + out += ["
Raw USB descriptors " + f"({len(raw)} bytes, as lsusb -v decodes them)", "", + "```", hexdump(raw), "```", "", "
", ""] + facts = {"id": ident, "vid": vid, "pid": pid, "node": node, + "writable": bool(access and "read+write" in access), + "supported": (vid, pid) in SUPPORTED} + return "\n".join(out), facts + + +def hints(facts): + """What to do next; printed on stderr so `> report.md` keeps only the report.""" + lines = [] + for fact in facts: + if fact["supported"]: + if not fact["writable"]: + lines.append(f"{fact['id']}: you cannot open {fact['node']}; install the udev " + "rules (`ragnaros install-system` or the .deb/AUR package) and " + "reboot once.") + continue + if fact["vid"] != protocol.VID: + continue + lines.append(f"{fact['id']} is not supported yet. To try it with the SS552 protocol " + "(experimental; README, \"Reporting a new device\"):") + steps = [] + if not fact["writable"] and fact["node"]: + steps.append((f"sudo chown \"$USER\" {fact['node']}", "until you unplug it")) + steps += [("systemctl --user stop ragnarosd", "if it runs"), + (f"ragnarosd --experimental-pid 0x{fact['pid']:04x}", "stays in the foreground"), + ("ragnarosctl watch --grab", "in a second terminal; press every key")] + lines += [f" {command:<40} # {note}" for command, note in steps] + return lines + + +def report(root=None, device=None): + """(markdown, hints, found): the report for vendor 0200, or one VID:PID.""" + root = ROOT if root is None else root + vid, pid = device if device else (protocol.VID, None) + quirks = usbhid_quirks(root) + wanted = f"{vid:04x}:{pid:04x}" if pid is not None else f"vendor {vid:04x}" + out = ["### Ragnaros device report", "", table(environment_rows(root, quirks)), ""] + facts = [] + devices = find_devices(root, vid, pid) + for directory in devices: + text, fact = device_section(root, directory, quirks) + out.append(text) + facts.append(fact) + if not devices: + out += [f"No USB device found for {wanted}.", ""] + notes = [f"No USB device found for {wanted}: is it plugged in? `lsusb` lists them" + + ("." if device else "; --device VID:PID reports another vendor's.")] + else: + notes = hints(facts) + return "\n".join(out).rstrip() + "\n", notes, bool(devices) + + +def run(args): + text, notes, found = report(device=args.device) + sys.stdout.write(text) + sys.stdout.flush() + for line in notes: + print(line, file=sys.stderr) + return 0 if found else 1 diff --git a/ragnaros/protocol.py b/ragnaros/protocol.py index 5449c3d..db33f27 100644 --- a/ragnaros/protocol.py +++ b/ragnaros/protocol.py @@ -1,11 +1,17 @@ import ctypes import ctypes.util +import os +import sys import time from .inputmap import parse_input VID = 0x0200 PID = 0x3001 +# Opt-in only: another product ID under the same vendor, driven with the +# SS552 protocol so an owner of an untested deck can see what works. +EXPERIMENTAL_ENV = "RAGNAROS_EXPERIMENTAL_PID" +LIBUSB_ERROR_NOT_FOUND = -5 PACKET_SIZE = 1024 READ_SIZE = 512 @@ -23,7 +29,37 @@ CRT = (0x43, 0x52, 0x54) -def activate_input(): +def parse_pid(text): + """A product ID under vendor 0200, in hex: '0x2000', '2000' or '0200:2000'.""" + vendor, sep, product = str(text).strip().lower().rpartition(":") + try: + vid = int(vendor, 16) if sep else VID + pid = int(product, 16) + except ValueError: + raise ValueError(f"not a USB product ID: {text!r} (hex, e.g. 0x2000)") from None + if vid != VID: + raise ValueError(f"only vendor {VID:04x} devices can be tried, not {vid:04x}") + if not 0 < pid <= 0xFFFF: + raise ValueError(f"not a USB product ID: {text!r} (hex, e.g. 0x2000)") + return pid + + +def product_ids(experimental=None): + """Product IDs to open, in order: the SS552, then the untested one a + tester opted into (--experimental-pid, else $RAGNAROS_EXPERIMENTAL_PID).""" + if experimental is None: + experimental = os.environ.get(EXPERIMENTAL_ENV, "") + if not str(experimental).strip(): + return (PID,) + pid = parse_pid(experimental) + return (PID,) if pid == PID else (PID, pid) + + +def device_id(pid): + return f"{VID:04x}:{pid:04x}" + + +def activate_input(pids=(PID,)): library = ctypes.util.find_library("usb-1.0") if not library: raise RuntimeError("libusb-1.0 is required to activate Ragnaros input") @@ -67,12 +103,21 @@ def activate_input(): raise RuntimeError(f"libusb initialization failed ({result})") handle = None + pid = None detached = [] claimed = [] try: - handle = usb.libusb_open_device_with_vid_pid(context, VID, PID) + for pid in pids: + handle = usb.libusb_open_device_with_vid_pid(context, VID, pid) + if handle: + break if not handle: - raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") + if tuple(pids) == (PID,): + raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") + raise RuntimeError( + f"cannot open {' or '.join(device_id(p) for p in pids)}: not plugged in, " + "or no permission (`ragnaros diagnose` shows the device node)") + experimental = pid != PID for interface in (0, 1): if usb.libusb_kernel_driver_active(handle, interface) == 1: result = usb.libusb_detach_kernel_driver(handle, interface) @@ -81,6 +126,8 @@ def activate_input(): detached.append(interface) for interface in (0, 1): result = usb.libusb_claim_interface(handle, interface) + if result == LIBUSB_ERROR_NOT_FOUND and experimental and interface: + continue # an untested deck may have a single interface if result < 0: raise RuntimeError(f"cannot claim HID interface {interface} ({result})") claimed.append(interface) @@ -93,13 +140,18 @@ def activate_input(): usb.libusb_close(handle) usb.libusb_exit(context) raise - return usb, context, handle, detached, claimed + return usb, context, handle, detached, claimed, pid class Ragnaros: - def __init__(self): - self.usb, self.context, self.handle, self.detached, self.claimed = activate_input() - self.path = f"libusb:{VID:04x}:{PID:04x}" + pid = PID + experimental = False # an untested product ID, opened on request + + def __init__(self, pids=(PID,)): + (self.usb, self.context, self.handle, self.detached, self.claimed, + self.pid) = activate_input(pids) + self.experimental = self.pid != PID + self.path = f"libusb:{device_id(self.pid)}" self._initialized = False def close(self): @@ -254,4 +306,8 @@ def poll(self, timeout=10): data = self.transfer(0x82, bytes(READ_SIZE), timeout) if data is None: return None + if self.experimental: + # the raw report, so a tester can show what an unknown deck sends + print(f"ragnaros: experimental input ({len(data)} bytes): " + f"{data.rstrip(bytes(1))[:64].hex(' ')}", file=sys.stderr) return parse_input(data) diff --git a/tests/test_bundled_config.py b/tests/test_bundled_config.py index d7eeffd..cd0a248 100644 --- a/tests/test_bundled_config.py +++ b/tests/test_bundled_config.py @@ -373,7 +373,7 @@ def test_gif_tools_write_what_the_panels_show(): class RecordingDev: """Stands in for protocol.Ragnaros and keeps the last image per slot.""" - def __init__(self): + def __init__(self, pids=None): self.keys, self.strip = {}, {} def send_image(self, slot, data, strip=False): diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 00856e9..c9a9bd5 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -15,10 +15,11 @@ def close(self): def test_wait_for_device_retries_until_present(monkeypatch): - calls = {"n": 0} + calls = {"n": 0, "pids": []} - def flaky(): + def flaky(pids): calls["n"] += 1 + calls["pids"].append(pids) if calls["n"] < 3: raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") return FakeDev() @@ -27,11 +28,13 @@ def flaky(): monkeypatch.setattr(ragnarosd.time, "sleep", lambda s: None) deck = ragnarosd.Deck() assert calls["n"] == 3 + assert calls["pids"] == [(0x3001,)] * 3 # only the SS552 unless asked assert isinstance(deck.dev, FakeDev) + assert deck.experimental_id is None deck.dev.close() def test_wait_for_device_exits_without_libusb(monkeypatch): - def broken(): + def broken(pids): raise RuntimeError("libusb-1.0 is required to activate Ragnaros input") monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", broken) @@ -43,8 +46,69 @@ def broken(): raise AssertionError("expected SystemExit when libusb is missing") +def test_an_untested_device_is_marked_experimental(monkeypatch, capsys): + class UntestedDev(FakeDev): + pid = 0x2000 + experimental = True + + monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda pids: UntestedDev()) + deck = ragnarosd.Deck((0x3001, 0x2000)) + assert deck.experimental_id == "0200:2000" + assert "EXPERIMENTAL: opened untested device 0200:2000" in capsys.readouterr().err + + +@pytest.fixture +def decks_built(monkeypatch): + """main() up to the point it builds the Deck: which product IDs it got.""" + built = [] + + class StubDeck: + def __init__(self, pids): + built.append(pids) + + def run(self): + pass + + monkeypatch.setattr(ragnarosd, "Deck", StubDeck) + monkeypatch.delenv(ragnarosd.protocol.EXPERIMENTAL_ENV, raising=False) + return built + + +def test_main_opens_only_the_ss552_by_default(decks_built, capsys): + assert ragnarosd.main([]) == 0 + assert decks_built == [(0x3001,)] + assert "EXPERIMENTAL" not in capsys.readouterr().err + + +@pytest.mark.parametrize("argv", [["--experimental-pid", "0x2000"], + ["--experimental-pid=2000"]]) +def test_main_takes_an_experimental_pid(decks_built, capsys, argv): + assert ragnarosd.main(argv) == 0 + assert decks_built == [(0x3001, 0x2000)] + assert "EXPERIMENTAL: also accepting 0200:2000" in capsys.readouterr().err + + +def test_main_reads_the_experimental_pid_from_the_environment(decks_built, monkeypatch): + monkeypatch.setenv(ragnarosd.protocol.EXPERIMENTAL_ENV, "0x2000") + assert ragnarosd.main([]) == 0 + assert decks_built == [(0x3001, 0x2000)] + + +@pytest.mark.parametrize("argv", [["--experimental-pid", "nope"], ["--experimental-pid"]]) +def test_main_refuses_a_bad_experimental_pid(decks_built, capsys, argv): + assert ragnarosd.main(argv) == 2 + assert decks_built == [] + assert "ragnarosd:" in capsys.readouterr().err + + +def test_help_documents_the_experimental_pid(capsys): + assert ragnarosd.main(["--help"]) == 0 + out = capsys.readouterr().out + assert "--experimental-pid" in out and "RAGNAROS_EXPERIMENTAL_PID" in out + + def test_reconnect_uses_wait_loop(monkeypatch): - monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda: FakeDev()) + monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda pids: FakeDev()) monkeypatch.setattr(ragnarosd.time, "sleep", lambda s: None) class BareDeck(ragnarosd.Deck): @@ -276,6 +340,10 @@ def test_control_status_and_errors(monkeypatch): deck = bare_deck(PROFILE) status = deck.handle_control(["status"]) assert status["ok"] and status["layer"] == "base" and status["strip"] == "keys" + assert "experimental" not in status + deck.experimental_id = "0200:2000" + assert deck.handle_control(["status"])["experimental"].startswith("0200:2000 (untested") + deck.experimental_id = None assert deck.handle_control([])["ok"] is False assert deck.handle_control(["nonsense"])["ok"] is False assert deck.handle_control(["profile", "ghost"])["ok"] is False diff --git a/tests/test_diagnose.py b/tests/test_diagnose.py new file mode 100644 index 0000000..e5756ed --- /dev/null +++ b/tests/test_diagnose.py @@ -0,0 +1,238 @@ +"""`ragnaros diagnose` against a fake sysfs: read-only, and nothing private.""" +import builtins +import getpass +import os +import platform +import socket + +import pytest + +from ragnaros import cli, diagnose, paths + +# The SS552's configuration descriptors as sysfs keeps them: device + config, +# a vendor HID interface (EP 0x82 in, 0x03 out) and a boot keyboard (EP 0x81). +SS552_DESCRIPTORS = bytes.fromhex(""" + 12 01 00 02 00 00 00 40 00 02 01 30 02 00 01 02 + 03 01 09 02 42 00 02 01 00 80 e1 09 04 00 00 02 + 03 00 00 00 09 21 00 02 00 01 22 36 00 07 05 82 + 03 00 02 01 07 05 03 03 00 04 01 09 04 01 00 01 + 03 01 01 00 09 21 00 02 00 01 22 3f 00 07 05 81 + 03 08 00 0a""") +REPORT_DESCRIPTOR = bytes.fromhex("06 00 ff 09 01 a1 01 09 02 15 00 26 ff 00 75 08 95 40 81 02 c0") +SERIAL = "SECRET-SERIAL-0042" + + +def write(path, content): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb" if isinstance(content, bytes) else "w") as f: + f.write(content) + + +def endpoint(interface, address, direction, size, b_interval, interval): + for key, value in {"bEndpointAddress": address, "direction": direction, + "type": "Interrupt", "wMaxPacketSize": size, + "bInterval": b_interval, "interval": interval}.items(): + write(interface / f"ep_{address}" / key, value + "\n") + + +def plug(root, name="5-1.4", vid="0200", pid="2000", busnum=5, devnum=2, driver="usbhid"): + """A deck in the fake sysfs; with usbhid bound it gets hidraw nodes.""" + dev = root / "sys" / "bus" / "usb" / "devices" / name + for key, value in {"idVendor": vid, "idProduct": pid, "manufacturer": "HOTSPOTEKUSB", + "product": "HOTSPOTEKUSB HID DEMO", "serial": SERIAL, + "busnum": str(busnum), "devnum": str(devnum), "speed": "480", + "version": " 2.00", "bcdDevice": "0002", "bDeviceClass": "00", + "bNumConfigurations": "1", "bNumInterfaces": " 2", + "bMaxPower": "450mA"}.items(): + write(dev / key, value + "\n") + raw = bytearray(SS552_DESCRIPTORS) + raw[10:12] = int(pid, 16).to_bytes(2, "little") + write(dev / "descriptors", bytes(raw)) + drivers = root / "sys" / "bus" / "usb" / "drivers" / driver + drivers.mkdir(parents=True, exist_ok=True) + for number, (subclass, protocol) in enumerate([("00", "00"), ("01", "01")]): + intf = dev / f"{name}:1.{number}" + for key, value in {"bInterfaceNumber": f"{number:02x}", "bInterfaceClass": "03", + "bInterfaceSubClass": subclass, + "bInterfaceProtocol": protocol}.items(): + write(intf / key, value + "\n") + os.symlink(drivers, intf / "driver") + if driver == "usbhid": + hid = intf / f"0003:{vid.upper()}:{pid.upper()}.000{number + 5}" + write(hid / "report_descriptor", REPORT_DESCRIPTOR) + (hid / "hidraw" / f"hidraw{number + 3}").mkdir(parents=True) + write(root / "dev" / f"hidraw{number + 3}", "") + endpoint(dev / f"{name}:1.0", "82", "in", "0200", "01", "125us") + endpoint(dev / f"{name}:1.0", "03", "out", "0400", "01", "125us") + endpoint(dev / f"{name}:1.1", "81", "in", "0008", "0a", "64ms") + write(root / "dev" / "bus" / "usb" / f"{busnum:03d}" / f"{devnum:03d}", "") + return dev + + +@pytest.fixture +def root(tmp_path, monkeypatch): + """A machine with the SS552's udev rule and quirk installed and no deck yet.""" + monkeypatch.delenv("RAGNAROS_EXPERIMENTAL_PID", raising=False) + with open(os.path.join(paths.bundled("udev"), "70-ragnaros.rules")) as f: + write(tmp_path / "etc" / "udev" / "rules.d" / "70-ragnaros.rules", f.read()) + write(tmp_path / "usr" / "lib" / "udev" / "rules.d" / "50-udev-default.rules", + 'SUBSYSTEM=="usb", ENV{DEVTYPE}=="usb_device", MODE="0664"\n') + write(tmp_path / "sys" / "module" / "usbhid" / "parameters" / "quirks", + "0x0200:0x3001:0x00000004,(null),(null),(null)\n") + write(tmp_path / "etc" / "os-release", 'NAME="Debian"\nPRETTY_NAME="Debian GNU/Linux 13"\n') + hub = tmp_path / "sys" / "bus" / "usb" / "devices" / "usb5" + write(hub / "idVendor", "1d6b\n") + write(hub / "idProduct", "0002\n") + return tmp_path + + +def test_an_untested_sibling_gets_everything_needed_to_add_it(root): + plug(root) + text, notes, found = diagnose.report(str(root)) + assert found + assert "#### 0200:2000 HOTSPOTEKUSB HID DEMO\n\nNot supported yet." in text + assert ("| 0 | 03/00/00 HID | usbhid | 0x03 OUT | Interrupt | 1024 | 125us (bInterval 01) |\n" + "| | | | 0x82 IN | Interrupt | 512 | 125us (bInterval 01) |") in text + assert "| 1 | 03/01/01 HID, boot keyboard | usbhid | 0x81 IN | Interrupt | 8 |" in text + assert ("HID report descriptor, interface 0 (54 bytes declared, HID 2.00), " + "`/dev/hidraw3` ") in text + assert "```\n06 00 ff 09 01 a1 01 09 02 15 00 26 ff 00 75 08\n95 40 81 02 c0\n```" in text + assert "| device node | `/dev/bus/usb/005/002` " in text + assert "| udev rules | none match |" in text # 70-ragnaros.rules is for 3001 only + assert "| usbhid quirk | none |" in text + assert "| distribution | Debian GNU/Linux 13 |" in text + assert "| usbhid quirks | `0x0200:0x3001:0x00000004` |" in text + assert "12 01 00 02 00 00 00 40 00 02 00 20 02 00 01 02" in text # raw descriptors + assert "1d6b" not in text # only vendor 0200 by default + assert any("ragnarosd --experimental-pid 0x2000" in line for line in notes) + + +def test_the_report_leaves_out_serial_host_and_user(root, monkeypatch): + plug(root) + monkeypatch.setattr(platform, "node", lambda: "secret-hostname") + monkeypatch.setattr(socket, "gethostname", lambda: "secret-hostname") + text, notes, _ = diagnose.report(str(root)) + assert "| serial number | present, not shown |" in text + for private in (SERIAL, "secret-hostname", str(root)): + assert private not in text and not any(private in line for line in notes) + if os.getuid() != 0: + assert getpass.getuser() not in text + + +def test_accounts_are_named_only_when_they_are_system_ones(): + names = {0: "root", 46: "plugdev", 1000: "alice", 1001: "bob"} + + def lookup(ident): + return (names[ident],) + + assert diagnose._account(0, lookup, 1000, True) == "root" + assert diagnose._account(46, lookup, 1000, False) == "plugdev" + assert diagnose._account(1000, lookup, 1000, True) == "you" + assert diagnose._account(1001, lookup, 1000, True) == "another user" + assert diagnose._account(1001, lookup, 1000, False) == "a user group" + + +def test_the_ss552_without_a_hid_driver(root): + plug(root, pid="3001", devnum=3, driver="usbfs") + text, notes, found = diagnose.report(str(root)) + assert found and notes == [] # writable, supported: nothing to do + assert "Supported (Redragon SS552)." in text + assert ("HID report descriptor, interface 0 (54 bytes declared, HID 2.00): " + "no HID driver bound (driver: usbfs).") in text + assert "| usbhid quirk | `0x00000004` (IGNORE) |" in text + assert "| udev rules | `/etc/udev/rules.d/70-ragnaros.rules` (hidraw, usb) |" in text + + +def test_device_picks_one_vid_pid(root): + plug(root) + plug(root, name="3-2", vid="046d", pid="c08c", busnum=3, devnum=2) + text, notes, found = diagnose.report(str(root), (0x046D, 0xC08C)) + assert found and "#### 046d:c08c" in text and "0200:2000" not in text + assert notes == [] # the experimental test is for vendor 0200 only + text, notes, found = diagnose.report(str(root), (0x0200, 0x1234)) + assert not found and "No USB device found for 0200:1234." in text + + +def test_the_hint_grants_access_when_the_node_is_locked(root): + if os.getuid() == 0: + pytest.skip("root can open anything") + plug(root) + os.chmod(root / "dev" / "bus" / "usb" / "005" / "002", 0o444) + text, notes, _ = diagnose.report(str(root)) + assert "you: read only" in text + assert any(line.startswith(' sudo chown "$USER" /dev/bus/usb/005/002 ') + and line.endswith("# until you unplug it") for line in notes) + + +def test_an_opted_in_device_says_so(root, monkeypatch): + plug(root) + monkeypatch.setenv("RAGNAROS_EXPERIMENTAL_PID", "0x2000") + text, _, _ = diagnose.report(str(root)) + assert "| `RAGNAROS_EXPERIMENTAL_PID` | `0x2000` |" in text + assert "opts it into the experimental test" in text + + +def test_it_never_writes_or_opens_a_device_node(root, monkeypatch): + plug(root) + opened = [] + real_open = builtins.open + + def guarded(path, mode="r", *args, **kwargs): + opened.append((str(path), mode)) + return real_open(path, mode, *args, **kwargs) + + monkeypatch.setattr(diagnose, "open", guarded, raising=False) + diagnose.report(str(root)) + assert opened and all(mode in ("r", "rb") for _, mode in opened) + assert not [p for p, _ in opened if os.path.relpath(p, root).startswith("dev")] + assert not [p for p, _ in opened if p.endswith("serial")] + + +def test_hid_descriptor_lengths_come_from_the_raw_descriptors(): + assert diagnose.hid_descriptors(SS552_DESCRIPTORS) == {0: ("2.00", 54), 1: ("2.00", 63)} + assert diagnose.hid_descriptors(b"") == {} + assert diagnose.hid_descriptors(b"\x00\x00") == {} + + +@pytest.mark.parametrize("rule,expected", [ + ('SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="3001", MODE="0660"', []), + ('SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="3001|2000"', ["usb"]), + ('SUBSYSTEM=="hidraw", ATTRS{idVendor}=="0200", ATTRS{idProduct}=="2*"', ["hidraw"]), + ('ATTRS{idVendor}=="0200", TAG+="uaccess"', ["any"]), + ('ENV{ID_VENDOR_ID}=="0200", ENV{ID_MODEL_ID}=="2000"', ["any"]), + ('# SUBSYSTEM=="usb", ATTR{idVendor}=="0200"', []), + ('SUBSYSTEM=="usb", ATTR{idVendor}=="046d"', []), + ('SUBSYSTEM=="usb", \\\n ATTR{idVendor}=="0200", ATTR{idProduct}=="2000"', ["usb"]), +]) +def test_udev_rule_matching(rule, expected): + assert diagnose.rule_matches(rule, 0x0200, 0x2000) == expected + + +def test_etc_rules_override_packaged_ones(root): + write(root / "usr" / "lib" / "udev" / "rules.d" / "70-ragnaros.rules", + 'SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="2000"\n') + assert diagnose.udev_coverage(str(root), 0x0200, 0x2000) == [] + os.remove(root / "etc" / "udev" / "rules.d" / "70-ragnaros.rules") + assert diagnose.udev_coverage(str(root), 0x0200, 0x2000) == [ + "`/usr/lib/udev/rules.d/70-ragnaros.rules` (usb)"] + + +def test_cli_diagnose_report_on_stdout_hints_on_stderr(root, monkeypatch, capsys): + plug(root) + monkeypatch.setattr(diagnose, "ROOT", str(root)) + assert cli.main(["diagnose"]) == 0 + out, err = capsys.readouterr() + assert out.startswith("### Ragnaros device report\n") + assert "ragnarosd --experimental-pid" not in out + assert "ragnarosd --experimental-pid 0x2000" in err + assert cli.main(["diagnose", "--device", "0200:1234"]) == 1 + assert "No USB device found for 0200:1234" in capsys.readouterr().err + with pytest.raises(SystemExit): + cli.main(["diagnose", "--device", "nope"]) + + +def test_parse_device(): + assert diagnose.parse_device("0200:2000") == (0x0200, 0x2000) + for bad in ("0200", "0200:", "zz:2000", "0200:12345"): + with pytest.raises(ValueError): + diagnose.parse_device(bad) diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 0978620..8852a00 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,3 +1,7 @@ +import types + +import pytest + from ragnaros import protocol @@ -81,3 +85,104 @@ def test_send_image_rejects_unknown_key(): pass else: raise AssertionError("expected ValueError for unknown key") + + +# -- opt-in untested product IDs ------------------------------------------------ +def test_parse_pid_accepts_the_usual_spellings(): + assert protocol.parse_pid("0x2000") == 0x2000 + assert protocol.parse_pid("2000") == 0x2000 + assert protocol.parse_pid(" 0200:2000 ") == 0x2000 + for bad in ("", "zz", "0x10000", "0", "046d:c08c"): + with pytest.raises(ValueError): + protocol.parse_pid(bad) + + +def test_only_the_ss552_unless_a_tester_opts_in(monkeypatch): + monkeypatch.delenv(protocol.EXPERIMENTAL_ENV, raising=False) + assert protocol.product_ids() == (0x3001,) + monkeypatch.setenv(protocol.EXPERIMENTAL_ENV, "0x2000") + assert protocol.product_ids() == (0x3001, 0x2000) + assert protocol.product_ids("0x1000") == (0x3001, 0x1000) # the flag wins + assert protocol.product_ids("3001") == (0x3001,) + monkeypatch.setenv(protocol.EXPERIMENTAL_ENV, " ") + assert protocol.product_ids() == (0x3001,) + + +def fake_libusb(monkeypatch, present, interfaces=(0, 1)): + """Enough of libusb-1.0 for activate_input(): which PIDs are plugged in and + which interfaces they have. Records the calls that touch a device.""" + calls = [] + + def open_device(context, vid, pid): + calls.append(("open", vid, pid)) + return 1 if pid in present else None + + def claim(handle, interface): + if interface not in interfaces: + return protocol.LIBUSB_ERROR_NOT_FOUND + calls.append(("claim", interface)) + return 0 + + usb = types.SimpleNamespace( + libusb_init=lambda context: 0, + libusb_open_device_with_vid_pid=open_device, + libusb_kernel_driver_active=lambda handle, interface: 0, + libusb_detach_kernel_driver=lambda handle, interface: 0, + libusb_claim_interface=claim, + libusb_release_interface=lambda handle, interface: calls.append(("release", interface)), + libusb_attach_kernel_driver=lambda handle, interface: 0, + libusb_control_transfer=lambda *args: 0, + libusb_interrupt_transfer=lambda *args: 0, + libusb_close=lambda handle: calls.append(("close",)), + libusb_exit=lambda context: None, + ) + monkeypatch.setattr(protocol.ctypes.util, "find_library", lambda name: "libusb-1.0.so.0") + monkeypatch.setattr(protocol.ctypes, "CDLL", lambda name: usb) + return calls + + +def test_the_default_opens_only_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x3001, 0x2000}) + dev = protocol.Ragnaros() + assert [c for c in calls if c[0] == "open"] == [("open", 0x0200, 0x3001)] + assert (dev.pid, dev.experimental, dev.path) == (0x3001, False, "libusb:0200:3001") + + +def test_the_default_error_is_unchanged(monkeypatch): + fake_libusb(monkeypatch, present={0x2000}) + with pytest.raises(RuntimeError, match="cannot open Ragnaros USB device"): + protocol.Ragnaros() + + +def test_an_opted_in_pid_is_tried_after_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x2000}, interfaces=(0,)) + dev = protocol.Ragnaros((0x3001, 0x2000)) + assert [c for c in calls if c[0] == "open"] == [("open", 0x0200, 0x3001), + ("open", 0x0200, 0x2000)] + assert (dev.pid, dev.experimental, dev.path) == (0x2000, True, "libusb:0200:2000") + assert dev.claimed == [0] # a single-interface deck is still worth a try + + +def test_a_missing_interface_still_fails_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x3001}, interfaces=(0,)) + with pytest.raises(RuntimeError, match="cannot claim HID interface 1"): + protocol.Ragnaros((0x3001, 0x2000)) + assert ("release", 0) in calls and ("close",) in calls + + +def test_nothing_to_open_names_both_ids(monkeypatch): + fake_libusb(monkeypatch, present=set()) + with pytest.raises(RuntimeError, match="0200:3001 or 0200:2000.*ragnaros diagnose"): + protocol.Ragnaros((0x3001, 0x2000)) + + +def test_experimental_input_is_logged_raw(capsys): + report = b"ACK\x00\x00OK\x00\x00\x0b\x01" + bytes(501) + dev = make_dev() + dev.transfer = lambda endpoint, payload, timeout: report + assert dev.poll() == ("unknown", 0x0B, 1) + assert capsys.readouterr().err == "" + dev.experimental = True + dev.poll() + err = capsys.readouterr().err + assert "experimental input (512 bytes): 41 43 4b 00 00 4f 4b 00 00 0b 01\n" in err