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

feat: add SearX search plugin and alert backend

Browse Source 
Add standalone !searx command for on-demand SearXNG search (top 3 results).
Add SearX as a third backend (sx) to the alert plugin for keyword monitoring.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-15 15:28:00 +01:00
parent 4c9dffaaf2
commit 10f62631be
6 changed files with 516 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
tests/test_alert.py
View File

@@ -28,6 +28,7 @@ from plugins.alert import ( # noqa: E402
_pollers,
_restore,
_save,
_search_searx,
_search_twitch,
_search_youtube,
_start_poller,
@@ -117,6 +118,15 @@ GQL_RESPONSE = {
},
}
# SearXNG search response
SEARX_RESPONSE = {
"results": [
{"title": "SearX Result 1", "url": "https://example.com/sx1", "content": "Snippet 1"},
{"title": "SearX Result 2", "url": "https://example.com/sx2", "content": "Snippet 2"},
{"title": "SearX Result 3", "url": "https://example.com/sx3", "content": "Snippet 3"},
],
}
# -- Helpers -----------------------------------------------------------------
@@ -218,7 +228,22 @@ def _fake_tw_error(keyword):
raise ConnectionError("Twitch down")
_FAKE_BACKENDS = {"yt": _fake_yt, "tw": _fake_tw}
def _fake_sx(keyword):
"""Fake SearX backend returning two results."""
return [
{"id": "https://example.com/sx1", "title": "SX Result 1",
"url": "https://example.com/sx1", "extra": ""},
{"id": "https://example.com/sx2", "title": "SX Result 2",
"url": "https://example.com/sx2", "extra": ""},
]
def _fake_sx_error(keyword):
"""Fake SearX backend that raises."""
raise ConnectionError("SearX down")
_FAKE_BACKENDS = {"yt": _fake_yt, "tw": _fake_tw, "sx": _fake_sx}
# ---------------------------------------------------------------------------
@@ -512,6 +537,7 @@ class TestCmdAlertAdd:
assert "minecraft speedrun" in bot.replied[0]
assert "2 yt" in bot.replied[0]
assert "2 tw" in bot.replied[0]
assert "2 sx" in bot.replied[0]
data = _load(bot, "#test:mc-speed")
assert data is not None
assert data["name"] == "mc-speed"
@@ -519,6 +545,7 @@ class TestCmdAlertAdd:
assert data["channel"] == "#test"
assert len(data["seen"]["yt"]) == 2
assert len(data["seen"]["tw"]) == 2
assert len(data["seen"]["sx"]) == 2
assert "#test:mc-speed" in _pollers
_stop_poller("#test:mc-speed")
await asyncio.sleep(0)
@@ -590,7 +617,7 @@ class TestCmdAlertAdd:
"""If a backend fails during seeding, seen list is empty for that backend."""
_clear()
bot = _FakeBot(admin=True)
backends = {"yt": _fake_yt, "tw": _fake_tw_error}
backends = {"yt": _fake_yt, "tw": _fake_tw_error, "sx": _fake_sx}
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
@@ -600,6 +627,7 @@ class TestCmdAlertAdd:
assert data is not None
assert len(data["seen"]["yt"]) == 2
assert len(data["seen"]["tw"]) == 0
assert len(data["seen"]["sx"]) == 2
_stop_poller("#test:partial")
await asyncio.sleep(0)
@@ -722,7 +750,10 @@ class TestCmdAlertCheck:
bot = _FakeBot()
data = {
"keyword": "test", "name": "chk", "channel": "#test",
"interval": 300, "seen": {"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"]},
"interval": 300, "seen": {
"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"],
"sx": ["https://example.com/sx1", "https://example.com/sx2"],
},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:chk", data)
@@ -751,11 +782,11 @@ class TestCmdAlertCheck:
bot = _FakeBot()
data = {
"keyword": "test", "name": "errchk", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:errchk", data)
backends = {"yt": _fake_yt_error, "tw": _fake_tw}
backends = {"yt": _fake_yt_error, "tw": _fake_tw, "sx": _fake_sx}
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
@@ -769,7 +800,7 @@ class TestCmdAlertCheck:
bot = _FakeBot()
data = {
"keyword": "test", "name": "news", "channel": "#test",
"interval": 300, "seen": {"yt": ["yt1"], "tw": []},
"interval": 300, "seen": {"yt": ["yt1"], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:news", data)
@@ -777,12 +808,14 @@ class TestCmdAlertCheck:
async def inner():
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await cmd_alert(bot, _msg("!alert check news"))
# yt2 is new for yt, both tw results are new
# yt2 is new for yt, both tw and sx results are new
announcements = [s for t, s in bot.sent if t == "#test"]
yt_msgs = [m for m in announcements if "/yt]" in m]
tw_msgs = [m for m in announcements if "/tw]" in m]
sx_msgs = [m for m in announcements if "/sx]" in m]
assert len(yt_msgs) == 1 # yt2 only
assert len(tw_msgs) == 2 # both tw results
assert len(sx_msgs) == 2 # both sx results
asyncio.run(inner())
@@ -803,7 +836,7 @@ class TestPollOnce:
bot = _FakeBot()
data = {
"keyword": "test", "name": "poll", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:poll"
@@ -814,9 +847,10 @@ class TestPollOnce:
with patch.object(_mod, "_BACKENDS", _FAKE_BACKENDS):
await _poll_once(bot, key, announce=True)
messages = [s for t, s in bot.sent if t == "#test"]
assert len(messages) == 4 # 2 yt + 2 tw
assert len(messages) == 6 # 2 yt + 2 tw + 2 sx
assert "[poll/yt]" in messages[0]
assert "[poll/tw]" in messages[2]
assert "[poll/sx]" in messages[4]
asyncio.run(inner())
@@ -826,7 +860,10 @@ class TestPollOnce:
data = {
"keyword": "test", "name": "dedup", "channel": "#test",
"interval": 300,
"seen": {"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"]},
"seen": {
"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"],
"sx": ["https://example.com/sx1", "https://example.com/sx2"],
},
"last_poll": "", "last_error": "",
}
key = "#test:dedup"
@@ -854,7 +891,7 @@ class TestPollOnce:
data = {
"keyword": "test", "name": "many", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:many"
@@ -876,20 +913,22 @@ class TestPollOnce:
bot = _FakeBot()
data = {
"keyword": "test", "name": "partial", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:partial"
_save(bot, key, data)
_subscriptions[key] = data
backends = {"yt": _fake_yt_error, "tw": _fake_tw}
backends = {"yt": _fake_yt_error, "tw": _fake_tw, "sx": _fake_sx}
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
await _poll_once(bot, key, announce=True)
# Twitch results should still be announced
# Twitch and SearX results should still be announced
tw_msgs = [s for t, s in bot.sent if t == "#test" and "/tw]" in s]
sx_msgs = [s for t, s in bot.sent if t == "#test" and "/sx]" in s]
assert len(tw_msgs) == 2
assert len(sx_msgs) == 2
# Error counter should be incremented
assert _errors[key] == 1
updated = _load(bot, key)
@@ -902,7 +941,7 @@ class TestPollOnce:
bot = _FakeBot()
data = {
"keyword": "test", "name": "quiet", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:quiet"
@@ -916,6 +955,7 @@ class TestPollOnce:
updated = _load(bot, key)
assert len(updated["seen"]["yt"]) == 2
assert len(updated["seen"]["tw"]) == 2
assert len(updated["seen"]["sx"]) == 2
asyncio.run(inner())
@@ -932,7 +972,7 @@ class TestPollOnce:
data = {
"keyword": "test", "name": "cap", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:cap"
@@ -954,13 +994,13 @@ class TestPollOnce:
bot = _FakeBot()
data = {
"keyword": "test", "name": "allerr", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:allerr"
_save(bot, key, data)
_subscriptions[key] = data
backends = {"yt": _fake_yt_error, "tw": _fake_tw_error}
backends = {"yt": _fake_yt_error, "tw": _fake_tw_error, "sx": _fake_sx_error}
async def inner():
with patch.object(_mod, "_BACKENDS", backends):
@@ -975,7 +1015,10 @@ class TestPollOnce:
bot = _FakeBot()
data = {
"keyword": "test", "name": "clrerr", "channel": "#test",
"interval": 300, "seen": {"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"]},
"interval": 300, "seen": {
"yt": ["yt1", "yt2"], "tw": ["stream:tw1", "vod:tw2"],
"sx": ["https://example.com/sx1", "https://example.com/sx2"],
},
"last_poll": "", "last_error": "old error",
}
key = "#test:clrerr"
@@ -1003,7 +1046,7 @@ class TestRestore:
bot = _FakeBot()
data = {
"keyword": "test", "name": "restored", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:restored", data)
@@ -1023,7 +1066,7 @@ class TestRestore:
bot = _FakeBot()
data = {
"keyword": "test", "name": "active", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:active", data)
@@ -1043,7 +1086,7 @@ class TestRestore:
bot = _FakeBot()
data = {
"keyword": "test", "name": "done", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:done", data)
@@ -1077,7 +1120,7 @@ class TestRestore:
bot = _FakeBot()
data = {
"keyword": "test", "name": "conn", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
_save(bot, "#test:conn", data)
@@ -1102,7 +1145,7 @@ class TestPollerManagement:
bot = _FakeBot()
data = {
"keyword": "test", "name": "mgmt", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:mgmt"
@@ -1125,7 +1168,7 @@ class TestPollerManagement:
bot = _FakeBot()
data = {
"keyword": "test", "name": "idem", "channel": "#test",
"interval": 300, "seen": {"yt": [], "tw": []},
"interval": 300, "seen": {"yt": [], "tw": [], "sx": []},
"last_poll": "", "last_error": "",
}
key = "#test:idem"
@@ -1163,3 +1206,44 @@ class TestCmdAlertUsage:
bot = _FakeBot()
asyncio.run(cmd_alert(bot, _msg("!alert foobar")))
assert "Usage:" in bot.replied[0]
# ---------------------------------------------------------------------------
# TestSearchSearx
# ---------------------------------------------------------------------------
class TestSearchSearx:
def test_parses_response(self):
class FakeResp:
def read(self):
return json.dumps(SEARX_RESPONSE).encode()
def close(self):
pass
with patch("urllib.request.urlopen", return_value=FakeResp()):
results = _search_searx("test query")
assert len(results) == 3
assert results[0]["id"] == "https://example.com/sx1"
assert results[0]["title"] == "SearX Result 1"
assert results[0]["url"] == "https://example.com/sx1"
assert results[0]["extra"] == ""
def test_empty_results(self):
empty = {"results": []}
class FakeResp:
def read(self):
return json.dumps(empty).encode()
def close(self):
pass
with patch("urllib.request.urlopen", return_value=FakeResp()):
results = _search_searx("nothing")
assert results == []
def test_http_error_propagates(self):
import pytest
with patch("urllib.request.urlopen", side_effect=ConnectionError("fail")):
with pytest.raises(ConnectionError):
_search_searx("test")

243
tests/test_searx.py Normal file
View File

@@ -0,0 +1,243 @@
"""Tests for the SearXNG search plugin."""
import asyncio
import importlib.util
import json
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.searx", Path(__file__).resolve().parent.parent / "plugins" / "searx.py",
)
_mod = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = _mod
_spec.loader.exec_module(_mod)
from plugins.searx import ( # noqa: E402
_MAX_QUERY_LEN,
_MAX_RESULTS,
_search,
_truncate,
cmd_searx,
)
# -- Fixtures ----------------------------------------------------------------
SEARX_RESPONSE = {
"results": [
{"title": "Result One", "url": "https://example.com/1", "content": "First snippet"},
{"title": "Result Two", "url": "https://example.com/2", "content": "Second snippet"},
{"title": "Result Three", "url": "https://example.com/3", "content": "Third snippet"},
{"title": "Result Four", "url": "https://example.com/4", "content": "Fourth snippet"},
{"title": "Result Five", "url": "https://example.com/5", "content": "Fifth snippet"},
],
}
SEARX_EMPTY = {"results": []}
# -- Helpers -----------------------------------------------------------------
class _FakeBot:
"""Minimal bot stand-in that captures sent/replied messages."""
def __init__(self):
self.sent: list[tuple[str, str]] = []
self.replied: list[str] = []
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 _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={},
)
def _pm(text: str, nick: str = "alice") -> Message:
"""Create a private PRIVMSG."""
return Message(
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
command="PRIVMSG", params=["botname", text], tags={},
)
class _FakeResp:
"""Fake HTTP response returning preset JSON."""
def __init__(self, data):
self._data = data
def read(self):
return json.dumps(self._data).encode()
def close(self):
pass
# ---------------------------------------------------------------------------
# TestTruncate
# ---------------------------------------------------------------------------
class TestTruncate:
def test_short_text_unchanged(self):
assert _truncate("hello", 80) == "hello"
def test_exact_length_unchanged(self):
text = "a" * 80
assert _truncate(text, 80) == text
def test_long_text_truncated(self):
text = "a" * 100
result = _truncate(text, 80)
assert len(result) == 80
assert result.endswith("...")
def test_default_max_length(self):
text = "a" * 100
result = _truncate(text)
assert len(result) == 80
def test_trailing_space_stripped(self):
text = "word " * 20
result = _truncate(text, 20)
assert not result.endswith(" ...")
# ---------------------------------------------------------------------------
# TestSearch
# ---------------------------------------------------------------------------
class TestSearch:
def test_success(self):
with patch("urllib.request.urlopen", return_value=_FakeResp(SEARX_RESPONSE)):
results = _search("test query")
assert len(results) == 5
assert results[0]["title"] == "Result One"
assert results[0]["url"] == "https://example.com/1"
assert results[0]["snippet"] == "First snippet"
def test_empty_results(self):
with patch("urllib.request.urlopen", return_value=_FakeResp(SEARX_EMPTY)):
results = _search("nothing")
assert results == []
def test_http_error_propagates(self):
import pytest
with patch("urllib.request.urlopen", side_effect=ConnectionError("down")):
with pytest.raises(ConnectionError):
_search("test")
def test_snippet_fallback(self):
"""Falls back to 'snippet' key when 'content' is absent."""
data = {"results": [
{"title": "T", "url": "http://x.com", "snippet": "fallback"},
]}
with patch("urllib.request.urlopen", return_value=_FakeResp(data)):
results = _search("test")
assert results[0]["snippet"] == "fallback"
# ---------------------------------------------------------------------------
# TestCmdSearx
# ---------------------------------------------------------------------------
class TestCmdSearx:
def test_results_displayed(self):
bot = _FakeBot()
async def inner():
with patch("urllib.request.urlopen", return_value=_FakeResp(SEARX_RESPONSE)):
await cmd_searx(bot, _msg("!searx test query"))
assert len(bot.replied) == _MAX_RESULTS
assert "Result One" in bot.replied[0]
assert "https://example.com/1" in bot.replied[0]
assert " -- " in bot.replied[0]
asyncio.run(inner())
def test_no_results(self):
bot = _FakeBot()
async def inner():
with patch("urllib.request.urlopen", return_value=_FakeResp(SEARX_EMPTY)):
await cmd_searx(bot, _msg("!searx nothing"))
assert len(bot.replied) == 1
assert "No results for:" in bot.replied[0]
asyncio.run(inner())
def test_error_handling(self):
bot = _FakeBot()
async def inner():
with patch("urllib.request.urlopen", side_effect=ConnectionError("fail")):
await cmd_searx(bot, _msg("!searx broken"))
assert len(bot.replied) == 1
assert "Search failed:" in bot.replied[0]
asyncio.run(inner())
def test_pm_rejected(self):
bot = _FakeBot()
asyncio.run(cmd_searx(bot, _pm("!searx test")))
assert "Use this command in a channel" in bot.replied[0]
def test_no_query(self):
bot = _FakeBot()
asyncio.run(cmd_searx(bot, _msg("!searx")))
assert "Usage:" in bot.replied[0]
def test_empty_query(self):
bot = _FakeBot()
asyncio.run(cmd_searx(bot, _msg("!searx ")))
assert "Usage:" in bot.replied[0]
def test_query_too_long(self):
bot = _FakeBot()
long_query = "x" * (_MAX_QUERY_LEN + 1)
asyncio.run(cmd_searx(bot, _msg(f"!searx {long_query}")))
assert "too long" in bot.replied[0]
def test_title_truncation(self):
"""Long titles are truncated to _MAX_TITLE_LEN."""
long_title = "A" * 100
data = {"results": [
{"title": long_title, "url": "http://x.com", "content": "s"},
]}
bot = _FakeBot()
async def inner():
with patch("urllib.request.urlopen", return_value=_FakeResp(data)):
await cmd_searx(bot, _msg("!searx test"))
assert len(bot.replied) == 1
title_part = bot.replied[0].split(" -- ")[0]
assert len(title_part) <= 80
assert title_part.endswith("...")
asyncio.run(inner())
def test_no_title_fallback(self):
"""Empty title shows '(no title)'."""
data = {"results": [
{"title": "", "url": "http://x.com", "content": "s"},
]}
bot = _FakeBot()
async def inner():
with patch("urllib.request.urlopen", return_value=_FakeResp(data)):
await cmd_searx(bot, _msg("!searx test"))
assert "(no title)" in bot.replied[0]
asyncio.run(inner())