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)
2026-02-27 18:40:13 -05:00
1 parent dbdafabcce
commit 1552d73412
@@ -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()
@@ -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()