fix: kept file protection, skip/autoplay, TTS routing, video ID expansion

- _cleanup_track: never delete files from kept directory (data/music/)
  even when track.keep=False -- fixes kept files vanishing on replay
- !kept rm: skip to next track if removing the currently playing one
- !skip: silent (no reply), restarts play loop for autoplay on empty queue
- TTS plays through merlin's own connection instead of derp's, preventing
  choppy audio when music and TTS compete for the same output buffer
- !play recognizes bare YouTube video IDs (11-char alphanumeric)
- !kept rm <id> subcommand for removing individual kept tracks

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-22 18:00:23 +01:00
parent 36da191b45
commit 068734d931
5 changed files with 163 additions and 34 deletions

View File

@@ -126,6 +126,34 @@ class TestMumbleGuard:
assert bot.replied == []
# ---------------------------------------------------------------------------
# TestExpandVideoId
# ---------------------------------------------------------------------------
class TestExpandVideoId:
def test_bare_video_id(self):
assert _mod._expand_video_id("U1yQMjFZ6j4") == \
"https://www.youtube.com/watch?v=U1yQMjFZ6j4"
def test_id_with_hyphens_underscores(self):
assert _mod._expand_video_id("dQw4w9WgXcQ") == \
"https://www.youtube.com/watch?v=dQw4w9WgXcQ"
def test_too_short_not_expanded(self):
assert _mod._expand_video_id("abc") == "abc"
def test_too_long_not_expanded(self):
assert _mod._expand_video_id("abcdefghijkl") == "abcdefghijkl"
def test_full_url_not_expanded(self):
url = "https://www.youtube.com/watch?v=U1yQMjFZ6j4"
assert _mod._expand_video_id(url) == url
def test_search_query_not_expanded(self):
assert _mod._expand_video_id("hello world") == "hello world"
# ---------------------------------------------------------------------------
# TestPlayCommand
# ---------------------------------------------------------------------------
@@ -235,10 +263,10 @@ class TestSkipCommand:
msg = _Msg(text="!skip")
with patch.object(_mod, "_ensure_loop"):
asyncio.run(_mod.cmd_skip(bot, msg))
assert any("Skipped" in r for r in bot.replied)
assert not bot.replied # skip is silent when queue has next track
mock_task.cancel.assert_called_once()
def test_skip_empty_queue(self):
def test_skip_empty_queue_restarts_loop(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(url="a", title="Only", requester="x")
@@ -246,8 +274,9 @@ class TestSkipCommand:
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!skip")
asyncio.run(_mod.cmd_skip(bot, msg))
assert any("empty" in r.lower() for r in bot.replied)
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_skip(bot, msg))
mock_loop.assert_called_once() # loop restarted for autoplay
# ---------------------------------------------------------------------------
@@ -1507,6 +1536,20 @@ class TestCleanupTrack:
track = _mod._Track(url="x", title="t", requester="a")
_mod._cleanup_track(track) # should not raise
def test_cleanup_preserves_kept_dir_files(self, tmp_path):
"""Cleanup never deletes files from the kept music directory."""
music_dir = tmp_path / "music"
music_dir.mkdir()
f = music_dir / "kept-track.opus"
f.write_bytes(b"audio")
track = _mod._Track(
url="x", title="t", requester="a",
local_path=f, keep=False,
)
with patch.object(_mod, "_MUSIC_DIR", music_dir):
_mod._cleanup_track(track)
assert f.exists()
# ---------------------------------------------------------------------------
# TestKeepCommand
@@ -1683,6 +1726,64 @@ class TestKeptCommand:
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("MISSING" in r for r in bot.replied)
def test_kept_rm(self, tmp_path):
bot = _FakeBot()
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "abc123.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"title": "Test Track", "filename": "abc123.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir):
msg = _Msg(text="!kept rm 1")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Removed #1" in r for r in bot.replied)
assert bot.state.get("music", "keep:1") is None
assert not (music_dir / "abc123.opus").exists()
def test_kept_rm_with_hash(self, tmp_path):
bot = _FakeBot()
bot.state.set("music", "keep:3", json.dumps({
"title": "Track 3", "filename": "t3.opus", "id": 3,
}))
with patch.object(_mod, "_MUSIC_DIR", tmp_path):
msg = _Msg(text="!kept rm #3")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Removed #3" in r for r in bot.replied)
assert bot.state.get("music", "keep:3") is None
def test_kept_rm_skips_if_playing(self, tmp_path):
bot = _FakeBot()
ps = _mod._ps(bot)
music_dir = tmp_path / "music"
music_dir.mkdir()
f = music_dir / "abc123.opus"
f.write_bytes(b"audio")
track = _mod._Track(url="x", title="t", requester="a",
local_path=f, keep=True)
ps["current"] = track
bot.state.set("music", "keep:1", json.dumps({
"title": "Test Track", "filename": "abc123.opus", "id": 1,
}))
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_ensure_loop") as mock_loop:
msg = _Msg(text="!kept rm 1")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Removed #1" in r for r in bot.replied)
mock_loop.assert_called_once() # restarts loop for autoplay
def test_kept_rm_missing_id(self):
bot = _FakeBot()
msg = _Msg(text="!kept rm")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Usage" in r for r in bot.replied)
def test_kept_rm_not_found(self):
bot = _FakeBot()
msg = _Msg(text="!kept rm 99")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("No kept track" in r for r in bot.replied)
def test_kept_non_mumble(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!kept")