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

feat: add URL shortening to subscription announcements

Browse Source 
Bot.shorten_url() method delegates to flaskpaste plugin when loaded.
RSS, YouTube, and pastemoni announcements auto-shorten links.
Includes test_flaskpaste.py (9 cases) and FakeBot updates in 3 test files.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-21 17:35:03 +01:00
parent 9abf8dce64
commit aebe1589d2
8 changed files with 238 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/pastemoni.py
View File

@@ -275,6 +275,8 @@ async def _poll_once(bot, key: str, announce: bool = True) -> None:
title = item.get("title") or "(untitled)"
snippet = item.get("snippet", "")
url = item.get("url", "")
if url:
url = await bot.shorten_url(url)
parts = [f"[{tag}] {title}"]
if snippet:
parts.append(snippet)

2
plugins/rss.py
View File

@@ -272,6 +272,8 @@ async def _poll_once(bot, key: str, announce: bool = True) -> None:
for item in shown:
title = _truncate(item["title"]) if item["title"] else "(no title)"
link = item["link"]
if link:
link = await bot.shorten_url(link)
date = item.get("date", "")
line = f"[{name}] {title}"
if date:

2
plugins/youtube.py
View File

@@ -394,6 +394,8 @@ async def _poll_once(bot, key: str, announce: bool = True) -> None:
for item in shown:
title = _truncate(item["title"]) if item["title"] else "(no title)"
link = item["link"]
if link:
link = await bot.shorten_url(link)
# Build metadata suffix
parts = []
dur = durations.get(item["id"], 0)

11
src/derp/bot.py
View File

@@ -510,6 +510,17 @@ class Bot:
"""Send a CTCP ACTION (/me) to a target."""
await self.send(target, f"\x01ACTION {text}\x01")
async def shorten_url(self, url: str) -> str:
"""Shorten a URL via FlaskPaste. Returns original on failure."""
fp = self.registry._modules.get("flaskpaste")
if not fp:
return url
loop = asyncio.get_running_loop()
try:
return await loop.run_in_executor(None, fp.shorten_url, self, url)
except Exception:
return url
async def join(self, channel: str) -> None:
"""Join an IRC channel."""
await self.conn.send(format_msg("JOIN", channel))

212
tests/test_flaskpaste.py Normal file
View File

