Connect to multiple IRC servers concurrently from a single config file. Plugins are loaded once and shared; per-server state is isolated via separate SQLite databases and per-bot runtime state (bot._pstate). - Add build_server_configs() for [servers.*] config layout - Bot.__init__ gains name parameter, _pstate dict for plugin isolation - cli.py runs multiple bots via asyncio.gather - 9 stateful plugins migrated from module-level dicts to _ps(bot) pattern - Backward compatible: legacy [server] config works unchanged Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
176 lines
6.2 KiB
Python
176 lines
6.2 KiB
Python
"""Tests for Bot.long_reply() paste overflow behaviour."""
|
|
|
|
import asyncio
|
|
import types
|
|
|
|
from derp.bot import Bot
|
|
from derp.irc import Message
|
|
from derp.plugin import PluginRegistry
|
|
|
|
# -- Helpers -----------------------------------------------------------------
|
|
|
|
def _make_bot(*, paste_threshold: int = 4, flaskpaste_mod=None) -> Bot:
|
|
"""Build a Bot with minimal config and a captured send log."""
|
|
config = {
|
|
"server": {
|
|
"host": "localhost", "port": 6667, "tls": False,
|
|
"nick": "testbot", "user": "testbot", "realname": "test",
|
|
},
|
|
"bot": {
|
|
"prefix": "!",
|
|
"channels": ["#test"],
|
|
"plugins_dir": "plugins",
|
|
"rate_limit": 100.0,
|
|
"rate_burst": 100,
|
|
"paste_threshold": paste_threshold,
|
|
"admins": [],
|
|
},
|
|
}
|
|
registry = PluginRegistry()
|
|
if flaskpaste_mod is not None:
|
|
registry._modules["flaskpaste"] = flaskpaste_mod
|
|
bot = Bot("test", config, registry)
|
|
bot._sent: list[tuple[str, str]] = [] # type: ignore[attr-defined]
|
|
|
|
async def _capturing_send(target: str, text: str) -> None:
|
|
bot._sent.append((target, text))
|
|
|
|
bot.send = _capturing_send # type: ignore[assignment]
|
|
return bot
|
|
|
|
|
|
def _msg(text: str = "", target: str = "#test", nick: str = "alice") -> Message:
|
|
"""Create a channel PRIVMSG."""
|
|
return Message(
|
|
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
|
|
command="PRIVMSG", params=[target, text], tags={},
|
|
)
|
|
|
|
|
|
def _pm(text: str = "", nick: str = "alice") -> Message:
|
|
"""Create a private PRIVMSG (target = bot nick)."""
|
|
return Message(
|
|
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
|
|
command="PRIVMSG", params=["testbot", text], tags={},
|
|
)
|
|
|
|
|
|
def _make_fp_mod(*, paste_url: str | None = "https://paste.example/abc"):
|
|
"""Build a fake flaskpaste module with create_paste()."""
|
|
mod = types.ModuleType("flaskpaste")
|
|
mod.create_paste = lambda bot, content: paste_url # type: ignore[attr-defined]
|
|
return mod
|
|
|
|
|
|
# -- Tests -------------------------------------------------------------------
|
|
|
|
class TestShortReply:
|
|
def test_sends_all(self):
|
|
"""Lines <= threshold are sent individually, no paste."""
|
|
bot = _make_bot(paste_threshold=4)
|
|
msg = _msg()
|
|
lines = ["line 1", "line 2", "line 3"]
|
|
asyncio.run(bot.long_reply(msg, lines))
|
|
assert len(bot._sent) == 3
|
|
assert bot._sent[0] == ("#test", "line 1")
|
|
assert bot._sent[1] == ("#test", "line 2")
|
|
assert bot._sent[2] == ("#test", "line 3")
|
|
|
|
|
|
class TestLongReply:
|
|
def test_creates_paste(self):
|
|
"""Lines > threshold creates paste, sends preview + URL."""
|
|
fp = _make_fp_mod(paste_url="https://paste.example/xyz")
|
|
bot = _make_bot(paste_threshold=3, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
lines = ["line 1", "line 2", "line 3", "line 4", "line 5"]
|
|
asyncio.run(bot.long_reply(msg, lines, label="results"))
|
|
# preview_count = min(2, threshold-1) = min(2, 2) = 2
|
|
assert len(bot._sent) == 3
|
|
assert bot._sent[0] == ("#test", "line 1")
|
|
assert bot._sent[1] == ("#test", "line 2")
|
|
assert "3 more lines" in bot._sent[2][1]
|
|
assert "(results)" in bot._sent[2][1]
|
|
assert "https://paste.example/xyz" in bot._sent[2][1]
|
|
|
|
def test_fallback_no_flaskpaste(self):
|
|
"""No flaskpaste module loaded -- falls back to sending all lines."""
|
|
bot = _make_bot(paste_threshold=2)
|
|
msg = _msg()
|
|
lines = ["a", "b", "c", "d"]
|
|
asyncio.run(bot.long_reply(msg, lines))
|
|
assert len(bot._sent) == 4
|
|
assert [t for _, t in bot._sent] == ["a", "b", "c", "d"]
|
|
|
|
def test_fallback_paste_fails(self):
|
|
"""create_paste returns None -- falls back to sending all lines."""
|
|
fp = _make_fp_mod(paste_url=None)
|
|
bot = _make_bot(paste_threshold=2, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
lines = ["a", "b", "c"]
|
|
asyncio.run(bot.long_reply(msg, lines))
|
|
assert len(bot._sent) == 3
|
|
assert [t for _, t in bot._sent] == ["a", "b", "c"]
|
|
|
|
def test_label_in_overflow_message(self):
|
|
"""Label appears in the overflow message."""
|
|
fp = _make_fp_mod()
|
|
bot = _make_bot(paste_threshold=2, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
lines = ["a", "b", "c"]
|
|
asyncio.run(bot.long_reply(msg, lines, label="history"))
|
|
overflow = bot._sent[-1][1]
|
|
assert "(history)" in overflow
|
|
|
|
def test_no_label(self):
|
|
"""Overflow message omits label suffix when label is empty."""
|
|
fp = _make_fp_mod()
|
|
bot = _make_bot(paste_threshold=2, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
lines = ["a", "b", "c"]
|
|
asyncio.run(bot.long_reply(msg, lines))
|
|
overflow = bot._sent[-1][1]
|
|
assert "more lines:" in overflow
|
|
assert "()" not in overflow
|
|
|
|
|
|
class TestThreshold:
|
|
def test_configurable(self):
|
|
"""Custom threshold from config controls overflow point."""
|
|
fp = _make_fp_mod()
|
|
bot = _make_bot(paste_threshold=10, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
|
|
# 10 lines == threshold -> no paste
|
|
lines_at = [f"line {i}" for i in range(10)]
|
|
asyncio.run(bot.long_reply(msg, lines_at))
|
|
assert len(bot._sent) == 10
|
|
|
|
def test_over_threshold_pastes(self):
|
|
"""Lines exceeding threshold triggers paste."""
|
|
fp = _make_fp_mod()
|
|
bot = _make_bot(paste_threshold=10, flaskpaste_mod=fp)
|
|
msg = _msg()
|
|
lines_over = [f"line {i}" for i in range(11)]
|
|
asyncio.run(bot.long_reply(msg, lines_over))
|
|
assert len(bot._sent) == 3 # 2 preview + overflow msg
|
|
|
|
|
|
class TestEdgeCases:
|
|
def test_empty_lines_noop(self):
|
|
"""Empty list produces no output."""
|
|
bot = _make_bot()
|
|
msg = _msg()
|
|
asyncio.run(bot.long_reply(msg, []))
|
|
assert bot._sent == []
|
|
|
|
def test_pm_uses_nick(self):
|
|
"""Private messages use nick as target."""
|
|
bot = _make_bot(paste_threshold=4)
|
|
msg = _pm()
|
|
lines = ["x", "y"]
|
|
asyncio.run(bot.long_reply(msg, lines))
|
|
assert len(bot._sent) == 2
|
|
assert bot._sent[0] == ("alice", "x")
|
|
assert bot._sent[1] == ("alice", "y")
|