10f62631be
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>
244 lines
7.7 KiB
Python
244 lines
7.7 KiB
Python
"""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())
|