@@ -0,0 +1,212 @@
"""Tests for the FlaskPaste plugin and Bot.shorten_url method."""
import asyncio
import importlib.util
import sys
from pathlib import Path
from unittest.mock import patch
from derp.irc import Message
# plugins/ is not a Python package -- load the module from file path
_spec = importlib.util.spec_from_file_location(
"plugins.flaskpaste",
Path(__file__).resolve().parent.parent / "plugins" / "flaskpaste.py",
)
_mod = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = _mod
_spec.loader.exec_module(_mod)
from plugins.flaskpaste import ( # noqa: E402
cmd_paste,
cmd_shorten,
create_paste,
shorten_url,
)
# -- Helpers -----------------------------------------------------------------
class _FakeState:
"""In-memory stand-in for bot.state."""
def __init__(self):
self._store: dict[str, dict[str, str]] = {}
def get(self, plugin: str, key: str, default: str | None = None) -> str | None:
return self._store.get(plugin, {}).get(key, default)
def set(self, plugin: str, key: str, value: str) -> None:
self._store.setdefault(plugin, {})[key] = value
def delete(self, plugin: str, key: str) -> bool:
try:
del self._store[plugin][key]
return True
except KeyError:
return False
def keys(self, plugin: str) -> list[str]:
return sorted(self._store.get(plugin, {}).keys())
class _FakeRegistry:
"""Minimal registry stand-in."""
def __init__(self):
self._modules: dict = {}
class _FakeBot:
"""Minimal bot stand-in that captures sent/replied messages."""
def __init__(self, *, admin: bool = False):
self.sent: list[tuple[str, str]] = []
self.replied: list[str] = []
self.state = _FakeState()
self.registry = _FakeRegistry()
self._admin = admin
self.config: dict = {}
async def send(self, target: str, text: str) -> None:
self.sent.append((target, text))
async def reply(self, message, text: str) -> None:
self.replied.append(text)
def _is_admin(self, message) -> bool:
return self._admin
def _msg(text: str, nick: str = "alice", target: str = "#test") -> Message:
"""Create a channel PRIVMSG."""
return Message(
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
command="PRIVMSG", params=[target, text], tags={},
)
# ---------------------------------------------------------------------------
# TestCmdShorten
# ---------------------------------------------------------------------------
class TestCmdShorten:
def test_missing_url(self):
bot = _FakeBot()
asyncio.run(cmd_shorten(bot, _msg("!shorten")))
assert "Usage:" in bot.replied[0]
def test_invalid_scheme(self):
bot = _FakeBot()
asyncio.run(cmd_shorten(bot, _msg("!shorten ftp://example.com")))
assert "http://" in bot.replied[0]
def test_success(self):
bot = _FakeBot()
async def inner():
with patch.object(
_mod, "_shorten_url",
return_value="https://paste.mymx.me/s/abc123",
):
await cmd_shorten(bot, _msg("!shorten https://example.com/long"))
assert "paste.mymx.me/s/abc123" in bot.replied[0]
asyncio.run(inner())
def test_failure(self):
bot = _FakeBot()
async def inner():
with patch.object(
_mod, "_shorten_url",
side_effect=ConnectionError("down"),
):
await cmd_shorten(bot, _msg("!shorten https://example.com/long"))
assert "shorten failed" in bot.replied[0]
asyncio.run(inner())
# ---------------------------------------------------------------------------
# TestCmdPaste
# ---------------------------------------------------------------------------
class TestCmdPaste:
def test_missing_text(self):
bot = _FakeBot()
asyncio.run(cmd_paste(bot, _msg("!paste")))
assert "Usage:" in bot.replied[0]
def test_success(self):
bot = _FakeBot()
async def inner():
with patch.object(
_mod, "_create_paste",
return_value="https://paste.mymx.me/xyz789",
):
await cmd_paste(bot, _msg("!paste hello world"))
assert "paste.mymx.me/xyz789" in bot.replied[0]
asyncio.run(inner())
def test_failure(self):
bot = _FakeBot()
async def inner():
with patch.object(
_mod, "_create_paste",
side_effect=ConnectionError("down"),
):
await cmd_paste(bot, _msg("!paste hello world"))
assert "paste failed" in bot.replied[0]
asyncio.run(inner())
# ---------------------------------------------------------------------------
# TestShortenUrlHelper
# ---------------------------------------------------------------------------
class TestShortenUrlHelper:
def test_returns_short_url(self):
bot = _FakeBot()
with patch.object(
_mod, "_shorten_url",
return_value="https://paste.mymx.me/s/short",
):
result = shorten_url(bot, "https://example.com/long")
assert result == "https://paste.mymx.me/s/short"
def test_returns_original_on_error(self):
bot = _FakeBot()
with patch.object(
_mod, "_shorten_url",
side_effect=ConnectionError("fail"),
):
result = shorten_url(bot, "https://example.com/long")
assert result == "https://example.com/long"
# ---------------------------------------------------------------------------
# TestCreatePasteHelper
# ---------------------------------------------------------------------------
class TestCreatePasteHelper:
def test_returns_paste_url(self):
bot = _FakeBot()
with patch.object(
_mod, "_create_paste",
return_value="https://paste.mymx.me/abc",
):
result = create_paste(bot, "hello")
assert result == "https://paste.mymx.me/abc"
def test_returns_none_on_error(self):
bot = _FakeBot()
with patch.object(
_mod, "_create_paste",
side_effect=ConnectionError("fail"),
):
result = create_paste(bot, "hello")
assert result is None

3
tests/test_pastemoni.py
View File

@@ -140,6 +140,9 @@ class _FakeBot:
async def reply(self, message, text: str) -> None:
self.replied.append(text)
async def shorten_url(self, url: str) -> str:
return url
def _is_admin(self, message) -> bool:
return self._admin

3
tests/test_rss.py
View File

@@ -166,6 +166,9 @@ class _FakeBot:
async def reply(self, message, text: str) -> None:
self.replied.append(text)
async def shorten_url(self, url: str) -> str:
return url
def _is_admin(self, message) -> bool:
return self._admin

3
tests/test_youtube.py
View File

@@ -171,6 +171,9 @@ class _FakeBot:
async def reply(self, message, text: str) -> None:
self.replied.append(text)
async def shorten_url(self, url: str) -> str:
return url
def _is_admin(self, message) -> bool:
return self._admin