username/derp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: auto-resume music on reconnect, sorcerer tier, cert auth

Browse Source 
Auto-resume: save playback position on stream errors and cancellation,
restore automatically after reconnect or container restart once the
channel is silent. Plugin lifecycle hook (on_connected) ensures the
reconnect watcher starts without waiting for user commands.

Sorcerer tier: new permission level between oper and admin. Configured
via [mumble] sorcerers list in derp.toml.

Mumble cert auth: pass certfile/keyfile to pymumble for client
certificate authentication.

Fixes: stream_audio now re-raises CancelledError and Exception so
_play_loop detects failures correctly. Subprocess cleanup uses 3s
timeout. Graceful shutdown cancels background tasks before stopping
pymumble. Safe getattr for _opers in core plugin.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-22 02:14:43 +01:00
parent f899241d73
commit ec55c2aef1
10 changed files with 764 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docker-compose.yml
View File

@@ -16,4 +16,4 @@ services:
- ./config/derp.toml:/app/config/derp.toml:ro,Z
- ./data:/app/data:Z
- ./secrets:/app/secrets:ro,Z
command: ["--verbose"]
command: ["--verbose", "--cprofile"]

7
docs/CHEATSHEET.md
View File

@@ -557,12 +557,19 @@ HTML stripped on receive, escaped on send. IRC-only commands are no-ops.
!np # Now playing
!volume # Show current volume
!volume 75 # Set volume (0-100, default 50)
!duck # Show ducking status
!duck on # Enable voice ducking
!duck off # Disable voice ducking
!duck floor 5 # Set duck floor volume (0-100, default 1)
!duck silence 20 # Set silence timeout seconds (default 15)
!duck restore 45 # Set restore ramp duration seconds (default 30)
```
Requires: `yt-dlp`, `ffmpeg`, `libopus` on the host.
Max 50 tracks in queue. Playlists auto-expand; excess truncated at limit.
Volume ramps smoothly over ~1s (no abrupt jumps mid-playback).
`!resume` restores position across restarts (persisted via `bot.state`).
Auto-resumes on reconnect if channel is silent (waits up to 60s for silence).
Mumble-only: `!play` replies with error on other adapters, others silently no-op.
## Plugin Template

50
docs/USAGE.md
View File

@@ -1592,3 +1592,53 @@ and voice transmission.
to text commands during streaming
- `!resume` continues from where playback was interrupted (`!stop`/`!skip`);
position is persisted via `bot.state` and survives bot restarts
### Auto-Resume on Reconnect
If the bot disconnects while music is playing (network hiccup, server
restart), it saves the current track and position. On reconnect, it
automatically resumes playback -- but only after the channel is silent
(using the same silence threshold as voice ducking, default 15s).
- Resume state is saved on both explicit stop/skip and on stream errors
(disconnect)
- Works across container restarts (cold boot) and network reconnections
- The bot waits up to 60s for silence; if the channel stays active, it
aborts and the saved state remains for manual `!resume`
- No chat message is sent on auto-resume; playback resumes silently
- The reconnect watcher starts via the `on_connected` plugin lifecycle hook
### Voice Ducking
When other users speak in the Mumble channel, the music volume automatically
ducks (lowers) to a configurable floor. After a configurable silence period,
volume gradually restores to the user-set level in small steps.
```
!duck Show ducking status and settings
!duck on Enable voice ducking
!duck off Disable voice ducking
!duck floor <0-100> Set floor volume % (default: 1)
!duck silence <sec> Set silence timeout in seconds (default: 15)
!duck restore <sec> Set restore ramp duration in seconds (default: 30)
```
Behavior:
- Enabled by default; voice is detected via pymumble's sound callback
- When someone speaks, volume drops immediately to the floor value
- After `silence` seconds of no voice, volume restores via a single
smooth linear ramp over `restore` seconds (default 30s)
- The per-frame volume ramp in `stream_audio` further smooths the
transition, eliminating audible steps
- Ducking resets when playback stops, skips, or the queue empties
Configuration (optional):
```toml
[music]
duck_enabled = true # Enable voice ducking (default: true)
duck_floor = 1 # Floor volume % during ducking (default: 1)
duck_silence = 15 # Seconds of silence before restoring (default: 15)
duck_restore = 30 # Seconds for smooth volume restore (default: 30)
```

11
plugins/core.py
View File

@@ -145,7 +145,8 @@ async def cmd_whoami(bot, message):
prefix = message.prefix or "unknown"
tier = bot._get_tier(message)
tags = [tier]
if message.prefix and message.prefix in bot._opers:
opers = getattr(bot, "_opers", set())
if message.prefix and message.prefix in opers:
tags.append("IRCOP")
await bot.reply(message, f"{prefix} [{', '.join(tags)}]")
@@ -158,12 +159,16 @@ async def cmd_admins(bot, message):
parts.append(f"Admin: {', '.join(bot._admins)}")
else:
parts.append("Admin: (none)")
sorcerers = getattr(bot, "_sorcerers", [])
if sorcerers:
parts.append(f"Sorcerer: {', '.join(sorcerers)}")
if bot._operators:
parts.append(f"Oper: {', '.join(bot._operators)}")
if bot._trusted:
parts.append(f"Trusted: {', '.join(bot._trusted)}")
if bot._opers:
parts.append(f"IRCOPs: {', '.join(sorted(bot._opers))}")
opers = getattr(bot, "_opers", set())
if opers:
parts.append(f"IRCOPs: {', '.join(sorted(opers))}")
else:
parts.append("IRCOPs: (none)")
await bot.reply(message, " | ".join(parts))

274
plugins/music.py
View File

@@ -7,6 +7,7 @@ import json
import logging
import random
import subprocess
import time
from dataclasses import dataclass
from derp.plugin import command
@@ -22,6 +23,7 @@ class _Track:
url: str
title: str
requester: str
origin: str = "" # original user-provided URL for re-resolution
# -- Per-bot runtime state ---------------------------------------------------
@@ -29,12 +31,20 @@ class _Track:
def _ps(bot):
"""Per-bot plugin runtime state."""
cfg = getattr(bot, "config", {}).get("music", {})
return bot._pstate.setdefault("music", {
"queue": [],
"current": None,
"volume": 50,
"task": None,
"done_event": None,
"duck_enabled": cfg.get("duck_enabled", True),
"duck_floor": cfg.get("duck_floor", 1),
"duck_silence": cfg.get("duck_silence", 15),
"duck_restore": cfg.get("duck_restore", 30),
"duck_vol": None,
"duck_task": None,
"_watcher_task": None,
})
@@ -70,7 +80,7 @@ def _fmt_time(seconds: float) -> str:
def _save_resume(bot, track: _Track, elapsed: float) -> None:
"""Persist current track and elapsed position for later resumption."""
data = json.dumps({
"url": track.url,
"url": track.origin or track.url,
"title": track.title,
"requester": track.requester,
"elapsed": round(elapsed, 2),
@@ -128,12 +138,143 @@ def _resolve_tracks(url: str, max_tracks: int = _MAX_QUEUE) -> list[tuple[str, s
return [(url, url)]
# -- Duck monitor ------------------------------------------------------------
async def _duck_monitor(bot) -> None:
"""Background task: duck volume when voice is detected, restore on silence.
Ducking is immediate (snap to floor). Restoration is a single smooth
linear ramp from floor to user volume over ``duck_restore`` seconds.
The per-frame volume ramp in ``stream_audio`` further smooths each
1-second update, eliminating audible steps.
"""
ps = _ps(bot)
restore_start: float = 0.0 # monotonic ts when restore began
restore_from: float = 0.0 # duck_vol at restore start
try:
while True:
await asyncio.sleep(1)
if not ps["duck_enabled"]:
if ps["duck_vol"] is not None:
ps["duck_vol"] = None
restore_start = 0.0
continue
ts = getattr(bot, "_last_voice_ts", 0.0)
if ts == 0.0:
continue
silence = time.monotonic() - ts
if silence < ps["duck_silence"]:
# Voice active -- duck immediately
if ps["duck_vol"] is None:
log.info("duck: voice detected, ducking to %d%%",
ps["duck_floor"])
ps["duck_vol"] = float(ps["duck_floor"])
restore_start = 0.0
elif ps["duck_vol"] is not None:
# Silence exceeded -- smooth linear restore
if restore_start == 0.0:
restore_start = time.monotonic()
restore_from = ps["duck_vol"]
log.info("duck: restoring %d%% -> %d%% over %ds",
int(restore_from), ps["volume"],
ps["duck_restore"])
elapsed = time.monotonic() - restore_start
dur = ps["duck_restore"]
if dur <= 0 or elapsed >= dur:
ps["duck_vol"] = None
restore_start = 0.0
else:
target = ps["volume"]
ps["duck_vol"] = restore_from + (target - restore_from) * (elapsed / dur)
except asyncio.CancelledError:
ps["duck_vol"] = None
# -- Auto-resume on reconnect ------------------------------------------------
async def _auto_resume(bot) -> None:
"""Wait for silence after reconnect, then resume saved playback."""
ps = _ps(bot)
if ps["current"] is not None:
return
data = _load_resume(bot)
if data is None:
return
# Let pymumble fully stabilize after reconnect
await asyncio.sleep(10)
deadline = time.monotonic() + 60
silence_needed = ps.get("duck_silence", 15)
while time.monotonic() < deadline:
await asyncio.sleep(2)
ts = getattr(bot, "_last_voice_ts", 0.0)
if ts == 0.0:
break
if time.monotonic() - ts >= silence_needed:
break
else:
log.info("music: auto-resume aborted, channel not silent after 60s")
return
# Re-check after waiting -- someone may have started playback manually
if ps["current"] is not None:
return
data = _load_resume(bot)
if data is None:
return
elapsed = data.get("elapsed", 0.0)
track = _Track(
url=data["url"],
title=data.get("title", data["url"]),
requester=data.get("requester", "?"),
)
ps["queue"].insert(0, track)
_clear_resume(bot)
log.info("music: auto-resuming '%s' from %s",
track.title, _fmt_time(elapsed))
_ensure_loop(bot, seek=elapsed)
async def _reconnect_watcher(bot) -> None:
"""Poll for reconnections and trigger auto-resume.
Also handles cold-start resume: if saved state exists on first
run, waits for the connection to stabilize then resumes.
"""
last_seen = getattr(bot, "_connect_count", 0)
boot_checked = False
while True:
await asyncio.sleep(2)
count = getattr(bot, "_connect_count", 0)
# Cold-start: resume saved state after first connection
if not boot_checked and count >= 1:
boot_checked = True
if _load_resume(bot) is not None:
log.info("music: saved state found on boot, attempting auto-resume")
await _auto_resume(bot)
continue
if count > last_seen and count > 1:
last_seen = count
log.info("music: reconnection detected, attempting auto-resume")
await _auto_resume(bot)
last_seen = max(last_seen, count)
# -- Play loop ---------------------------------------------------------------
async def _play_loop(bot, *, seek: float = 0.0) -> None:
"""Pop tracks from queue and stream them sequentially."""
ps = _ps(bot)
duck_task = bot._spawn(_duck_monitor(bot), name="music-duck-monitor")
ps["duck_task"] = duck_task
first = True
try:
while ps["queue"]:
@@ -150,7 +291,11 @@ async def _play_loop(bot, *, seek: float = 0.0) -> None:
try:
await bot.stream_audio(
track.url,
volume=lambda: ps["volume"] / 100.0,
volume=lambda: (
ps["duck_vol"]
if ps["duck_vol"] is not None
else ps["volume"]
) / 100.0,
on_done=done,
seek=cur_seek,
progress=progress,
@@ -162,15 +307,24 @@ async def _play_loop(bot, *, seek: float = 0.0) -> None:
raise
except Exception:
log.exception("music: stream error for %s", track.url)
elapsed = cur_seek + progress[0] * 0.02
if elapsed > 1.0:
_save_resume(bot, track, elapsed)
break
_clear_resume(bot)
await done.wait()
if progress[0] > 0:
_clear_resume(bot)
except asyncio.CancelledError:
pass
finally:
if duck_task and not duck_task.done():
duck_task.cancel()
ps["current"] = None
ps["done_event"] = None
ps["task"] = None
ps["duck_vol"] = None
ps["duck_task"] = None
def _ensure_loop(bot, *, seek: float = 0.0) -> None:
@@ -228,9 +382,12 @@ async def cmd_play(bot, message):
was_idle = ps["current"] is None
requester = message.nick or "?"
added = 0
# Only set origin for direct URLs (not searches) so resume uses the
# resolved video URL rather than an ephemeral search query
origin = url if not is_search else ""
for track_url, track_title in resolved[:remaining]:
ps["queue"].append(_Track(url=track_url, title=track_title,
requester=requester))
requester=requester, origin=origin))
added += 1
total_resolved = len(resolved)
@@ -269,6 +426,7 @@ async def cmd_stop(bot, message):
ps["current"] = None
ps["task"] = None
ps["done_event"] = None
ps["duck_vol"] = None
await bot.reply(message, "Stopped")
@@ -328,6 +486,7 @@ async def cmd_skip(bot, message):
skipped = ps["current"]
ps["current"] = None
ps["task"] = None
ps["duck_vol"] = None
if ps["queue"]:
_ensure_loop(bot)
@@ -441,3 +600,110 @@ async def cmd_volume(bot, message):
ps["volume"] = val
await bot.reply(message, f"Volume set to {val}%")
@command("duck", help="Music: !duck [on|off|floor N|silence N|restore N]")
async def cmd_duck(bot, message):
"""Configure voice-activated volume ducking.
Usage:
!duck Show ducking status and settings
!duck on Enable voice ducking
!duck off Disable voice ducking
!duck floor <0-100> Set floor volume %
!duck silence <sec> Set silence timeout (seconds)
!duck restore <sec> Set restore ramp duration (seconds)
"""
if not _is_mumble(bot):
await bot.reply(message, "Mumble-only feature")
return
ps = _ps(bot)
parts = message.text.split()
if len(parts) < 2:
state = "on" if ps["duck_enabled"] else "off"
ducking = ""
if ps["duck_vol"] is not None:
ducking = f", ducked to {int(ps['duck_vol'])}%"
await bot.reply(
message,
f"Duck: {state} | floor={ps['duck_floor']}%"
f" silence={ps['duck_silence']}s"
f" restore={ps['duck_restore']}s{ducking}",
)
return
sub = parts[1].lower()
if sub == "on":
ps["duck_enabled"] = True
await bot.reply(message, "Voice ducking enabled")
elif sub == "off":
ps["duck_enabled"] = False
ps["duck_vol"] = None
await bot.reply(message, "Voice ducking disabled")
elif sub == "floor":
if len(parts) < 3:
await bot.reply(message, "Usage: !duck floor <0-100>")
return
try:
val = int(parts[2])
except ValueError:
await bot.reply(message, "Usage: !duck floor <0-100>")
return
if val < 0 or val > 100:
await bot.reply(message, "Floor must be 0-100")
return
ps["duck_floor"] = val
await bot.reply(message, f"Duck floor set to {val}%")
elif sub == "silence":
if len(parts) < 3:
await bot.reply(message, "Usage: !duck silence <seconds>")
return
try:
val = int(parts[2])
except ValueError:
await bot.reply(message, "Usage: !duck silence <seconds>")
return
if val < 1:
await bot.reply(message, "Silence timeout must be >= 1")
return
ps["duck_silence"] = val
await bot.reply(message, f"Duck silence set to {val}s")
elif sub == "restore":
if len(parts) < 3:
await bot.reply(message, "Usage: !duck restore <seconds>")
return
try:
val = int(parts[2])
except ValueError:
await bot.reply(message, "Usage: !duck restore <seconds>")
return
if val < 1:
await bot.reply(message, "Restore duration must be >= 1")
return
ps["duck_restore"] = val
await bot.reply(message, f"Duck restore set to {val}s")
else:
await bot.reply(
message, "Usage: !duck [on|off|floor N|silence N|restore N]",
)
# -- Plugin lifecycle --------------------------------------------------------
async def on_connected(bot) -> None:
"""Called by MumbleBot after each (re)connection.
Ensures the reconnect watcher is running -- triggers boot-resume
and reconnect-resume without waiting for a user command.
"""
if not _is_mumble(bot):
return
ps = _ps(bot)
if ps["_watcher_task"] is None and hasattr(bot, "_spawn"):
ps["_watcher_task"] = bot._spawn(
_reconnect_watcher(bot), name="music-reconnect-watcher",
)

4
src/derp/bot.py
View File

@@ -96,6 +96,7 @@ class Bot:
self._tasks: set[asyncio.Task] = set()
self._reconnect_delay: float = 5.0
self._admins: list[str] = config.get("bot", {}).get("admins", [])
self._sorcerers: list[str] = config.get("bot", {}).get("sorcerers", [])
self._operators: list[str] = config.get("bot", {}).get("operators", [])
self._trusted: list[str] = config.get("bot", {}).get("trusted", [])
self._opers: set[str] = set() # hostmasks of known IRC operators
@@ -377,6 +378,9 @@ class Bot:
for pattern in self._admins:
if fnmatch.fnmatch(msg.prefix, pattern):
return "admin"
for pattern in self._sorcerers:
if fnmatch.fnmatch(msg.prefix, pattern):
return "sorcerer"
for pattern in self._operators:
if fnmatch.fnmatch(msg.prefix, pattern):
return "oper"

70
src/derp/mumble.py
View File

@@ -16,6 +16,7 @@ import pymumble_py3 as pymumble
from pymumble_py3.constants import (
PYMUMBLE_CLBK_CONNECTED,
PYMUMBLE_CLBK_DISCONNECTED,
PYMUMBLE_CLBK_SOUNDRECEIVED,
PYMUMBLE_CLBK_TEXTMESSAGERECEIVED,
)
@@ -135,6 +136,8 @@ class MumbleBot:
self._port: int = mu_cfg.get("port", 64738)
self._username: str = mu_cfg.get("username", "derp")
self._password: str = mu_cfg.get("password", "")
self._certfile: str | None = mu_cfg.get("certfile")
self._keyfile: str | None = mu_cfg.get("keyfile")
self.nick: str = self._username
self.prefix: str = (
mu_cfg.get("prefix")
@@ -144,6 +147,7 @@ class MumbleBot:
self._started: float = time.monotonic()
self._tasks: set[asyncio.Task] = set()
self._admins: list[str] = [str(x) for x in mu_cfg.get("admins", [])]
self._sorcerers: list[str] = [str(x) for x in mu_cfg.get("sorcerers", [])]
self._operators: list[str] = [str(x) for x in mu_cfg.get("operators", [])]
self._trusted: list[str] = [str(x) for x in mu_cfg.get("trusted", [])]
self.state = StateStore(f"data/state-{name}.db")
@@ -151,6 +155,8 @@ class MumbleBot:
# pymumble state
self._mumble: pymumble.Mumble | None = None
self._loop: asyncio.AbstractEventLoop | None = None
self._last_voice_ts: float = 0.0
self._connect_count: int = 0
rate_cfg = config.get("bot", {})
self._bucket = _TokenBucket(
@@ -165,6 +171,7 @@ class MumbleBot:
self._mumble = pymumble.Mumble(
self._host, self._username,
port=self._port, password=self._password,
certfile=self._certfile, keyfile=self._keyfile,
reconnect=True,
)
self._mumble.callbacks.set_callback(
@@ -179,19 +186,54 @@ class MumbleBot:
PYMUMBLE_CLBK_DISCONNECTED,
self._on_disconnected,
)
self._mumble.set_receive_sound(False)
self._mumble.callbacks.set_callback(
PYMUMBLE_CLBK_SOUNDRECEIVED,
self._on_sound_received,
)
self._mumble.set_receive_sound(True)
self._mumble.start()
self._mumble.is_ready()
def _on_connected(self) -> None:
"""Callback from pymumble thread: connection established."""
self._connect_count += 1
kind = "reconnected" if self._connect_count > 1 else "connected"
session = getattr(self._mumble.users, "myself_session", "?")
log.info("mumble: connected as %s on %s:%d (session=%s)",
self._username, self._host, self._port, session)
log.info("mumble: %s as %s on %s:%d (session=%s)",
kind, self._username, self._host, self._port, session)
if self._loop:
asyncio.run_coroutine_threadsafe(
self._notify_plugins_connected(), self._loop,
)
async def _notify_plugins_connected(self) -> None:
"""Call on_connected(bot) in each loaded plugin that defines it."""
for name, mod in self.registry._modules.items():
fn = getattr(mod, "on_connected", None)
if fn is None or not asyncio.iscoroutinefunction(fn):
continue
try:
await fn(self)
except Exception:
log.exception("mumble: on_connected hook failed in %s", name)
def _on_disconnected(self) -> None:
"""Callback from pymumble thread: connection lost."""
log.warning("mumble: disconnected")
self._last_voice_ts = 0.0
def _on_sound_received(self, user, sound_chunk) -> None:
"""Callback from pymumble thread: voice audio received.
Updates the timestamp used by the music plugin's duck monitor.
When this callback is registered, pymumble passes decoded PCM
directly and does not queue it -- no memory buildup.
"""
prev = self._last_voice_ts
self._last_voice_ts = time.monotonic()
if prev == 0.0:
name = user["name"] if isinstance(user, dict) else "?"
log.info("mumble: first voice packet from %s", name)
def _on_text_message(self, message) -> None:
"""Callback from pymumble thread: text message received.
@@ -250,6 +292,11 @@ class MumbleBot:
while self._running:
await asyncio.sleep(1)
finally:
# Cancel background tasks first so play-loop can save resume state
for task in list(self._tasks):
task.cancel()
if self._tasks:
await asyncio.gather(*self._tasks, return_exceptions=True)
if self._mumble:
self._mumble.stop()
self._mumble = None
@@ -322,6 +369,9 @@ class MumbleBot:
for name in self._admins:
if msg.prefix == name:
return "admin"
for name in self._sorcerers:
if msg.prefix == name:
return "sorcerer"
for name in self._operators:
if msg.prefix == name:
return "oper"
@@ -526,17 +576,25 @@ class MumbleBot:
await asyncio.sleep(0.1)
log.info("stream_audio: finished, %d frames", frames)
except asyncio.CancelledError:
self._mumble.sound_output.clear_buffer()
try:
self._mumble.sound_output.clear_buffer()
except Exception:
pass
log.info("stream_audio: cancelled at frame %d", frames)
raise
except Exception:
log.exception("stream_audio: error at frame %d", frames)
raise
finally:
try:
proc.kill()
except ProcessLookupError:
pass
stderr_out = await proc.stderr.read()
await proc.wait()
try:
stderr_out = await asyncio.wait_for(proc.stderr.read(), timeout=3)
await asyncio.wait_for(proc.wait(), timeout=3)
except (asyncio.TimeoutError, asyncio.CancelledError):
stderr_out = b""
if stderr_out:
log.warning("stream_audio: subprocess stderr: %s",
stderr_out.decode(errors="replace")[:500])

2
src/derp/plugin.py
View File

@@ -12,7 +12,7 @@ from typing import Any, Callable
log = logging.getLogger(__name__)
TIERS: tuple[str, ...] = ("user", "trusted", "oper", "admin")
TIERS: tuple[str, ...] = ("user", "trusted", "oper", "sorcerer", "admin")
@dataclass(slots=True)

2
tests/test_acl.py
View File

@@ -107,7 +107,7 @@ def _msg(text: str, prefix: str = "nick!user@host") -> Message:
class TestTierConstants:
def test_tier_order(self):
assert TIERS == ("user", "trusted", "oper", "admin")
assert TIERS == ("user", "trusted", "oper", "sorcerer", "admin")
def test_index_comparison(self):
assert TIERS.index("user") < TIERS.index("trusted")

358
tests/test_music.py
View File

@@ -3,6 +3,7 @@
import asyncio
import importlib.util
import sys
import time
from unittest.mock import AsyncMock, MagicMock, patch
# -- Load plugin module directly ---------------------------------------------
@@ -40,6 +41,7 @@ class _FakeBot:
self.sent: list[tuple[str, str]] = []
self.replied: list[str] = []
self.state = _FakeState()
self.config: dict = {}
self._pstate: dict = {}
self._tasks: set[asyncio.Task] = set()
if mumble:
@@ -648,3 +650,359 @@ class TestFmtTime:
def test_large_value(self):
assert _mod._fmt_time(3661) == "61:01"
# ---------------------------------------------------------------------------
# TestDuckCommand
# ---------------------------------------------------------------------------
class TestDuckCommand:
def test_show_status(self):
bot = _FakeBot()
msg = _Msg(text="!duck")
asyncio.run(_mod.cmd_duck(bot, msg))
assert any("Duck:" in r for r in bot.replied)
assert any("floor=1%" in r for r in bot.replied)
assert any("restore=30s" in r for r in bot.replied)
def test_toggle_on(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = False
msg = _Msg(text="!duck on")
asyncio.run(_mod.cmd_duck(bot, msg))
assert ps["duck_enabled"] is True
assert any("enabled" in r for r in bot.replied)
def test_toggle_off(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_vol"] = 5.0
msg = _Msg(text="!duck off")
asyncio.run(_mod.cmd_duck(bot, msg))
assert ps["duck_enabled"] is False
assert ps["duck_vol"] is None
assert any("disabled" in r for r in bot.replied)
def test_set_floor(self):
bot = _FakeBot()
msg = _Msg(text="!duck floor 10")
asyncio.run(_mod.cmd_duck(bot, msg))
ps = _mod._ps(bot)
assert ps["duck_floor"] == 10
assert any("10%" in r for r in bot.replied)
def test_set_floor_invalid(self):
bot = _FakeBot()
msg = _Msg(text="!duck floor 200")
asyncio.run(_mod.cmd_duck(bot, msg))
assert any("0-100" in r for r in bot.replied)
def test_set_silence(self):
bot = _FakeBot()
msg = _Msg(text="!duck silence 30")
asyncio.run(_mod.cmd_duck(bot, msg))
ps = _mod._ps(bot)
assert ps["duck_silence"] == 30
assert any("30s" in r for r in bot.replied)
def test_set_restore(self):
bot = _FakeBot()
msg = _Msg(text="!duck restore 45")
asyncio.run(_mod.cmd_duck(bot, msg))
ps = _mod._ps(bot)
assert ps["duck_restore"] == 45
assert any("45s" in r for r in bot.replied)
def test_non_mumble(self):
bot = _FakeBot(mumble=False)
msg = _Msg(text="!duck")
asyncio.run(_mod.cmd_duck(bot, msg))
assert any("Mumble-only" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestDuckMonitor
# ---------------------------------------------------------------------------
class TestDuckMonitor:
def test_voice_detected_ducks_to_floor(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 5
bot._last_voice_ts = time.monotonic()
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(1.5)
assert ps["duck_vol"] == 5.0
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_silence_begins_smooth_restore(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 1
ps["duck_restore"] = 10 # 10s total restore
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
ps["duck_vol"] = 1.0 # already ducked
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(1.5)
# After ~1s into a 10s ramp from 1->50, vol should be ~5-6
vol = ps["duck_vol"]
assert vol is not None and vol > 1.0
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_full_restore_sets_none(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 1
ps["duck_restore"] = 1 # 1s restore -- completes quickly
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
ps["duck_vol"] = 1.0
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
# First tick starts restore, second tick sees elapsed >= dur
await asyncio.sleep(2.5)
assert ps["duck_vol"] is None
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_reduck_during_restore(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = True
ps["duck_floor"] = 5
ps["duck_restore"] = 30
ps["volume"] = 50
bot._last_voice_ts = time.monotonic() - 100
ps["duck_vol"] = 30.0 # mid-restore
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(0.5)
# Simulate voice arriving now
bot._last_voice_ts = time.monotonic()
await asyncio.sleep(1.5)
assert ps["duck_vol"] == 5.0 # re-ducked to floor
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_disabled_no_ducking(self):
bot = _FakeBot()
ps = _mod._ps(bot)
ps["duck_enabled"] = False
bot._last_voice_ts = time.monotonic()
async def _check():
task = asyncio.create_task(_mod._duck_monitor(bot))
await asyncio.sleep(1.5)
assert ps["duck_vol"] is None
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
# ---------------------------------------------------------------------------
# TestAutoResume
# ---------------------------------------------------------------------------
class TestAutoResume:
def test_resume_on_silence(self):
"""Auto-resume loads saved state when channel is silent."""
bot = _FakeBot()
bot._connect_count = 2
bot._last_voice_ts = 0.0
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 120.0)
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._auto_resume(bot))
mock_loop.assert_called_once_with(bot, seek=120.0)
ps = _mod._ps(bot)
assert len(ps["queue"]) == 1
assert ps["queue"][0].url == "https://example.com/a"
# Resume state cleared after loading
assert _mod._load_resume(bot) is None
def test_no_resume_if_playing(self):
"""Auto-resume returns early when already playing."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["current"] = _mod._Track(url="x", title="Playing", requester="a")
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 60.0)
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._auto_resume(bot))
mock_loop.assert_not_called()
def test_no_resume_if_no_state(self):
"""Auto-resume returns early when nothing is saved."""
bot = _FakeBot()
bot._last_voice_ts = 0.0
with patch.object(_mod, "_ensure_loop") as mock_loop:
asyncio.run(_mod._auto_resume(bot))
mock_loop.assert_not_called()
def test_abort_if_voice_active(self):
"""Auto-resume aborts if voice never goes silent within deadline."""
bot = _FakeBot()
now = time.monotonic()
bot._last_voice_ts = now
ps = _mod._ps(bot)
ps["duck_silence"] = 15
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 60.0)
async def _check():
# Patch monotonic to jump past the 60s deadline; keep voice active
mono_val = [now]
_real_sleep = asyncio.sleep
def _fast_mono():
return mono_val[0]
async def _fast_sleep(s):
mono_val[0] += s
bot._last_voice_ts = mono_val[0]
await _real_sleep(0)
with patch.object(time, "monotonic", side_effect=_fast_mono):
with patch("asyncio.sleep", side_effect=_fast_sleep):
with patch.object(_mod, "_ensure_loop") as mock_loop:
await _mod._auto_resume(bot)
mock_loop.assert_not_called()
asyncio.run(_check())
def test_reconnect_watcher_triggers_resume(self):
"""Watcher detects connect_count increment and calls _auto_resume."""
bot = _FakeBot()
bot._connect_count = 1
async def _check():
with patch.object(_mod, "_auto_resume", new_callable=AsyncMock) as mock_ar:
task = asyncio.create_task(_mod._reconnect_watcher(bot))
await asyncio.sleep(0.5)
# Simulate reconnection
bot._connect_count = 2
await asyncio.sleep(3)
mock_ar.assert_called_once_with(bot)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_watcher_ignores_first_connect(self):
"""Watcher does not trigger on initial connection (count 0->1) without saved state."""
bot = _FakeBot()
bot._connect_count = 0
async def _check():
with patch.object(_mod, "_auto_resume", new_callable=AsyncMock) as mock_ar:
task = asyncio.create_task(_mod._reconnect_watcher(bot))
await asyncio.sleep(0.5)
bot._connect_count = 1
await asyncio.sleep(3)
mock_ar.assert_not_called()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_watcher_boot_resume_with_saved_state(self):
"""Watcher triggers boot-resume on first connect when state exists."""
bot = _FakeBot()
bot._connect_count = 0
track = _mod._Track(url="https://example.com/a", title="Song", requester="Alice")
_mod._save_resume(bot, track, 30.0)
async def _check():
with patch.object(_mod, "_auto_resume", new_callable=AsyncMock) as mock_ar:
task = asyncio.create_task(_mod._reconnect_watcher(bot))
await asyncio.sleep(0.5)
bot._connect_count = 1
await asyncio.sleep(3)
mock_ar.assert_called_once_with(bot)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_check())
def test_on_connected_starts_watcher(self):
"""on_connected() starts the reconnect watcher task."""
bot = _FakeBot()
spawned = []
def fake_spawn(coro, *, name=None):
task = MagicMock()
task.done.return_value = False
spawned.append(name)
# Close the coroutine to avoid RuntimeWarning
coro.close()
return task
bot._spawn = fake_spawn
asyncio.run(_mod.on_connected(bot))
assert "music-reconnect-watcher" in spawned
ps = _mod._ps(bot)
assert ps["_watcher_task"] is not None
def test_on_connected_no_double_start(self):
"""on_connected() does not start a second watcher."""
bot = _FakeBot()
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))
asyncio.run(_mod.on_connected(bot))
assert spawned.count("music-reconnect-watcher") == 1