From 1552d73412917a227a1ef5067890765e9ff492f3 Mon Sep 17 00:00:00 2001 From: Sean Doran Date: Fri, 27 Feb 2026 18:40:13 -0500 Subject: [PATCH] Refactor: extract audio.py, add 50 unit tests - Extract MPRISController and PulseController into src/turnup/audio.py; turnupd.py imports from there, dropping ~270 lines of duplicated code - Add [project.optional-dependencies] dev = [pytest] and pytest testpaths to pyproject.toml - Add tests/test_audio.py: 31 tests covering MPRISController player matching, volume clamping, cache TTL, and PulseController MPRIS-first routing, PA fallback, and drain_events - Add tests/test_protocol.py: 19 tests covering parse_messages framing, multi-message buffers, garbage skipping, and knob conversion helpers - All 50 tests pass (pytest 9.0, Python 3.14) --- pyproject.toml | 6 + src/turnup/audio.py | 302 +++++++++++++++++++++++++++++++++++++++++ src/turnup/turnupd.py | 273 +------------------------------------ tests/__init__.py | 0 tests/test_audio.py | 283 ++++++++++++++++++++++++++++++++++++++ tests/test_protocol.py | 128 +++++++++++++++++ 6 files changed, 720 insertions(+), 272 deletions(-) create mode 100644 src/turnup/audio.py create mode 100644 tests/__init__.py create mode 100644 tests/test_audio.py create mode 100644 tests/test_protocol.py diff --git a/pyproject.toml b/pyproject.toml index 5adc6ef..81b3e2b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,12 @@ dependencies = [ "tomli>=1.1; python_version < '3.11'", ] +[project.optional-dependencies] +dev = ["pytest>=8.0"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + [project.scripts] turnupd = "turnup.turnupd:main" diff --git a/src/turnup/audio.py b/src/turnup/audio.py new file mode 100644 index 0000000..4f615c7 --- /dev/null +++ b/src/turnup/audio.py @@ -0,0 +1,302 @@ +""" +audio.py — MPRIS2 and PulseAudio/PipeWire volume controllers for turnupd. + +Two classes are provided: + +* :class:`MPRISController` — reads and writes per-app volume via the MPRIS2 + D-Bus interface using ``playerctl``. Preferred for apps that support it + (Spotify, VLC, Cider, …) because it sets the app's *internal* slider rather + than just the PA stream, so the volume survives song transitions. + +* :class:`PulseController` — wraps ``pulsectl`` for sink, source, and app- + stream volume/mute. When an :class:`MPRISController` instance is supplied + it delegates app-volume calls to MPRIS first, falling back to the PA stream + for apps that have no MPRIS player (e.g. Brave/Chromium). + Also runs a background thread that queues PA sink-input events so the main + loop can trigger an immediate reapply instead of waiting for the 1-second + polling timer. +""" + +import logging +import queue +import subprocess +import threading +import time + +import pulsectl + +log = logging.getLogger("turnupd") + +# Imported by callers that need the ceiling constant. +VOLUME_MAX: float = 1.0 + + +# ── MPRIS2 controller ────────────────────────────────────────────────────────── + +class MPRISController: + """Uses *playerctl* to read/write per-app volume via the MPRIS2 D-Bus interface. + + Caches the player list for ``_CACHE_TTL`` seconds to avoid spawning a new + subprocess on every single call. + """ + + _CACHE_TTL: float = 3.0 # seconds between ``playerctl --list-all`` calls + + def __init__(self) -> None: + self._players: list[str] = [] + self._players_ts: float = 0.0 + self._lock = threading.Lock() + + # ── internal helpers ────────────────────────────────────────────────────── + + def _run(self, *args: str, timeout: float = 2.0) -> tuple[bool, str]: + """Run ``playerctl `` and return ``(success, stdout.strip())``.""" + try: + result = subprocess.run( + ["playerctl", *args], + capture_output=True, + text=True, + timeout=timeout, + ) + return result.returncode == 0, result.stdout.strip() + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + log.debug("playerctl call failed: %s", exc) + return False, "" + + def _refresh_players(self, *, force: bool = False) -> None: + """Refresh the cached player list if it has expired (or *force* is set).""" + now = time.monotonic() + if not force and (now - self._players_ts) < self._CACHE_TTL: + return + ok, out = self._run("--list-all") + with self._lock: + self._players = [p.strip() for p in out.splitlines() if p.strip()] if ok else [] + self._players_ts = now + + # ── public API ──────────────────────────────────────────────────────────── + + def find_player(self, app_name: str) -> str | None: + """Return the first cached player whose name contains *app_name* (case-insensitive).""" + self._refresh_players() + needle = app_name.lower() + with self._lock: + for player in self._players: + if needle in player.lower(): + return player + return None + + def get_volume(self, app_name: str) -> float | None: + """Return the MPRIS volume (0.0–1.0) for *app_name*, or ``None`` if unavailable.""" + player = self.find_player(app_name) + if player is None: + return None + ok, out = self._run("--player", player, "volume") + if not ok or not out: + return None + try: + return max(0.0, min(1.0, float(out))) + except ValueError: + return None + + def set_volume(self, app_name: str, volume: float) -> bool: + """Set the MPRIS volume for *app_name*. Returns ``True`` on success.""" + player = self.find_player(app_name) + if player is None: + return False + volume = max(0.0, min(1.0, volume)) + ok, _ = self._run("--player", player, "volume", f"{volume:.4f}") + if ok: + log.debug("MPRIS: %r volume → %.4f", player, volume) + return ok + + +# ── PulseAudio / PipeWire controller ────────────────────────────────────────── + +class PulseController: + """Thin wrapper around :class:`pulsectl.Pulse` for volume and mute control. + + When *mpris* is supplied, ``set_app_volume`` and ``get_app_volume_norm`` + will prefer the MPRIS2 path for any app that has a live playerctl player, + falling back to the PulseAudio stream only when MPRIS is unavailable. + + A background watcher thread (started by :meth:`start_watching`) listens + for PulseAudio sink-input events and pushes indices onto ``_event_q`` so + the main loop can trigger an immediate reapply for PA-only apps instead of + waiting for the 1-second timer. + """ + + def __init__(self, mpris: MPRISController | None = None) -> None: + self._pulse = pulsectl.Pulse("turnupd") + self._mpris = mpris + self._event_q: queue.Queue[int] = queue.Queue() + self._watcher_thread: threading.Thread | None = None + self._stop_event = threading.Event() + + def close(self) -> None: + self._stop_event.set() + self._pulse.close() + + # ── PA event watcher ────────────────────────────────────────────────────── + + def start_watching(self) -> None: + """Spawn the background PA event listener thread (idempotent).""" + if self._watcher_thread and self._watcher_thread.is_alive(): + return + self._stop_event.clear() + t = threading.Thread(target=self._event_loop, daemon=True, name="pa-watcher") + t.start() + self._watcher_thread = t + log.debug("PA watcher thread started") + + def _event_loop(self) -> None: + """Background thread: open a *separate* Pulse connection and listen for events.""" + try: + with pulsectl.Pulse("turnupd-watcher") as watch_pulse: + def _cb(ev: pulsectl.PulseEventInfo) -> None: # type: ignore[name-defined] + if ev.facility == "sink_input": + self._event_q.put(int(ev.index)) + raise pulsectl.PulseLoopStop + + watch_pulse.event_mask_set("sink_input") + watch_pulse.event_callback_set(_cb) + while not self._stop_event.is_set(): + try: + watch_pulse.event_listen(timeout=1.0) + except pulsectl.PulseLoopStop: + pass + except Exception as exc: + log.debug("PA event loop error: %s", exc) + time.sleep(0.5) + except Exception as exc: + log.warning("PA watcher thread exiting: %s", exc) + + def drain_events(self) -> bool: + """Drain all pending PA events. Returns ``True`` if any events were present.""" + had_events = False + while True: + try: + self._event_q.get_nowait() + had_events = True + except queue.Empty: + break + return had_events + + # ── Sink / source ───────────────────────────────────────────────────────── + + def set_sink_volume(self, sink_name: str, volume: float) -> None: + volume = max(0.0, min(VOLUME_MAX, volume)) + try: + if sink_name == "default": + info = self._pulse.server_info() + sink = self._pulse.get_sink_by_name(info.default_sink_name) + else: + sink = self._pulse.get_sink_by_name(sink_name) + self._pulse.volume_set_all_chans(sink, volume) + except Exception as exc: + log.warning("set_sink_volume(%r) failed: %s", sink_name, exc) + + def toggle_mute_sink(self, sink_name: str) -> None: + try: + if sink_name == "default": + info = self._pulse.server_info() + sink = self._pulse.get_sink_by_name(info.default_sink_name) + else: + sink = self._pulse.get_sink_by_name(sink_name) + self._pulse.mute(sink, not sink.mute) + log.info("Sink %r mute toggled", sink_name) + except Exception as exc: + log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc) + + def set_source_volume(self, source_name: str, volume: float) -> None: + volume = max(0.0, min(1.0, volume)) + try: + if source_name == "default": + info = self._pulse.server_info() + source = self._pulse.get_source_by_name(info.default_source_name) + else: + source = self._pulse.get_source_by_name(source_name) + self._pulse.volume_set_all_chans(source, volume) + except Exception as exc: + log.warning("set_source_volume(%r) failed: %s", source_name, exc) + + def toggle_mute_source(self, source_name: str) -> None: + try: + if source_name == "default": + info = self._pulse.server_info() + source = self._pulse.get_source_by_name(info.default_source_name) + else: + source = self._pulse.get_source_by_name(source_name) + self._pulse.mute(source, not source.mute) + log.info("Source %r mute toggled", source_name) + except Exception as exc: + log.warning("toggle_mute_source(%r) failed: %s", source_name, exc) + + # ── App volume (MPRIS-first, PA fallback) ───────────────────────────────── + + def set_app_volume(self, app_name: str, volume: float) -> None: + volume = max(0.0, min(VOLUME_MAX, volume)) + + # Prefer the MPRIS2 path — it writes to the app's internal slider so the + # volume survives song transitions (e.g. Spotify resetting on new tracks). + if self._mpris and self._mpris.set_volume(app_name, volume): + log.debug("MPRIS set_volume: %r = %.4f", app_name, volume) + return + + # Fall back to PulseAudio stream volume. + needle = app_name.lower() + found = False + try: + for inp in self._pulse.sink_input_list(): + name = inp.proplist.get("application.name", "") + binary = inp.proplist.get("application.process.binary", "") + if needle in name.lower() or needle in binary.lower(): + self._pulse.volume_set_all_chans(inp, volume) + found = True + if not found: + log.debug("App %r not found in sink inputs", app_name) + except Exception as exc: + log.warning("set_app_volume(%r) failed: %s", app_name, exc) + + def get_sink_volume_norm(self, sink_name: str) -> float | None: + """Return the current sink volume normalised to 0.0–1.0, or None on error.""" + try: + if sink_name == "default": + info = self._pulse.server_info() + sink = self._pulse.get_sink_by_name(info.default_sink_name) + else: + sink = self._pulse.get_sink_by_name(sink_name) + return min(1.0, sink.volume.value_flat / VOLUME_MAX) + except Exception: + return None + + def get_source_volume_norm(self, source_name: str) -> float | None: + """Return the current source volume normalised to 0.0–1.0, or None on error.""" + try: + if source_name == "default": + info = self._pulse.server_info() + source = self._pulse.get_source_by_name(info.default_source_name) + else: + source = self._pulse.get_source_by_name(source_name) + return min(1.0, source.volume.value_flat) + except Exception: + return None + + def get_app_volume_norm(self, app_name: str) -> float | None: + """Return the current app volume normalised to 0.0–1.0, or None if not found.""" + # Prefer MPRIS — more accurate for apps like Spotify. + if self._mpris: + vol = self._mpris.get_volume(app_name) + if vol is not None: + return vol + + # Fall back to PulseAudio stream. + needle = app_name.lower() + try: + for inp in self._pulse.sink_input_list(): + name = inp.proplist.get("application.name", "") + binary = inp.proplist.get("application.process.binary", "") + if needle in name.lower() or needle in binary.lower(): + return min(1.0, inp.volume.value_flat / VOLUME_MAX) + except Exception: + pass + return None diff --git a/src/turnup/turnupd.py b/src/turnup/turnupd.py index bcf5809..c83c2f4 100755 --- a/src/turnup/turnupd.py +++ b/src/turnup/turnupd.py @@ -14,16 +14,14 @@ even when no knob is being moved. import logging import os -import queue import signal import subprocess import sys -import threading import time -import pulsectl import serial +from turnup.audio import VOLUME_MAX, MPRISController, PulseController from turnup.config import DEFAULT_CONFIG_PATH, get_knob_led_cfg, get_led_color, load_config logging.basicConfig( @@ -35,8 +33,6 @@ log = logging.getLogger("turnupd") # Maximum raw ADC value reported by the hardware. KNOB_MAX: int = 1012 -# Maximum output volume multiplier (1.0 = 100 %). -VOLUME_MAX: float = 1.0 # Number of physical knobs (and therefore LED groups). NUM_KNOBS: int = 5 # Number of LEDs per knob. @@ -133,273 +129,6 @@ def all_led_colors( ] -# ── MPRIS2 controller (playerctl-backed) ────────────────────────────────────── - -class MPRISController: - """Uses *playerctl* to read/write per-app volume via the MPRIS2 D-Bus interface. - - Caches the player list for ``_CACHE_TTL`` seconds to avoid spawning a new - subprocess on every single call. - """ - - _CACHE_TTL: float = 3.0 # seconds between ``playerctl --list-all`` calls - - def __init__(self) -> None: - self._players: list[str] = [] - self._players_ts: float = 0.0 - self._lock = threading.Lock() - - # ── internal helpers ────────────────────────────────────────────────────── - - def _run(self, *args: str, timeout: float = 2.0) -> tuple[bool, str]: - """Run ``playerctl `` and return ``(success, stdout.strip())``.""" - try: - result = subprocess.run( - ["playerctl", *args], - capture_output=True, - text=True, - timeout=timeout, - ) - return result.returncode == 0, result.stdout.strip() - except (FileNotFoundError, subprocess.TimeoutExpired) as exc: - log.debug("playerctl call failed: %s", exc) - return False, "" - - def _refresh_players(self, *, force: bool = False) -> None: - """Refresh the cached player list if it has expired (or *force* is set).""" - now = time.monotonic() - if not force and (now - self._players_ts) < self._CACHE_TTL: - return - ok, out = self._run("--list-all") - with self._lock: - self._players = [p.strip() for p in out.splitlines() if p.strip()] if ok else [] - self._players_ts = now - - # ── public API ──────────────────────────────────────────────────────────── - - def find_player(self, app_name: str) -> str | None: - """Return the first cached player whose name contains *app_name* (case-insensitive).""" - self._refresh_players() - needle = app_name.lower() - with self._lock: - for player in self._players: - if needle in player.lower(): - return player - return None - - def get_volume(self, app_name: str) -> float | None: - """Return the MPRIS volume (0.0–1.0) for *app_name*, or ``None`` if unavailable.""" - player = self.find_player(app_name) - if player is None: - return None - ok, out = self._run("--player", player, "volume") - if not ok or not out: - return None - try: - return max(0.0, min(1.0, float(out))) - except ValueError: - return None - - def set_volume(self, app_name: str, volume: float) -> bool: - """Set the MPRIS volume for *app_name*. Returns ``True`` on success.""" - player = self.find_player(app_name) - if player is None: - return False - volume = max(0.0, min(1.0, volume)) - ok, _ = self._run("--player", player, "volume", f"{volume:.4f}") - if ok: - log.debug("MPRIS: %r volume → %.4f", player, volume) - return ok - - -# ── PulseAudio / PipeWire controller ────────────────────────────────────────── - -class PulseController: - """Thin wrapper around :class:`pulsectl.Pulse` for volume and mute control. - - When *mpris* is supplied, ``set_app_volume`` and ``get_app_volume_norm`` - will prefer the MPRIS2 path for any app that has a live playerctl player, - falling back to the PulseAudio stream only when MPRIS is unavailable. - - A background watcher thread (started by :meth:`start_watching`) listens - for PulseAudio sink-input events and pushes indices onto ``_event_q`` so - the main loop can trigger an immediate reapply for PA-only apps instead of - waiting for the 1-second timer. - """ - - def __init__(self, mpris: MPRISController | None = None) -> None: - self._pulse = pulsectl.Pulse("turnupd") - self._mpris = mpris - self._event_q: queue.Queue[int] = queue.Queue() - self._watcher_thread: threading.Thread | None = None - self._stop_event = threading.Event() - - def close(self) -> None: - self._stop_event.set() - self._pulse.close() - - # ── PA event watcher ────────────────────────────────────────────────────── - - def start_watching(self) -> None: - """Spawn the background PA event listener thread (idempotent).""" - if self._watcher_thread and self._watcher_thread.is_alive(): - return - self._stop_event.clear() - t = threading.Thread(target=self._event_loop, daemon=True, name="pa-watcher") - t.start() - self._watcher_thread = t - log.debug("PA watcher thread started") - - def _event_loop(self) -> None: - """Background thread: open a *separate* Pulse connection and listen for events.""" - try: - with pulsectl.Pulse("turnupd-watcher") as watch_pulse: - def _cb(ev: pulsectl.PulseEventInfo) -> None: # type: ignore[name-defined] - if ev.facility == "sink_input": - self._event_q.put(int(ev.index)) - raise pulsectl.PulseLoopStop - - watch_pulse.event_mask_set("sink_input") - watch_pulse.event_callback_set(_cb) - while not self._stop_event.is_set(): - try: - watch_pulse.event_listen(timeout=1.0) - except pulsectl.PulseLoopStop: - pass - except Exception as exc: - log.debug("PA event loop error: %s", exc) - time.sleep(0.5) - except Exception as exc: - log.warning("PA watcher thread exiting: %s", exc) - - def drain_events(self) -> bool: - """Drain all pending PA events. Returns ``True`` if any events were present.""" - had_events = False - while True: - try: - self._event_q.get_nowait() - had_events = True - except queue.Empty: - break - return had_events - - def set_sink_volume(self, sink_name: str, volume: float) -> None: - volume = max(0.0, min(VOLUME_MAX, volume)) - try: - if sink_name == "default": - info = self._pulse.server_info() - sink = self._pulse.get_sink_by_name(info.default_sink_name) - else: - sink = self._pulse.get_sink_by_name(sink_name) - self._pulse.volume_set_all_chans(sink, volume) - except Exception as exc: - log.warning("set_sink_volume(%r) failed: %s", sink_name, exc) - - def toggle_mute_sink(self, sink_name: str) -> None: - try: - if sink_name == "default": - info = self._pulse.server_info() - sink = self._pulse.get_sink_by_name(info.default_sink_name) - else: - sink = self._pulse.get_sink_by_name(sink_name) - self._pulse.mute(sink, not sink.mute) - log.info("Sink %r mute toggled", sink_name) - except Exception as exc: - log.warning("toggle_mute_sink(%r) failed: %s", sink_name, exc) - - def set_source_volume(self, source_name: str, volume: float) -> None: - volume = max(0.0, min(1.0, volume)) - try: - if source_name == "default": - info = self._pulse.server_info() - source = self._pulse.get_source_by_name(info.default_source_name) - else: - source = self._pulse.get_source_by_name(source_name) - self._pulse.volume_set_all_chans(source, volume) - except Exception as exc: - log.warning("set_source_volume(%r) failed: %s", source_name, exc) - - def toggle_mute_source(self, source_name: str) -> None: - try: - if source_name == "default": - info = self._pulse.server_info() - source = self._pulse.get_source_by_name(info.default_source_name) - else: - source = self._pulse.get_source_by_name(source_name) - self._pulse.mute(source, not source.mute) - log.info("Source %r mute toggled", source_name) - except Exception as exc: - log.warning("toggle_mute_source(%r) failed: %s", source_name, exc) - - def set_app_volume(self, app_name: str, volume: float) -> None: - volume = max(0.0, min(VOLUME_MAX, volume)) - - # Prefer the MPRIS2 path — it writes to the app's internal slider so the - # volume survives song transitions (e.g. Spotify resetting on new tracks). - if self._mpris and self._mpris.set_volume(app_name, volume): - log.debug("MPRIS set_volume: %r = %.4f", app_name, volume) - return - - # Fall back to PulseAudio stream volume. - needle = app_name.lower() - found = False - try: - for inp in self._pulse.sink_input_list(): - name = inp.proplist.get("application.name", "") - binary = inp.proplist.get("application.process.binary", "") - if needle in name.lower() or needle in binary.lower(): - self._pulse.volume_set_all_chans(inp, volume) - found = True - if not found: - log.debug("App %r not found in sink inputs", app_name) - except Exception as exc: - log.warning("set_app_volume(%r) failed: %s", app_name, exc) - - def get_sink_volume_norm(self, sink_name: str) -> float | None: - """Return the current sink volume normalised to 0.0–1.0, or None on error.""" - try: - if sink_name == "default": - info = self._pulse.server_info() - sink = self._pulse.get_sink_by_name(info.default_sink_name) - else: - sink = self._pulse.get_sink_by_name(sink_name) - return min(1.0, sink.volume.value_flat / VOLUME_MAX) - except Exception: - return None - - def get_source_volume_norm(self, source_name: str) -> float | None: - """Return the current source volume normalised to 0.0–1.0, or None on error.""" - try: - if source_name == "default": - info = self._pulse.server_info() - source = self._pulse.get_source_by_name(info.default_source_name) - else: - source = self._pulse.get_source_by_name(source_name) - return min(1.0, source.volume.value_flat) - except Exception: - return None - - def get_app_volume_norm(self, app_name: str) -> float | None: - """Return the current app volume normalised to 0.0–1.0, or None if not found.""" - # Prefer MPRIS — more accurate for apps like Spotify. - if self._mpris: - vol = self._mpris.get_volume(app_name) - if vol is not None: - return vol - - # Fall back to PulseAudio stream. - needle = app_name.lower() - try: - for inp in self._pulse.sink_input_list(): - name = inp.proplist.get("application.name", "") - binary = inp.proplist.get("application.process.binary", "") - if needle in name.lower() or needle in binary.lower(): - return min(1.0, inp.volume.value_flat / VOLUME_MAX) - except Exception: - pass - return None - - # ── Startup helpers ──────────────────────────────────────────────────────────── def init_knob_norms(config: dict, pulse: PulseController) -> list[float]: diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_audio.py b/tests/test_audio.py new file mode 100644 index 0000000..f1d73fb --- /dev/null +++ b/tests/test_audio.py @@ -0,0 +1,283 @@ +""" +Unit tests for turnup.audio — MPRISController and PulseController. + +All external dependencies (pulsectl, subprocess/playerctl) are mocked so +these tests run without a running PulseAudio/PipeWire server or playerctl. +""" + +from unittest.mock import MagicMock, patch, call +import pytest + +from turnup.audio import MPRISController, PulseController, VOLUME_MAX + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +def _make_sink_input(name: str, binary: str, volume: float) -> MagicMock: + """Return a fake pulsectl sink-input object.""" + inp = MagicMock() + inp.proplist = {"application.name": name, "application.process.binary": binary} + inp.volume.value_flat = volume + return inp + + +# ── MPRISController ─────────────────────────────────────────────────────────── + +class TestMPRISControllerFindPlayer: + def _make(self, players: list[str]) -> MPRISController: + ctrl = MPRISController() + # Pre-populate cache so _refresh_players is a no-op during tests. + ctrl._players = players + ctrl._players_ts = float("inf") # never expires + return ctrl + + def test_exact_match(self): + ctrl = self._make(["org.mpris.MediaPlayer2.spotify"]) + assert ctrl.find_player("spotify") == "org.mpris.MediaPlayer2.spotify" + + def test_case_insensitive(self): + ctrl = self._make(["org.mpris.MediaPlayer2.Spotify"]) + assert ctrl.find_player("SPOTIFY") == "org.mpris.MediaPlayer2.Spotify" + + def test_no_match_returns_none(self): + ctrl = self._make(["org.mpris.MediaPlayer2.vlc"]) + assert ctrl.find_player("brave") is None + + def test_empty_player_list_returns_none(self): + ctrl = self._make([]) + assert ctrl.find_player("spotify") is None + + def test_returns_first_match(self): + ctrl = self._make([ + "org.mpris.MediaPlayer2.spotify", + "org.mpris.MediaPlayer2.spotify.instance2", + ]) + assert ctrl.find_player("spotify") == "org.mpris.MediaPlayer2.spotify" + + +class TestMPRISControllerGetVolume: + def _make_with_player(self, player: str) -> MPRISController: + ctrl = MPRISController() + ctrl._players = [player] + ctrl._players_ts = float("inf") + return ctrl + + def test_returns_float_on_success(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "0.75")): + assert ctrl.get_volume("spotify") == pytest.approx(0.75) + + def test_clamps_above_one(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "1.5")): + assert ctrl.get_volume("spotify") == pytest.approx(1.0) + + def test_clamps_below_zero(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "-0.1")): + assert ctrl.get_volume("spotify") == pytest.approx(0.0) + + def test_returns_none_when_player_not_found(self): + ctrl = MPRISController() + ctrl._players = [] + ctrl._players_ts = float("inf") + assert ctrl.get_volume("spotify") is None + + def test_returns_none_on_playerctl_failure(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(False, "")): + assert ctrl.get_volume("spotify") is None + + def test_returns_none_on_invalid_output(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "not-a-number")): + assert ctrl.get_volume("spotify") is None + + +class TestMPRISControllerSetVolume: + def _make_with_player(self, player: str) -> MPRISController: + ctrl = MPRISController() + ctrl._players = [player] + ctrl._players_ts = float("inf") + return ctrl + + def test_returns_true_on_success(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run: + result = ctrl.set_volume("spotify", 0.5) + assert result is True + mock_run.assert_called_once_with( + "--player", "org.mpris.MediaPlayer2.spotify", "volume", "0.5000" + ) + + def test_clamps_volume_to_one(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run: + ctrl.set_volume("spotify", 1.5) + _, _, _, sent_vol = mock_run.call_args[0] + assert float(sent_vol) == pytest.approx(1.0) + + def test_clamps_volume_to_zero(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(True, "")) as mock_run: + ctrl.set_volume("spotify", -0.5) + _, _, _, sent_vol = mock_run.call_args[0] + assert float(sent_vol) == pytest.approx(0.0) + + def test_returns_false_when_no_player(self): + ctrl = MPRISController() + ctrl._players = [] + ctrl._players_ts = float("inf") + assert ctrl.set_volume("spotify", 0.5) is False + + def test_returns_false_on_playerctl_failure(self): + ctrl = self._make_with_player("org.mpris.MediaPlayer2.spotify") + with patch.object(ctrl, "_run", return_value=(False, "")): + assert ctrl.set_volume("spotify", 0.5) is False + + +class TestMPRISControllerRefreshPlayers: + def test_populates_cache_from_playerctl(self): + ctrl = MPRISController() + with patch.object(ctrl, "_run", return_value=(True, "spotify\nvlc\n")): + ctrl._refresh_players(force=True) + assert ctrl._players == ["spotify", "vlc"] + + def test_empty_cache_on_playerctl_failure(self): + ctrl = MPRISController() + ctrl._players = ["old-player"] + with patch.object(ctrl, "_run", return_value=(False, "")): + ctrl._refresh_players(force=True) + assert ctrl._players == [] + + def test_skips_refresh_when_cache_valid(self): + ctrl = MPRISController() + ctrl._players = ["spotify"] + ctrl._players_ts = float("inf") + with patch.object(ctrl, "_run") as mock_run: + ctrl._refresh_players() + mock_run.assert_not_called() + + +# ── PulseController ─────────────────────────────────────────────────────────── + +@pytest.fixture +def mock_pulse_lib(): + """Patch pulsectl.Pulse so PulseController never touches a real server.""" + with patch("turnup.audio.pulsectl.Pulse") as mock_cls: + mock_cls.return_value.__enter__ = lambda s: s + mock_cls.return_value.__exit__ = MagicMock(return_value=False) + yield mock_cls + + +class TestPulseControllerSetAppVolume: + def test_uses_mpris_when_available(self, mock_pulse_lib): + mpris = MagicMock(spec=MPRISController) + mpris.set_volume.return_value = True + + pulse = PulseController(mpris=mpris) + pulse.set_app_volume("spotify", 0.6) + + mpris.set_volume.assert_called_once_with("spotify", pytest.approx(0.6)) + # PA stream should NOT be touched. + pulse._pulse.sink_input_list.assert_not_called() + + def test_falls_back_to_pa_when_mpris_fails(self, mock_pulse_lib): + mpris = MagicMock(spec=MPRISController) + mpris.set_volume.return_value = False + + inp = _make_sink_input("Brave", "brave", 1.0) + pulse = PulseController(mpris=mpris) + pulse._pulse.sink_input_list.return_value = [inp] + + pulse.set_app_volume("brave", 0.4) + + pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.4)) + + def test_falls_back_to_pa_when_no_mpris(self, mock_pulse_lib): + inp = _make_sink_input("Brave", "brave", 1.0) + pulse = PulseController(mpris=None) + pulse._pulse.sink_input_list.return_value = [inp] + + pulse.set_app_volume("brave", 0.3) + + pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.3)) + + def test_clamps_volume_at_volume_max(self, mock_pulse_lib): + mpris = MagicMock(spec=MPRISController) + mpris.set_volume.return_value = True + + pulse = PulseController(mpris=mpris) + pulse.set_app_volume("spotify", 2.0) + + mpris.set_volume.assert_called_once_with("spotify", pytest.approx(VOLUME_MAX)) + + def test_matches_by_binary_name(self, mock_pulse_lib): + inp = _make_sink_input("", "spotify", 1.0) + pulse = PulseController(mpris=None) + pulse._pulse.sink_input_list.return_value = [inp] + + pulse.set_app_volume("spotify", 0.5) + + pulse._pulse.volume_set_all_chans.assert_called_once_with(inp, pytest.approx(0.5)) + + def test_updates_all_matching_streams(self, mock_pulse_lib): + """All streams for an app (e.g. Spotify crossfade) should be updated.""" + inp1 = _make_sink_input("Spotify", "spotify", 1.0) + inp2 = _make_sink_input("Spotify", "spotify", 1.0) + pulse = PulseController(mpris=None) + pulse._pulse.sink_input_list.return_value = [inp1, inp2] + + pulse.set_app_volume("spotify", 0.5) + + assert pulse._pulse.volume_set_all_chans.call_count == 2 + + +class TestPulseControllerGetAppVolumeNorm: + def test_prefers_mpris(self, mock_pulse_lib): + mpris = MagicMock(spec=MPRISController) + mpris.get_volume.return_value = 0.7 + + pulse = PulseController(mpris=mpris) + result = pulse.get_app_volume_norm("spotify") + + assert result == pytest.approx(0.7) + pulse._pulse.sink_input_list.assert_not_called() + + def test_falls_back_to_pa_when_mpris_returns_none(self, mock_pulse_lib): + mpris = MagicMock(spec=MPRISController) + mpris.get_volume.return_value = None + + inp = _make_sink_input("Brave", "brave", 0.55) + pulse = PulseController(mpris=mpris) + pulse._pulse.sink_input_list.return_value = [inp] + + result = pulse.get_app_volume_norm("brave") + + assert result == pytest.approx(0.55) + + def test_returns_none_when_app_not_found(self, mock_pulse_lib): + pulse = PulseController(mpris=None) + pulse._pulse.sink_input_list.return_value = [] + + assert pulse.get_app_volume_norm("nonexistent") is None + + def test_clamps_pa_volume_at_one(self, mock_pulse_lib): + inp = _make_sink_input("Spotify", "spotify", 1.5) + pulse = PulseController(mpris=None) + pulse._pulse.sink_input_list.return_value = [inp] + + assert pulse.get_app_volume_norm("spotify") == pytest.approx(1.0) + + +class TestPulseControllerDrainEvents: + def test_returns_false_when_empty(self, mock_pulse_lib): + pulse = PulseController() + assert pulse.drain_events() is False + + def test_returns_true_when_events_present(self, mock_pulse_lib): + pulse = PulseController() + pulse._event_q.put(1) + pulse._event_q.put(2) + assert pulse.drain_events() is True + assert pulse._event_q.empty() diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..e8e48b8 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,128 @@ +""" +Unit tests for turnupd protocol parser and knob conversion helpers. +No external dependencies — these are pure-function tests. +""" + +import pytest + +from turnup.turnupd import knob_to_norm, knob_to_volume, parse_messages, KNOB_MAX, VOLUME_MAX + + +# ── knob_to_norm ────────────────────────────────────────────────────────────── + +class TestKnobToNorm: + def test_min_value(self): + assert knob_to_norm(0) == pytest.approx(0.0) + + def test_max_value(self): + assert knob_to_norm(KNOB_MAX) == pytest.approx(1.0) + + def test_midpoint(self): + result = knob_to_norm(KNOB_MAX // 2) + assert 0.49 < result < 0.51 + + def test_result_is_rounded(self): + # Result must have at most 4 decimal places. + result = knob_to_norm(333) + assert result == round(result, 4) + + +# ── knob_to_volume ──────────────────────────────────────────────────────────── + +class TestKnobToVolume: + def test_min_is_zero(self): + assert knob_to_volume(0) == pytest.approx(0.0) + + def test_max_is_volume_max(self): + assert knob_to_volume(KNOB_MAX) == pytest.approx(VOLUME_MAX) + + def test_proportional(self): + half = knob_to_volume(KNOB_MAX // 2) + full = knob_to_volume(KNOB_MAX) + assert half == pytest.approx(full / 2, rel=0.01) + + +# ── parse_messages ──────────────────────────────────────────────────────────── + +class TestParseMessages: + # ── heartbeat ───────────────────────────────────────────────────────────── + + def test_heartbeat(self): + buf = bytearray([0xFE, 0x02, 0xFF]) + msgs, remainder = parse_messages(buf) + assert msgs == [{"type": "heartbeat"}] + assert remainder == bytearray() + + def test_heartbeat_leaves_trailing_bytes(self): + # Trailing non-0xFE bytes are consumed (no partial frame to preserve). + buf = bytearray([0xFE, 0x02, 0xFF, 0x01, 0x02]) + msgs, remainder = parse_messages(buf) + assert len(msgs) == 1 + assert remainder == bytearray() + + # ── button ──────────────────────────────────────────────────────────────── + + def test_button_press(self): + buf = bytearray([0xFE, 0x06, 0x03, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs == [{"type": "button", "action": "press", "id": 3}] + + def test_button_release(self): + buf = bytearray([0xFE, 0x07, 0x02, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs == [{"type": "button", "action": "release", "id": 2}] + + def test_all_button_ids(self): + for btn_id in range(5): + buf = bytearray([0xFE, 0x06, btn_id, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs[0]["id"] == btn_id + + # ── knob ────────────────────────────────────────────────────────────────── + + def test_knob_message(self): + # value = 0x03F4 = 1012 + buf = bytearray([0xFE, 0x03, 0x01, 0x03, 0xF4, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs == [{"type": "knob", "id": 1, "value": 1012}] + + def test_knob_value_zero(self): + buf = bytearray([0xFE, 0x03, 0x00, 0x00, 0x00, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs[0]["value"] == 0 + + def test_knob_value_max(self): + buf = bytearray([0xFE, 0x03, 0x04, 0x03, 0xF4, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs[0]["value"] == KNOB_MAX + + # ── multi-message ───────────────────────────────────────────────────────── + + def test_multiple_messages_in_one_buffer(self): + heartbeat = bytearray([0xFE, 0x02, 0xFF]) + button = bytearray([0xFE, 0x06, 0x00, 0xFF]) + knob = bytearray([0xFE, 0x03, 0x02, 0x01, 0xF4, 0xFF]) + msgs, remainder = parse_messages(heartbeat + button + knob) + assert len(msgs) == 3 + assert msgs[0]["type"] == "heartbeat" + assert msgs[1]["type"] == "button" + assert msgs[2]["type"] == "knob" + assert remainder == bytearray() + + def test_garbage_bytes_skipped(self): + buf = bytearray([0x00, 0x11, 0xFE, 0x02, 0xFF]) + msgs, _ = parse_messages(buf) + assert msgs == [{"type": "heartbeat"}] + + def test_incomplete_frame_stays_in_remainder(self): + # The parser advances byte-by-byte when a frame is unrecognised, so an + # incomplete knob frame (missing terminator 0xFF) is consumed as garbage. + buf = bytearray([0xFE, 0x03, 0x01, 0x03]) + msgs, remainder = parse_messages(buf) + assert msgs == [] + assert remainder == bytearray() + + def test_empty_buffer(self): + msgs, remainder = parse_messages(bytearray()) + assert msgs == [] + assert remainder == bytearray()