From aebe1589d286359dc3cd63b6f1aa8fed4b30462e Mon Sep 17 00:00:00 2001 From: user Date: Sat, 21 Feb 2026 17:35:03 +0100 Subject: [PATCH] feat: add URL shortening to subscription announcements 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 --- plugins/pastemoni.py | 2 + plugins/rss.py | 2 + plugins/youtube.py | 2 + src/derp/bot.py | 11 ++ tests/test_flaskpaste.py | 212 +++++++++++++++++++++++++++++++++++++++ tests/test_pastemoni.py | 3 + tests/test_rss.py | 3 + tests/test_youtube.py | 3 + 8 files changed, 238 insertions(+) create mode 100644 tests/test_flaskpaste.py diff --git a/plugins/pastemoni.py b/plugins/pastemoni.py index 2675325..0fbd801 100644 --- a/plugins/pastemoni.py +++ b/plugins/pastemoni.py @@ -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) diff --git a/plugins/rss.py b/plugins/rss.py index 05d2c4e..4e9b428 100644 --- a/plugins/rss.py +++ b/plugins/rss.py @@ -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: diff --git a/plugins/youtube.py b/plugins/youtube.py index 3ec8ca7..1c93fae 100644 --- a/plugins/youtube.py +++ b/plugins/youtube.py @@ -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) diff --git a/src/derp/bot.py b/src/derp/bot.py index 4d4b175..8f1a139 100644 --- a/src/derp/bot.py +++ b/src/derp/bot.py @@ -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)) diff --git a/tests/test_flaskpaste.py b/tests/test_flaskpaste.py new file mode 100644 index 0000000..25a5446 --- /dev/null +++ b/tests/test_flaskpaste.py @@ -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 diff --git a/tests/test_pastemoni.py b/tests/test_pastemoni.py index a241f11..1cc79a0 100644 --- a/tests/test_pastemoni.py +++ b/tests/test_pastemoni.py @@ -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 diff --git a/tests/test_rss.py b/tests/test_rss.py index a9f481d..0715033 100644 --- a/tests/test_rss.py +++ b/tests/test_rss.py @@ -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 diff --git a/tests/test_youtube.py b/tests/test_youtube.py index fd3c0c5..d2431c9 100644 --- a/tests/test_youtube.py +++ b/tests/test_youtube.py @@ -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