feat: fade-out on skip/stop/prev, song metadata on keep
Some checks failed
CI / gitleaks (push) Failing after 4s
CI / lint (push) Successful in 24s
CI / test (3.11) (push) Failing after 30s
CI / test (3.13) (push) Failing after 34s
CI / test (3.12) (push) Failing after 36s
CI / build (push) Has been skipped

- Add fade_step parameter to stream_audio for fast volume ramps
- _fade_and_cancel helper: smooth ~0.8s fade before track switch
- !skip, !stop, !seek now fade out instead of cutting instantly
- !prev command: go back to previous track (10-track history stack)
- !keep fetches title/artist/duration via yt-dlp, stores in bot.state
- !kept displays metadata (title, artist, duration, file size)
- !kept clear also removes stored metadata
- 29 new tests for fade, prev, history, keep metadata

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-22 06:38:25 +01:00
parent de2d1fdf15
commit 8f1df167b9
5 changed files with 487 additions and 43 deletions

View File

@@ -2,6 +2,7 @@
import asyncio
import importlib.util
import json
import sys
import time
from unittest.mock import AsyncMock, MagicMock, patch
@@ -53,6 +54,11 @@ class _FakeBot:
async def reply(self, message, text: str) -> None:
self.replied.append(text)
async def long_reply(self, message, lines: list[str], *,
label: str = "") -> None:
for line in lines:
self.replied.append(line)
def _is_admin(self, message) -> bool:
return False
@@ -1358,3 +1364,295 @@ class TestVolumePersistence:
asyncio.run(_mod.on_connected(bot))
ps = _mod._ps(bot)
assert ps["volume"] == 50 # default unchanged
# ---------------------------------------------------------------------------
# TestFadeAndCancel
# ---------------------------------------------------------------------------
class TestFadeAndCancel:
def test_sets_fade_state(self):
"""_fade_and_cancel sets fade_vol=0 and a fast fade_step."""
bot = _FakeBot()
ps = _mod._ps(bot)
# Create a fake task that stays "running"
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
async def _check():
task = asyncio.create_task(_mod._fade_and_cancel(bot, duration=0.1))
# Let the fade start
await asyncio.sleep(0.02)
assert ps["fade_vol"] == 0
assert ps["fade_step"] is not None
assert ps["fade_step"] > 0.01
await task
asyncio.run(_check())
def test_noop_when_no_task(self):
"""_fade_and_cancel returns immediately if no task is running."""
bot = _FakeBot()
_mod._ps(bot)
asyncio.run(_mod._fade_and_cancel(bot, duration=0.1))
def test_clears_fade_step_after_cancel(self):
"""fade_step is reset to None after cancellation."""
bot = _FakeBot()
ps = _mod._ps(bot)
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
asyncio.run(_mod._fade_and_cancel(bot, duration=0.1))
assert ps["fade_step"] is None
# ---------------------------------------------------------------------------
# TestPrevCommand
# ---------------------------------------------------------------------------
class TestPrevCommand:
def test_prev_no_history(self):
bot = _FakeBot()
msg = _Msg(text="!prev")
asyncio.run(_mod.cmd_prev(bot, msg))
assert any("No previous" in r for r in bot.replied)
def test_prev_pops_history(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(url="a", title="Current", requester="x")
ps["history"] = [
_mod._Track(url="b", title="Previous", requester="y"),
]
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!prev")
with patch.object(_mod, "_fade_and_cancel", new_callable=AsyncMock):
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_prev(bot, msg))
assert any("Previous" in r for r in bot.replied)
# History should be empty (popped the only entry)
assert len(ps["history"]) == 0
# Queue should have: prev track, then current track
assert len(ps["queue"]) == 2
assert ps["queue"][0].title == "Previous"
assert ps["queue"][1].title == "Current"
def test_prev_non_mumble(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!prev")
asyncio.run(_mod.cmd_prev(bot, msg))
assert bot.replied == []
def test_prev_no_current_track(self):
"""!prev with history but nothing currently playing."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["history"] = [
_mod._Track(url="b", title="Previous", requester="y"),
]
msg = _Msg(text="!prev")
with patch.object(_mod, "_fade_and_cancel", new_callable=AsyncMock):
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_prev(bot, msg))
# Only the prev track in queue (no current to re-queue)
assert len(ps["queue"]) == 1
assert ps["queue"][0].title == "Previous"
# ---------------------------------------------------------------------------
# TestHistoryTracking
# ---------------------------------------------------------------------------
class TestHistoryTracking:
def test_skip_pushes_to_history(self):
"""Skipping a track adds it to history."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(url="a", title="First", requester="x")
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!skip")
with patch.object(_mod, "_fade_and_cancel", new_callable=AsyncMock):
asyncio.run(_mod.cmd_skip(bot, msg))
assert len(ps["history"]) == 1
assert ps["history"][0].title == "First"
def test_history_capped(self):
"""History does not exceed _MAX_HISTORY entries."""
bot = _FakeBot()
ps = _mod._ps(bot)
# Fill history to max
for i in range(_mod._MAX_HISTORY):
ps["history"].append(
_mod._Track(url=f"u{i}", title=f"T{i}", requester="a"),
)
assert len(ps["history"]) == _mod._MAX_HISTORY
# Skip another track, pushing to history
ps["current"] = _mod._Track(url="new", title="New", requester="x")
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!skip")
with patch.object(_mod, "_fade_and_cancel", new_callable=AsyncMock):
asyncio.run(_mod.cmd_skip(bot, msg))
assert len(ps["history"]) == _mod._MAX_HISTORY
assert ps["history"][-1].title == "New"
def test_ps_has_history(self):
"""_ps initializes with empty history list."""
bot = _FakeBot()
ps = _mod._ps(bot)
assert ps["history"] == []
# ---------------------------------------------------------------------------
# TestFadeState
# ---------------------------------------------------------------------------
class TestFadeState:
def test_ps_fade_fields_initialized(self):
"""_ps initializes fade_vol and fade_step to None."""
bot = _FakeBot()
ps = _mod._ps(bot)
assert ps["fade_vol"] is None
assert ps["fade_step"] is None
# ---------------------------------------------------------------------------
# TestKeepMetadata
# ---------------------------------------------------------------------------
class TestKeepMetadata:
def test_keep_stores_metadata(self, tmp_path):
"""!keep stores metadata JSON in bot.state."""
bot = _FakeBot()
ps = _mod._ps(bot)
f = tmp_path / "abc123.opus"
f.write_bytes(b"audio")
track = _mod._Track(
url="https://example.com/v", title="Test Song",
requester="a", local_path=f,
)
ps["current"] = track
msg = _Msg(text="!keep")
meta = {"title": "My Song", "artist": "Artist", "duration": 195.0}
with patch.object(_mod, "_fetch_metadata", return_value=meta):
asyncio.run(_mod.cmd_keep(bot, msg))
assert track.keep is True
raw = bot.state.get("music", "keep:abc123.opus")
assert raw is not None
stored = json.loads(raw)
assert stored["title"] == "My Song"
assert stored["artist"] == "Artist"
assert stored["duration"] == 195.0
assert any("My Song" in r for r in bot.replied)
assert any("Artist" in r for r in bot.replied)
assert any("3:15" in r for r in bot.replied)
def test_keep_no_artist(self, tmp_path):
"""!keep with empty artist omits the artist field."""
bot = _FakeBot()
ps = _mod._ps(bot)
f = tmp_path / "def456.opus"
f.write_bytes(b"audio")
track = _mod._Track(
url="https://example.com/v", title="Song",
requester="a", local_path=f,
)
ps["current"] = track
msg = _Msg(text="!keep")
meta = {"title": "Song", "artist": "NA", "duration": 60.0}
with patch.object(_mod, "_fetch_metadata", return_value=meta):
asyncio.run(_mod.cmd_keep(bot, msg))
# Should not contain "NA" as artist
assert not any("NA" in r and "--" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestKeptMetadata
# ---------------------------------------------------------------------------
class TestKeptMetadata:
def test_kept_shows_metadata(self, tmp_path):
"""!kept displays metadata from bot.state when available."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"x" * 2048)
bot.state.set("music", "keep:abc123.opus", json.dumps({
"title": "Cool Song", "artist": "DJ Test", "duration": 225.0,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Cool Song" in r for r in bot.replied)
assert any("DJ Test" in r for r in bot.replied)
assert any("3:45" in r for r in bot.replied)
def test_kept_fallback_no_metadata(self, tmp_path):
"""!kept falls back to filename when no metadata stored."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "xyz789.webm").write_bytes(b"x" * 1024)
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("xyz789.webm" in r for r in bot.replied)
def test_kept_clear_removes_metadata(self, tmp_path):
"""!kept clear also removes stored metadata."""
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"audio")
bot.state.set("music", "keep:abc123.opus", json.dumps({
"title": "Song", "artist": "", "duration": 0,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept clear")
asyncio.run(_mod.cmd_kept(bot, msg))
assert bot.state.get("music", "keep:abc123.opus") is None
assert any("Deleted 1 file(s)" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestFetchMetadata
# ---------------------------------------------------------------------------
class TestFetchMetadata:
def test_success(self):
result = MagicMock()
result.stdout = "My Song\nArtist Name\n195.5\n"
with patch("subprocess.run", return_value=result):
meta = _mod._fetch_metadata("https://example.com/v")
assert meta["title"] == "My Song"
assert meta["artist"] == "Artist Name"
assert meta["duration"] == 195.5
def test_partial_output(self):
result = MagicMock()
result.stdout = "Only Title\n"
with patch("subprocess.run", return_value=result):
meta = _mod._fetch_metadata("https://example.com/v")
assert meta["title"] == "Only Title"
assert meta["artist"] == ""
assert meta["duration"] == 0
def test_error_returns_empty(self):
with patch("subprocess.run", side_effect=Exception("fail")):
meta = _mod._fetch_metadata("https://example.com/v")
assert meta["title"] == ""
assert meta["artist"] == ""
assert meta["duration"] == 0