Logan CusanoandClaude Opus 5.5
419d7c4d8c
- Rewritten as RadioManager reconciling C2's retained
nodes/{id}/discord/desired (token ids) and publishing
nodes/{id}/discord/actual {applied_seq, sessions, streams}.
- One gateway session per leased token, one voice connection per
(token, guild); a token no stream needs is logged OUT (client.close()),
which is what lets C2 free it safely (item 1).
- Watchdog is a separate reconcile loop; recovering a dead session can no
longer cancel it (item 2). Presence kept across rejoins.
- No message-content intent and no @bot leave/join mention controls (item 3).
- Local dashboard: /api/discord/join returns 409 with an explanation instead
of a TypeError 500; /api/discord/leave pauses locally; GET /api/discord
shows state (item 4). Join/leave results are reported to C2 (item 5).
- Token strings fetched over HTTPS from C2 (lease holder only), memory only.
- MQTT-loss fail-safe: logs every bot out after
DISCORD_MQTT_LOSS_STOP_SECONDS (300, below C2's 600s lease grace).
- Legacy discord_join/discord_leave MQTT commands are ignored.
- Audio path unchanged: FFmpeg -f pulse on settings.pulse_source per
connection, never Icecast.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
261 lines
8.7 KiB
Python
261 lines
8.7 KiB
Python
"""
|
|
Edge Discord radio manager (node-26#25): desired-state reconcile, logout on
|
|
release, watchdog that survives a dead session, MQTT-loss fail-safe.
|
|
|
|
Discord itself is faked: FakeClient logs "in" by firing on_ready, guilds and
|
|
channels are plain objects, voice connections record play/disconnect.
|
|
"""
|
|
import asyncio
|
|
|
|
import discord
|
|
import pytest
|
|
|
|
from app.internal import discord_radio
|
|
from app.internal.discord_radio import RadioManager
|
|
|
|
|
|
class FakeVC:
|
|
def __init__(self, channel):
|
|
self.channel = channel
|
|
self.connected = True
|
|
self.playing = False
|
|
self.plays = 0
|
|
|
|
def is_connected(self):
|
|
return self.connected
|
|
|
|
def is_playing(self):
|
|
return self.playing
|
|
|
|
def play(self, source, after=None):
|
|
self.playing = True
|
|
self.plays += 1
|
|
|
|
def stop(self):
|
|
self.playing = False
|
|
|
|
async def disconnect(self, force=False):
|
|
self.connected = False
|
|
self.playing = False
|
|
self.channel.guild.voice_client = None
|
|
|
|
async def move_to(self, channel):
|
|
self.channel = channel
|
|
channel.guild.voice_client = self
|
|
|
|
|
|
class FakeChannel:
|
|
def __init__(self, guild, cid, kind=discord.ChannelType.voice):
|
|
self.guild, self.id, self.type = guild, cid, kind
|
|
|
|
async def connect(self):
|
|
vc = FakeVC(self)
|
|
self.guild.voice_client = vc
|
|
return vc
|
|
|
|
|
|
class FakeGuild:
|
|
def __init__(self, gid, channel_ids):
|
|
self.id = gid
|
|
self.voice_client = None
|
|
self.channels = {c: FakeChannel(self, c) for c in channel_ids}
|
|
|
|
def get_channel(self, cid):
|
|
return self.channels.get(cid)
|
|
|
|
|
|
class FakeClient:
|
|
instances: list = []
|
|
|
|
def __init__(self, guilds, fail_login=False):
|
|
self.guilds = guilds
|
|
self.fail_login = fail_login
|
|
self.closed = False
|
|
self.handlers = {}
|
|
self.token = None
|
|
self.user = "radio-bot"
|
|
FakeClient.instances.append(self)
|
|
|
|
def event(self, coro):
|
|
self.handlers[coro.__name__] = coro
|
|
return coro
|
|
|
|
async def start(self, token):
|
|
self.token = token
|
|
if self.fail_login:
|
|
raise discord.LoginFailure("Improper token has been passed.")
|
|
await self.handlers["on_ready"]()
|
|
while not self.closed:
|
|
await asyncio.sleep(0.01)
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
|
|
def is_closed(self):
|
|
return self.closed
|
|
|
|
def get_guild(self, gid):
|
|
return self.guilds.get(gid)
|
|
|
|
async def change_presence(self, activity=None):
|
|
self.presence = activity
|
|
|
|
|
|
@pytest.fixture
|
|
def env(monkeypatch):
|
|
FakeClient.instances = []
|
|
guilds = {1: FakeGuild(1, [11, 12]), 2: FakeGuild(2, [21])}
|
|
published = []
|
|
fetched = []
|
|
state = {"mqtt": True, "fail_login": False}
|
|
|
|
async def fetch(token_id):
|
|
fetched.append(token_id)
|
|
return f"secret-{token_id}"
|
|
|
|
async def publish(payload):
|
|
published.append(payload)
|
|
|
|
async def pulse_ready(timeout=None):
|
|
return True
|
|
monkeypatch.setattr(discord_radio.pulse, "wait_until_ready", pulse_ready)
|
|
monkeypatch.setattr(discord_radio.pulse, "is_ready", lambda: True)
|
|
|
|
mgr = RadioManager(
|
|
client_factory=lambda: FakeClient(guilds, fail_login=state["fail_login"]),
|
|
token_fetcher=fetch, publisher=publish, audio_factory=lambda: object(),
|
|
mqtt_up=lambda: state["mqtt"], ready_timeout=1,
|
|
)
|
|
yield type("Env", (), {"mgr": mgr, "guilds": guilds, "published": published,
|
|
"fetched": fetched, "state": state})
|
|
# Leave no background tasks behind.
|
|
if mgr._watchdog:
|
|
mgr._watchdog.cancel()
|
|
|
|
|
|
def _desired(seq, *streams):
|
|
return {"seq": seq, "streams": [
|
|
{"stream_id": f"{g}:{c}", "guild_id": str(g), "channel_id": str(c), "token_id": t, "system_name": "County"}
|
|
for g, c, t in streams
|
|
]}
|
|
|
|
|
|
async def test_apply_connects_and_reports(env):
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
actual = env.published[-1]
|
|
assert actual["applied_seq"] == 1
|
|
assert actual["sessions"] == ["t1"]
|
|
assert actual["streams"] == {"1:11": {"state": "connected", "error": None, "code": None}}
|
|
assert env.mgr.is_connected
|
|
vc = env.guilds[1].voice_client
|
|
assert vc.plays == 1 and vc.channel.id == 11
|
|
|
|
|
|
async def test_one_session_serves_two_guilds(env):
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
|
|
assert len(FakeClient.instances) == 1
|
|
assert env.published[-1]["sessions"] == ["t1"]
|
|
assert env.guilds[1].voice_client.is_connected() and env.guilds[2].voice_client.is_connected()
|
|
|
|
|
|
async def test_dropping_the_last_stream_logs_the_bot_out(env):
|
|
"""node-26#25 item 1: leave() used to keep the gateway session alive."""
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1"), (2, 21, "t1")))
|
|
client = FakeClient.instances[0]
|
|
await env.mgr.apply_desired(_desired(2, (2, 21, "t1")))
|
|
assert not client.closed # still needed for guild 2
|
|
assert env.guilds[1].voice_client is None
|
|
await env.mgr.apply_desired(_desired(3))
|
|
assert client.closed # logged OUT, not just out of voice
|
|
assert env.published[-1] == {"applied_seq": 3, "sessions": [], "streams": {}}
|
|
|
|
|
|
async def test_stale_desired_is_ignored(env):
|
|
await env.mgr.apply_desired(_desired(5, (1, 11, "t1")))
|
|
await env.mgr.apply_desired(_desired(4))
|
|
assert env.mgr.snapshot()["applied_seq"] == 5 and env.mgr.is_connected
|
|
|
|
|
|
async def test_dead_session_is_recovered_by_the_watchdog(env):
|
|
"""node-26#25 item 2: the old watchdog cancelled itself on rejoin."""
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
first = FakeClient.instances[0]
|
|
first.closed = True # bot task dies
|
|
await asyncio.sleep(0.05)
|
|
env.guilds[1].voice_client.connected = False
|
|
await env.mgr.tick()
|
|
assert len(FakeClient.instances) == 2 and not FakeClient.instances[1].closed
|
|
assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
|
|
# The watchdog task itself is still alive after the recovery.
|
|
assert env.mgr._watchdog is not None and not env.mgr._watchdog.done()
|
|
|
|
|
|
async def test_dropped_voice_is_reconnected(env):
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
env.guilds[1].voice_client.connected = False
|
|
env.guilds[1].voice_client = None
|
|
await env.mgr.tick()
|
|
assert env.guilds[1].voice_client.is_connected()
|
|
|
|
|
|
async def test_bot_not_in_guild_is_reported_not_swallowed(env):
|
|
await env.mgr.apply_desired(_desired(1, (9, 91, "t1")))
|
|
st = env.published[-1]["streams"]["9:91"]
|
|
assert st["state"] == "error" and st["code"] == "not_in_guild"
|
|
|
|
|
|
async def test_login_failure_is_reported(env):
|
|
env.state["fail_login"] = True
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
st = env.published[-1]["streams"]["1:11"]
|
|
assert st["state"] == "error" and st["code"] == "login_failed"
|
|
assert env.published[-1]["sessions"] == []
|
|
|
|
|
|
async def test_mqtt_loss_failsafe_logs_out_then_resumes(env, monkeypatch):
|
|
monkeypatch.setattr(discord_radio.settings, "discord_mqtt_loss_stop_seconds", 300)
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
client = FakeClient.instances[0]
|
|
env.state["mqtt"] = False
|
|
await env.mgr.tick(now=1000.0)
|
|
assert not client.closed # inside the window
|
|
await env.mgr.tick(now=1299.0)
|
|
assert not client.closed
|
|
await env.mgr.tick(now=1300.0)
|
|
assert client.closed and env.mgr.snapshot()["sessions"] == []
|
|
# Token strings are not kept across a fail-safe logout: re-login refetches
|
|
# from C2, which refuses if the lease has moved on.
|
|
assert env.mgr._tokens == {}
|
|
env.state["mqtt"] = True
|
|
before = len(env.fetched)
|
|
await env.mgr.tick(now=1400.0)
|
|
assert len(env.fetched) == before + 1
|
|
assert env.mgr.snapshot()["streams"]["1:11"]["state"] == "connected"
|
|
|
|
|
|
async def test_token_fetch_refused_reports_error(env):
|
|
async def refuse(token_id):
|
|
return None
|
|
env.mgr._fetch_token = refuse
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
st = env.published[-1]["streams"]["1:11"]
|
|
assert st["state"] == "error" and FakeClient.instances == []
|
|
|
|
|
|
async def test_local_pause_logs_out_until_next_desired(env):
|
|
await env.mgr.apply_desired(_desired(1, (1, 11, "t1")))
|
|
await env.mgr.pause_locally()
|
|
assert FakeClient.instances[0].closed
|
|
assert env.published[-1]["streams"]["1:11"]["state"] == "paused_locally"
|
|
await env.mgr.apply_desired(_desired(2, (1, 11, "t1")))
|
|
assert env.mgr.is_connected
|
|
|
|
|
|
async def test_no_message_content_intent_on_real_client():
|
|
client = discord_radio._default_client_factory()
|
|
try:
|
|
assert client.intents.message_content is False
|
|
assert client.intents.voice_states and client.intents.guilds
|
|
finally:
|
|
await client.close()
|