feat: add !seek command and persist volume across restarts

Seek to absolute or relative positions mid-track via !seek. Supports
M:SS and plain seconds with +/- prefixes. Volume is now saved to
bot.state and restored on connect.
This commit is contained in:
user
2026-02-22 03:31:35 +01:00
parent 7c099d8cf0
commit c493583a71
2 changed files with 282 additions and 0 deletions

View File

@@ -1181,3 +1181,181 @@ class TestKeptCommand:
msg = _Msg(text="!kept")
asyncio.run(_mod.cmd_kept(bot, msg))
assert any("Mumble-only" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestParseSeek
# ---------------------------------------------------------------------------
class TestParseSeek:
def test_absolute_seconds(self):
assert _mod._parse_seek("90") == ("abs", 90.0)
def test_absolute_mss(self):
assert _mod._parse_seek("1:30") == ("abs", 90.0)
def test_relative_forward(self):
assert _mod._parse_seek("+30") == ("rel", 30.0)
def test_relative_backward(self):
assert _mod._parse_seek("-30") == ("rel", -30.0)
def test_relative_mss(self):
assert _mod._parse_seek("+1:30") == ("rel", 90.0)
def test_relative_backward_mss(self):
assert _mod._parse_seek("-1:30") == ("rel", -90.0)
def test_invalid_raises(self):
import pytest
with pytest.raises(ValueError):
_mod._parse_seek("abc")
def test_empty_raises(self):
import pytest
with pytest.raises(ValueError):
_mod._parse_seek("")
# ---------------------------------------------------------------------------
# TestSeekCommand
# ---------------------------------------------------------------------------
class TestSeekCommand:
def test_seek_nothing_playing(self):
bot = _FakeBot()
msg = _Msg(text="!seek 1:30")
asyncio.run(_mod.cmd_seek(bot, msg))
assert any("Nothing playing" in r for r in bot.replied)
def test_seek_non_mumble(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!seek 1:30")
asyncio.run(_mod.cmd_seek(bot, msg))
assert bot.replied == []
def test_seek_no_arg(self):
bot = _FakeBot()
msg = _Msg(text="!seek")
asyncio.run(_mod.cmd_seek(bot, msg))
assert any("Usage" in r for r in bot.replied)
def test_seek_invalid_arg(self):
bot = _FakeBot()
msg = _Msg(text="!seek xyz")
asyncio.run(_mod.cmd_seek(bot, msg))
assert any("Usage" in r for r in bot.replied)
def test_seek_absolute(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!seek 1:30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
mock_loop.assert_called_once_with(bot, seek=90.0)
assert ps["queue"][0] is track
assert any("1:30" in r for r in bot.replied)
mock_task.cancel.assert_called_once()
def test_seek_relative_forward(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
ps["progress"] = [1500] # 1500 * 0.02 = 30s
ps["cur_seek"] = 60.0 # started at 60s
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!seek +30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 60 + 30 = 90, target = 90 + 30 = 120
mock_loop.assert_called_once_with(bot, seek=120.0)
def test_seek_relative_backward_clamps(self):
bot = _FakeBot()
ps = _mod._ps(bot)
track = _mod._Track(url="x", title="Song", requester="a")
ps["current"] = track
ps["progress"] = [500] # 500 * 0.02 = 10s
ps["cur_seek"] = 0.0
mock_task = MagicMock()
mock_task.done.return_value = False
ps["task"] = mock_task
msg = _Msg(text="!seek -30")
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod.cmd_seek(bot, msg))
# elapsed = 0 + 10 = 10, target = 10 - 30 = -20, clamped to 0
mock_loop.assert_called_once_with(bot, seek=0.0)
# ---------------------------------------------------------------------------
# TestVolumePersistence
# ---------------------------------------------------------------------------
class TestVolumePersistence:
def test_volume_persists_to_state(self):
bot = _FakeBot()
msg = _Msg(text="!volume 75")
asyncio.run(_mod.cmd_volume(bot, msg))
assert bot.state.get("music", "volume") == "75"
def test_volume_loads_on_connect(self):
bot = _FakeBot()
bot.state.set("music", "volume", "80")
spawned = []
def fake_spawn(coro, *, name=None):
task = MagicMock()
task.done.return_value = False
spawned.append(name)
coro.close()
return task
bot._spawn = fake_spawn
asyncio.run(_mod.on_connected(bot))
ps = _mod._ps(bot)
assert ps["volume"] == 80
def test_volume_loads_clamps_high(self):
bot = _FakeBot()
bot.state.set("music", "volume", "200")
spawned = []
def fake_spawn(coro, *, name=None):
task = MagicMock()
task.done.return_value = False
spawned.append(name)
coro.close()
return task
bot._spawn = fake_spawn
asyncio.run(_mod.on_connected(bot))
ps = _mod._ps(bot)
assert ps["volume"] == 100
def test_volume_loads_ignores_invalid(self):
bot = _FakeBot()
bot.state.set("music", "volume", "notanumber")
spawned = []
def fake_spawn(coro, *, name=None):
task = MagicMock()
task.done.return_value = False
spawned.append(name)
coro.close()
return task
bot._spawn = fake_spawn
asyncio.run(_mod.on_connected(bot))
ps = _mod._ps(bot)
assert ps["volume"] == 50 # default unchanged