feat: implement volume boost functionality with signal-based multiplier and audio processing override

tests/test_api_surface.py

@@ -20,6 +20,7 @@ EXPECTED_ENDPOINTS = {
     ("post", "/api/skip"),
     ("post", "/api/stop"),
     ("post", "/api/volume"),
+    ("post", "/api/volume_boost"),
     ("post", "/api/loop"),
     ("post", "/api/loop_playlist"),
     ("post", "/api/loop_cycle"),

tests/test_music_player.py

@@ -7,6 +7,7 @@ import tempfile
 import unittest
 
 from tests.fakes import FakeTrack, build_player
+from zyrarbot.music.guild_music_state import VOLUME_BOOST_MULTIPLIER
 from zyrarbot.music.loop_mode import LoopMode
 from zyrarbot.storage.queue_repository import QueueRepository
 
@@ -235,6 +236,37 @@ class MusicPlayerQueueTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual(track.volume, 0.25)
 
+    async def test_volume_boost_multiplies_effective_volume_by_multiplier(self):
+        player, _ = build_player()
+        player.set_volume(GUILD, 0.2)
+        player.set_volume_boost(GUILD, True)
+
+        self.assertEqual(player.state(GUILD).volume, 0.2)
+        self.assertAlmostEqual(
+            player.state(GUILD).effective_volume, 0.2 * VOLUME_BOOST_MULTIPLIER
+        )
+
+        track = FakeTrack("boosted")
+        await player.enqueue(GUILD, track)
+        self.assertAlmostEqual(track.volume, 0.2 * VOLUME_BOOST_MULTIPLIER)
+
+        player.set_volume(GUILD, 1.0)
+        self.assertAlmostEqual(
+            player.state(GUILD).effective_volume, 1.0 * VOLUME_BOOST_MULTIPLIER
+        )
+
+    async def test_toggle_volume_boost(self):
+        player, _ = build_player()
+        self.assertFalse(player.state(GUILD).volume_boost)
+
+        res = player.toggle_volume_boost(GUILD)
+        self.assertTrue(res)
+        self.assertTrue(player.state(GUILD).volume_boost)
+
+        res2 = player.toggle_volume_boost(GUILD)
+        self.assertFalse(res2)
+        self.assertFalse(player.state(GUILD).volume_boost)
+
     async def test_default_plays_are_applied_per_guild(self):
         player, _ = build_player()
         player._config.get_default_plays.side_effect = lambda guild_id: {1: 2, 2: 7}.get(

tests/test_ytdl_source.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import struct
+import unittest
+from unittest.mock import MagicMock
+
+import discord
+
+from zyrarbot.audio.ytdl_source import FRAME_LENGTH_SECONDS, YTDLSource
+
+
+class DummyPCM(discord.AudioSource):
+    def __init__(self, data: bytes):
+        self._data = data
+        self.cleanup_called = False
+
+    def read(self) -> bytes:
+        data = self._data
+        self._data = b""
+        return data
+
+    def cleanup(self) -> None:
+        self.cleanup_called = True
+
+
+class TestYTDLSource(unittest.TestCase):
+    def test_read_advances_elapsed_and_returns_data(self):
+        # 1 sample of 16-bit stereo = 4 bytes
+        pcm = struct.pack("<hh", 1000, 2000)
+        source = DummyPCM(pcm)
+        ytdl = YTDLSource(source, data={"title": "test"}, volume=1.0)
+
+        self.assertEqual(ytdl.elapsed, 0.0)
+        data = ytdl.read()
+        self.assertEqual(data, pcm)
+        self.assertAlmostEqual(ytdl.elapsed, FRAME_LENGTH_SECONDS)
+
+        # Subsequent read returns empty and does not advance elapsed further
+        empty = ytdl.read()
+        self.assertEqual(empty, b"")
+        self.assertAlmostEqual(ytdl.elapsed, FRAME_LENGTH_SECONDS)
+
+    def test_read_with_volume_boost_over_2_does_not_clamp_to_2(self):
+        # Sample value 5000:
+        # At volume 2.0 (discord limit): 5000 * 2 = 10000
+        # At volume 5.0: 5000 * 5 = 25000
+        # At volume 10.0: 5000 * 10 = 50000 -> clipped to 32767
+        pcm = struct.pack("<hh", 5000, -5000)
+        source = DummyPCM(pcm)
+        ytdl = YTDLSource(source, data={"title": "distortion test"}, volume=5.0)
+
+        data = ytdl.read()
+        left, right = struct.unpack("<hh", data)
+        self.assertEqual(left, 25000)
+        self.assertEqual(right, -25000)
+
+    def test_read_distortion_clips_at_pcm_limits(self):
+        # Overdriving large amplitude samples causes clipping / przester at [-32768, 32767]
+        pcm = struct.pack("<hh", 10000, -10000)
+        source = DummyPCM(pcm)
+        ytdl = YTDLSource(source, data={"title": "przester"}, volume=10.0)
+
+        data = ytdl.read()
+        left, right = struct.unpack("<hh", data)
+        self.assertEqual(left, 32767)
+        self.assertEqual(right, -32768)
+
+
+if __name__ == "__main__":
+    unittest.main()